Description
The new Kafka Streams Scala DSL provides transform function with following signature
def transform[K1, V1](transformer: Transformer[K, V, (K1, V1)], stateStoreNames: String*): KStream[K1, V1]
the provided 'transformer' (will refer to it as scala-transformer) instance is than used to derive java Transformer instance and in turn a TransformerSupplier that is passed to the underlying java DSL. However that causes all the tasks to share the same instance of the scala-transformer. This introduce all sort of issues. The simplest way to reproduce is to implement simplest transformer of the following shape:
.transform(new Transformer[String, String, (String, String)] {
var context: ProcessorContext = _
def init(pc: ProcessorContext) = { context = pc}
def transform(k: String, v: String): (String, String) = {
context.timestamp()
...
}}}{{})
the call to timestmap will die with exception "This should not happen as timestamp() should only be called while a record is processed" due to record context not being set - while the update of record context was actually performed, but due to shared nature of the scala-transformer the local reference to the processor context is pointing to the one of the last initialized task rather than the current task.
The solution is to accept a function in following manner:
def transform[K1, V1](getTransformer: () => Transformer[K, V, (K1, V1)], stateStoreNames: String*): KStream[K1, V1]
or TransformerSupplier - like the transformValues DSL function does.
Attachments
Issue Links
- is duplicated by
-
KAFKA-7882 StateStores are frequently closed during the 'transform' method
-
- Resolved
-
- links to