Interface ProcessorContext<KForward,VForward>
-
- Type Parameters:
KForward
- a bound on the types of keys that may be forwardedVForward
- a bound on the types of values that may be forwarded
- All Known Implementing Classes:
MockProcessorContext
public interface ProcessorContext<KForward,VForward>
Processor context interface.
-
-
Method Summary
All Methods Instance Methods Abstract Methods Modifier and Type Method 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 extends KForward,V extends VForward>
voidforward(Record<K,V> record)
Forwards a record to all child processors.<K extends KForward,V extends VForward>
voidforward(Record<K,V> record, String childName)
Forwards a record to the specified child processor.<S extends StateStore>
SgetStateStore(String name)
Get the state store given the store name.Serde<?>
keySerde()
Returns the default key serde.StreamsMetrics
metrics()
Returns Metrics instance.Optional<RecordMetadata>
recordMetadata()
The metadata of the source record, if is one.Cancellable
schedule(Duration interval, PunctuationType type, Punctuator callback)
Schedules a periodic operation for processors.File
stateDir()
Returns the state directory for the partition.TaskId
taskId()
Returns the task id.Serde<?>
valueSerde()
Returns the default value serde.
-
-
-
Method Detail
-
applicationId
String applicationId()
Returns the application id.- Returns:
- the application id
-
taskId
TaskId taskId()
Returns the task id.- Returns:
- the task id
-
recordMetadata
Optional<RecordMetadata> recordMetadata()
The metadata of the source record, if is one. Processors may be invoked to process a source record from an input topic, to run a scheduled punctuation (seeschedule(Duration, PunctuationType, Punctuator)
), or because a parent processor calledforward(Record)
.In the case of a punctuation, there is no source record, so this metadata would be undefined. Note that when a punctuator invokes
forward(Record)
, downstream processors will receive the forwarded record as a regularProcessor.process(Record)
invocation. In other words, it wouldn't be apparent to downstream processors whether or not the record being processed came from an input topic or punctuation and therefore whether or not this metadata is defined. This is why the return type of this method isOptional
.If there is any possibility of punctuators upstream, any access to this field should consider the case of "
recordMetadata().isPresent() == false
". Of course, it would be safest to always guard this condition.
-
keySerde
Serde<?> keySerde()
Returns the default key serde.- Returns:
- the key serializer
-
valueSerde
Serde<?> valueSerde()
Returns the default value serde.- Returns:
- the value serializer
-
stateDir
File stateDir()
Returns the state directory for the partition.- Returns:
- the state directory
-
metrics
StreamsMetrics metrics()
Returns Metrics instance.- Returns:
- StreamsMetrics
-
getStateStore
<S extends StateStore> S getStateStore(String name)
Get the state store given the store name.- Type Parameters:
S
- The type or interface of the store to return- Parameters:
name
- The store name- Returns:
- The state store instance
- Throws:
ClassCastException
- if the return type isn't a type or interface of the actual returned store.
-
schedule
Cancellable schedule(Duration interval, PunctuationType type, Punctuator callback)
Schedules a periodic operation for processors. A processor may call this method duringinitialization
orProcessor.process(Record)
processing} 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, ...
- 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
- if the interval is not representable in milliseconds
-
forward
<K extends KForward,V extends VForward> void forward(Record<K,V> record)
Forwards a record to all child processors.Note that the forwarded
Record
is shared between the parent and child processors. And of course, the parent may forward the same object to multiple children, and the child may forward it to grandchildren, etc. Therefore, you should be mindful of mutability.The
Record
class itself is immutable (all the setter-style methods return an independent copy of the instance). However, the key, value, and headers referenced by the Record may themselves be mutable.Some programs may opt to make use of this mutability for high performance, in which case the input record may be mutated and then forwarded by each
Processor
. However, most applications should instead favor safety.Forwarding records safely simply means to make a copy of the record before you mutate it. This is trivial when using the
Record.withKey(Object)
,Record.withValue(Object)
, andRecord.withTimestamp(long)
methods, as each of these methods make a copy of the record as a matter of course. But a little extra care must be taken with headers, since theHeader
class is mutable. The easiest way to safely handle headers is to use theRecord
constructors to make a copy before modifying headers.In other words, this would be considered unsafe:
process(Record inputRecord) { inputRecord.headers().add(...); context.forward(inputRecord); }
This is unsafe because the parent, and potentially siblings, grandparents, etc., all will see this modification to their shared Headers reference. This is a violation of causality and could lead to undefined behavior.A safe usage would look like this:
process(Record inputRecord) { // makes a copy of the headers Record toForward = inputRecord.withHeaders(inputRecord.headers()); // Other options to create a safe copy are: // * use any copy-on-write method, which makes a copy of all fields: // toForward = inputRecord.withValue(); // * explicitly copy all fields: // toForward = new Record(inputRecord.key(), inputRecord.value(), inputRecord.timestamp(), inputRecord.headers()); // * create a fresh, empty Headers: // toForward = new Record(inputRecord.key(), inputRecord.value(), inputRecord.timestamp()); // * etc. // now, we are modifying our own independent copy of the headers. toForward.headers().add(...); context.forward(toForward); }
- Parameters:
record
- The record to forward to all children
-
forward
<K extends KForward,V extends VForward> void forward(Record<K,V> record, String childName)
Forwards a record to the specified child processor. Seeforward(Record)
for considerations.- Parameters:
record
- The record to forwardchildName
- The name of the child processor to receive the record- See Also:
forward(Record)
-
commit
void commit()
Requests a commit.
-
appConfigs
Map<String,Object> appConfigs()
Returns all the application config properties as key/value pairs.The config properties are defined in the
StreamsConfig
object and associated to the ProcessorContext.The type of the values is dependent on the
type
of the property (e.g. the value ofDEFAULT_KEY_SERDE_CLASS_CONFIG
will be of typeClass
, even if it was specified as a String toStreamsConfig(Map)
).- Returns:
- all the key/values from the StreamsConfig properties
-
appConfigsWithPrefix
Map<String,Object> appConfigsWithPrefix(String prefix)
Returns all the application config properties with the given key prefix, as key/value pairs stripping the prefix.The config properties are defined in the
StreamsConfig
object and associated to the ProcessorContext.- Parameters:
prefix
- the properties prefix- Returns:
- the key/values matching the given prefix from the StreamsConfig properties.
-
-