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)
