Package org.apache.kafka.streams.processor.api

Interface ProcessingContext

All Known Subinterfaces:

FixedKeyProcessorContext<KForward,VForward>, ProcessorContext<KForward,VForward>

All Known Implementing Classes:

MockProcessorContext

public interface ProcessingContext

Processor context interface.

Method Summary

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)	Return all the application config properties with the given key prefix, as key/value pairs stripping the prefix.
String	applicationId()	Return the application id.
void	commit()	Request a commit.
long	currentStreamTimeMs()	Return the current stream-time in milliseconds.
long	currentSystemTimeMs()	Return the current system timestamp (also called wall-clock time) in milliseconds.
<S extends StateStore> S	getStateStore(String name)	Get the state store given the store name.
Serde<?>	keySerde()	Return the default key serde.
StreamsMetrics	metrics()	Return Metrics instance.
Optional<RecordMetadata>	recordMetadata()	Return the metadata of the current record if available.
Cancellable	schedule(Duration interval, PunctuationType type, Punctuator callback)	Schedule a periodic operation for processors.
File	stateDir()	Return the state directory for the partition.
TaskId	taskId()	Return the task id.
Serde<?>	valueSerde()	Return the default value serde.

Method Details

applicationId

String applicationId()

Return the application id.

Returns:

the application id

taskId

TaskId taskId()

Return the task id.

Returns:

the task id

recordMetadata

Optional<RecordMetadata> recordMetadata()

Return the metadata of the current record if available. Processors may be invoked to process a source record from an input topic, to run a scheduled punctuation (see 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) or FixedKeyProcessor.process(FixedKeyRecord) invocation. In other words, it wouldn't be apparent to downstream processors whether the record being processed came from an input topic or punctuation and therefore whether 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.

keySerde

Serde<?> keySerde()

Return the default key serde.

Returns:

the key serializer

valueSerde

Serde<?> valueSerde()

Return the default value serde.

Returns:

the value serializer

stateDir

File stateDir()

Return the state directory for the partition.

Returns:

the state directory

metrics

StreamsMetrics metrics()

Return 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)

Schedule a periodic operation for processors. A processor may call this method during initialization, processing, 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 arrive
• PunctuationType.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

Skipping punctuations: Punctuations will not be triggered more than once at any given timestamp. This means that "missed" punctuation will be skipped. It's possible to "miss" a punctuation if:

• 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

commit

void commit()

Request a commit. Note that calling commit() is only a request for a commit, but it does not execute one. Hence, when commit() returns, no commit was executed yet. However, Kafka Streams will commit as soon as possible, instead of waiting for next commit.interval.ms to pass.

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 of DEFAULT_KEY_SERDE_CLASS_CONFIG will be of type Class, even if it was specified as a String to StreamsConfig(Map)).

Returns:

all the key/values from the StreamsConfig properties

appConfigsWithPrefix

Map<String,Object> appConfigsWithPrefix(String prefix)

Return 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.

currentSystemTimeMs

long currentSystemTimeMs()

Return the current system timestamp (also called wall-clock time) in milliseconds.

Note: this method returns the internally cached system timestamp from the Kafka Stream runtime. Thus, it may return a different value compared to System.currentTimeMillis().

Returns:

the current system timestamp in milliseconds

currentStreamTimeMs

long currentStreamTimeMs()

Return the current stream-time in milliseconds.

Stream-time is the maximum observed record timestamp so far (including the currently processed record), i.e., it can be considered a high-watermark. Stream-time is tracked on a per-task basis and is preserved across restarts and during task migration.

Note: this method is not supported for global processors (cf. Topology.addGlobalStore(org.apache.kafka.streams.state.StoreBuilder<?>, java.lang.String, org.apache.kafka.common.serialization.Deserializer<K>, org.apache.kafka.common.serialization.Deserializer<V>, java.lang.String, java.lang.String, org.apache.kafka.streams.processor.ProcessorSupplier<K, V>) (...) and StreamsBuilder.addGlobalStore(org.apache.kafka.streams.state.StoreBuilder<?>, java.lang.String, org.apache.kafka.streams.kstream.Consumed<K, V>, org.apache.kafka.streams.processor.ProcessorSupplier<K, V>) (...), because there is no concept of stream-time for this case. Calling this method in a global processor will result in an UnsupportedOperationException.

Returns:

the current stream-time in milliseconds