Michal created KAFKA-7250: ----------------------------- Summary: Kafka-Streams-Scala DSL transform shares transformer instance Key: KAFKA-7250 URL: https://issues.apache.org/jira/browse/KAFKA-7250 Project: Kafka Issue Type: Bug Reporter: Michal
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. -- This message was sent by Atlassian JIRA (v7.6.3#76005)