Hello Bill, Thanks for the updated KIP. I made a pass on the StreamJoined section. Just a quick question from user's perspective: if a user wants to provide a customized store-supplier, she is forced to also provide a name via the store-supplier. So there's no way to say "I want to provide my own store engine but let the library decide its name", is that right?
Guozhang On Tue, Sep 17, 2019 at 8:53 AM Bill Bejeck <bbej...@gmail.com> wrote: > Bumping this discussion as we need to re-vote before the KIP deadline. > > On Fri, Sep 13, 2019 at 10:29 AM Bill Bejeck <bbej...@gmail.com> wrote: > > > Hi All, > > > > While working on the implementation of KIP-479, some issues came to light > > that the KIP as written won't work. I have updated the KIP with a > solution > > I believe will solve the original problem as well as address the > > impediment to the initial approach. > > > > This update is a significant change, so please review the updated KIP > > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-479%3A+Add+StreamJoined+config+object+to+Join > and > > provide feedback. After we conclude the discussion, there will be a > > re-vote. > > > > Thanks! > > Bill > > > > On Wed, Jul 17, 2019 at 7:01 PM Guozhang Wang <wangg...@gmail.com> > wrote: > > > >> Hi Bill, thanks for your explanations. I'm on board with your decision > >> too. > >> > >> > >> Guozhang > >> > >> On Wed, Jul 17, 2019 at 10:20 AM Bill Bejeck <bbej...@gmail.com> wrote: > >> > >> > Thanks for the response, John. > >> > > >> > > If I can offer my thoughts, it seems better to just document on the > >> > > Stream join javadoc for the Materialized parameter that it will not > >> > > make the join result queriable. I'm not opposed to the queriable > flag > >> > > in general, but introducing it is a much larger consideration that > has > >> > > previously derailed this KIP discussion. In the interest of just > >> > > closing the gap and keeping the API change small, it seems better to > >> > > just go with documentation for now. > >> > > >> > I agree with your statement here. IMHO the most important goal of > this > >> KIP > >> > is to not breaking existing users and gain some consistency of the > API. > >> > > >> > I'll update the KIP accordingly. > >> > > >> > -Bill > >> > > >> > On Tue, Jul 16, 2019 at 11:55 AM John Roesler <j...@confluent.io> > >> wrote: > >> > > >> > > Hi Bill, > >> > > > >> > > Thanks for driving this KIP toward a conclusion. I'm on board with > >> > > your decision. > >> > > > >> > > You didn't mention whether you're still proposing to add the > >> > > "queriable" flag to the Materialized config object, or just document > >> > > that a Stream join is never queriable. Both options have come up > >> > > earlier in the discussion, so it would be good to pin this down. > >> > > > >> > > If I can offer my thoughts, it seems better to just document on the > >> > > Stream join javadoc for the Materialized parameter that it will not > >> > > make the join result queriable. I'm not opposed to the queriable > flag > >> > > in general, but introducing it is a much larger consideration that > has > >> > > previously derailed this KIP discussion. In the interest of just > >> > > closing the gap and keeping the API change small, it seems better to > >> > > just go with documentation for now. > >> > > > >> > > Thanks again, > >> > > -John > >> > > > >> > > On Thu, Jul 11, 2019 at 2:45 PM Bill Bejeck <bbej...@gmail.com> > >> wrote: > >> > > > > >> > > > Thanks all for the great discussion so far. > >> > > > > >> > > > Everyone has made excellent points, and I appreciate the detail > >> > everyone > >> > > > has put into their arguments. > >> > > > > >> > > > However, after carefully evaluating all the points made so far, > >> > creating > >> > > an > >> > > > overload with Materialized is still my #1 option. > >> > > > My reasoning for saying so is two-fold: > >> > > > > >> > > > 1. It's a small change, and IMHO since it's consistent with our > >> > > current > >> > > > API concerning state store usage, the cognitive load on users > >> will > >> > be > >> > > > minimal. > >> > > > 2. It achieves the most important goal of this KIP, namely to > >> close > >> > > the > >> > > > gap of naming state stores independently of the join operator > >> name. > >> > > > > >> > > > Additionally, I agree with the points made by Matthias earlier (I > >> > realize > >> > > > there is some overlap here). > >> > > > > >> > > > > - the main purpose of this KIP is to close the naming gap what > we > >> > > achieve > >> > > > > - we can allow people to use the new in-memory store > >> > > > > - we allow people to enable/disable caching > >> > > > > - we unify the API > >> > > > > - we decouple querying from naming > >> > > > > - it's a small API change > >> > > > > >> > > > Although it's not a perfect solution, IMHO the positives of using > >> > > > Materialize far outweigh the negatives, and from what we've > >> discussed > >> > so > >> > > > far, anything we implement seems to involve an additional change > >> down > >> > the > >> > > > road. > >> > > > > >> > > > If others are still strongly opposed to using Materialized, my > other > >> > > > preferences would be > >> > > > > >> > > > 1. Add a "withStoreName" to Joined. Although I agree with > >> Guozhang > >> > > that > >> > > > having a parameter that only applies to one use-case would be > >> > clumsy. > >> > > > 2. Add a String overload for naming the store, but this would > be > >> my > >> > > > least favorite option as IMHO this seems to be a step backward > >> from > >> > > why we > >> > > > introduced configuration objects in the first place. > >> > > > > >> > > > Thanks, > >> > > > Bill > >> > > > > >> > > > On Thu, Jun 27, 2019 at 4:45 PM Matthias J. Sax < > >> matth...@confluent.io > >> > > > >> > > > wrote: > >> > > > > >> > > > > Thanks for the KIP Bill! > >> > > > > > >> > > > > Great discussion to far. > >> > > > > > >> > > > > About John's idea about querying upstream stores and don't > >> > materialize > >> > > a > >> > > > > store: I agree with Bill that this seems to be an orthogonal > >> > question, > >> > > > > and it might be better to treat it as an independent > optimization > >> and > >> > > > > exclude from this KIP. > >> > > > > > >> > > > > > What should be the behavior if there is no store > >> > > > > > configured (e.g., if Materialized with only serdes) and > >> querying is > >> > > > > > enabled? > >> > > > > > >> > > > > IMHO, this could be an error case. If one wants to query a > store, > >> > they > >> > > > > need to provide a name -- if you don't know the name, how would > >> you > >> > > > > actually query the store (even if it would be possible to get > the > >> > name > >> > > > > from the `TopologyDescription`, it seems clumsy). > >> > > > > > >> > > > > If we don't want to throw an error, materializing seems to be > the > >> > right > >> > > > > option, to exclude "query optimization" from this KIP. I would > be > >> ok > >> > > > > with this option, even if it's clumsy to get the name from > >> > > > > `TopologyDescription`; hence, I would prefer to treat it as an > >> error. > >> > > > > > >> > > > > > To get back to the current behavior, users would have to > >> > > > > > add a "bytes store supplier" to the Materialized to indicate > >> that, > >> > > > > > yes, they really want a state store there. > >> > > > > > >> > > > > This sound like a quite subtle semantic difference on how to use > >> the > >> > > > > API. Might be hard to explain to users. I would prefer to not > >> > > introduce it. > >> > > > > > >> > > > > > >> > > > > > >> > > > > About Guozhang's points: > >> > > > > > >> > > > > 1a) That is actually a good point. However, I believe we cannot > >> get > >> > > > > around this issue easily, and it seems ok to me, to expose the > >> actual > >> > > > > store type we are using. (More thoughts later.) > >> > > > > > >> > > > > 1b) I don't see an issue with allowing users to query all > stores? > >> > What > >> > > > > is the rational behind it? What do we gain by not allowing it? > >> > > > > > >> > > > > 2) While I understand what you are saying, we also want/need to > >> have > >> > a > >> > > > > way in the PAPI to allow users adding "internal/private" > >> > non-queryable > >> > > > > stores to a topology. That's possible via > >> > > > > `Materialized#withQueryingDisabled()`. We could also update > >> > > > > `Topology#addStateStore(StoreBuilder, boolean isQueryable, > >> > String...)` > >> > > > > to address this. Again, I agree with Bill that the current API > is > >> > built > >> > > > > in a certain way, and if we want to change it, it should be a > >> > separate > >> > > > > KIP, as it seems to be an orthogonal concern. > >> > > > > > >> > > > > > Instead, we just restrict KIP-307 to NOT > >> > > > > > use the Joined.name for state store names and always use > >> internal > >> > > names > >> > > > > as > >> > > > > > well, which admittedly indeed leaves a hole of not being able > to > >> > > cover > >> > > > > all > >> > > > > > internal names here > >> > > > > > >> > > > > I think it's important to close this gap. Naming entities seems > >> to a > >> > > > > binary feature: if there is a gap, the feature is more or less > >> > useless, > >> > > > > rendering KIP-307 void. > >> > > > > > >> > > > > > >> > > > > > >> > > > > I like John's detailed list of required features and what > >> > > > > Materialized/WindowByteStoreSuppliers offers. My take is, that > >> adding > >> > > > > Materialized including the required run-time checks is the best > >> > option > >> > > > > we have, for the following reasons: > >> > > > > > >> > > > > - the main purpose of this KIP is to close the naming gap what > we > >> > > achieve > >> > > > > - we can allow people to use the new in-memory store > >> > > > > - we allow people to enable/disable caching > >> > > > > - we unify the API > >> > > > > - we decouple querying from naming > >> > > > > - it's a small API change > >> > > > > > >> > > > > Adding an overload and only passing in a name, would address the > >> main > >> > > > > purpose of the KIP. However, it falls short on all the other > >> > "goodies". > >> > > > > As you mentioned, passing in `Materialized` might not be perfect > >> and > >> > > > > maybe we need to deprecate is at some point; but this is also > true > >> > for > >> > > > > passing in just a name. > >> > > > > > >> > > > > I am also not convinced, that a `StreamJoinStore` would resolve > >> all > >> > the > >> > > > > issues. In the end, as long as we are using a `WindowedStore` > >> > > > > internally, we need to expose this "implemenation detail" to > >> users to > >> > > > > allow them to plug in a custom store. Adding `Materialized` seem > >> to > >> > be > >> > > > > the best short-term fix from my point of view. > >> > > > > > >> > > > > > >> > > > > -Matthias > >> > > > > > >> > > > > > >> > > > > On 6/27/19 9:56 AM, Guozhang Wang wrote: > >> > > > > > Hi John, > >> > > > > > > >> > > > > > I actually feels better about a new interface but I'm not sure > >> if > >> > we > >> > > > > would > >> > > > > > need the full configuration of store / log / cache, now or in > >> the > >> > > future > >> > > > > > ever for stream-stream join. > >> > > > > > > >> > > > > > Right now I feel that 1) we want to improve our implementation > >> of > >> > > > > > stream-stream join, and potentially also allow users to > >> customize > >> > > this > >> > > > > > implementation but with a more suitable interface than the > >> current > >> > > > > > WindowStore interface, how to do that is less clear and > >> > > execution-wise > >> > > > > it's > >> > > > > > (arguably..) not urgent; 2) we want to close the last gap > >> > > (Stream-stream > >> > > > > > join) of allowing users to specify all internal names to help > on > >> > > backward > >> > > > > > compatibility, which is urgent. > >> > > > > > > >> > > > > > Therefore if we want to unblock 2) from 1) in the near term, I > >> feel > >> > > > > > slightly inclined to just add overload functions that takes > in a > >> > > store > >> > > > > name > >> > > > > > for stream-stream joins only -- and admittedly, in the future > >> this > >> > > > > function > >> > > > > > maybe deprecated -- i.e. if we have to do something that we > "may > >> > > regret" > >> > > > > in > >> > > > > > the future, I'd like to pick the least intrusive option. > >> > > > > > > >> > > > > > About `Joined#withStoreName`: since the Joined class itself is > >> also > >> > > used > >> > > > > in > >> > > > > > other join types, I feel less comfortable to have a > >> > > > > `Joined#withStoreName` > >> > > > > > which is only going to be used by stream-stream join. Or > maybe I > >> > miss > >> > > > > > something here about the "latter" case that you are referring > >> to? > >> > > > > > > >> > > > > > > >> > > > > > > >> > > > > > Guozhang > >> > > > > > > >> > > > > > On Mon, Jun 24, 2019 at 12:16 PM John Roesler < > >> j...@confluent.io> > >> > > wrote: > >> > > > > > > >> > > > > >> Thanks Guozhang, > >> > > > > >> > >> > > > > >> Yep. Maybe we can consider just exactly what the join needs: > >> > > > > >> > >> > > > > >>> the WindowStore<Bytes, byte[]> itself is fine, if overly > >> broad, > >> > > > > >>> since the only two methods we need are `window.put(key, > value, > >> > > > > >>> context().timestamp())` and `WindowStoreIterator<V2> iter = > >> > > > > >>> window.fetch(key, timeFrom, timeTo)`. > >> > > > > >> > >> > > > > >> One "middle ground" would be to extract _this_ into a new > store > >> > > > > >> interface, which only supports these API calls, like > >> > > > > >> StreamJoinStore<K, V>. This would give us the latitude we > need > >> to > >> > > > > >> efficiently support the exact operation without concerning > >> > ourselves > >> > > > > >> with all the other things a WindowStore can do (which are > >> > > unreachable > >> > > > > >> for the join use case). It would also let us drop "store > >> > duplicates" > >> > > > > >> from the main WindowStore interface, since it only exists to > >> > support > >> > > > > >> the join use case. > >> > > > > >> > >> > > > > >> If we were to add a new StreamJoinStore interface, then it'd > be > >> > > > > >> straightforward how we could add also > >> > > > > >> `Materialized.as(StreamJoinBytesStoreSupplier)` and use > >> > > Materialized, > >> > > > > >> or alternatively add the ability to set the bytes store on > >> Joined. > >> > > > > >> > >> > > > > >> Personally, I'm kind of leaning toward the latter (and also > >> doing > >> > > > > >> `Joined#withStoreName`), since adding the new interface to > >> > > > > >> Materialized then also pollutes the interface for its > _actual_ > >> use > >> > > > > >> case of materializing a table view. Of course, to solve the > >> > > immediate > >> > > > > >> problem, all we need is the store name, but we might feel > >> better > >> > > about > >> > > > > >> adding the store name to Joined if we _also_ feel like in the > >> > > future, > >> > > > > >> we would add store/log/cache configuration to Joined as well. > >> > > > > >> > >> > > > > >> -John > >> > > > > >> > >> > > > > >> On Mon, Jun 24, 2019 at 12:56 PM Guozhang Wang < > >> > wangg...@gmail.com> > >> > > > > wrote: > >> > > > > >>> > >> > > > > >>> Hello John, > >> > > > > >>> > >> > > > > >>> My main concern is exactly the first point at the bottom of > >> your > >> > > > > analysis > >> > > > > >>> here: "* configure the bytes store". I'm not sure if using a > >> > window > >> > > > > bytes > >> > > > > >>> store would be ideal for stream-stream windowed join; e.g. > we > >> > could > >> > > > > >>> consider two dimensional list sorted by timestamps and then > by > >> > > keys to > >> > > > > do > >> > > > > >>> the join, whereas a windowed bytes store is basically sorted > >> by > >> > key > >> > > > > >> first, > >> > > > > >>> then by timestamp. If we expose the Materialized to let user > >> pass > >> > > in a > >> > > > > >>> windowed bytes store, then we would need to change that if > we > >> > want > >> > > to > >> > > > > >>> replace it with a different implementation interface. > >> > > > > >>> > >> > > > > >>> > >> > > > > >>> Guozhang > >> > > > > >>> > >> > > > > >>> On Mon, Jun 24, 2019 at 8:59 AM John Roesler < > >> j...@confluent.io> > >> > > > > wrote: > >> > > > > >>> > >> > > > > >>>> Hey Guozhang and Bill, > >> > > > > >>>> > >> > > > > >>>> For what it's worth, I agree with you both! > >> > > > > >>>> > >> > > > > >>>> I think it might help the discussion to look concretely at > >> what > >> > > > > >>>> Materialized does: > >> > > > > >>>> * set a WindowBytesStoreSupplier > >> > > > > >>>> * set a name > >> > > > > >>>> * set the key/value serdes > >> > > > > >>>> * disable/enable/configure change-logging > >> > > > > >>>> * disable/enable caching > >> > > > > >>>> * configure retention > >> > > > > >>>> > >> > > > > >>>> Further, looking into the WindowBytesStoreSupplier, the > >> > interface > >> > > lets > >> > > > > >> you: > >> > > > > >>>> * get the segment interval > >> > > > > >>>> * get the window size > >> > > > > >>>> * get whether "duplicates" are enabled > >> > > > > >>>> * get the retention period > >> > > > > >>>> * (obviously) get a WindowStore<Bytes, byte[]> > >> > > > > >>>> > >> > > > > >>>> We know that Materialized isn't exactly what we need for > >> stream > >> > > joins, > >> > > > > >>>> but we can see how close Materialized is to what we need. > If > >> it > >> > is > >> > > > > >>>> close, maybe we can use it and document the gaps, and if it > >> is > >> > not > >> > > > > >>>> close, then maybe we should just add what we need to > Joined. > >> > > > > >>>> Stream Join's requirements for its stores: > >> > > > > >>>> * a multimap store (i.e., it keeps duplicates) for storing > >> > general > >> > > > > >>>> (not windowed) keyed records associated with their > insertion > >> > > time, and > >> > > > > >>>> allows efficient time-bounded lookups and also efficient > >> purges > >> > > of old > >> > > > > >>>> data. > >> > > > > >>>> ** Note, a properly configured WindowBytesStoreSupplier > >> > satisfies > >> > > this > >> > > > > >>>> requirement, and the interface supports the queries we need > >> to > >> > > verify > >> > > > > >>>> the configuration at run-time > >> > > > > >>>> * set a name for the store > >> > > > > >>>> * do _not_ set the serdes (they are already set in Joined) > >> > > > > >>>> * logging could be configurable (set to enabled now) > >> > > > > >>>> * caching could be configurable (set to enabled now) > >> > > > > >>>> * do _not_ configure retention (determined by JoinWindows) > >> > > > > >>>> > >> > > > > >>>> So, out of six capabilities for Materialized, there are two > >> we > >> > > don't > >> > > > > >>>> want (serdes and retention). These would become run-time > >> checks > >> > > if we > >> > > > > >>>> use it. > >> > > > > >>>> > >> > > > > >>>> A third questionable capability is to provide a > >> > > > > >>>> WindowBytesStoreSupplier. Looking at whether the > >> > > > > >>>> WindowBytesStoreSupplier is the right interface for Stream > >> Join: > >> > > > > >>>> * configuring segment interval is fine > >> > > > > >>>> * should _not_ configure window size (it's determined by > >> > > JoinWindows) > >> > > > > >>>> * duplicates _must_ be enabled > >> > > > > >>>> * retention should be _at least_ windowSize + gracePeriod, > >> but > >> > > note > >> > > > > >>>> that (unlike for Table window stores) there is no utility > in > >> > > having a > >> > > > > >>>> longer retention time. > >> > > > > >>>> * the WindowStore<Bytes, byte[]> itself is fine, if overly > >> > broad, > >> > > > > >>>> since the only two methods we need are `window.put(key, > >> value, > >> > > > > >>>> context().timestamp())` and `WindowStoreIterator<V2> iter = > >> > > > > >>>> window.fetch(key, timeFrom, timeTo)`. > >> > > > > >>>> > >> > > > > >>>> Thus, flattening out the overlap for > WindowBytesStoreSupplier > >> > > onto the > >> > > > > >>>> overlap for Materialized, we have 9 capabilities total > (note > >> > > retention > >> > > > > >>>> is duplicated), we have 4 that we don't want: > >> > > > > >>>> * do _not_ set the serdes (they are already set in Joined) > >> > > > > >>>> * do _not_ configure retention (determined by JoinWindows) > >> > > > > >>>> * should _not_ configure window size (it's determined by > >> > > JoinWindows) > >> > > > > >>>> * duplicates _must_ be enabled > >> > > > > >>>> > >> > > > > >>>> These gaps would have to be covered with run-time checks if > >> we > >> > > re-use > >> > > > > >>>> Materialized and WindowStoreBytesStoreSupplier both. Maybe > >> this > >> > > sounds > >> > > > > >>>> bad, but consider the other side, that we get 5 new > >> capabilities > >> > > we > >> > > > > >>>> don't require, but are still pretty nice: > >> > > > > >>>> * configure the bytes store > >> > > > > >>>> * set a name for the store > >> > > > > >>>> * configure caching > >> > > > > >>>> * configure logging > >> > > > > >>>> * configure segment interval > >> > > > > >>>> > >> > > > > >>>> Not sure where this nets us out, but it's food for thought. > >> > > > > >>>> -John > >> > > > > >>>> > >> > > > > >>>> On Sun, Jun 23, 2019 at 7:52 PM Guozhang Wang < > >> > wangg...@gmail.com > >> > > > > >> > > > > >> wrote: > >> > > > > >>>>> > >> > > > > >>>>> Hi Bill, > >> > > > > >>>>> > >> > > > > >>>>> I think by giving a Materialized param into stream-stream > >> join, > >> > > it's > >> > > > > >> okay > >> > > > > >>>>> (though still ideal) to say "we still would not expose the > >> > store > >> > > for > >> > > > > >>>>> queries", but it would sound a bit awkward to say "we > would > >> > also > >> > > > > >> ignore > >> > > > > >>>>> whatever the passed in store supplier but just use our > >> default > >> > > ones" > >> > > > > >> -- > >> > > > > >>>>> again the concern is that, if in the future we'd want to > >> change > >> > > the > >> > > > > >>>> default > >> > > > > >>>>> implementation of join algorithm which no longer rely on a > >> > window > >> > > > > >> store > >> > > > > >>>>> with deduping enabled, then we need to change this API > >> again by > >> > > > > >> changing > >> > > > > >>>>> the store supplier type. > >> > > > > >>>>> > >> > > > > >>>>> If we do want to fill this hole for stream-stream join, I > >> feel > >> > > just > >> > > > > >>>> adding > >> > > > > >>>>> a String typed store-name would even be less > >> future-intrusive > >> > if > >> > > we > >> > > > > >>>> expect > >> > > > > >>>>> this parameter to be modified later. > >> > > > > >>>>> > >> > > > > >>>>> Does that makes sense? > >> > > > > >>>>> > >> > > > > >>>>> > >> > > > > >>>>> Guozhang > >> > > > > >>>>> > >> > > > > >>>>> On Sat, Jun 22, 2019 at 12:51 PM Bill Bejeck < > >> > bbej...@gmail.com> > >> > > > > >> wrote: > >> > > > > >>>>> > >> > > > > >>>>>> Thanks for the comments John and Guozhang, I'll address > >> each > >> > > one of > >> > > > > >>>> your > >> > > > > >>>>>> comments in turn. > >> > > > > >>>>>> > >> > > > > >>>>>> John, > >> > > > > >>>>>> > >> > > > > >>>>>>> I'm wondering about a missing quadrant from the truth > >> table > >> > > > > >> involving > >> > > > > >>>>>>> whether a Materialized is stored or not and querying is > >> > > > > >>>>>>> enabled/disabled... What should be the behavior if there > >> is > >> > no > >> > > > > >> store > >> > > > > >>>>>>> configured (e.g., if Materialized with only serdes) and > >> > > querying > >> > > > > >> is > >> > > > > >>>>>> enabled? > >> > > > > >>>>>> > >> > > > > >>>>>>> It seems we have two choices: > >> > > > > >>>>>>> 1. we can force creation of a state store in this case, > so > >> > the > >> > > > > >> store > >> > > > > >>>>>>> can be used to serve the queries > >> > > > > >>>>>>> 2. we can provide just a queriable view, basically > >> letting IQ > >> > > > > >> query > >> > > > > >>>>>>> into the "KTableValueGetter", which would transparently > >> > > > > >> construct the > >> > > > > >>>>>>> query response by applying the operator logic to the > >> upstream > >> > > > > >> state > >> > > > > >>>> if > >> > > > > >>>>>>> the operator state isn't already stored. > >> > > > > >>>>>> > >> > > > > >>>>>> > >> > > > > >>>>>> I agree with your assertion about a missing quadrant from > >> the > >> > > truth > >> > > > > >>>> table. > >> > > > > >>>>>> Additionally, I too like the concept of a queriable view. > >> > But I > >> > > > > >> think > >> > > > > >>>> that > >> > > > > >>>>>> goes a bit beyond the scope of this KIP and would like to > >> > pursue > >> > > > > >> that > >> > > > > >>>>>> feature as follow-on work. Also thinking about this KIP > >> some > >> > > > > >> more, I'm > >> > > > > >>>>>> thinking of the changes to Materialized might be a reach > as > >> > > well. > >> > > > > >>>>>> Separating the naming from a store and its queryable > state > >> > seems > >> > > > > >> like a > >> > > > > >>>>>> complex issue in and of itself and should be treated > >> > > accordingly. > >> > > > > >>>>>> > >> > > > > >>>>>> So here's what I'm thinking now. We add Materialzied to > >> Join, > >> > > but > >> > > > > >> for > >> > > > > >>>> now, > >> > > > > >>>>>> we internally disable querying. I know this breaks our > >> > current > >> > > > > >>>> semantic > >> > > > > >>>>>> approach, but I think it's crucial that we do two things > in > >> > this > >> > > > > >> KIP > >> > > > > >>>>>> > >> > > > > >>>>>> 1. Break the naming of the state stores from Joined to > >> > > > > >>>> Materialized, so > >> > > > > >>>>>> the naming of state stores follows our current pattern > >> and > >> > > > > >> enables > >> > > > > >>>>>> upgrades > >> > > > > >>>>>> from 2.3 to 2.4 > >> > > > > >>>>>> 2. Offer the ability to configure the state stores of > >> the > >> > > join, > >> > > > > >> even > >> > > > > >>>>>> providing a different implementation (i.e. in-memory) > if > >> > > > > >> desired. > >> > > > > >>>>>> > >> > > > > >>>>>> With that in mind I'm considering changing the KIP to > >> remove > >> > the > >> > > > > >>>> changes to > >> > > > > >>>>>> Materialized, and we document very clearly that by > >> providing a > >> > > > > >>>> Materialized > >> > > > > >>>>>> object with a name is only for naming the state store, > >> hence > >> > the > >> > > > > >>>> changelog > >> > > > > >>>>>> topics and any possible configurations of the store, but > >> this > >> > > store > >> > > > > >>>> *will > >> > > > > >>>>>> not be available for IQ.* > >> > > > > >>>>>> > >> > > > > >>>>>> WDYT? > >> > > > > >>>>>> > >> > > > > >>>>>> Guozhang, > >> > > > > >>>>>> > >> > > > > >>>>>>> 1. About not breaking compatibility of stream-stream > join > >> > > > > >>>> materialized > >> > > > > >>>>>>> stores: I think this is a valid issue to tackle, but > after > >> > > > > >> thinking > >> > > > > >>>> about > >> > > > > >>>>>>> it once more I'm not sure if exposing Materialized would > >> be a > >> > > > > >> good > >> > > > > >>>>>> solution > >> > > > > >>>>>>> here. My rationles: > >> > > > > >>>>>>> > >> > > > > >>>>>>> 1.a For stream-stream join, our current usage of > >> window-store > >> > > is > >> > > > > >> not > >> > > > > >>>>>> ideal, > >> > > > > >>>>>>> and we want to modify it in the near future to be more > >> > > > > >> efficient. Not > >> > > > > >>>>>>> allowing users to override such state store backend > gives > >> us > >> > > such > >> > > > > >>>> freedom > >> > > > > >>>>>>> (which was also considered in the original DSL design), > >> > whereas > >> > > > > >>>> getting a > >> > > > > >>>>>>> Materialized<WindowStore> basically kicks out that > freedom > >> > out > >> > > > > >> of the > >> > > > > >>>>>>> window. > >> > > > > >>>>>>> 1.b For strema-stream join, in our original design we > >> intend > >> > to > >> > > > > >>>> "never" > >> > > > > >>>>>>> want users to query the state, since it is just for > >> buffering > >> > > the > >> > > > > >>>>>> upcoming > >> > > > > >>>>>>> records from the stream. Now I know that some users may > >> > indeed > >> > > > > >> want > >> > > > > >>>> to > >> > > > > >>>>>>> query it from the debugging perspective, but still I > >> > concerned > >> > > > > >> about > >> > > > > >>>>>>> whether leveraging IQ for debugging purposes would be > the > >> > right > >> > > > > >>>> solution > >> > > > > >>>>>>> here. And adding Materialized object opens the door to > let > >> > > users > >> > > > > >>>> query > >> > > > > >>>>>>> about it (unless we did something intentionally to still > >> > > forbids > >> > > > > >> it), > >> > > > > >>>>>> which > >> > > > > >>>>>>> also restricts us in the future. > >> > > > > >>>>>>> > >> > > > > >>>>>>> 2. About the coupling between Materialized.name() and > >> > > queryable: > >> > > > > >>>> again I > >> > > > > >>>>>>> think this is a valid issue. But I'm not sure if the > >> current > >> > > > > >>>>>>> "withQuerryingDisabled / Enabled" at Materialized is the > >> best > >> > > > > >>>> approach. > >> > > > > >>>>>>> Here I think I agree with John, that generally speaking > >> it's > >> > > > > >> better > >> > > > > >>>> be a > >> > > > > >>>>>>> control function on the `KTable` itself, rather than on > >> > > > > >>>> `Materialized`, > >> > > > > >>>>>> so > >> > > > > >>>>>>> fixing it via adding functions through `Materialized` > >> seems > >> > > not a > >> > > > > >>>> natural > >> > > > > >>>>>> approach either. > >> > > > > >>>>>> > >> > > > > >>>>>> I understand your thoughts here, and up to a point, I > agree > >> > with > >> > > > > >> you. > >> > > > > >>>>>> But concerning not providing Materialized as it may > >> restrict > >> > us > >> > > in > >> > > > > >> the > >> > > > > >>>>>> future for delivering different implementations, I'm > >> wondering > >> > > if > >> > > > > >> we > >> > > > > >>>> are > >> > > > > >>>>>> doing some premature optimization here. > >> > > > > >>>>>> My rationale for saying so > >> > > > > >>>>>> > >> > > > > >>>>>> 1. I think the cost of not allowing the naming of > state > >> > > stores > >> > > > > >> for > >> > > > > >>>> joins > >> > > > > >>>>>> is too big of a gap to leave. IMHO for joins to > follow > >> > the > >> > > > > >> current > >> > > > > >>>>>> pattern of using Materialized for naming state stores > >> would > >> > > be > >> > > > > >> what > >> > > > > >>>> most > >> > > > > >>>>>> users would expect to use. As I said in my comments > >> > above, I > >> > > > > >> think > >> > > > > >>>> we > >> > > > > >>>>>> should *not include* the changes to Materialized and > >> > enforce > >> > > > > >> named > >> > > > > >>>>>> stores for joins as unavailable for IQ. > >> > > > > >>>>>> 2. We'll still have the join methods available > without a > >> > > > > >>>> Materialized > >> > > > > >>>>>> allowing us to do something different internally if a > >> > > > > >> Materialized > >> > > > > >>>> is > >> > > > > >>>>>> not > >> > > > > >>>>>> provided. > >> > > > > >>>>>> > >> > > > > >>>>>> > >> > > > > >>>>>>> Overall, I'm thinking maybe we should still use two > stones > >> > > rather > >> > > > > >>>> than > >> > > > > >>>>>> one > >> > > > > >>>>>>> to kill these two birds, and probably for this KIP we > just > >> > > focus > >> > > > > >> on > >> > > > > >>>> 1) > >> > > > > >>>>>>> above. And for that I'd like to not expose the > >> Materialized > >> > > > > >> either > >> > > > > >>>> for > >> > > > > >>>>>>> rationales that I've listed above. Instead, we just > >> restrict > >> > > > > >> KIP-307 > >> > > > > >>>> to > >> > > > > >>>>>> NOT > >> > > > > >>>>>>> use the Joined.name for state store names and always use > >> > > internal > >> > > > > >>>> names > >> > > > > >>>>>> as > >> > > > > >>>>>>> well, which admittedly indeed leaves a hole of not being > >> able > >> > > to > >> > > > > >>>> cover > >> > > > > >>>>>> all > >> > > > > >>>>>>> internal names here, but now I feel this `hole` may > >> better be > >> > > > > >> filled > >> > > > > >>>> by, > >> > > > > >>>>>>> e.g. not creating changelog topics but just use the > >> upstream > >> > to > >> > > > > >>>>>>> re-bootstrap the materialized store, more concretely: > when > >> > > > > >>>> materializing > >> > > > > >>>>>>> the store, try to piggy-back the changelog topic on an > >> > existing > >> > > > > >>>> topic, > >> > > > > >>>>>> e.g. > >> > > > > >>>>>>> a) if the stream is coming directly from some source > topic > >> > > > > >> (including > >> > > > > >>>>>>> repartition topic), make that as changelog topic and if > >> it is > >> > > > > >>>> repartition > >> > > > > >>>>>>> topic change the retention / data purging policy > >> necessarily > >> > as > >> > > > > >>>> well; b) > >> > > > > >>>>>> if > >> > > > > >>>>>>> the stream is coming from some stateless operators, > >> delegate > >> > > that > >> > > > > >>>>>> stateless > >> > > > > >>>>>>> operator to the parent stream similar as a); if the > >> stream is > >> > > > > >> coming > >> > > > > >>>> from > >> > > > > >>>>>> a > >> > > > > >>>>>>> stream-stream join which is the only stateful operator > >> that > >> > can > >> > > > > >>>> result in > >> > > > > >>>>>> a > >> > > > > >>>>>>> stream, consider merging the join into multi-way joins > >> (yes, > >> > > > > >> this is > >> > > > > >>>> a > >> > > > > >>>>>> very > >> > > > > >>>>>>> hand-wavy thought, but the point here is that we do not > >> try > >> > to > >> > > > > >>>> tackle it > >> > > > > >>>>>>> now but leave it for a better solution :). > >> > > > > >>>>>> > >> > > > > >>>>>> I really like this idea! I agree with you in that this > >> > approach > >> > > > > >> to too > >> > > > > >>>>>> much for adding in this KIP, but we could pick it up > later > >> and > >> > > > > >>>> leverage the > >> > > > > >>>>>> Optimization framework to accomplish this re-use. > >> > > > > >>>>>> Again, while I agree we should break the naming of join > >> state > >> > > > > >> stores > >> > > > > >>>> from > >> > > > > >>>>>> KIP-307, IMHO it's something we should fix now as it will > >> be > >> > the > >> > > > > >> last > >> > > > > >>>> piece > >> > > > > >>>>>> we can provide to give users the ability to completely > make > >> > > their > >> > > > > >>>>>> topologies "upgrade proof" when adding additional > >> operations. > >> > > > > >>>>>> > >> > > > > >>>>>> Thanks again to both of you for comments and I look > >> forward to > >> > > > > >> hearing > >> > > > > >>>> back > >> > > > > >>>>>> from you. > >> > > > > >>>>>> > >> > > > > >>>>>> Regards, > >> > > > > >>>>>> Bill > >> > > > > >>>>>> > >> > > > > >>>>>> > >> > > > > >>>>>> > >> > > > > >>>>>> > >> > > > > >>>>>> > >> > > > > >>>>>> > >> > > > > >>>>>> > >> > > > > >>>>>> On Thu, Jun 20, 2019 at 2:33 PM Guozhang Wang < > >> > > wangg...@gmail.com> > >> > > > > >>>> wrote: > >> > > > > >>>>>> > >> > > > > >>>>>>> Hello Bill, > >> > > > > >>>>>>> > >> > > > > >>>>>>> Thanks for the KIP. Glad to see that we can likely > >> shooting > >> > two > >> > > > > >> birds > >> > > > > >>>>>> with > >> > > > > >>>>>>> one stone. I have some concerns though about those "two > >> > birds" > >> > > > > >>>>>> themselves: > >> > > > > >>>>>>> > >> > > > > >>>>>>> 1. About not breaking compatibility of stream-stream > join > >> > > > > >>>> materialized > >> > > > > >>>>>>> stores: I think this is a valid issue to tackle, but > after > >> > > > > >> thinking > >> > > > > >>>> about > >> > > > > >>>>>>> it once more I'm not sure if exposing Materialized would > >> be a > >> > > > > >> good > >> > > > > >>>>>> solution > >> > > > > >>>>>>> here. My rationles: > >> > > > > >>>>>>> > >> > > > > >>>>>>> 1.a For stream-stream join, our current usage of > >> window-store > >> > > is > >> > > > > >> not > >> > > > > >>>>>> ideal, > >> > > > > >>>>>>> and we want to modify it in the near future to be more > >> > > > > >> efficient. Not > >> > > > > >>>>>>> allowing users to override such state store backend > gives > >> us > >> > > such > >> > > > > >>>> freedom > >> > > > > >>>>>>> (which was also considered in the original DSL design), > >> > whereas > >> > > > > >>>> getting a > >> > > > > >>>>>>> Materialized<WindowStore> basically kicks out that > freedom > >> > out > >> > > > > >> of the > >> > > > > >>>>>>> window. > >> > > > > >>>>>>> 1.b For strema-stream join, in our original design we > >> intend > >> > to > >> > > > > >>>> "never" > >> > > > > >>>>>>> want users to query the state, since it is just for > >> buffering > >> > > the > >> > > > > >>>>>> upcoming > >> > > > > >>>>>>> records from the stream. Now I know that some users may > >> > indeed > >> > > > > >> want > >> > > > > >>>> to > >> > > > > >>>>>>> query it from the debugging perspective, but still I > >> > concerned > >> > > > > >> about > >> > > > > >>>>>>> whether leveraging IQ for debugging purposes would be > the > >> > right > >> > > > > >>>> solution > >> > > > > >>>>>>> here. And adding Materialized object opens the door to > let > >> > > users > >> > > > > >>>> query > >> > > > > >>>>>>> about it (unless we did something intentionally to still > >> > > forbids > >> > > > > >> it), > >> > > > > >>>>>> which > >> > > > > >>>>>>> also restricts us in the future. > >> > > > > >>>>>>> > >> > > > > >>>>>>> 2. About the coupling between Materialized.name() and > >> > > queryable: > >> > > > > >>>> again I > >> > > > > >>>>>>> think this is a valid issue. But I'm not sure if the > >> current > >> > > > > >>>>>>> "withQuerryingDisabled / Enabled" at Materialized is the > >> best > >> > > > > >>>> approach. > >> > > > > >>>>>>> Here I think I agree with John, that generally speaking > >> it's > >> > > > > >> better > >> > > > > >>>> be a > >> > > > > >>>>>>> control function on the `KTable` itself, rather than on > >> > > > > >>>> `Materialized`, > >> > > > > >>>>>> so > >> > > > > >>>>>>> fixing it via adding functions through `Materialized` > >> seems > >> > > not a > >> > > > > >>>> natural > >> > > > > >>>>>>> approach either. > >> > > > > >>>>>>> > >> > > > > >>>>>>> > >> > > > > >>>>>>> Overall, I'm thinking maybe we should still use two > stones > >> > > rather > >> > > > > >>>> than > >> > > > > >>>>>> one > >> > > > > >>>>>>> to kill these two birds, and probably for this KIP we > just > >> > > focus > >> > > > > >> on > >> > > > > >>>> 1) > >> > > > > >>>>>>> above. And for that I'd like to not expose the > >> Materialized > >> > > > > >> either > >> > > > > >>>> for > >> > > > > >>>>>>> rationales that I've listed above. Instead, we just > >> restrict > >> > > > > >> KIP-307 > >> > > > > >>>> to > >> > > > > >>>>>> NOT > >> > > > > >>>>>>> use the Joined.name for state store names and always use > >> > > internal > >> > > > > >>>> names > >> > > > > >>>>>> as > >> > > > > >>>>>>> well, which admittedly indeed leaves a hole of not being > >> able > >> > > to > >> > > > > >>>> cover > >> > > > > >>>>>> all > >> > > > > >>>>>>> internal names here, but now I feel this `hole` may > >> better be > >> > > > > >> filled > >> > > > > >>>> by, > >> > > > > >>>>>>> e.g. not creating changelog topics but just use the > >> upstream > >> > to > >> > > > > >>>>>>> re-bootstrap the materialized store, more concretely: > when > >> > > > > >>>> materializing > >> > > > > >>>>>>> the store, try to piggy-back the changelog topic on an > >> > existing > >> > > > > >>>> topic, > >> > > > > >>>>>> e.g. > >> > > > > >>>>>>> a) if the stream is coming directly from some source > topic > >> > > > > >> (including > >> > > > > >>>>>>> repartition topic), make that as changelog topic and if > >> it is > >> > > > > >>>> repartition > >> > > > > >>>>>>> topic change the retention / data purging policy > >> necessarily > >> > as > >> > > > > >>>> well; b) > >> > > > > >>>>>> if > >> > > > > >>>>>>> the stream is coming from some stateless operators, > >> delegate > >> > > that > >> > > > > >>>>>> stateless > >> > > > > >>>>>>> operator to the parent stream similar as a); if the > >> stream is > >> > > > > >> coming > >> > > > > >>>>>> from a > >> > > > > >>>>>>> stream-stream join which is the only stateful operator > >> that > >> > can > >> > > > > >>>> result > >> > > > > >>>>>> in a > >> > > > > >>>>>>> stream, consider merging the join into multi-way joins > >> (yes, > >> > > > > >> this is > >> > > > > >>>> a > >> > > > > >>>>>> very > >> > > > > >>>>>>> hand-wavy thought, but the point here is that we do not > >> try > >> > to > >> > > > > >>>> tackle it > >> > > > > >>>>>>> now but leave it for a better solution :). > >> > > > > >>>>>>> > >> > > > > >>>>>>> > >> > > > > >>>>>>> Guozhang > >> > > > > >>>>>>> > >> > > > > >>>>>>> > >> > > > > >>>>>>> > >> > > > > >>>>>>> On Wed, Jun 19, 2019 at 11:41 AM John Roesler < > >> > > j...@confluent.io > >> > > > > >>> > >> > > > > >>>> wrote: > >> > > > > >>>>>>> > >> > > > > >>>>>>>> Hi Bill, > >> > > > > >>>>>>>> > >> > > > > >>>>>>>> Thanks for the KIP! Awesome job catching this > unexpected > >> > > > > >>>> consequence > >> > > > > >>>>>>>> of the prior KIPs before it was released. > >> > > > > >>>>>>>> > >> > > > > >>>>>>>> The proposal looks good to me. On top of just fixing > the > >> > > > > >> problem, > >> > > > > >>>> it > >> > > > > >>>>>>>> seems to address two other pain points: > >> > > > > >>>>>>>> * that naming a state store automatically causes it to > >> > become > >> > > > > >>>>>> queriable. > >> > > > > >>>>>>>> * that there's currently no way to configure the bytes > >> store > >> > > > > >> for > >> > > > > >>>> join > >> > > > > >>>>>>>> windows. > >> > > > > >>>>>>>> > >> > > > > >>>>>>>> It's awesome that we can fix this issue and two others > >> with > >> > > one > >> > > > > >>>>>> feature. > >> > > > > >>>>>>>> > >> > > > > >>>>>>>> I'm wondering about a missing quadrant from the truth > >> table > >> > > > > >>>> involving > >> > > > > >>>>>>>> whether a Materialized is stored or not and querying is > >> > > > > >>>>>>>> enabled/disabled... What should be the behavior if > there > >> is > >> > no > >> > > > > >>>> store > >> > > > > >>>>>>>> configured (e.g., if Materialized with only serdes) and > >> > > > > >> querying is > >> > > > > >>>>>>>> enabled? > >> > > > > >>>>>>>> > >> > > > > >>>>>>>> It seems we have two choices: > >> > > > > >>>>>>>> 1. we can force creation of a state store in this case, > >> so > >> > the > >> > > > > >>>> store > >> > > > > >>>>>>>> can be used to serve the queries > >> > > > > >>>>>>>> 2. we can provide just a queriable view, basically > >> letting > >> > IQ > >> > > > > >> query > >> > > > > >>>>>>>> into the "KTableValueGetter", which would transparently > >> > > > > >> construct > >> > > > > >>>> the > >> > > > > >>>>>>>> query response by applying the operator logic to the > >> > upstream > >> > > > > >>>> state if > >> > > > > >>>>>>>> the operator state isn't already stored. > >> > > > > >>>>>>>> > >> > > > > >>>>>>>> Offhand, it seems like the second is actually a pretty > >> > awesome > >> > > > > >>>>>>>> capability. But it might have an awkward interaction > with > >> > the > >> > > > > >>>> current > >> > > > > >>>>>>>> semantics. Presently, if I provide a > >> Materialized.withName, > >> > it > >> > > > > >>>> implies > >> > > > > >>>>>>>> that querying should be enabled AND that the view > should > >> > > > > >> actually > >> > > > > >>>> be > >> > > > > >>>>>>>> stored in a state store. Under option 2 above, this > >> behavior > >> > > > > >> would > >> > > > > >>>>>>>> change to NOT provision a state store and instead just > >> > consult > >> > > > > >> the > >> > > > > >>>>>>>> ValueGetter. To get back to the current behavior, users > >> > would > >> > > > > >> have > >> > > > > >>>> to > >> > > > > >>>>>>>> add a "bytes store supplier" to the Materialized to > >> indicate > >> > > > > >> that, > >> > > > > >>>>>>>> yes, they really want a state store there. > >> > > > > >>>>>>>> > >> > > > > >>>>>>>> Behavior changes are always kind of scary, but I think > in > >> > this > >> > > > > >>>> case, > >> > > > > >>>>>>>> it might actually be preferable. In the event where > only > >> the > >> > > > > >> name > >> > > > > >>>> is > >> > > > > >>>>>>>> provided, it means that people just wanted to make the > >> > > > > >> operation > >> > > > > >>>>>>>> result queriable. If we automatically convert this to a > >> > > > > >> non-stored > >> > > > > >>>>>>>> view, then simply upgrading results in the same > >> observable > >> > > > > >> behavior > >> > > > > >>>>>>>> and semantics, but a linear reduction in local storage > >> > > > > >> requirements > >> > > > > >>>>>>>> and disk i/o, as well as a corresponding linear > >> reduction in > >> > > > > >> memory > >> > > > > >>>>>>>> usage both on and off heap. > >> > > > > >>>>>>>> > >> > > > > >>>>>>>> What do you think? > >> > > > > >>>>>>>> -John > >> > > > > >>>>>>>> > >> > > > > >>>>>>>> On Tue, Jun 18, 2019 at 9:21 PM Bill Bejeck < > >> > > bbej...@gmail.com > >> > > > > >>> > >> > > > > >>>> wrote: > >> > > > > >>>>>>>>> > >> > > > > >>>>>>>>> All, > >> > > > > >>>>>>>>> > >> > > > > >>>>>>>>> I'd like to start a discussion for adding a > Materialized > >> > > > > >>>>>> configuration > >> > > > > >>>>>>>>> object to KStream.join for naming state stores > involved > >> in > >> > > > > >> joins. > >> > > > > >>>>>>>>> > >> > > > > >>>>>>>>> > >> > > > > >>>>>>>> > >> > > > > >>>>>>> > >> > > > > >>>>>> > >> > > > > >>>> > >> > > > > >> > >> > > > > > >> > > > >> > > >> > https://cwiki.apache.org/confluence/display/KAFKA/KIP-479%3A+Add+Materialized+to+Join > >> > > > > >>>>>>>>> > >> > > > > >>>>>>>>> Your comments and suggestions are welcome. > >> > > > > >>>>>>>>> > >> > > > > >>>>>>>>> Thanks, > >> > > > > >>>>>>>>> Bill > >> > > > > >>>>>>>> > >> > > > > >>>>>>> > >> > > > > >>>>>>> > >> > > > > >>>>>>> -- > >> > > > > >>>>>>> -- Guozhang > >> > > > > >>>>>>> > >> > > > > >>>>>> > >> > > > > >>>>> > >> > > > > >>>>> > >> > > > > >>>>> -- > >> > > > > >>>>> -- Guozhang > >> > > > > >>>> > >> > > > > >>> > >> > > > > >>> > >> > > > > >>> -- > >> > > > > >>> -- Guozhang > >> > > > > >> > >> > > > > > > >> > > > > > > >> > > > > > >> > > > > > >> > > > >> > > >> > >> > >> -- > >> -- Guozhang > >> > > > -- -- Guozhang
