[jira] [Comment Edited] (KAFKA-7523) TransformerSupplier/ProcessorSupplier enhancements

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

    [ https://issues.apache.org/jira/browse/KAFKA-7523?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16691122#comment-16691122 ] 

Guozhang Wang edited comment on KAFKA-7523 at 11/18/18 11:06 PM:
-----------------------------------------------------------------

[~pgwhalen] Sorry for being late on this. I think you have a good point that since in {{StreamsBuilder#build()}} the {{connectProcessorAndStateStore}} will be called, which checked if the state store has already been added to the topology. Hence if the state store is not added yet, it will throw an exception, and hence the awkwardness that you described in the above.

Note that in {{StreamsBuilder}}, you can also directly add a state store in {{StreamsBuilder#addStateStore}}, it is added in version 1.0.0 as part of https://issues.apache.org/jira/browse/KAFKA-5873. So you can actually do:

{code}
builder.addStateStore(... "store-name")
builder.stream(...).transform(..., "store-name")
{code}


Does that work for you?


was (Author: guozhang):
[~pgwhalen] Sorry for being late on this. I think you have a good point that since in {{StreamsBuilder#build()}} the {{connectProcessorAndStateStore}} will be called, which checked if the state store has already been added to the topology. Hence if the state store is not added yet, it will throw an exception, and hence the awkwardness that you described in the above.

Note that in {{StreamsBuilder}}, you can also directly add a state store in {{StreamsBuilder#addStateStore}}, it is added in version 1.0.0 as part of https://issues.apache.org/jira/browse/KAFKA-5873. Does that work for you?

> TransformerSupplier/ProcessorSupplier enhancements
> --------------------------------------------------
>
>                 Key: KAFKA-7523
>                 URL: https://issues.apache.org/jira/browse/KAFKA-7523
>             Project: Kafka
>          Issue Type: Improvement
>          Components: streams
>            Reporter: Paul Whalen
>            Priority: Minor
>              Labels: needs-kip
>
> I have found that when writing "low level" {{Processors}} and {{Transformers}} that are stateful, often I want these processors to "own" one or more state stores, the details of which are not important to the business logic of the application.  However, when incorporating these into the topologies defined by the high level API, using {{KStream::transform}} or {{KStream::process}}, I'm forced to specify the stores so the topology is wired up correctly.  This creates an unfortunate pattern where the {{TransformerSupplier}} or {{ProcessorSupplier,}} who (according to the pattern I've been following) holds the information about the name of the state stores, must be defined above the "high level" "fluent API"-style pipeline, which makes it hard to understand the business logic data flow.
>  
> What I currently have to do:
> {code:java}
> TransformerSupplier transformerSupplier = new TransformerSupplierWithState(topology, val -> businessLogic(val));
> builder.stream("in.topic")
>         .transform(transformerSupplier, transformerSupplier.stateStoreNames())
>         .to("out.topic");{code}
> I have to both define the {{TransformerSupplier}} above the "fluent block", and pass the topology in so I can call {{topology.addStateStore()}} inside the {{TransformerSupplier}} constructor and tell the {{StreamsBuilder}} what the state store names are for that point in the topology. The lambda {{val -> businessLogic(val)}} is really what I want to see in-line because that's the crux of what is happening, along with the name of some factory method describing what the transformer is doing for me internally. This issue is obviously exacerbated when the "fluent block" is much longer than this example - It gets worse the farther away {{val -> businessLogic(val)}} is from {{KStream::transform}}.
>  
> An improvement:
> {code:java}
> builder.stream("in.topic")
>         .transform(transformerSupplierWithState(topology, val -> businessLogic(val)))
>         .to("out.topic");{code}
> Which implies the existence of a {{KStream::transform}} that takes a single argument that adheres to this interface:
> {code:java}
> interface TransformerSupplierWithState<K, V, R> {
>     Transformer<K, V, R> get();
>     String[] stateStoreNames();
> }{code}
> Or better yet, I wouldn't have to pass in the topology, the caller of {{TransformerSupplierWithState}} could also handle the job of "adding" its state stores to the topology:
> {code:java}
> interface TransformerSupplierWithState<K, V, R> {
>     Transformer<K, V, R> get();
>     Map<String, StoreBuilder> stateStores();
> }{code}
> Which would enable my ideal:
> {code:java}
> builder.stream("in.topic")
>         .transform(transformerSupplierWithState(val -> businessLogic(val)))
>         .to("out.topic");{code}
> I think this would be a huge improvement in the usability of low-level processors with the high-level DSL.
> Please let me know if I'm missing something as to why this cannot or should not happen, or if there is a better forum for this suggestion (presumably it would require a KIP?). I'd be happy to build it as well if there is a chance of it being merged, it doesn't seem like a huge challenge to me.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)