Great thoughts Matthias, thanks! I think we're all agreed that naming and
documentation/education are the biggest hurdles for this KIP, and in light
of that, I think it makes sense for me to just take a stab at a full
fledged PR with documentation to convince us that it's possible to do it
with enough clarity.
In response to your specific thoughts:
1) StateStoreConnector as a name: Really good point about defining the
difference between "adding" and "connecting." Guozhang suggested
StateStoreConnector which was definitely an improvement over my
StateStoresSupplier, but I think you're right that we need to be careful to
make it clear that it's really accomplishing both. Thinking about it now,
one problem with Connector is that the implementer of the interface is not
really doing any connecting, it's providing/supplying the store that will
be both added and connected. StoreProvider seems reasonable to me and
probably the best candidate at the moment, but it would be nice if the name
could convey that it's providing the store specifically so the caller can
add it to the topology and connect it to the associated transformer.
In general I think that really calling out what "adding" versus
"connecting" is in the documentation will help make the entire purpose of
this feature more clear to the user.
2) Default method vs new interface: The choice of a default method was
influenced by Guozhang's fear about API bloat/discoverability. I can
definitely see it both ways Would the separate interface be a
sub-interface of Processor/TransformerSupplier or standalone? It seems
like you're suggesting standalone and I think that's what I favor. My only
concern there is that the interface wouldn't actually be a type to any
public API which sort of hurts discoverability. You would have to read the
javadocs for stream.process/transform() to discover that implementing the
interface in addition to Processor/TransformerSupplier would add and
connect the store for you. But that added burden actually probably helps
us in terms of making sure people don't mix and match, like you said.
3) Returning null instead of empty: Seems fair to me. I always worry about
returning null when an empty collection can be used instead, but given that
the library is the caller rather than the client I think your point makes
sense here.
4) Returning Set instead of Collection: Agreed, don't see why not to make
it more specific.
Paul
On Fri, Apr 26, 2019 at 2:30 AM Matthias J. Sax <matth...@confluent.io>
wrote:
> Hi, sorry for the long pause. Just trying to catch up here.
>
> I think it save to allow `addStateStore()` to be idempotent for the same
> `StoreBuilder` object. In fact, the `name` is "hard coded" and thus it's
> not really possible to use the same `StoreBuilder` object to create
> different stores.
>
> I also agree with the concern, that only allowing a single store (as
> proposed by Ivan) might be too restrictive.
>
> Overall, the current KIP version LGTM. I don't have mayor concerns about
> user education for this case, but I agree that we need to document this
> clearly.
>
> Some further comments:
>
> (1) I am not sure if `StateStoreConnector` is the best name for the new
> interface. Note, that there are two concepts about stores:
>
> - adding a store: this makes the store available in the topology in
> general (however, the store is still "dangling", and not used)
> - connecting a store: this allows a processor etc to use a store
>
> The new interface does both, but its name only indicates that second
> part what might be confusing. It might be especially confusing because
> we want to disallow to mix the exiting "manually add and connect"
> pattern, with a new pattern to "auto add+connect". If the new interface
> name indicates the connect part only, user might think they need to add
> stores manually and can connect automatically.
>
> Unfortunately, I don't have a much better suggestion for a name either.
> The only idea that came to my mind was `StoreProvider`: to me, a
> provider is a "service" interface that does work for us, ie, it adds and
> connects a store. Not sure if this is too subtle, if we consider that
> there is already the `StoreSupplier` interface?
>
> But maybe somebody else might still have a good idea on how the improve
> the name.
>
> In any case, I would suggest to shorten the name to `StoreConnector`
> instead of `StateStoreConnector`, because we also have `StoreSupplier`
> and `StoreBuilder`.
>
>
>
> (2) The KIP proposes to add the new interface to `ProcessorSupplier` etc
> and to add a default implementation for the new method. Hence, user
> would need to overwrite this default implementation to op-in to the
> feature. I am wonder if it might be better to not add the new interface
> to `ProcessorSupplier` etc and to just provide a new interface with no
> default implementation. Users would opt-in by adding the interface
> explicitly to their existing `ProcessorSupplier` implementation.
> Overwriting a default method and getting different behavior seems to be
> a little subtle to me, especially, because we don't want to allow to
> mix-and-match the old and new approaches. Think: I only overwrite a
> default method and my code breaks.
>
> Thoughts?
>
>
>
> (3) If we keep the current default implementation for the new method, I
> am wondering if it should return `null` instead of an empty collection?
> This might be saver to detect bugs in user code for which, per accident,
> an empty collection could be returned.
>
>
>
> (4) Should the new method return a `Set` instead of a `Collection` to
> indicate the semantics clearly (ie, returning the same `StoreBuilder`
> multiple times is idempotent and one cannot add+connect to it twice).
>
>
>
> -Matthias
>
>
>
>
> On 4/6/19 12:27 PM, Paul Whalen wrote:
> > Ivan and Guozhang,
> >
> > Thanks for the thoughts! Ivan's use case is definitely interesting. The
> > way I see it, if we can achieve the main goal of the KIP (allowing
> > Processor/TransformerSuppliers to encapsulate their usage of state
> stores),
> > we will enable this kind of thing in "user space" very easily.
> >
> > I will say that I'm not totally sure that most use cases of transform()
> use
> > just one state store. It's hard to know since I haven't seen many
> examples
> > in public, but my team's usages almost exclusively require multiple state
> > stores. We only reach for the low level processor API when we need that
> > complexity, and it's somewhat hard to imagine many use cases that only
> need
> > one state store, since the high level DSL can usually accomplish those
> > tasks. The example Ivan presented for instance looks like a
> > stream.groupByKey().reduce(...) to me. Ivan, I'd be curious what sort of
> > other usages you're imagining.
> >
> > That being said, perhaps the Processor API should really just be
> considered
> > a separate paradigm in Streams, not just a lower level that we reach to
> > when necessary. In which case it would be beneficial to make the simple
> > use cases easier. I've definitely talked about this with my own team -
> if
> > you're less familiar with the kind of functional style that the high
> level
> > DSL offers, it might be easier to "see" your state and interact with it
> > directly.
> >
> > Anyway, I've updated the KIP to reflect my current PR with Guozhang's
> > suggestions. It seems like there is at least some interest in that on
> its
> > own and not a ton of pushback, so I think I will try to start a vote.
> >
> > Paul
> >
> > On Sat, Mar 30, 2019 at 10:03 AM Ivan Ponomarev <iponoma...@mail.ru>
> wrote:
> >
> >> Hi all!
> >>
> >> I was about to write another KIP, but found out that KIP-401 addresses
> >> exactly the problem I faced. So let me jump into your discussion and ask
> >> you to assess another idea.
> >>
> >> I fully agree with the KIP-401's motivation part. E. g in my project I
> had
> >> to invent a wrapper class that hides the details of KeyValueStore
> >> management from business logic. Of course this should be done better in
> >> KStreams API.
> >>
> >> But I was about to look at this problem from another side and propose a
> >> simple alternative in high-level DSL, that will not fit all the cases,
> but
> >> most of them. Hence my idea does not exclude the Paul's proposal.
> >>
> >> What if we restrict ourselves to *only one* KeyValueStore and propose a
> >> method that resembles `aggregate` and `reduce` methods, like this:
> >>
> >> stream
> >> .map(...)
> >> .filter(...)
> >> .transform ((k, v, s)->{....}, Transformed.with(....))
> >>
> >> where
> >> * k, v -- input key & value
> >> * s -- a KeyValueStore provided as an argument
> >> * return value of the lambda should be KeyValue.pair(...)
> >> * Transformed.with... is a builder, used in order to define the
> >> Transformer and KeyValueStore building parameters. Some of these
> parameters
> >> should be:
> >> ** store's KeySerde,
> >> ** store's ValueSerde,
> >> ** whether the store is persistent or in-memory,
> >> ** store's name -- optional parameter, the system should be able to
> devise
> >> the name of the store transparently for the user, if we don't want to
> >> devise it ourselves/share the store between processors.
> >> ** scheduled punctuation.
> >>
> >> Imagine we have a KStream<String, Integer>, and we need to calculate a
> >> `derivative` stream, that is, a stream of 'deltas' of the provided
> integer
> >> values.
> >>
> >> This could be achieved as simple as
> >>
> >> stream.transform((key, value, stateStore) -> {
> >> int previousValue =
> >> Optional.ofNullable(stateStore.get(key)).orElse(0);
> >> stateStore.put(key, value);
> >> return KeyValue.pair(key, value - previousValue);
> >> }
> >> //we do not need to bother with store name, punctuation etc.
> >> //may be even Serde part can be omitted, since we can inherit
> the
> >> serdes from stream by default
> >> , Transformed.with(Serdes.String(), Serdes.Integer())
> >> }
> >>
> >> The hard part of it is that new `transform` method definition should be
> >> parameterized by six type parameters:
> >>
> >> * input/output/KeyValueStore key type,
> >> * input/output/KeyValueStore value type.
> >>
> >> However, it seems that all these types can be inferred from the provided
> >> lambda and Transformed.with instances.
> >>
> >> What do you think about this?
> >>
> >> Regards,
> >>
> >> Ivan
> >>
> >>
> >> 27.03.2019 20:45, Guozhang Wang пишет:
> >>
> >> Hello Paul,
> >>
> >> Thanks for the uploaded PR and the detailed description! I've made a
> pass
> >> on it and left some comments.
> >>
> >> Overall I think I agree with you that passing in the storebuilder
> directly
> >> that store name is more convienent as it does not require another
> >> `addStore` call, but we just need to spend some more documentation
> effort
> >> on educating users about the two ways of connecting their stores. I'm
> >> slightly concerned about this education curve but I can be convinced if
> >> most people felt it is worthy.
> >>
> >>
> >> Guozhang
> >>
> >> On Sat, Mar 23, 2019 at 5:15 PM Paul Whalen <pgwha...@gmail.com> <
> pgwha...@gmail.com> wrote:
> >>
> >>
> >> I'd like to resurrect this discussion with a cursory, proof-of-concept
> >> implementation of the KIP which combines many of our ideas:
> https://github.com/apache/kafka/pull/6496. I tried to keep the diff as
> >> small as possible for now, just using it to convey the main ideas. But
> >> I'll separately address some of our earlier discussion:
> >>
> >> - Will there be a new, separate interface for users to implement for
> the
> >> new functionality? No, to hopefully keep things simple, all of the
> >> Processor/TransformerSupplier interfaces will just extend
> >> StateStoresSupplier, allowing users to opt in to this functionality
> by
> >> overriding the default implementation that gives an empty list.
> >> - Will the interface allow users to specify the store name, or the
> >> entire StoreBuilder? The entire StoreBuilder, so the
> >> Processor/TransformerSupplier can completely encapsulate name and
> >> implementation of a state store if desired.
> >> - Will the old way of specifying store names alongside the supplier
> when
> >> calling stream.process/transform() be deprecated? No, this is still a
> >> legitimate way to wire up Processors/Transformers and their stores.
> But
> >> I
> >> would recommend not allowing stream.process/transform() calls that
> use
> >> both
> >> store declaration mechanisms (this restriction is not in the proof of
> >> concept)
> >> - How will we handle adding the same state store to the topology
> >> multiple times because different Processor/TransformerSuppliers
> declare
> >> it?
> >> topology.addStateStore() will be slightly relaxed for convenience,
> and
> >> will
> >> allow adding the same StoreBuilder multiple times as long as the
> exact
> >> same
> >> StoreBuilder instance is being added for the same store name. This
> >> seems
> >> to prevent in practice the issue of accidentally making two state
> stores
> >> one by adding with the same name. For additional safety, if we
> wanted
> >> to
> >> (not in the proof of concept), we could allow for this relaxation
> only
> >> for
> >> internal callers of topology.addStateStore().
> >>
> >> So, in summary, the use cases look like:
> >>
> >> - 1 transformer/processor that owns its store: Using the new
> >> StateStoresSupplier interface method to supply its StoreBuilders that
> >> will
> >> be added to the topology automatically.
> >> - Multiple transformer/processors that share the same store: Either
> >>
> >>
> >> 1. The old way: the StoreBuilder is defined "far away" from the
> >> Transformer/Processor implementations, and is added to the topology
> >> manually by the user
> >> 2. The new way: the StoreBuilder is defined closer to the
> >> Transformer/Processor implementations, and the same instance is
> >> returned by
> >> all Transformer/ProcessorSuppliers that need it
> >>
> >>
> >> This makes the KIP wiki a bit stale; I'll update if we want to bring
> this
> >> design to a vote.
> >>
> >> Thanks!
> >> Paul
> >>
> >> On Sun, Dec 16, 2018 at 6:04 PM Guozhang Wang <wangg...@gmail.com> <
> wangg...@gmail.com> wrote:
> >>
> >>
> >> Matthias / Paul,
> >>
> >> The concern I had about introducing `StoreBuilderSupplier` is simply
> >> because it is another XXSupplier to the public API, so I'd like to ask
> if
> >> we really have to add it :)
> >>
> >> The difference between encapsulating the store name and encapsulating
> the
> >> full state store builder is that, in the former:
> >>
> >> -----------
> >>
> >> String storeName = "store1";
> >> builder.addStore(new MyStoreBuilder(storeName));
> >> stream1.transform(new MyTransformerSupplier(storeName)); // following
> >>
> >> my
> >>
> >> proposal, that the store name can be passed in and used for both
> >> `listStores` and in the `Transformer#init`; so the Transformer function
> >> does not need to get the constant string name again.
> >>
> >> // one caveat to admit, is that
> >> MyTransofmerSupplier logic may be just unique to `store1` so it cannot
> be
> >> reused with a different store name anyways.
> >> -----------
> >>
> >> While in the latter:
> >>
> >> -----------
> >>
> >> stream1.transform(new MyTransformerSupplierForStore1); // the name is
> >> just indicating that we may have one such supplier for each store.
> >>
> >> -----------
> >>
> >> I understand the latter introduce more convenience from the API, but the
> >> cost is that since we still cannot completely `builder.addStore`, but
> >>
> >> only
> >>
> >> reduce its semantic scope to shared state stores only,; hence users need
> >>
> >> to
> >>
> >> learn two ways of creating state stores for those two patterns.
> >>
> >> My argument is that more public APIs requires longer learning curve for
> >> users, and introduces more usage patterns that may confuse users (the
> >> proposal I had tries to replace one with another completely).
> >>
> >>
> >> Guozhang
> >>
> >> On Sun, Dec 16, 2018 at 2:58 PM Paul Whalen <pgwha...@gmail.com> <
> pgwha...@gmail.com> wrote:
> >>
> >>
> >> Thanks for the great thoughts Matthias and Guozhang!
> >>
> >> If I'm not mistaken, Guozhang's suggestion is what my second
> >>
> >> alternative
> >>
> >> on
> >>
> >> the KIP is ("Have the added method on the Supplier interfaces only
> >>
> >> return
> >>
> >> store names, not builders"). I do think it would be a worthwhile
> >>
> >> usability
> >>
> >> improvement on its own, but to Matthias's point, it doesn't achieve the
> >> full goal of completing encapsulating a state store and it's processor
> >>
> >> -
> >>
> >> it
> >>
> >> encapsulates the name, but not the StateStoreBuilder.
> >>
> >> I'm really intrigued by Matthias's idea that forgoes the default
> >>
> >> interface
> >>
> >> method I proposed. Having smaller, separate interfaces is a powerful
> >>
> >> idea
> >>
> >> and I think a cleaner API than what I proposed. The non-shared store
> >>
> >> use
> >>
> >> case is handled well here, and the shared store use case is possible,
> >> though maybe still not as graceful as we would like (having to add the
> >> StoreBuilderSupplier before the StoreNameSupplier seems maybe too
> >>
> >> subtle
> >>
> >> to
> >>
> >> me).
> >>
> >> We're all agreed that one of the big problems with the shared store use
> >> case is how to deal with adding the same store to the topology multiple
> >> times. Catching the "store already added" exception is risky. Here's
> >>
> >> a
> >>
> >> maybe radical idea: change `topology.addStateStore()` to be idempotent
> >>
> >> for
> >>
> >> adding a given state store name and `StoreBuilder`. In other words,
> >> `addStateStore` would not throw the "store already added" exception if
> >>
> >> the
> >>
> >> `StoreBuilder` being added for a given name has the same identity as
> >>
> >> the
> >>
> >> one that has already been added. Does this eliminate all the bugs
> >>
> >> we're
> >>
> >> worried about? Thinking about it for a few minutes, it seems to
> >>
> >> eliminate
> >>
> >> most at least (would a user really use the exact same StoreBuilder when
> >> they intend there to be two stores?). It might make the API slightly
> >> harder to use if a user isn't immediately aware of that subtlety, but a
> >> good error message should ease the pain, and it would happen
> >>
> >> immediately
> >>
> >> during development.
> >>
> >> And with regards to Matthias's comment about whether we need to
> >>
> >> deprecate
> >>
> >> existing varargs transform methods - I don't think we need to, but it
> >>
> >> might
> >>
> >> be nice for there only to be one way to do things, assuming whatever we
> >> come up with supports all existing use cases. I don't feel strongly
> >>
> >> about
> >>
> >> this, but if we don't deprecate, I do think it's important to add
> >>
> >> checks
> >>
> >> that prevent users from trying to do the same thing in two different
> >>
> >> ways,
> >>
> >> as we've discussed.
> >>
> >> Paul
> >>
> >> On Sun, Dec 16, 2018 at 5:36 AM Matthias J. Sax <matth...@confluent.io
> >>
> >> wrote:
> >>
> >>
> >> Guozhang,
> >>
> >>
> >> Regarding the last option to catch "store exist already" exception
> >>
> >> and
> >>
> >> fallback to connect stores, I'm a bit concerned it may be hiding
> >>
> >> actual
> >>
> >> user bugs.
> >>
> >> I agree with this concern. From my original email:
> >>
> >>
> >> The only disadvantage I see, might be
> >> potential bugs about sharing state if two different stores are
> >>
> >> named
> >>
> >> the
> >>
> >> same by mistake (this would not be detected).
> >>
> >> For your new proposal: I am not sure if it addresses Paul's original
> >> idea -- I hope Paul can clarify. From my understanding, the idea was
> >>
> >> to
> >>
> >> encapsulate a store and its processor. As many stores are not shared,
> >> this seems to be quite useful. Your proposal falls a little short to
> >> support encapsulation for none-shared stores.
> >>
> >>
> >> -Matthias
> >>
> >>
> >>
> >> On 12/15/18 1:40 AM, Guozhang Wang wrote:
> >>
> >> Matthias,
> >>
> >> Thanks for your feedbacks.
> >>
> >> Regarding the last option to catch "store exist already" exception
> >>
> >> and
> >>
> >> fallback to connect stores, I'm a bit concerned it may be hiding
> >>
> >> actual
> >>
> >> user bugs.
> >>
> >> Thinking about Paul's proposal and your suggestion again, I'd like
> >>
> >> to
> >>
> >> propose another alternative somewhere in the middle of your
> >>
> >> approaches,
> >>
> >> i.e. we still let users to create sharable state stores via
> >> `addStateStore`, and we allow the TransformerSupplier to return a
> >>
> >> list
> >>
> >> of
> >>
> >> state stores that it needs, i.e.:
> >>
> >> public interface TransformerSupplier<K, V, R> {
> >> Transformer<K, V, R> get();
> >> default List<String> stateStoreNames() {
> >> return Collections.emptyList();
> >> <https://cwiki.apache.org/confluence/pages/Collections.emptyList()
> >>
> >> ;>
> >>
> >> }
> >> }
> >>
> >> by doing this users can still "consolidate" the references of store
> >>
> >> names
> >>
> >> in a single place in the transform call, e.g.:
> >>
> >> public class MyTransformerSupplier<K, V, R> {
> >> private String storeName;
> >>
> >> public class MyTransformer<K, V, R> {
> >>
> >> ....
> >>
> >> init() {
> >> store = context.getStateStore(storeName);
> >> }
> >> }
> >>
> >> default List<String> stateStoreNames() {
> >> return Collections.singletonList(storeName);
> >> <https://cwiki.apache.org/confluence/pages/Collections.emptyList()
> >>
> >> ;>
> >>
> >> }
> >> }
> >>
> >> Basically, we move the parameters from the caller of `transform` to
> >>
> >> inside
> >>
> >> the TransformSuppliers. DSL implementations would not change much,
> >>
> >> simply
> >>
> >> calling `connectStateStore` by getting the list of names from the
> >>
> >> provided
> >>
> >> function.
> >>
> >> Guozhang
> >>
> >>
> >> On Thu, Dec 13, 2018 at 7:27 AM Matthias J. Sax <
> >>
> >> matth...@confluent.io
> >>
> >> wrote:
> >>
> >>
> >> Just a meta comment: do we really need to deprecate existing
> >> `transform()` etc methods?
> >>
> >> The last argument is a vararg, and thus, just keeping the existing
> >>
> >> API
> >>
> >> for this part seems to work too, allowing to implement both
> >>
> >> patterns?
> >>
> >> Also, instead of adding a default method, we could also add a new
> >> interface `StoreBuilderSupplier` with method `List<StoreBuilder>
> >> stateStores()` -- users could implement `TransformerSupplier` and
> >> `StoreBuilderSupplier` at once; and for this case, we require that
> >>
> >> users
> >>
> >> don't provide store name in `transform()`.
> >>
> >> Similar, we could add an interface `StoreNameSupplier` with method
> >> `List<String> stateStores()`. This allows to "auto-wire" a
> >>
> >> transformer
> >>
> >> to existing stores (to avoid the issue to add the same store
> >>
> >> multiple
> >>
> >> times).
> >>
> >> Hence, for shared stores, there would be one "main" transformer
> >>
> >> that
> >>
> >> implements `StoreBuilderSupplier` and that must be added first to
> >>
> >> the
> >>
> >> topology. The other transformers would implement
> >>
> >> `StoreNameSupplier`
> >>
> >> and
> >>
> >> just connect to those stores.
> >>
> >> Another possibility to avoid the issue of adding the same stores
> >> multiple times would be, that the DSL always calls
> >>
> >> `addStateStore()`
> >>
> >> but
> >>
> >> catches a potential "store exists already" exception and falls
> >>
> >> back
> >>
> >> to
> >>
> >> `connectProcessorAndStateStore()` for this case. Thus, we would
> >>
> >> not
> >>
> >> need
> >>
> >> the `StoreNameSupplier` interface and the order in which
> >>
> >> transformers
> >>
> >> are added would not matter either. The only disadvantage I see,
> >>
> >> might
> >>
> >> be
> >>
> >> potential bugs about sharing state if two different stores are
> >>
> >> named
> >>
> >> the
> >>
> >> same by mistake (this would not be detected).
> >>
> >>
> >>
> >> Just some ideas I wanted to share. What do you think?
> >>
> >>
> >>
> >> -Matthias
> >>
> >> On 12/11/18 3:46 AM, Paul Whalen wrote:
> >>
> >> Ah yes of course, this was an oversight, I completely ignored the
> >>
> >> multiple
> >>
> >> processors sharing the same state store when writing up the KIP.
> >>
> >> Which
> >>
> >> is
> >>
> >> funny, because I've actually done this (different processors
> >>
> >> sharing
> >>
> >> state
> >>
> >> stores) a fair amount myself, and I've settled on a pattern
> >>
> >> where I
> >>
> >> group
> >>
> >> the Processors in an enclosing class, and that enclosing class
> >>
> >> handles
> >>
> >> as
> >>
> >> much as possible. Here's a gist showing the rough structure,
> >>
> >> just
> >>
> >> for
> >>
> >> context:
> >>
> >> https://gist.github.com/pgwhalen/57a00dcc2269b7610e1aaeb1549b3b65
> >>
> >> . Note how it adds the stores to the topology, as well as
> >>
> >> providing a
> >>
> >> public method with the store names.
> >>
> >> I don't think my proposal completely conflicts with the multiple
> >>
> >> processors
> >>
> >> sharing state stores use case, since you can create a supplier
> >>
> >> that
> >>
> >> provides the store name you want, somewhat independently of your
> >>
> >> actual
> >>
> >> Processor logic. The issue I do see though, is that
> >> topology.addStateStore() can only be called once for a given
> >>
> >> store.
> >>
> >> So
> >>
> >> for
> >>
> >> your example, if the there was a single TransformerSupplier that
> >>
> >> was
> >>
> >> passed
> >>
> >> into both transform() calls, "store1" would be added (under the
> >>
> >> hood)
> >>
> >> to
> >>
> >> the topology twice, which is no good.
> >>
> >> Perhaps this suggests that one of my alternatives on the KIP
> >>
> >> might
> >>
> >> be
> >>
> >> desirable: either not having the suppliers return StoreBuilders
> >>
> >> (just
> >>
> >> store
> >>
> >> names), or not deprecating the old methods that take "String...
> >> stateStoreNames". I'll have to think about it a bit.
> >>
> >> Paul
> >>
> >> On Sun, Dec 9, 2018 at 11:57 PM Guozhang Wang <
> >>
> >> wangg...@gmail.com>
> >>
> >> wrote:
> >>
> >> Hello Paul,
> >>
> >> Thanks for the great writeup (very detailed and crystal
> >>
> >> motivation
> >>
> >> sections!).
> >>
> >> This is quite an interesting idea and I do like the API
> >>
> >> cleanness
> >>
> >> you
> >>
> >> proposed. The original motivation of letting StreamsTopology to
> >>
> >> add
> >>
> >> state
> >>
> >> stores though, is to allow different processors to share the
> >>
> >> state
> >>
> >> store.
> >>
> >> For example:
> >>
> >> builder.addStore("store1");
> >>
> >> // a path of stream transformations that leads to KStream
> >>
> >> stream1.
> >>
> >> stream1.transform(..., "store1");
> >>
> >> // another path that generates a KStream stream2.
> >> stream2.transform(..., "store1");
> >>
> >> Behind the scene, Streams will make sure stream1 / stream2
> >>
> >> transformations
> >>
> >> will always be grouped together as a single group of tasks, each
> >>
> >> of
> >>
> >> which
> >>
> >> will be executed by a single thread and hence there's no
> >>
> >> concurrency
> >>
> >> issues
> >>
> >> on accessing the store from different operators within the same
> >>
> >> task.
> >>
> >> I'm
> >>
> >> not sure how common this use case is, but I'd like to hear if
> >>
> >> you
> >>
> >> have
> >>
> >> any
> >>
> >> thoughts maintaining this since the current proposal seems
> >>
> >> exclude
> >>
> >> this
> >>
> >> possibility.
> >>
> >>
> >> Guozhang
> >>
> >>
> >> On Sun, Dec 9, 2018 at 4:18 PM Paul Whalen <pgwha...@gmail.com> <
> pgwha...@gmail.com>
> >>
> >> wrote:
> >>
> >> Here's KIP-401 for discussion, a minor Kafka Streams API change
> >>
> >> that
> >>
> >> I
> >>
> >> think could greatly increase the usability of the low-level
> >>
> >> processor
> >>
> >> API.
> >>
> >> I have some code written but will wait to see if there is buy
> >>
> >> in
> >>
> >> before
> >>
> >> going all out and creating a pull request. It seems like most
> >>
> >> of
> >>
> >> the
> >>
> >> work
> >>
> >> would be in updating documentation and tests.
> >>
> >>
> >>
> >>
> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=97553756
> >>
> >> Thanks!
> >> Paul
> >>
> >>
> >> --
> >> -- Guozhang
> >>
> >>
> >> --
> >> -- Guozhang
> >>
> >>
> >>
> >>
> >
>
>
