Current state: Accepted
Discussion thread: here
JIRA: KAFKA-6849
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
In Kafka Streams the KStream interface has several transformValues methods, where as the KTable interface does not. The related mapValues method is available on both KTable and KStream. The transformValue method is more flexible than mapValues:
transformValues accepts a supplier of transformers, rather than the mapper instance mapValues accepts. The supplier is used to instantiate a transformer per stream task, meaning the implementation does not need to be thread-safe. Conversely, mapValues shares the passed in mapper instance across stream tasks, so the implementations must be thread-safe.transformValues allows access to existing state-stores, where as mapValues does not. ValueTransformer interface is much richer than the ValueMapper, supporting both init() and close() calls.There is no conceptual reason not to support the richer transformValues on the KTable interface.
The following methods would be added to the Java KTable interface:
<VR> KTable<K, VR> transformValues(final ValueTransformerWithKeySupplier<? super K, ? super V, ? extends VR> valueTransformerSupplier,
final String... stateStoreNames);
<VR> KTable<K, VR> transformValues(final ValueTransformerWithKeySupplier<? super K, ? super V, ? extends VR> valueTransformerSupplier,
final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized,
final String... stateStoreNames); |
The following methods would be added to the Scala KTable class.
def transformValues[VR](valueTransformerSupplier: ValueTransformerWithKeySupplier[K, V, VR],
stateStoreNames: String*): KTable[K, VR]
def transformValues[VR](valueTransformerSupplier: ValueTransformerWithKeySupplier[K, V, VR],
materialized: Materialized[K, VR, KeyValueStore[Bytes, Array[Byte]]],
stateStoreNames: String*): KTable[K, VR] |
The new methods on `KTableImpl` will add a new KTableTransformValues processor node and attach any state stores. The new KTableTransformValues will be implemented in a similar manner to other processors, instantiating the user supplied transformer once per task.
The `ProcessorContext` passed to the `init()` method of transformer implementations will be restricted: any call to any variant of the `forward()` method will throw a `StreamsException`. This will stop implementations outputting values with a new key.
Users must upgrade to new version if they want to use this functionality.
Unit tests to cover new classes and methods. Integration or system test are not required.
Include overloads that take a `ValueTransformerSupplier`, matching the overloads available on `KStream`. These were no included as it keeps the interface more succinct, users can ignore the key value if they do not need it, and likely these overloads on the `KStream` interface will be deprecated, in favour of the 'WithKey' variants, in time.