KForward - a bound on the types of keys that may be forwardedVForward - a bound on the types of values that may be forwardedpublic interface ProcessorContext<KForward,VForward>
| 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 extends KForward,V extends VForward> |
forward(Record<K,V> record)
Forwards a record to all child processors.
|
<K extends KForward,V extends VForward> |
forward(Record<K,V> record,
String childName)
Forwards a record to the specified child processor.
|
<S extends StateStore> |
getStateStore(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.
|
String applicationId()
TaskId taskId()
Optional<RecordMetadata> recordMetadata()
schedule(Duration, PunctuationType, Punctuator) ),
or because a parent processor called forward(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 regular
Processor.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 is Optional.
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.
Serde<?> keySerde()
Serde<?> valueSerde()
File stateDir()
StreamsMetrics metrics()
<S extends StateStore> S getStateStore(String name)
S - The type or interface of the store to returnname - The store nameClassCastException - if the return type isn't a type or interface of the actual returned store.Cancellable schedule(Duration interval, PunctuationType type, Punctuator callback)
initialization or
Processor.process(Record) 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_TIMEcallback - a function consuming timestamps representing the current stream or system timeIllegalArgumentException - if the interval is not representable in milliseconds<K extends KForward,V extends VForward> void forward(Record<K,V> record)
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),
and Record.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
the Header class is mutable. The easiest way to
safely handle headers is to use the Record 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);
}
record - The record to forward to all children<K extends KForward,V extends VForward> void forward(Record<K,V> record, String childName)
forward(Record) for considerations.record - The record to forwardchildName - The name of the child processor to receive the recordforward(Record)void commit()
Map<String,Object> appConfigs()
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 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)
The config properties are defined in the StreamsConfig
object and associated to the ProcessorContext.
prefix - the properties prefix