org.nuxeo.lib.stream.pattern.consumer.internals

Class ConsumerRunner<M extends Message>

java.lang.Object

org.nuxeo.lib.stream.pattern.consumer.internals.ConsumerRunner<M>

All Implemented Interfaces:

Callable<ConsumerStatus>, RebalanceListener

public class ConsumerRunner<M extends Message>
extends Object
implements Callable<ConsumerStatus>, RebalanceListener

Read messages from a tailer and drive a consumer according to its policy.

Since:

9.1

Field Summary

Fields

Modifier and Type	Field and Description
protected long	acceptCounter
protected boolean	alreadySalted
protected long	batchCommitCounter
protected long	batchFailureCounter
protected long	committedCounter
protected Consumer<M>	consumer
protected String	consumerId
protected BatchPolicy	currentBatchPolicy
protected ConsumerFactory<M>	factory
protected com.codahale.metrics.Timer	globalAcceptTimer
protected com.codahale.metrics.Timer	globalBatchCommitTimer
protected com.codahale.metrics.Counter	globalBatchFailureCounter
protected com.codahale.metrics.Counter	globalCommittedCounter
protected com.codahale.metrics.Counter	globalConsumersCounter
static String	NUXEO_METRICS_REGISTRY_NAME
protected ConsumerPolicy	policy
protected com.codahale.metrics.MetricRegistry	registry
protected LogTailer<M>	tailer
protected String	threadName

Constructor Summary

Constructors

Constructor and Description
ConsumerRunner(ConsumerFactory<M> factory, ConsumerPolicy policy, LogManager manager, List<LogPartition> defaultAssignments)

Method Summary

Modifier and Type	Method and Description
protected BatchState	acceptBatch()
protected void	addSalt()
protected void	beginBatch()
ConsumerStatus	call()
protected void	commitBatch(BatchState state)
protected void	consumerLoop()
protected LogTailer<M>	createTailer(LogManager manager, List<LogPartition> defaultAssignments)
void	onPartitionsAssigned(Collection<LogPartition> partitions)
void	onPartitionsRevoked(Collection<LogPartition> partitions)
protected boolean	processBatch()
protected boolean	processBatchWithRetry(net.jodah.failsafe.Execution execution)
protected void	restoreBatchPolicy()
protected void	rollbackBatch(Exception e)
protected void	setBatchRetryPolicy()
protected void	setMetrics()
protected void	setTailerPosition(LogManager manager)
protected void	setThreadName(String message)

Methods inherited from class java.lang.Object

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

Field Detail

NUXEO_METRICS_REGISTRY_NAME

public static final String NUXEO_METRICS_REGISTRY_NAME

See Also:

Constant Field Values

factory

protected final ConsumerFactory<M extends Message> factory

policy

protected final ConsumerPolicy policy

tailer

protected final LogTailer<M extends Message> tailer

consumerId

protected String consumerId

currentBatchPolicy

protected BatchPolicy currentBatchPolicy

threadName

protected String threadName

consumer

protected Consumer<M extends Message> consumer

registry

protected final com.codahale.metrics.MetricRegistry registry

acceptCounter

protected long acceptCounter

committedCounter

protected long committedCounter

batchCommitCounter

protected long batchCommitCounter

batchFailureCounter

protected long batchFailureCounter

alreadySalted

protected boolean alreadySalted

globalAcceptTimer

protected com.codahale.metrics.Timer globalAcceptTimer

globalCommittedCounter

protected com.codahale.metrics.Counter globalCommittedCounter

globalBatchCommitTimer

protected com.codahale.metrics.Timer globalBatchCommitTimer

globalBatchFailureCounter

protected com.codahale.metrics.Counter globalBatchFailureCounter

globalConsumersCounter

protected com.codahale.metrics.Counter globalConsumersCounter

Constructor Detail

ConsumerRunner

public ConsumerRunner(ConsumerFactory<M> factory,
                      ConsumerPolicy policy,
                      LogManager manager,
                      List<LogPartition> defaultAssignments)

Method Detail

createTailer

protected LogTailer<M> createTailer(LogManager manager,
                                    List<LogPartition> defaultAssignments)

call

public ConsumerStatus call()
                    throws Exception

Specified by:

call in interface Callable<ConsumerStatus>

Throws:

Exception

setMetrics

protected void setMetrics()

addSalt

protected void addSalt()
                throws InterruptedException

Throws:

InterruptedException

setTailerPosition

protected void setTailerPosition(LogManager manager)

consumerLoop

protected void consumerLoop()
                     throws InterruptedException

Throws:

InterruptedException

processBatchWithRetry

protected boolean processBatchWithRetry(net.jodah.failsafe.Execution execution)
                                 throws InterruptedException

Throws:

InterruptedException

setBatchRetryPolicy

protected void setBatchRetryPolicy()

restoreBatchPolicy

protected void restoreBatchPolicy()

processBatch

protected boolean processBatch()
                        throws InterruptedException

Throws:

InterruptedException

beginBatch

protected void beginBatch()

commitBatch

protected void commitBatch(BatchState state)

rollbackBatch

protected void rollbackBatch(Exception e)

acceptBatch

protected BatchState acceptBatch()
                          throws InterruptedException

Throws:

InterruptedException

setThreadName

protected void setThreadName(String message)

onPartitionsRevoked

public void onPartitionsRevoked(Collection<LogPartition> partitions)

Specified by:

onPartitionsRevoked in interface RebalanceListener

onPartitionsAssigned

public void onPartitionsAssigned(Collection<LogPartition> partitions)

Specified by:

onPartitionsAssigned in interface RebalanceListener