[
https://issues.apache.org/jira/browse/KAFKA-7250?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16570897#comment-16570897
]
Michal Dziemianko commented on KAFKA-7250:
------------------------------------------
[~guozhang] it seems like the API is really inconsistent:
def transform[K1, V1](transformer: Transformer[K, V, (K1, V1)],
stateStoreNames: String*): KStream[K1, V1]
def transformValues[VR](valueTransformerSupplier: ValueTransformerSupplier[V,
VR],stateStoreNames: String*): KStream[K, VR]
def transformValues[VR](valueTransformerSupplier:
ValueTransformerWithKeySupplier[K, V, VR], stateStoreNames: String*):
KStream[K, VR]
def process(processorSupplier: () => Processor[K, V], stateStoreNames:
String*): Unit
Personally, I think the `process` is the best option (and it matches all other
operations like map, join, aggregate etc.)
I tried a quick implementation but there are some issues with transformValues
due to type erasure so I just did PR for the transform as it is currently not
very usable, while I will go over the API and try to make it a bit more
consistent later...
> 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
>
> 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)
