org.apache.kafka.streams.kstream

Interface ValueTransformer<V,VR>

Type Parameters:

V - value type

VR - transformed value type

@InterfaceStability.Unstable
public interface ValueTransformer<V,VR>

The ValueTransformer interface for stateful mapping of a value to a new value (with possible new type). This is a stateful record-by-record operation, i.e, transform(Object) is invoked individually for each record of a stream and can access and modify a state that is available beyond a single call of transform(Object) (cf. ValueMapper for stateless value transformation). Additionally, the interface can be called in regular intervals based on the processing progress (cf. punctuate(long). If ValueTransformer is applied to a KeyValue pair record the record's key is preserved.

Use ValueTransformerSupplier to provide new instances of ValueTransformer to Kafka Stream's runtime.

If a record's key and value should be modified Transformer can be used.

See Also:

ValueTransformerSupplier, KStream.transformValues(ValueTransformerSupplier, String...), Transformer

Method Summary

Methods

Modifier and Type	Method and Description
void	close() Close this processor and clean up any resources.
void	init(ProcessorContext context) Initialize this transformer.
VR	punctuate(long timestamp) Perform any periodic operations if this processor schedule itself with the context during initialization.
VR	transform(V value) Transform the given value to a new value.

Method Detail

init

void init(ProcessorContext context)

Initialize this transformer. This is called once per instance when the topology gets initialized.

The provided context can be used to access topology and record meta data, to schedule itself for periodical calls (cf. punctuate(long)), and to access attached StateStores.

Note that ProcessorContext is updated in the background with the current record's meta data. Thus, it only contains valid record meta data when accessed within transform(Object).

Note that using ProcessorContext.forward(Object, Object), ProcessorContext.forward(Object, Object, int), or ProcessorContext.forward(Object, Object, String) is not allowed within any method of ValueTransformer and will result in an exception.

Parameters:

context - the context

transform

VR transform(V value)

Transform the given value to a new value. Additionally, any StateStore that is attached to this operator can be accessed and modified arbitrarily (cf. ProcessorContext.getStateStore(String)).

Note, that using ProcessorContext.forward(Object, Object), ProcessorContext.forward(Object, Object, int), and ProcessorContext.forward(Object, Object, String) is not allowed within transform and will result in an exception.

Parameters:

value - the value to be transformed

Returns:

the new value

punctuate

VR punctuate(long timestamp)

Perform any periodic operations if this processor schedule itself with the context during initialization.

It is not possible to return any new output records within punctuate. Using ProcessorContext.forward(Object, Object), ProcessorContext.forward(Object, Object, int), or ProcessorContext.forward(Object, Object, String) will result in an exception. Furthermore, punctuate must return null.

Note, that punctuate is called base on stream time (i.e., time progress with regard to timestamps return by the used TimestampExtractor) and not based on wall-clock time.

Parameters:

timestamp - the stream time when punctuate is being called

Returns:

must return null—otherwise, an exception will be thrown

close

void close()

Close this processor and clean up any resources.

It is not possible to return any new output records within close(). Using ProcessorContext.forward(Object, Object), ProcessorContext.forward(Object, Object, int), or ProcessorContext.forward(Object, Object, String) will result in an exception.