Interface ValueTransformer<V,VR>
- Type Parameters:
V- value typeVR- transformed value type
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, this ValueTransformer can schedule
a method to be called periodically with the provided context.
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:
-
Method Summary
-
Method Details
-
init
Deprecated.Initialize this transformer. This is called once per instance when the topology gets initialized. When the framework is done with the transformer,close()will be called on it; the framework may later re-use the transformer by callinginit(ProcessorContext)again.The provided
contextcan be used to access topology and record meta data, toschedulea method to becalled periodicallyand to access attachedStateStores.Note that
ProcessorContextis updated in the background with the current record's meta data. Thus, it only contains valid record meta data when accessed withintransform(Object).Note that using
ProcessorContext.forward(Object, Object)orProcessorContext.forward(Object, Object, To)is not allowed within any method ofValueTransformerand will result in anexception.- Parameters:
context- the context- Throws:
IllegalStateException- If store gets registered after initialization is already finishedStreamsException- if the store's change log does not contain the partition
-
transform
Deprecated.Transform the given value to a new value. Additionally, anyStateStorethat isattachedto this operator can be accessed and modified arbitrarily (cf.ProcessorContext.getStateStore(String)).Note, that using
ProcessorContext.forward(Object, Object)orProcessorContext.forward(Object, Object, To)is not allowed withintransformand will result in anexception.- Parameters:
value- the value to be transformed- Returns:
- the new value
-
close
void close()Deprecated.Close this transformer and clean up any resources. The framework may later re-use this transformer by callinginit(ProcessorContext)on it again.It is not possible to return any new output records within
close(). UsingProcessorContext.forward(Object, Object)orProcessorContext.forward(Object, Object, To)will result in anexception.
-
FixedKeyProcessorinstead.