@InterfaceStability.Evolving public interface ProcessorContext
Modifier and Type | Method and Description |
---|---|
Map<String,Object> |
appConfigs()
Returns all the application config properties as key/value pairs.
|
Map<String,Object> |
appConfigsWithPrefix(String prefix)
Returns all the application config properties with the given key prefix, as key/value pairs
stripping the prefix.
|
String |
applicationId()
Returns the application id
|
void |
commit()
Requests a commit
|
<K,V> void |
forward(K key,
V value)
Forwards a key/value pair to all downstream processors.
|
<K,V> void |
forward(K key,
V value,
int childIndex)
Deprecated.
please use
forward(Object, Object, To) instead |
<K,V> void |
forward(K key,
V value,
String childName)
Deprecated.
please use
forward(Object, Object, To) instead |
<K,V> void |
forward(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.
|
Headers |
headers()
Returns the headers of the current input record; could be null if it is not available
|
Serde<?> |
keySerde()
Returns the default key serde
|
StreamsMetrics |
metrics()
Returns Metrics instance
|
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)
|
void |
register(StateStore store,
StateRestoreCallback stateRestoreCallback)
Registers and possibly restores the specified storage engine.
|
Cancellable |
schedule(Duration interval,
PunctuationType type,
Punctuator callback)
Schedules a periodic operation for processors.
|
Cancellable |
schedule(long intervalMs,
PunctuationType type,
Punctuator callback)
Deprecated.
Use
schedule(Duration, PunctuationType, Punctuator) instead |
File |
stateDir()
Returns the state directory for the partition.
|
TaskId |
taskId()
Returns the task id
|
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)
|
Serde<?> |
valueSerde()
Returns the default value serde
|
String applicationId()
TaskId taskId()
Serde<?> keySerde()
Serde<?> valueSerde()
File stateDir()
StreamsMetrics metrics()
void register(StateStore store, StateRestoreCallback stateRestoreCallback)
store
- the storage enginestateRestoreCallback
- the restoration callback logic for log-backed state stores upon restartIllegalStateException
- If store gets registered after initialized is already finishedStreamsException
- if the store's change log does not contain the partitionStateStore getStateStore(String name)
name
- The store name@Deprecated Cancellable schedule(long intervalMs, PunctuationType type, Punctuator callback)
schedule(Duration, PunctuationType, Punctuator)
insteadinitialization
or
processing
to
schedule a periodic callback — called a punctuation — to Punctuator.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 the TimestampExtractor
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 completePunctuationType.STREAM_TIME
, when stream time advances more than intervalPunctuationType.WALL_CLOCK_TIME
, on GC pause, too short interval, ...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 timeCancellable schedule(Duration interval, PunctuationType type, Punctuator callback) throws IllegalArgumentException
initialization
or
processing
to
schedule a periodic callback — called a punctuation — to Punctuator.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 the TimestampExtractor
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 completePunctuationType.STREAM_TIME
, when stream time advances more than intervalPunctuationType.WALL_CLOCK_TIME
, on GC pause, too short interval, ...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 timeIllegalArgumentException
<K,V> void forward(K key, V value)
key
- keyvalue
- value<K,V> void forward(K key, V value, To to)
key
- keyvalue
- valueto
- the options to use when forwarding@Deprecated <K,V> void forward(K key, V value, int childIndex)
forward(Object, Object, To)
insteadkey
- keyvalue
- valuechildIndex
- index in list of children of this node@Deprecated <K,V> void forward(K key, V value, String childName)
forward(Object, Object, To)
insteadkey
- keyvalue
- valuechildName
- name of downstream processorvoid commit()
String topic()
int partition()
long offset()
Headers headers()
long timestamp()
ConsumerRecord
by TimestampExtractor
.
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 smallest among all its input stream partition timestamps.Map<String,Object> appConfigs()
StreamsConfig
object and associated to the ProcessorContext.
The type of the values is dependent on the type
of the property
(e.g. the value of DEFAULT_KEY_SERDE_CLASS_CONFIG
will be of type Class
, even if it was specified as a String to
StreamsConfig(Map)
).
Map<String,Object> appConfigsWithPrefix(String prefix)
StreamsConfig
object and associated to the ProcessorContext.prefix
- the properties prefix