Class ProcessorContextImpl
- java.lang.Object
-
- org.apache.kafka.streams.processor.internals.AbstractProcessorContext
-
- org.apache.kafka.streams.processor.internals.ProcessorContextImpl
-
- All Implemented Interfaces:
InternalProcessorContext
,RecordCollector.Supplier
,ProcessorContext
public class ProcessorContextImpl extends AbstractProcessorContext implements RecordCollector.Supplier
-
-
Field Summary
-
Fields inherited from class org.apache.kafka.streams.processor.internals.AbstractProcessorContext
cache, currentNode, NONEXIST_TOPIC, recordContext
-
Fields inherited from interface org.apache.kafka.streams.processor.internals.InternalProcessorContext
BYTEARRAY_VALUE_SERIALIZER, BYTES_KEY_SERIALIZER
-
-
Constructor Summary
Constructors Constructor Description ProcessorContextImpl(TaskId id, StreamsConfig config, ProcessorStateManager stateMgr, StreamsMetricsImpl metrics, ThreadCache cache)
-
Method Summary
All Methods Instance Methods Concrete Methods Deprecated Methods Modifier and Type Method Description void
commit()
Requests a commit.ProcessorNode<?,?>
currentNode()
Get the currentProcessorNode
<K,V>
voidforward(K key, V value)
Forwards a key/value pair to all downstream processors.<K,V>
voidforward(K key, V value, int childIndex)
Deprecated.<K,V>
voidforward(K key, V value, String childName)
Deprecated.<K,V>
voidforward(K key, V value, To to)
Forwards a key/value pair to the specified downstream processors.StateStore
getStateStore(String name)
Get the state store given the store name.void
logChange(String storeName, org.apache.kafka.common.utils.Bytes key, byte[] value, long timestamp)
long
offset()
Returns the offset of the current input record; could be -1 if it is not available (for example, if this method is invoked from the punctuate call).int
partition()
Returns the partition id of the current input record; could be -1 if it is not available (for example, if this method is invoked from the punctuate call).RecordCollector
recordCollector()
Get the record collector.ProcessorRecordContext
recordContext()
Returns the currentRecordContext
void
register(StateStore store, StateRestoreCallback stateRestoreCallback)
Registers and possibly restores the specified storage engine.void
registerCacheFlushListener(String namespace, ThreadCache.DirtyEntryFlushListener listener)
Register a dirty entry flush listener for a particular namespaceCancellable
schedule(long intervalMs, PunctuationType type, Punctuator callback)
Deprecated.Cancellable
schedule(Duration interval, PunctuationType type, Punctuator callback)
Schedules a periodic operation for processors.void
setRecordContext(ProcessorRecordContext recordContext)
ProcessorStateManager
stateManager()
long
timestamp()
Returns the current timestamp.String
topic()
Returns the topic name of the current input record; could be null if it is not available (for example, if this method is invoked from the punctuate call).void
transitionToActive(StreamTask streamTask, RecordCollector recordCollector, ThreadCache newCache)
Transition to active task and register a new task and cache to this processor contextvoid
transitionToStandby(ThreadCache newCache)
Transition to standby task and register a dummy cache to this processor context-
Methods inherited from class org.apache.kafka.streams.processor.internals.AbstractProcessorContext
appConfigs, appConfigsWithPrefix, applicationId, applicationInternalStream, cache, currentSystemTimeMs, headers, initialize, keySerde, metrics, setCurrentNode, setSystemTimeMs, stateDir, taskId, taskType, uninitialize, valueSerde
-
Methods inherited from class java.lang.Object
clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait
-
Methods inherited from interface org.apache.kafka.streams.processor.internals.InternalProcessorContext
getStateStore
-
-
-
-
Constructor Detail
-
ProcessorContextImpl
public ProcessorContextImpl(TaskId id, StreamsConfig config, ProcessorStateManager stateMgr, StreamsMetricsImpl metrics, ThreadCache cache)
-
-
Method Detail
-
transitionToActive
public void transitionToActive(StreamTask streamTask, RecordCollector recordCollector, ThreadCache newCache)
Description copied from interface:InternalProcessorContext
Transition to active task and register a new task and cache to this processor context- Specified by:
transitionToActive
in interfaceInternalProcessorContext
-
transitionToStandby
public void transitionToStandby(ThreadCache newCache)
Description copied from interface:InternalProcessorContext
Transition to standby task and register a dummy cache to this processor context- Specified by:
transitionToStandby
in interfaceInternalProcessorContext
-
registerCacheFlushListener
public void registerCacheFlushListener(String namespace, ThreadCache.DirtyEntryFlushListener listener)
Description copied from interface:InternalProcessorContext
Register a dirty entry flush listener for a particular namespace- Specified by:
registerCacheFlushListener
in interfaceInternalProcessorContext
-
stateManager
public ProcessorStateManager stateManager()
-
register
public void register(StateStore store, StateRestoreCallback stateRestoreCallback)
Description copied from interface:ProcessorContext
Registers and possibly restores the specified storage engine.- Specified by:
register
in interfaceProcessorContext
- Overrides:
register
in classAbstractProcessorContext
- Parameters:
store
- the storage enginestateRestoreCallback
- the restoration callback logic for log-backed state stores upon restart
-
recordCollector
public RecordCollector recordCollector()
Description copied from interface:RecordCollector.Supplier
Get the record collector.- Specified by:
recordCollector
in interfaceRecordCollector.Supplier
- Returns:
- the record collector
-
logChange
public void logChange(String storeName, org.apache.kafka.common.utils.Bytes key, byte[] value, long timestamp)
- Specified by:
logChange
in interfaceInternalProcessorContext
-
getStateStore
public StateStore getStateStore(String name)
Description copied from interface:ProcessorContext
Get the state store given the store name.- Specified by:
getStateStore
in interfaceProcessorContext
- Parameters:
name
- The store name- Returns:
- The state store instance
- Throws:
StreamsException
- if an attempt is made to access this state store from an unknown nodeUnsupportedOperationException
- if the current streamTask type is standby
-
forward
public <K,V> void forward(K key, V value)
Description copied from interface:ProcessorContext
Forwards a key/value pair to all downstream processors. Used the input record's timestamp as timestamp for the output record.- Specified by:
forward
in interfaceProcessorContext
- Parameters:
key
- keyvalue
- value
-
forward
@Deprecated public <K,V> void forward(K key, V value, int childIndex)
Deprecated.Description copied from interface:ProcessorContext
Forwards a key/value pair to one of the downstream processors designated by childIndex.- Specified by:
forward
in interfaceProcessorContext
- Parameters:
key
- keyvalue
- valuechildIndex
- index in list of children of this node
-
forward
@Deprecated public <K,V> void forward(K key, V value, String childName)
Deprecated.Description copied from interface:ProcessorContext
Forwards a key/value pair to one of the downstream processors designated by the downstream processor name.- Specified by:
forward
in interfaceProcessorContext
- Parameters:
key
- keyvalue
- valuechildName
- name of downstream processor
-
forward
public <K,V> void forward(K key, V value, To to)
Description copied from interface:ProcessorContext
Forwards a key/value pair to the specified downstream processors. Can be used to set the timestamp of the output record.- Specified by:
forward
in interfaceProcessorContext
- Parameters:
key
- keyvalue
- valueto
- the options to use when forwarding
-
commit
public void commit()
Description copied from interface:ProcessorContext
Requests a commit.- Specified by:
commit
in interfaceProcessorContext
-
schedule
@Deprecated public Cancellable schedule(long intervalMs, PunctuationType type, Punctuator callback)
Deprecated.Description copied from interface:ProcessorContext
Schedules a periodic operation for processors. A processor may call this method duringinitialization
orprocessing
to schedule a periodic callback — called a punctuation — toPunctuator.punctuate(long)
. The type parameter controls what notion of time is used for punctuation:PunctuationType.STREAM_TIME
— uses "stream time", which is advanced by the processing of messages in accordance with the timestamp as extracted by theTimestampExtractor
in use. The first punctuation will be triggered by the first record that is processed. NOTE: Only advanced if messages arrivePunctuationType.WALL_CLOCK_TIME
— uses system time (the wall-clock time), which is advanced independent of whether new messages arrive. The first punctuation will be triggered after interval has elapsed. NOTE: This is best effort only as its granularity is limited by how long an iteration of the processing loop takes to complete
- with
PunctuationType.STREAM_TIME
, when stream time advances more than interval - with
PunctuationType.WALL_CLOCK_TIME
, on GC pause, too short interval, ...
- Specified by:
schedule
in interfaceProcessorContext
- Parameters:
intervalMs
- the time interval between punctuations in millisecondstype
- one of:PunctuationType.STREAM_TIME
,PunctuationType.WALL_CLOCK_TIME
callback
- a function consuming timestamps representing the current stream or system time- Returns:
- a handle allowing cancellation of the punctuation schedule established by this method
-
schedule
public Cancellable schedule(Duration interval, PunctuationType type, Punctuator callback) throws IllegalArgumentException
Description copied from interface:ProcessorContext
Schedules a periodic operation for processors. A processor may call this method duringinitialization
orprocessing
to schedule a periodic callback — called a punctuation — toPunctuator.punctuate(long)
. The type parameter controls what notion of time is used for punctuation:PunctuationType.STREAM_TIME
— uses "stream time", which is advanced by the processing of messages in accordance with the timestamp as extracted by theTimestampExtractor
in use. The first punctuation will be triggered by the first record that is processed. NOTE: Only advanced if messages arrivePunctuationType.WALL_CLOCK_TIME
— uses system time (the wall-clock time), which is advanced independent of whether new messages arrive. The first punctuation will be triggered after interval has elapsed. NOTE: This is best effort only as its granularity is limited by how long an iteration of the processing loop takes to complete
- with
PunctuationType.STREAM_TIME
, when stream time advances more than interval - with
PunctuationType.WALL_CLOCK_TIME
, on GC pause, too short interval, ...
- Specified by:
schedule
in interfaceProcessorContext
- Parameters:
interval
- the time interval between punctuations (supported minimum is 1 millisecond)type
- one of:PunctuationType.STREAM_TIME
,PunctuationType.WALL_CLOCK_TIME
callback
- a function consuming timestamps representing the current stream or system time- Returns:
- a handle allowing cancellation of the punctuation schedule established by this method
- Throws:
IllegalArgumentException
-
topic
public String topic()
Description copied from interface:ProcessorContext
Returns the topic name of the current input record; could be null if it is not available (for example, if this method is invoked from the punctuate call).- Specified by:
topic
in interfaceProcessorContext
- Overrides:
topic
in classAbstractProcessorContext
- Returns:
- the topic name
-
partition
public int partition()
Description copied from interface:ProcessorContext
Returns the partition id of the current input record; could be -1 if it is not available (for example, if this method is invoked from the punctuate call).- Specified by:
partition
in interfaceProcessorContext
- Overrides:
partition
in classAbstractProcessorContext
- Returns:
- the partition id
-
offset
public long offset()
Description copied from interface:ProcessorContext
Returns the offset of the current input record; could be -1 if it is not available (for example, if this method is invoked from the punctuate call).- Specified by:
offset
in interfaceProcessorContext
- Overrides:
offset
in classAbstractProcessorContext
- Returns:
- the offset
-
timestamp
public long timestamp()
Description copied from interface:ProcessorContext
Returns the current timestamp.If it is triggered while processing a record streamed from the source processor, timestamp is defined as the timestamp of the current input record; the timestamp is extracted from
ConsumerRecord
byTimestampExtractor
.If it is triggered while processing a record generated not from the source processor (for example, if this method is invoked from the punctuate call), timestamp is defined as the current task's stream time, which is defined as the largest timestamp of any record processed by the task.
- Specified by:
timestamp
in interfaceProcessorContext
- Overrides:
timestamp
in classAbstractProcessorContext
- Returns:
- the timestamp
-
currentNode
public ProcessorNode<?,?> currentNode()
Description copied from interface:InternalProcessorContext
Get the currentProcessorNode
- Specified by:
currentNode
in interfaceInternalProcessorContext
- Overrides:
currentNode
in classAbstractProcessorContext
-
setRecordContext
public void setRecordContext(ProcessorRecordContext recordContext)
- Specified by:
setRecordContext
in interfaceInternalProcessorContext
- Overrides:
setRecordContext
in classAbstractProcessorContext
- Parameters:
recordContext
- theProcessorRecordContext
for the record about to be processes
-
recordContext
public ProcessorRecordContext recordContext()
Description copied from interface:InternalProcessorContext
Returns the currentRecordContext
- Specified by:
recordContext
in interfaceInternalProcessorContext
- Overrides:
recordContext
in classAbstractProcessorContext
- Returns:
- the current
RecordContext
-
-