[
https://issues.apache.org/jira/browse/KAFKA-7250?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16574974#comment-16574974
]
John Roesler commented on KAFKA-7250:
-------------------------------------
I've created [https://github.com/apache/kafka/pull/5481] to make transform
expect a TransformerSupplier instead of a Function[Transformer]. This is a
super minor change, but it does move us toward having one pattern for these
methods instead of two.
( I started the PR because I thought there was a way to fix the bug without
breaking the API, but instead my dreams were broken. Nevertheless, I think just
moving toward Supplier is still beneficial. )
> 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
> Components: streams
> Affects Versions: 2.0.0
> Reporter: Michal
> Assignee: Michal Dziemianko
> Priority: Major
> Labels: scala
> Fix For: 2.0.1, 2.1.0
>
>
> 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)
