Package org.nuxeo.lib.stream.computation.log

Class ComputationRunner

java.lang.Object

org.nuxeo.lib.stream.computation.log.ComputationRunner

All Implemented Interfaces:

Runnable, RebalanceListener

public class ComputationRunner
extends Object
implements Runnable, RebalanceListener

Thread driving a Computation

Since:

9.3

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
protected static class	ComputationRunner.CheckPointException
protected static class	ComputationRunner.ReturnCode

Field Summary

Fields

Modifier and Type	Field	Description
protected CountDownLatch	assignmentLatch
protected static int	CHECKPOINT_MAX_RETRY
protected static int	CHECKPOINT_PAUSE_MS
protected Computation	computation
protected ComputationContextImpl	context
protected long	counter
protected List<LogPartition>	defaultAssignment
protected boolean	drain
protected io.dropwizard.metrics5.Counter	failureCount
static String	GLOBAL_FAILURE_COUNT_REGISTRY_NAME
protected io.dropwizard.metrics5.Counter	globalFailureCount
protected static long	INACTIVITY_BREAK_MS
protected long	inCheckpointRecords
protected long	inRecords
protected long	lastReadTime
protected io.opencensus.trace.SpanContext	lastSpanContext
protected long	lastTimerExecution
protected WatermarkMonotonicInterval	lowWatermark
protected ComputationMetadataMapping	metadata
static String	NUXEO_METRICS_REGISTRY_NAME
protected long	outRecords
protected ComputationPolicy	policy
protected io.dropwizard.metrics5.Timer	processRecordTimer
protected io.dropwizard.metrics5.Timer	processTimerTimer
static Duration	READ_TIMEOUT
protected boolean	recordActivity
protected io.dropwizard.metrics5.Counter	recordSkippedCount
protected io.dropwizard.metrics5.MetricRegistry	registry
protected io.dropwizard.metrics5.Counter	runningCount
protected static AtomicInteger	skipFailures
static long	SLOW_COMPUTATION_THRESHOLD_NS
protected static long	STARVING_TIMEOUT_MS
protected boolean	stop
protected LogStreamManager	streamManager
protected Supplier<Computation>	supplier
protected LogTailer<Record>	tailer
protected String	threadName

Constructor Summary

Constructors

Constructor	Description
ComputationRunner(Supplier<Computation> supplier, ComputationMetadataMapping metadata, List<LogPartition> defaultAssignment, LogStreamManager streamManager, ComputationPolicy policy)

Method Summary

Modifier and Type	Method	Description
protected void	checkpoint()
protected void	checkpointIfNecessary()
protected void	checkRecordFlags(Record record)
protected void	checkSourceLowWatermark()
protected void	closeTailer()
protected boolean	continueLoop()
void	drain()
Watermark	getLowWatermark()
protected io.opencensus.trace.Span	getSpanFromRecord(Record record)
protected Duration	getTimeoutDuration()
void	onPartitionsAssigned(Collection<LogPartition> partitions)
void	onPartitionsRevoked(Collection<LogPartition> partitions)
protected void	processFallback(ComputationContextImpl context)
protected void	processLoop()
protected boolean	processRecord()
protected void	processRecordWithRetry(String from, Record record)
protected void	processRecordWithTracing(String from, Record record)
protected boolean	processTimer()
protected void	processTimerWithRetry(String key, Long value)
protected boolean	processTimerWithTracing(long now, LinkedHashMap<String,Long> sortedTimer)
protected void	registerMetrics()
void	run()
protected ComputationRunner.ReturnCode	runOnce()
protected void	saveOffsets()
protected void	saveState()
protected void	saveTimers()
protected void	sendRecords()
protected void	setThreadName(String message)
protected boolean	skipFailureForRecovery()
void	stop()
boolean	waitForAssignments(Duration timeout)

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.nuxeo.lib.stream.log.RebalanceListener

onPartitionsLost

Field Detail

READ_TIMEOUT

public static final Duration READ_TIMEOUT

STARVING_TIMEOUT_MS

protected static final long STARVING_TIMEOUT_MS

See Also:

Constant Field Values

INACTIVITY_BREAK_MS

protected static final long INACTIVITY_BREAK_MS

See Also:

Constant Field Values

CHECKPOINT_MAX_RETRY

protected static final int CHECKPOINT_MAX_RETRY

See Also:

Constant Field Values

CHECKPOINT_PAUSE_MS

protected static final int CHECKPOINT_PAUSE_MS

See Also:

Constant Field Values

SLOW_COMPUTATION_THRESHOLD_NS

public static final long SLOW_COMPUTATION_THRESHOLD_NS

See Also:

Constant Field Values

streamManager

protected final LogStreamManager streamManager

metadata

protected final ComputationMetadataMapping metadata

tailer

protected final LogTailer<Record> tailer

supplier

protected final Supplier<Computation> supplier

assignmentLatch

protected final CountDownLatch assignmentLatch

lowWatermark

protected final WatermarkMonotonicInterval lowWatermark

policy

protected final ComputationPolicy policy

context

protected ComputationContextImpl context

stop

protected volatile boolean stop

drain

protected volatile boolean drain

computation

protected Computation computation

counter

protected long counter

inRecords

protected long inRecords

inCheckpointRecords

protected long inCheckpointRecords

outRecords

protected long outRecords

lastReadTime

protected long lastReadTime

lastTimerExecution

protected long lastTimerExecution

threadName

protected String threadName

defaultAssignment

protected List<LogPartition> defaultAssignment

NUXEO_METRICS_REGISTRY_NAME

public static final String NUXEO_METRICS_REGISTRY_NAME

See Also:

Constant Field Values

GLOBAL_FAILURE_COUNT_REGISTRY_NAME

public static final String GLOBAL_FAILURE_COUNT_REGISTRY_NAME

registry

protected final io.dropwizard.metrics5.MetricRegistry registry

globalFailureCount

protected io.dropwizard.metrics5.Counter globalFailureCount

failureCount

protected io.dropwizard.metrics5.Counter failureCount

recordSkippedCount

protected io.dropwizard.metrics5.Counter recordSkippedCount

runningCount

protected io.dropwizard.metrics5.Counter runningCount

processRecordTimer

protected io.dropwizard.metrics5.Timer processRecordTimer

processTimerTimer

protected io.dropwizard.metrics5.Timer processTimerTimer

skipFailures

protected static AtomicInteger skipFailures

recordActivity

protected boolean recordActivity

lastSpanContext

protected io.opencensus.trace.SpanContext lastSpanContext

Constructor Detail

ComputationRunner

public ComputationRunner(Supplier<Computation> supplier,
                         ComputationMetadataMapping metadata,
                         List<LogPartition> defaultAssignment,
                         LogStreamManager streamManager,
                         ComputationPolicy policy)

Method Detail

stop

public void stop()

drain

public void drain()

waitForAssignments

public boolean waitForAssignments(Duration timeout)
                           throws InterruptedException

Throws:

InterruptedException

run

public void run()

Specified by:

run in interface Runnable

runOnce

protected ComputationRunner.ReturnCode runOnce()

registerMetrics

protected void registerMetrics()

closeTailer

protected void closeTailer()

processLoop

protected void processLoop()
                    throws InterruptedException

Throws:

InterruptedException

continueLoop

protected boolean continueLoop()

processTimer

protected boolean processTimer()

processTimerWithTracing

protected boolean processTimerWithTracing(long now,
                                          LinkedHashMap<String,Long> sortedTimer)

processTimerWithRetry

protected void processTimerWithRetry(String key,
                                     Long value)

processRecord

protected boolean processRecord()
                         throws InterruptedException

Throws:

InterruptedException

processRecordWithTracing

protected void processRecordWithTracing(String from,
                                        Record record)

getSpanFromRecord

protected io.opencensus.trace.Span getSpanFromRecord(Record record)

processRecordWithRetry

protected void processRecordWithRetry(String from,
                                      Record record)

processFallback

protected void processFallback(ComputationContextImpl context)

skipFailureForRecovery

protected boolean skipFailureForRecovery()

getTimeoutDuration

protected Duration getTimeoutDuration()

checkSourceLowWatermark

protected void checkSourceLowWatermark()

checkRecordFlags

protected void checkRecordFlags(Record record)

checkpointIfNecessary

protected void checkpointIfNecessary()

checkpoint

protected void checkpoint()

saveTimers

protected void saveTimers()

saveState

protected void saveState()

saveOffsets

protected void saveOffsets()

sendRecords

protected void sendRecords()

getLowWatermark

public Watermark getLowWatermark()

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