[jira] [Created] (KAFKA-7250) Kafka-Streams-Scala DSL transform shares transformer instance

["Michal (JIRA)" <ji...@apache.org>]

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)