[
https://issues.apache.org/jira/browse/KAFKA-7250?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16570608#comment-16570608
]
Guozhang Wang commented on KAFKA-7250:
--------------------------------------
Great! I've assigned you to the ticket, please feel free to follow the PR
guidelines:
https://cwiki.apache.org/confluence/display/KAFKA/Contributing+Code+Changes#ContributingCodeChanges-PullRequest
> 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
> Assignee: Michal Dziemianko
> Priority: Major
>
> 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)
