org.nuxeo.lib.stream.computation.log

Class LogStreamProcessor

java.lang.Object

org.nuxeo.lib.stream.computation.log.LogStreamProcessor

All Implemented Interfaces:

StreamProcessor

public class LogStreamProcessor
extends Object
implements StreamProcessor

Since:

9.3

Field Summary

Fields

Modifier and Type	Field and Description
protected LogManager	manager
protected List<ComputationPool>	pools
protected Settings	settings
protected Topology	topology

Constructor Summary

Constructors

Constructor and Description
LogStreamProcessor(LogManager manager)

Method Summary

Modifier and Type	Method and Description
boolean	drainAndStop(Duration timeout) Stop computations when input streams are empty.
protected Codec<Record>	getCodecForStreams(String name, Set<String> streams)
protected List<List<LogPartition>>	getDefaultAssignments(ComputationMetadataMapping meta)
Latency	getLatency(String computationName) Returns the latency for a computation.
long	getLowWatermark() Returns the low watermark for all the computations of the topology.
long	getLowWatermark(String computationName) Returns the low watermark for the computation.
StreamProcessor	init(Topology topology, Settings settings) Initialize streams, but don't run the computations
protected List<ComputationPool>	initPools()
protected void	initSourceAppenders()
protected void	initStreams()
boolean	isDone(long timestamp) Returns true if all messages with a lower timestamp has been processed by the topology.
boolean	isTerminated() True if there is no active processing threads.
void	shutdown() Shutdown immediately.
void	start() Run the initialized computations.
boolean	stop(Duration timeout) Try to stop computations gracefully after processing a record or a timer within the timeout duration.
boolean	waitForAssignments(Duration timeout) Wait for the computations to have assigned partitions ready to process records.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

manager

protected final LogManager manager

topology

protected Topology topology

settings

protected Settings settings

pools

protected List<ComputationPool> pools

Constructor Detail

LogStreamProcessor

public LogStreamProcessor(LogManager manager)

Method Detail

init

public StreamProcessor init(Topology topology,
                            Settings settings)

Description copied from interface: StreamProcessor

Initialize streams, but don't run the computations

Specified by:

init in interface StreamProcessor

start

public void start()

Description copied from interface: StreamProcessor

Run the initialized computations.

Specified by:

start in interface StreamProcessor

waitForAssignments

public boolean waitForAssignments(Duration timeout)
                           throws InterruptedException

Description copied from interface: StreamProcessor

Wait for the computations to have assigned partitions ready to process records. The processor must be started. This is useful for writing unit test.

Returns true if all computations have assigned partitions during the timeout delay.

Specified by:

waitForAssignments in interface StreamProcessor

Throws:

InterruptedException

isTerminated

public boolean isTerminated()

Description copied from interface: StreamProcessor

True if there is no active processing threads.

Specified by:

isTerminated in interface StreamProcessor

stop

public boolean stop(Duration timeout)

Description copied from interface: StreamProcessor

Try to stop computations gracefully after processing a record or a timer within the timeout duration. If this can not be done within the timeout, shutdown and returns false.

Specified by:

stop in interface StreamProcessor

drainAndStop

public boolean drainAndStop(Duration timeout)

Description copied from interface: StreamProcessor

Stop computations when input streams are empty. The timeout is applied for each computation, the total duration can be up to nb computations * timeout

Returns true if computations are stopped during the timeout delay.

Specified by:

drainAndStop in interface StreamProcessor

shutdown

public void shutdown()

Description copied from interface: StreamProcessor

Shutdown immediately.

Specified by:

shutdown in interface StreamProcessor

getLowWatermark

public long getLowWatermark()

Description copied from interface: StreamProcessor

Returns the low watermark for all the computations of the topology. Any message with an offset below the low watermark has been processed. The returned watermark is local to this processing node.

Specified by:

getLowWatermark in interface StreamProcessor

getLatency

public Latency getLatency(String computationName)

Description copied from interface: StreamProcessor

Returns the latency for a computation. This works also for distributed computations.

Specified by:

getLatency in interface StreamProcessor

getLowWatermark

public long getLowWatermark(String computationName)

Description copied from interface: StreamProcessor

Returns the low watermark for the computation. Any message with an offset below the low watermark has been processed by this computation and its ancestors. The returned watermark is local to this processing node, if the computation is distributed the global low watermark is the minimum of all nodes low watermark.

Specified by:

getLowWatermark in interface StreamProcessor

isDone

public boolean isDone(long timestamp)

Description copied from interface: StreamProcessor

Returns true if all messages with a lower timestamp has been processed by the topology.

Specified by:

isDone in interface StreamProcessor

initPools

protected List<ComputationPool> initPools()

getCodecForStreams

protected Codec<Record> getCodecForStreams(String name,
                                           Set<String> streams)

getDefaultAssignments

protected List<List<LogPartition>> getDefaultAssignments(ComputationMetadataMapping meta)

initStreams

protected void initStreams()

initSourceAppenders

protected void initSourceAppenders()