Hi John and Guozhang Ah yes, I lost that in the mix! Thanks for the convergent solutions - I do think that the attachment that John included makes for a better design. It should also help with overall performance as very high-cardinality foreign keyed data (say millions of events with the same entity) will be able to leverage the multiple nodes for join functionality instead of having it all performed in one node. There is still a bottleneck in the right table having to propagate all those events, but with slimmer structures, less IO and no need to perform the join I think the throughput will be much higher in those scenarios.
Okay, I am convinced. I will update the KIP accordingly to a Gliffy version of John's diagram and ensure that the example flow matches correctly. Then I can go back to working on the PR to match the diagram. Thanks both of you for all the help - very much appreciated. Adam On Mon, Dec 17, 2018 at 6:39 PM Guozhang Wang <wangg...@gmail.com> wrote: > Hi John, > > Just made a pass on your diagram (nice hand-drawing btw!), and obviously we > are thinking about the same thing :) A neat difference that I like, is that > in the pre-join repartition topic we can still send message in the format > of `K=k, V=(i=2)` while using "i" as the partition key in StreamsPartition, > this way we do not need to even augment the key for the repartition topic, > but just do a projection on the foreign key part but trim all other fields: > as long as we still materialize the store as `A-2` co-located with the > right KTable, that is fine. > > As I mentioned in my previous email, I also think this has a few advantages > on saving over-the-wire bytes as well as disk bytes. > > Guozhang > > > On Mon, Dec 17, 2018 at 3:17 PM John Roesler <j...@confluent.io> wrote: > > > Hi Guozhang, > > > > Thanks for taking a look! I think Adam's already addressed your questions > > as well as I could have. > > > > Hi Adam, > > > > Thanks for updating the KIP. It looks great, especially how all the > > need-to-know information is right at the top, followed by the details. > > > > Also, thanks for that high-level diagram. Actually, now that I'm looking > > at it, I think part of my proposal got lost in translation, although I do > > think that what you have there is also correct. > > > > I sketched up a crude diagram based on yours and attached it to the KIP > > (I'm not sure if attached or inline images work on the mailing list): > > > https://cwiki.apache.org/confluence/download/attachments/74684836/suggestion.png > > . It's also attached to this email for convenience. > > > > Hopefully, you can see how it's intended to line up, and which parts are > > modified. > > At a high level, instead of performing the join on the right-hand side, > > we're essentially just registering interest, like "LHS key A wishes to > > receive updates for RHS key 2". Then, when there is a new "interest" or > any > > updates to the RHS records, it "broadcasts" its state back to the LHS > > records who are interested in it. > > > > Thus, instead of sending the LHS values to the RHS joiner workers and > then > > sending the join results back to the LHS worke be co-partitioned and > > validated, we instead only send the LHS *keys* to the RHS workers and > then > > only the RHS k/v back to be joined by the LHS worker. > > > > I've been considering both your diagram and mine, and I *think* what I'm > > proposing has a few advantages. > > > > Here are some points of interest as you look at the diagram: > > * When we extract the foreign key and send it to the Pre-Join Repartition > > Topic, we can send only the FK/PK pair. There's no need to worry about > > custom partitioner logic, since we can just use the foreign key plainly > as > > the repartition record key. Also, we save on transmitting the LHS value, > > since we only send its key in this step. > > * We also only need to store the RHSKey:LHSKey mapping in the > > MaterializedSubscriptionStore, saving on disk. We can use the same rocks > > key format you proposed and the same algorithm involving range scans when > > the RHS records get updated. > > * Instead of joining on the right side, all we do is compose a > > re-repartition record so we can broadcast the RHS k/v pair back to the > > original LHS partition. (this is what the "rekey" node is doing) > > * Then, there is a special kind of Joiner that's co-resident in the same > > StreamTask as the LHS table, subscribed to the Post-Join Repartition > Topic. > > ** This Joiner is *not* triggered directly by any changes in the LHS > > KTable. Instead, LHS events indirectly trigger the join via the whole > > lifecycle. > > ** For each event arriving from the Post-Join Repartition Topic, the > > Joiner looks up the corresponding record in the LHS KTable. It validates > > the FK as you noted, discarding any inconsistent events. Otherwise, it > > unpacks the RHS K/V pair and invokes the ValueJoiner to obtain the join > > result > > ** Note that the Joiner itself is stateless, so materializing the join > > result is optional, just as with the 1:1 joins. > > > > So in summary: > > * instead of transmitting the LHS keys and values to the right and the > > JoinResult back to the left, we only transmit the LHS keys to the right > and > > the RHS values to the left. Assuming the average RHS value is on smaller > > than or equal to the average join result size, it's a clear win on broker > > traffic. I think this is actually a reasonable assumption, which we can > > discuss more if you're suspicious. > > * we only need one copy of the data (the left and right tables need to be > > materialized) and one extra copy of the PK:FK pairs in the Materialized > > Subscription Store. Materializing the join result is optional, just as > with > > the existing 1:1 joins. > > * we still need the fancy range-scan algorithm on the right to locate all > > interested LHS keys when a RHS value is updated, but we don't need a > custom > > partitioner for either repartition topic (this is of course a > modification > > we could make to your version as well) > > > > How does this sound to you? (And did I miss anything?) > > -John > > > > On Mon, Dec 17, 2018 at 9:00 AM Adam Bellemare <adam.bellem...@gmail.com > > > > wrote: > > > >> Hi John & Guozhang > >> > >> @John & @Guozhang Wang <wangg...@gmail.com> - I have cleaned up the > KIP, > >> pruned much of what I wrote and put a simplified diagram near the top to > >> illustrate the workflow. I encapsulated Jan's content at the bottom of > the > >> document. I believe it is simpler to read by far now. > >> > >> @Guozhang Wang <wangg...@gmail.com>: > >> > #1: rekey left table > >> > -> source from the left upstream, send to rekey-processor to > generate > >> combined key, and then sink to copartition topic. > >> Correct. > >> > >> > #2: first-join with right table > >> > -> source from the right table upstream, materialize the right > table. > >> > -> source from the co-partition topic, materialize the rekeyed left > >> table, join with the right table, rekey back, and then sink to the > >> rekeyed-back topic. > >> Almost - I cleared up the KIP. We do not rekey back yet, as I need the > >> Foreign-Key value generated in #1 above to compare in the resolution > >> stage. > >> > >> > #3: second join > >> > -> source from the rekeyed-back topic, materialize the rekeyed back > >> table. > >> > -> source from the left upstream, materialize the left table, join > >> with > >> the rekeyed back table. > >> Almost - As each event comes in, we just run it through a stateful > >> processor that checks the original ("This") KTable for the key. The > value > >> payload then has the foreignKeyExtractor applied again as in Part #1 > >> above, > >> and gets the current foreign key. Then we compare it to the joined event > >> that we are currently resolving. If they have the same foreign-key, > >> propagate the result out. If they don't, throw the event away. > >> > >> The end result is that we do need to materialize 2 additional tables > >> (left/this-combinedkey table, and the final Joined table) as I've > >> illustrated in the updated KIP. I hope the diagram clears it up a lot > >> better. Please let me know. > >> > >> Thanks again > >> Adam > >> > >> > >> > >> > >> On Sun, Dec 16, 2018 at 3:18 PM Guozhang Wang <wangg...@gmail.com> > wrote: > >> > >> > John, > >> > > >> > Thanks a lot for the suggestions on refactoring the wiki, I agree with > >> you > >> > that we should consider the KIP proposal to be easily understood by > >> anyone > >> > in the future to read, and hence should provide a good summary on the > >> > user-facing interfaces, as well as rejected alternatives to represent > >> > briefly "how we came a long way to this conclusion, and what we have > >> > argued, disagreed, and agreed about, etc" so that readers do not need > to > >> > dig into the DISCUSS thread to get all the details. We can, of course, > >> keep > >> > the implementation details like "workflows" on the wiki page as a > >> addendum > >> > section since it also has correlations. > >> > > >> > Regarding your proposal on comment 6): that's a very interesting idea! > >> Just > >> > to clarify that I understands it fully correctly: the proposal's > >> resulted > >> > topology is still the same as the current proposal, where we will > have 3 > >> > sub-topologies for this operator: > >> > > >> > #1: rekey left table > >> > -> source from the left upstream, send to rekey-processor to > generate > >> > combined key, and then sink to copartition topic. > >> > > >> > #2: first-join with right table > >> > -> source from the right table upstream, materialize the right > table. > >> > -> source from the co-partition topic, materialize the rekeyed left > >> > table, join with the right table, rekey back, and then sink to the > >> > rekeyed-back topic. > >> > > >> > #3: second join > >> > -> source from the rekeyed-back topic, materialize the rekeyed back > >> > table. > >> > -> source from the left upstream, materialize the left table, join > >> with > >> > the rekeyed back table. > >> > > >> > Sub-topology #1 and #3 may be merged to a single sub-topology since > >> both of > >> > them read from the left table source stream. In this workflow, we need > >> to > >> > materialize 4 tables (left table in #3, right table in #2, rekeyed > left > >> > table in #2, rekeyed-back table in #3), and 2 repartition topics > >> > (copartition topic, rekeyed-back topic). > >> > > >> > Compared with Adam's current proposal in the workflow overview, it has > >> the > >> > same num.materialize tables (left table, rekeyed left table, right > >> table, > >> > out-of-ordering resolver table), and same num.internal topics (two). > The > >> > advantage is that on the copartition topic, we can save bandwidth by > not > >> > sending value, and in #2 the rekeyed left table is smaller since we do > >> not > >> > have any values to materialize. Is that right? > >> > > >> > > >> > Guozhang > >> > > >> > > >> > > >> > On Wed, Dec 12, 2018 at 1:22 PM John Roesler <j...@confluent.io> > wrote: > >> > > >> > > Hi Adam, > >> > > > >> > > Given that the committers are all pretty busy right now, I think > that > >> it > >> > > would help if you were to refactor the KIP a little to reduce the > >> > workload > >> > > for reviewers. > >> > > > >> > > I'd recommend the following changes: > >> > > * relocate all internal details to a section at the end called > >> something > >> > > like "Implementation Notes" or something like that. > >> > > * rewrite the rest of the KIP to be a succinct as possible and > mention > >> > only > >> > > publicly-facing API changes. > >> > > ** for example, the interface that you've already listed there, as > >> well > >> > as > >> > > a textual description of the guarantees we'll be providing (join > >> result > >> > is > >> > > copartitioned with the LHS, and the join result is guaranteed > correct) > >> > > > >> > > A good target would be that the whole main body of the KIP, > including > >> > > Status, Motivation, Proposal, Justification, and Rejected > Alternatives > >> > all > >> > > fit "above the fold" (i.e., all fit on the screen at a comfortable > >> zoom > >> > > level). > >> > > I think the only real Rejected Alternative that bears mention at > this > >> > point > >> > > is KScatteredTable, which you could just include the executive > >> summary on > >> > > (no implementation details), and link to extra details in the > >> > > Implementation Notes section. > >> > > > >> > > Taking a look at the wiki page, ~90% of the text there is internal > >> > detail, > >> > > which is useful for the dubious, but doesn't need to be ratified in > a > >> > vote > >> > > (and would be subject to change without notice in the future > anyway). > >> > > There's also a lot of conflicting discussion, as you've very > >> respectfully > >> > > tried to preserve the original proposal from Jan while adding your > >> own. > >> > > Isolating all this information in a dedicated section at the bottom > >> frees > >> > > the voters up to focus on the public API part of the proposal, which > >> is > >> > > really all they need to consider. > >> > > > >> > > Plus, it'll be clear to future readers which parts of the document > are > >> > > enduring, and which parts are a snapshot of our implementation > >> thinking > >> > at > >> > > the time. > >> > > > >> > > I'm suggesting this because I suspect that the others haven't made > >> time > >> > to > >> > > review it partly because it seems daunting. If it seems like it > would > >> be > >> > a > >> > > huge time investment to review, people will just keep putting it > off. > >> But > >> > > if the KIP is a single page, then they'll be more inclined to give > it > >> a > >> > > read. > >> > > > >> > > Honestly, I don't think the KIP itself is that controversial (apart > >> from > >> > > the scattered table thing (sorry, Jan) ). Most of the discussion has > >> been > >> > > around the implementation, which we can continue more effectively in > >> a PR > >> > > once the KIP has passed. > >> > > > >> > > How does that sound? > >> > > -John > >> > > > >> > > On Mon, Dec 10, 2018 at 3:54 PM Adam Bellemare < > >> adam.bellem...@gmail.com > >> > > > >> > > wrote: > >> > > > >> > > > 1) I believe that the resolution mechanism John has proposed is > >> > > sufficient > >> > > > - it is clean and easy and doesn't require additional RocksDB > >> stores, > >> > > which > >> > > > reduces the footprint greatly. I don't think we need to resolve > >> based > >> > on > >> > > > timestamp or offset anymore, but if we decide to do to that would > be > >> > > within > >> > > > the bounds of the existing API. > >> > > > > >> > > > 2) Is the current API sufficient, or does it need to be altered to > >> go > >> > > back > >> > > > to vote? > >> > > > > >> > > > 3) KScatteredTable implementation can always be added in a future > >> > > revision. > >> > > > This API does not rule it out. This implementation of this > function > >> > would > >> > > > simply be replaced with `KScatteredTable.resolve()` while still > >> > > maintaining > >> > > > the existing API, thereby giving both features as Jan outlined > >> earlier. > >> > > > Would this work? > >> > > > > >> > > > > >> > > > Thanks Guozhang, John and Jan > >> > > > > >> > > > > >> > > > > >> > > > > >> > > > On Mon, Dec 10, 2018 at 10:39 AM John Roesler <j...@confluent.io> > >> > wrote: > >> > > > > >> > > > > Hi, all, > >> > > > > > >> > > > > >> In fact, we > >> > > > > >> can just keep a single final-result store with timestamps and > >> > reject > >> > > > > values > >> > > > > >> that have a smaller timestamp, is that right? > >> > > > > > >> > > > > > Which is the correct output should at least be decided on the > >> > offset > >> > > of > >> > > > > > the original message. > >> > > > > > >> > > > > Thanks for this point, Jan. > >> > > > > > >> > > > > KIP-258 is merely to allow embedding the record timestamp in > the > >> k/v > >> > > > > store, > >> > > > > as well as providing a storage-format upgrade path. > >> > > > > > >> > > > > I might have missed it, but I think we have yet to discuss > whether > >> > it's > >> > > > > safe > >> > > > > or desirable just to swap topic-ordering our for > >> timestamp-ordering. > >> > > This > >> > > > > is > >> > > > > a very deep topic, and I think it would only pollute the current > >> > > > > discussion. > >> > > > > > >> > > > > What Adam has proposed is safe, given the *current* ordering > >> > semantics > >> > > > > of the system. If we can agree on his proposal, I think we can > >> merge > >> > > the > >> > > > > feature well before the conversation about timestamp ordering > even > >> > > takes > >> > > > > place, much less reaches a conclusion. In the mean time, it > would > >> > seem > >> > > to > >> > > > > be unfortunate to have one join operator with different ordering > >> > > > semantics > >> > > > > from every other KTable operator. > >> > > > > > >> > > > > If and when that timestamp discussion takes place, many (all?) > >> KTable > >> > > > > operations > >> > > > > will need to be updated, rendering the many:one join a small > >> marginal > >> > > > cost. > >> > > > > > >> > > > > And, just to plug it again, I proposed an algorithm above that I > >> > > believe > >> > > > > provides > >> > > > > correct ordering without any additional metadata, and regardless > >> of > >> > the > >> > > > > ordering semantics. I didn't bring it up further, because I felt > >> the > >> > > KIP > >> > > > > only needs > >> > > > > to agree on the public API, and we can discuss the > implementation > >> at > >> > > > > leisure in > >> > > > > a PR... > >> > > > > > >> > > > > Thanks, > >> > > > > -John > >> > > > > > >> > > > > > >> > > > > On Mon, Dec 10, 2018 at 2:28 AM Jan Filipiak < > >> > jan.filip...@trivago.com > >> > > > > >> > > > > wrote: > >> > > > > > >> > > > > > > >> > > > > > > >> > > > > > On 10.12.2018 07:42, Guozhang Wang wrote: > >> > > > > > > Hello Adam / Jan / John, > >> > > > > > > > >> > > > > > > Sorry for being late on this thread! I've finally got some > >> time > >> > > this > >> > > > > > > weekend to cleanup a load of tasks on my queue (actually > I've > >> > also > >> > > > > > realized > >> > > > > > > there are a bunch of other things I need to enqueue while > >> > cleaning > >> > > > them > >> > > > > > up > >> > > > > > > --- sth I need to improve on my side). So here are my > >> thoughts: > >> > > > > > > > >> > > > > > > Regarding the APIs: I like the current written API in the > KIP. > >> > More > >> > > > > > > generally I'd prefer to keep the 1) one-to-many join > >> > > functionalities > >> > > > as > >> > > > > > > well as 2) other join types than inner as separate KIPs > since > >> 1) > >> > > may > >> > > > > > worth > >> > > > > > > a general API refactoring that can benefit not only > foreignkey > >> > > joins > >> > > > > but > >> > > > > > > collocate joins as well (e.g. an extended proposal of > >> > > > > > > > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > >> > https://cwiki.apache.org/confluence/display/KAFKA/KIP-150+-+Kafka-Streams+Cogroup > >> > > > > > ), > >> > > > > > > and I'm not sure if other join types would actually be > needed > >> > > (maybe > >> > > > > left > >> > > > > > > join still makes sense), so it's better to > >> > > > > wait-for-people-to-ask-and-add > >> > > > > > > than add-sth-that-no-one-uses. > >> > > > > > > > >> > > > > > > Regarding whether we enforce step 3) / 4) v.s. introducing a > >> > > > > > > KScatteredTable for users to inject their own optimization: > >> I'd > >> > > > prefer > >> > > > > to > >> > > > > > > do the current option as-is, and my main rationale is for > >> > > > optimization > >> > > > > > > rooms inside the Streams internals and the API succinctness. > >> For > >> > > > > advanced > >> > > > > > > users who may indeed prefer KScatteredTable and do their own > >> > > > > > optimization, > >> > > > > > > while it is too much of the work to use Processor API > >> directly, I > >> > > > think > >> > > > > > we > >> > > > > > > can still extend the current API to support it in the future > >> if > >> > it > >> > > > > > becomes > >> > > > > > > necessary. > >> > > > > > > >> > > > > > no internal optimization potential. it's a myth > >> > > > > > > >> > > > > > ¯\_(ツ)_/¯ > >> > > > > > > >> > > > > > :-) > >> > > > > > > >> > > > > > > > >> > > > > > > Another note about step 4) resolving out-of-ordering data, > as > >> I > >> > > > > mentioned > >> > > > > > > before I think with KIP-258 (embedded timestamp with > key-value > >> > > store) > >> > > > > we > >> > > > > > > can actually make this step simpler than the current > >> proposal. In > >> > > > fact, > >> > > > > > we > >> > > > > > > can just keep a single final-result store with timestamps > and > >> > > reject > >> > > > > > values > >> > > > > > > that have a smaller timestamp, is that right? > >> > > > > > > >> > > > > > Which is the correct output should at least be decided on the > >> > offset > >> > > of > >> > > > > > the original message. > >> > > > > > > >> > > > > > > > >> > > > > > > > >> > > > > > > That's all I have in mind now. Again, great appreciation to > >> Adam > >> > to > >> > > > > make > >> > > > > > > such HUGE progress on this KIP! > >> > > > > > > > >> > > > > > > > >> > > > > > > Guozhang > >> > > > > > > > >> > > > > > > On Wed, Dec 5, 2018 at 2:40 PM Jan Filipiak < > >> > > > jan.filip...@trivago.com> > >> > > > > > > wrote: > >> > > > > > > > >> > > > > > >> If they don't find the time: > >> > > > > > >> They usually take the opposite path from me :D > >> > > > > > >> so the answer would be clear. > >> > > > > > >> > >> > > > > > >> hence my suggestion to vote. > >> > > > > > >> > >> > > > > > >> > >> > > > > > >> On 04.12.2018 21:06, Adam Bellemare wrote: > >> > > > > > >>> Hi Guozhang and Matthias > >> > > > > > >>> > >> > > > > > >>> I know both of you are quite busy, but we've gotten this > KIP > >> > to a > >> > > > > point > >> > > > > > >>> where we need more guidance on the API (perhaps a bit of a > >> > > > > tie-breaker, > >> > > > > > >> if > >> > > > > > >>> you will). If you have anyone else you may think should > >> look at > >> > > > this, > >> > > > > > >>> please tag them accordingly. > >> > > > > > >>> > >> > > > > > >>> The scenario is as such: > >> > > > > > >>> > >> > > > > > >>> Current Option: > >> > > > > > >>> API: > >> > > > > > >>> > >> > > > > > >> > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > >> > https://cwiki.apache.org/confluence/display/KAFKA/KIP-213+Support+non-key+joining+in+KTable#KIP-213Supportnon-keyjoininginKTable-PublicInterfaces > >> > > > > > >>> 1) Rekey the data to CombinedKey, and shuffles it to the > >> > > partition > >> > > > > with > >> > > > > > >> the > >> > > > > > >>> foreignKey (repartition 1) > >> > > > > > >>> 2) Join the data > >> > > > > > >>> 3) Shuffle the data back to the original node (repartition > >> 2) > >> > > > > > >>> 4) Resolve out-of-order arrival / race condition due to > >> > > foreign-key > >> > > > > > >> changes. > >> > > > > > >>> > >> > > > > > >>> Alternate Option: > >> > > > > > >>> Perform #1 and #2 above, and return a KScatteredTable. > >> > > > > > >>> - It would be keyed on a wrapped key function: > >> <CombinedKey<KO, > >> > > K>, > >> > > > > VR> > >> > > > > > >> (KO > >> > > > > > >>> = Other Table Key, K = This Table Key, VR = Joined Result) > >> > > > > > >>> - KScatteredTable.resolve() would perform #3 and #4 but > >> > > otherwise a > >> > > > > > user > >> > > > > > >>> would be able to perform additional functions directly > from > >> the > >> > > > > > >>> KScatteredTable (TBD - currently out of scope). > >> > > > > > >>> - John's analysis 2-emails up is accurate as to the > >> tradeoffs. > >> > > > > > >>> > >> > > > > > >>> Current Option is coded as-is. Alternate option is > possible, > >> > but > >> > > > will > >> > > > > > >>> require for implementation details to be made in the API > and > >> > some > >> > > > > > >> exposure > >> > > > > > >>> of new data structures into the API (ie: CombinedKey). > >> > > > > > >>> > >> > > > > > >>> I appreciate any insight into this. > >> > > > > > >>> > >> > > > > > >>> Thanks. > >> > > > > > >>> > >> > > > > > >>> On Tue, Dec 4, 2018 at 2:59 PM Adam Bellemare < > >> > > > > > adam.bellem...@gmail.com> > >> > > > > > >>> wrote: > >> > > > > > >>> > >> > > > > > >>>> Hi John > >> > > > > > >>>> > >> > > > > > >>>> Thanks for your feedback and assistance. I think your > >> summary > >> > is > >> > > > > > >> accurate > >> > > > > > >>>> from my perspective. Additionally, I would like to add > that > >> > > there > >> > > > > is a > >> > > > > > >> risk > >> > > > > > >>>> of inconsistent final states without performing the > >> > resolution. > >> > > > This > >> > > > > > is > >> > > > > > >> a > >> > > > > > >>>> major concern for me as most of the data I have dealt > with > >> is > >> > > > > produced > >> > > > > > >> by > >> > > > > > >>>> relational databases. We have seen a number of cases > where > >> a > >> > > user > >> > > > in > >> > > > > > the > >> > > > > > >>>> Rails UI has modified the field (foreign key), realized > >> they > >> > > made > >> > > > a > >> > > > > > >>>> mistake, and then updated the field again with a new key. > >> The > >> > > > events > >> > > > > > are > >> > > > > > >>>> propagated out as they are produced, and as such we have > >> had > >> > > > > > real-world > >> > > > > > >>>> cases where these inconsistencies were propagated > >> downstream > >> > as > >> > > > the > >> > > > > > >> final > >> > > > > > >>>> values due to the race conditions in the fanout of the > >> data. > >> > > > > > >>>> > >> > > > > > >>>> This solution that I propose values correctness of the > >> final > >> > > > result > >> > > > > > over > >> > > > > > >>>> other factors. > >> > > > > > >>>> > >> > > > > > >>>> We could always move this function over to using a > >> > > KScatteredTable > >> > > > > > >>>> implementation in the future, and simply deprecate it > this > >> > join > >> > > > API > >> > > > > in > >> > > > > > >>>> time. I think I would like to hear more from some of the > >> other > >> > > > major > >> > > > > > >>>> committers on which course of action they would think is > >> best > >> > > > before > >> > > > > > any > >> > > > > > >>>> more coding is done. > >> > > > > > >>>> > >> > > > > > >>>> Thanks again > >> > > > > > >>>> > >> > > > > > >>>> Adam > >> > > > > > >>>> > >> > > > > > >>>> > >> > > > > > >>>> On Mon, Dec 3, 2018 at 8:24 PM John Roesler < > >> > j...@confluent.io> > >> > > > > > wrote: > >> > > > > > >>>> > >> > > > > > >>>>> Hi Jan and Adam, > >> > > > > > >>>>> > >> > > > > > >>>>> Wow, thanks for doing that test, Adam. Those results are > >> > > > > encouraging. > >> > > > > > >>>>> > >> > > > > > >>>>> Thanks for your performance experience as well, Jan. I > >> agree > >> > > that > >> > > > > > >> avoiding > >> > > > > > >>>>> unnecessary join outputs is especially important when > the > >> > > fan-out > >> > > > > is > >> > > > > > so > >> > > > > > >>>>> high. I suppose this could also be built into the > >> > > implementation > >> > > > > > we're > >> > > > > > >>>>> discussing, but it wouldn't have to be specified in the > >> KIP > >> > > > (since > >> > > > > > >> it's an > >> > > > > > >>>>> API-transparent optimization). > >> > > > > > >>>>> > >> > > > > > >>>>> As far as whether or not to re-repartition the data, I > >> didn't > >> > > > bring > >> > > > > > it > >> > > > > > >> up > >> > > > > > >>>>> because it sounded like the two of you agreed to leave > the > >> > KIP > >> > > > > as-is, > >> > > > > > >>>>> despite the disagreement. > >> > > > > > >>>>> > >> > > > > > >>>>> If you want my opinion, I feel like both approaches are > >> > > > reasonable. > >> > > > > > >>>>> It sounds like Jan values more the potential for > >> developers > >> > to > >> > > > > > optimize > >> > > > > > >>>>> their topologies to re-use the intermediate nodes, > whereas > >> > Adam > >> > > > > > places > >> > > > > > >>>>> more > >> > > > > > >>>>> value on having a single operator that people can use > >> without > >> > > > extra > >> > > > > > >> steps > >> > > > > > >>>>> at the end. > >> > > > > > >>>>> > >> > > > > > >>>>> Personally, although I do find it exceptionally annoying > >> > when a > >> > > > > > >> framework > >> > > > > > >>>>> gets in my way when I'm trying to optimize something, it > >> > seems > >> > > > > better > >> > > > > > >> to > >> > > > > > >>>>> go > >> > > > > > >>>>> for a single operation. > >> > > > > > >>>>> * Encapsulating the internal transitions gives us > >> significant > >> > > > > > latitude > >> > > > > > >> in > >> > > > > > >>>>> the implementation (for example, joining only at the > end, > >> not > >> > > in > >> > > > > the > >> > > > > > >>>>> middle > >> > > > > > >>>>> to avoid extra data copying and out-of-order resolution; > >> how > >> > we > >> > > > > > >> represent > >> > > > > > >>>>> the first repartition keys (combined keys vs. value > >> vectors), > >> > > > > etc.). > >> > > > > > >> If we > >> > > > > > >>>>> publish something like a KScatteredTable with the > >> > > > right-partitioned > >> > > > > > >> joined > >> > > > > > >>>>> data, then the API pretty much locks in the > >> implementation as > >> > > > well. > >> > > > > > >>>>> * The API seems simpler to understand and use. I do mean > >> > > "seems"; > >> > > > > if > >> > > > > > >>>>> anyone > >> > > > > > >>>>> wants to make the case that KScatteredTable is actually > >> > > simpler, > >> > > > I > >> > > > > > >> think > >> > > > > > >>>>> hypothetical usage code would help. From a relational > >> algebra > >> > > > > > >> perspective, > >> > > > > > >>>>> it seems like KTable.join(KTable) should produce a new > >> KTable > >> > > in > >> > > > > all > >> > > > > > >>>>> cases. > >> > > > > > >>>>> * That said, there might still be room in the API for a > >> > > different > >> > > > > > >>>>> operation > >> > > > > > >>>>> like what Jan has proposed to scatter a KTable, and then > >> do > >> > > > things > >> > > > > > like > >> > > > > > >>>>> join, re-group, etc from there... I'm not sure; I > haven't > >> > > thought > >> > > > > > >> through > >> > > > > > >>>>> all the consequences yet. > >> > > > > > >>>>> > >> > > > > > >>>>> This is all just my opinion after thinking over the > >> > discussion > >> > > so > >> > > > > > >> far... > >> > > > > > >>>>> -John > >> > > > > > >>>>> > >> > > > > > >>>>> On Mon, Dec 3, 2018 at 2:56 PM Adam Bellemare < > >> > > > > > >> adam.bellem...@gmail.com> > >> > > > > > >>>>> wrote: > >> > > > > > >>>>> > >> > > > > > >>>>>> Updated the PR to take into account John's feedback. > >> > > > > > >>>>>> > >> > > > > > >>>>>> I did some preliminary testing for the performance of > the > >> > > > > > prefixScan. > >> > > > > > >> I > >> > > > > > >>>>>> have attached the file, but I will also include the > text > >> in > >> > > the > >> > > > > body > >> > > > > > >>>>> here > >> > > > > > >>>>>> for archival purposes (I am not sure what happens to > >> > attached > >> > > > > > files). > >> > > > > > >> I > >> > > > > > >>>>>> also updated the PR and the KIP accordingly. > >> > > > > > >>>>>> > >> > > > > > >>>>>> Summary: It scales exceptionally well for scanning > large > >> > > values > >> > > > of > >> > > > > > >>>>>> records. As Jan mentioned previously, the real issue > >> would > >> > be > >> > > > more > >> > > > > > >>>>> around > >> > > > > > >>>>>> processing the resulting records after obtaining them. > >> For > >> > > > > instance, > >> > > > > > >> it > >> > > > > > >>>>>> takes approximately ~80-120 mS to flush the buffer and > a > >> > > further > >> > > > > > >>>>> ~35-85mS > >> > > > > > >>>>>> to scan 27.5M records, obtaining matches for 2.5M of > >> them. > >> > > > > Iterating > >> > > > > > >>>>>> through the records just to generate a simple count > >> takes ~ > >> > 40 > >> > > > > times > >> > > > > > >>>>> longer > >> > > > > > >>>>>> than the flush + scan combined. > >> > > > > > >>>>>> > >> > > > > > >>>>>> > >> > > > > > >>>>> > >> > > > > > >> > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > >> > ============================================================================================ > >> > > > > > >>>>>> Setup: > >> > > > > > >>>>>> > >> > > > > > >>>>>> > >> > > > > > >>>>> > >> > > > > > >> > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > >> > ============================================================================================ > >> > > > > > >>>>>> Java 9 with default settings aside from a 512 MB heap > >> > > (Xmx512m, > >> > > > > > >> Xms512m) > >> > > > > > >>>>>> CPU: i7 2.2 Ghz. > >> > > > > > >>>>>> > >> > > > > > >>>>>> Note: I am using a slightly-modified, > directly-accessible > >> > > Kafka > >> > > > > > >> Streams > >> > > > > > >>>>>> RocksDB > >> > > > > > >>>>>> implementation (RocksDB.java, basically just avoiding > the > >> > > > > > >>>>>> ProcessorContext). > >> > > > > > >>>>>> There are no modifications to the default RocksDB > values > >> > > > provided > >> > > > > in > >> > > > > > >> the > >> > > > > > >>>>>> 2.1/trunk release. > >> > > > > > >>>>>> > >> > > > > > >>>>>> > >> > > > > > >>>>>> keysize = 128 bytes > >> > > > > > >>>>>> valsize = 512 bytes > >> > > > > > >>>>>> > >> > > > > > >>>>>> Step 1: > >> > > > > > >>>>>> Write X positive matching events: (key = prefix + > >> > left-padded > >> > > > > > >>>>>> auto-incrementing integer) > >> > > > > > >>>>>> Step 2: > >> > > > > > >>>>>> Write 10X negative matching events (key = left-padded > >> > > > > > >> auto-incrementing > >> > > > > > >>>>>> integer) > >> > > > > > >>>>>> Step 3: > >> > > > > > >>>>>> Perform flush > >> > > > > > >>>>>> Step 4: > >> > > > > > >>>>>> Perform prefixScan > >> > > > > > >>>>>> Step 5: > >> > > > > > >>>>>> Iterate through return Iterator and validate the count > of > >> > > > expected > >> > > > > > >>>>> events. > >> > > > > > >>>>>> > >> > > > > > >>>>>> > >> > > > > > >>>>>> > >> > > > > > >>>>> > >> > > > > > >> > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > >> > ============================================================================================ > >> > > > > > >>>>>> Results: > >> > > > > > >>>>>> > >> > > > > > >>>>>> > >> > > > > > >>>>> > >> > > > > > >> > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > >> > ============================================================================================ > >> > > > > > >>>>>> X = 1k (11k events total) > >> > > > > > >>>>>> Flush Time = 39 mS > >> > > > > > >>>>>> Scan Time = 7 mS > >> > > > > > >>>>>> 6.9 MB disk > >> > > > > > >>>>>> > >> > > > > > >>>>>> > >> > > > > > >>>>> > >> > > > > > >> > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > >> > -------------------------------------------------------------------------------------------- > >> > > > > > >>>>>> X = 10k (110k events total) > >> > > > > > >>>>>> Flush Time = 45 mS > >> > > > > > >>>>>> Scan Time = 8 mS > >> > > > > > >>>>>> 127 MB > >> > > > > > >>>>>> > >> > > > > > >>>>>> > >> > > > > > >>>>> > >> > > > > > >> > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > >> > -------------------------------------------------------------------------------------------- > >> > > > > > >>>>>> X = 100k (1.1M events total) > >> > > > > > >>>>>> Test1: > >> > > > > > >>>>>> Flush Time = 60 mS > >> > > > > > >>>>>> Scan Time = 12 mS > >> > > > > > >>>>>> 678 MB > >> > > > > > >>>>>> > >> > > > > > >>>>>> Test2: > >> > > > > > >>>>>> Flush Time = 45 mS > >> > > > > > >>>>>> Scan Time = 7 mS > >> > > > > > >>>>>> 576 MB > >> > > > > > >>>>>> > >> > > > > > >>>>>> > >> > > > > > >>>>> > >> > > > > > >> > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > >> > -------------------------------------------------------------------------------------------- > >> > > > > > >>>>>> X = 1MB (11M events total) > >> > > > > > >>>>>> Test1: > >> > > > > > >>>>>> Flush Time = 52 mS > >> > > > > > >>>>>> Scan Time = 19 mS > >> > > > > > >>>>>> 7.2 GB > >> > > > > > >>>>>> > >> > > > > > >>>>>> Test2: > >> > > > > > >>>>>> Flush Time = 84 mS > >> > > > > > >>>>>> Scan Time = 34 mS > >> > > > > > >>>>>> 9.1 GB > >> > > > > > >>>>>> > >> > > > > > >>>>>> > >> > > > > > >>>>> > >> > > > > > >> > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > >> > -------------------------------------------------------------------------------------------- > >> > > > > > >>>>>> X = 2.5M (27.5M events total) > >> > > > > > >>>>>> Test1: > >> > > > > > >>>>>> Flush Time = 82 mS > >> > > > > > >>>>>> Scan Time = 63 mS > >> > > > > > >>>>>> 17GB - 276 sst files > >> > > > > > >>>>>> > >> > > > > > >>>>>> Test2: > >> > > > > > >>>>>> Flush Time = 116 mS > >> > > > > > >>>>>> Scan Time = 35 mS > >> > > > > > >>>>>> 23GB - 361 sst files > >> > > > > > >>>>>> > >> > > > > > >>>>>> Test3: > >> > > > > > >>>>>> Flush Time = 103 mS > >> > > > > > >>>>>> Scan Time = 82 mS > >> > > > > > >>>>>> 19 GB - 300 sst files > >> > > > > > >>>>>> > >> > > > > > >>>>>> > >> > > > > > >>>>> > >> > > > > > >> > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > >> > -------------------------------------------------------------------------------------------- > >> > > > > > >>>>>> > >> > > > > > >>>>>> I had to limit my testing on my laptop to X = 2.5M > >> events. I > >> > > > tried > >> > > > > > to > >> > > > > > >> go > >> > > > > > >>>>>> to X = 10M (110M events) but RocksDB was going into the > >> > 100GB+ > >> > > > > range > >> > > > > > >>>>> and my > >> > > > > > >>>>>> laptop ran out of disk. More extensive testing could be > >> done > >> > > > but I > >> > > > > > >>>>> suspect > >> > > > > > >>>>>> that it would be in line with what we're seeing in the > >> > results > >> > > > > > above. > >> > > > > > >>>>>> > >> > > > > > >>>>>> > >> > > > > > >>>>>> > >> > > > > > >>>>>> > >> > > > > > >>>>>> > >> > > > > > >>>>>> > >> > > > > > >>>>>> At this point in time, I think the only major > discussion > >> > point > >> > > > is > >> > > > > > >> really > >> > > > > > >>>>>> around what Jan and I have disagreed on: repartitioning > >> > back + > >> > > > > > >> resolving > >> > > > > > >>>>>> potential out of order issues or leaving that up to the > >> > client > >> > > > to > >> > > > > > >>>>> handle. > >> > > > > > >>>>>> > >> > > > > > >>>>>> > >> > > > > > >>>>>> Thanks folks, > >> > > > > > >>>>>> > >> > > > > > >>>>>> Adam > >> > > > > > >>>>>> > >> > > > > > >>>>>> > >> > > > > > >>>>>> On Mon, Dec 3, 2018 at 4:34 AM Jan Filipiak < > >> > > > > > jan.filip...@trivago.com > >> > > > > > >>> > >> > > > > > >>>>>> wrote: > >> > > > > > >>>>>> > >> > > > > > >>>>>>> > >> > > > > > >>>>>>> > >> > > > > > >>>>>>> On 29.11.2018 15:14, John Roesler wrote: > >> > > > > > >>>>>>>> Hi all, > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> Sorry that this discussion petered out... I think the > >> 2.1 > >> > > > > release > >> > > > > > >>>>>>> caused an > >> > > > > > >>>>>>>> extended distraction that pushed it off everyone's > >> radar > >> > > > (which > >> > > > > > was > >> > > > > > >>>>>>>> precisely Adam's concern). Personally, I've also had > >> some > >> > > > extend > >> > > > > > >>>>>>>> distractions of my own that kept (and continue to > >> keep) me > >> > > > > > >>>>> preoccupied. > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> However, calling for a vote did wake me up, so I > guess > >> Jan > >> > > was > >> > > > > on > >> > > > > > >> the > >> > > > > > >>>>>>> right > >> > > > > > >>>>>>>> track! > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> I've gone back and reviewed the whole KIP document > and > >> the > >> > > > prior > >> > > > > > >>>>>>>> discussion, and I'd like to offer a few thoughts: > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> API Thoughts: > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> 1. If I read the KIP right, you are proposing a > >> > many-to-one > >> > > > > join. > >> > > > > > >>>>> Could > >> > > > > > >>>>>>> we > >> > > > > > >>>>>>>> consider naming it manyToOneJoin? Or, if you prefer, > >> flip > >> > > the > >> > > > > > design > >> > > > > > >>>>>>> around > >> > > > > > >>>>>>>> and make it a oneToManyJoin? > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> The proposed name "joinOnForeignKey" disguises the > join > >> > > type, > >> > > > > and > >> > > > > > it > >> > > > > > >>>>>>> seems > >> > > > > > >>>>>>>> like it might trick some people into using it for a > >> > > one-to-one > >> > > > > > join. > >> > > > > > >>>>>>> This > >> > > > > > >>>>>>>> would work, of course, but it would be super > >> inefficient > >> > > > > compared > >> > > > > > to > >> > > > > > >>>>> a > >> > > > > > >>>>>>>> simple rekey-and-join. > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> 2. I might have missed it, but I don't think it's > >> > specified > >> > > > > > whether > >> > > > > > >>>>>>> it's an > >> > > > > > >>>>>>>> inner, outer, or left join. I'm guessing an outer > >> join, as > >> > > > > > >>>>> (neglecting > >> > > > > > >>>>>>> IQ), > >> > > > > > >>>>>>>> the rest can be achieved by filtering or by handling > >> it in > >> > > the > >> > > > > > >>>>>>> ValueJoiner. > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> 3. The arg list to joinOnForeignKey doesn't look > quite > >> > > right. > >> > > > > > >>>>>>>> 3a. Regarding Serialized: There are a few different > >> > > paradigms > >> > > > in > >> > > > > > >>>>> play in > >> > > > > > >>>>>>>> the Streams API, so it's confusing, but instead of > >> three > >> > > > > > Serialized > >> > > > > > >>>>>>> args, I > >> > > > > > >>>>>>>> think it would be better to have one that allows > >> > > (optionally) > >> > > > > > >> setting > >> > > > > > >>>>>>> the 4 > >> > > > > > >>>>>>>> incoming serdes. The result serde is defined by the > >> > > > > Materialized. > >> > > > > > >> The > >> > > > > > >>>>>>>> incoming serdes can be optional because they might > >> already > >> > > be > >> > > > > > >>>>> available > >> > > > > > >>>>>>> on > >> > > > > > >>>>>>>> the source KTables, or the default serdes from the > >> config > >> > > > might > >> > > > > be > >> > > > > > >>>>>>>> applicable. > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> 3b. Is the StreamPartitioner necessary? The other > joins > >> > > don't > >> > > > > > allow > >> > > > > > >>>>>>> setting > >> > > > > > >>>>>>>> one, and it seems like it might actually be harmful, > >> since > >> > > the > >> > > > > > rekey > >> > > > > > >>>>>>>> operation needs to produce results that are > >> co-partitioned > >> > > > with > >> > > > > > the > >> > > > > > >>>>>>> "other" > >> > > > > > >>>>>>>> KTable. > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> 4. I'm fine with the "reserved word" header, but I > >> didn't > >> > > > > actually > >> > > > > > >>>>>>> follow > >> > > > > > >>>>>>>> what Matthias meant about namespacing requiring > >> > > > "deserializing" > >> > > > > > the > >> > > > > > >>>>>>> record > >> > > > > > >>>>>>>> header. The headers are already Strings, so I don't > >> think > >> > > that > >> > > > > > >>>>>>>> deserialization is required. If we applied the > >> namespace > >> > at > >> > > > > source > >> > > > > > >>>>> nodes > >> > > > > > >>>>>>>> and stripped it at sink nodes, this would be > >> practically > >> > no > >> > > > > > >> overhead. > >> > > > > > >>>>>>> The > >> > > > > > >>>>>>>> advantage of the namespace idea is that no public API > >> > change > >> > > > wrt > >> > > > > > >>>>> headers > >> > > > > > >>>>>>>> needs to happen, and no restrictions need to be > placed > >> on > >> > > > users' > >> > > > > > >>>>>>> headers. > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> (Although I'm wondering if we can get away without > the > >> > > header > >> > > > at > >> > > > > > >>>>> all... > >> > > > > > >>>>>>>> stay tuned) > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> 5. I also didn't follow the discussion about the HWM > >> table > >> > > > > growing > >> > > > > > >>>>>>> without > >> > > > > > >>>>>>>> bound. As I read it, the HWM table is effectively > >> > > implementing > >> > > > > OCC > >> > > > > > >> to > >> > > > > > >>>>>>>> resolve the problem you noted with disordering when > the > >> > > rekey > >> > > > is > >> > > > > > >>>>>>>> reversed... particularly notable when the FK changes. > >> As > >> > > such, > >> > > > > it > >> > > > > > >>>>> only > >> > > > > > >>>>>>>> needs to track the most recent "version" (the offset > in > >> > the > >> > > > > source > >> > > > > > >>>>>>>> partition) of each key. Therefore, it should have the > >> same > >> > > > > number > >> > > > > > of > >> > > > > > >>>>>>> keys > >> > > > > > >>>>>>>> as the source table at all times. > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> I see that you are aware of KIP-258, which I think > >> might > >> > be > >> > > > > > relevant > >> > > > > > >>>>> in > >> > > > > > >>>>>>> a > >> > > > > > >>>>>>>> couple of ways. One: it's just about storing the > >> timestamp > >> > > in > >> > > > > the > >> > > > > > >>>>> state > >> > > > > > >>>>>>>> store, but the ultimate idea is to effectively use > the > >> > > > timestamp > >> > > > > > as > >> > > > > > >>>>> an > >> > > > > > >>>>>>> OCC > >> > > > > > >>>>>>>> "version" to drop disordered updates. You wouldn't > >> want to > >> > > use > >> > > > > the > >> > > > > > >>>>>>>> timestamp for this operation, but if you were to use > a > >> > > similar > >> > > > > > >>>>>>> mechanism to > >> > > > > > >>>>>>>> store the source offset in the store alongside the > >> > re-keyed > >> > > > > > values, > >> > > > > > >>>>> then > >> > > > > > >>>>>>>> you could avoid a separate table. > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> 6. You and Jan have been thinking about this for a > long > >> > > time, > >> > > > so > >> > > > > > >> I've > >> > > > > > >>>>>>>> probably missed something here, but I'm wondering if > we > >> > can > >> > > > > avoid > >> > > > > > >> the > >> > > > > > >>>>>>> HWM > >> > > > > > >>>>>>>> tracking at all and resolve out-of-order during a > final > >> > join > >> > > > > > >>>>> instead... > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> Let's say we're joining a left table (Integer K: > Letter > >> > FK, > >> > > > > (other > >> > > > > > >>>>>>> data)) > >> > > > > > >>>>>>>> to a right table (Letter K: (some data)). > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> Left table: > >> > > > > > >>>>>>>> 1: (A, xyz) > >> > > > > > >>>>>>>> 2: (B, asd) > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> Right table: > >> > > > > > >>>>>>>> A: EntityA > >> > > > > > >>>>>>>> B: EntityB > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> We could do a rekey as you proposed with a combined > >> key, > >> > but > >> > > > not > >> > > > > > >>>>>>>> propagating the value at all.. > >> > > > > > >>>>>>>> Rekey table: > >> > > > > > >>>>>>>> A-1: (dummy value) > >> > > > > > >>>>>>>> B-2: (dummy value) > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> Which we then join with the right table to produce: > >> > > > > > >>>>>>>> A-1: EntityA > >> > > > > > >>>>>>>> B-2: EntityB > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> Which gets rekeyed back: > >> > > > > > >>>>>>>> 1: A, EntityA > >> > > > > > >>>>>>>> 2: B, EntityB > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> And finally we do the actual join: > >> > > > > > >>>>>>>> Result table: > >> > > > > > >>>>>>>> 1: ((A, xyz), EntityA) > >> > > > > > >>>>>>>> 2: ((B, asd), EntityB) > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> The thing is that in that last join, we have the > >> > opportunity > >> > > > to > >> > > > > > >>>>> compare > >> > > > > > >>>>>>> the > >> > > > > > >>>>>>>> current FK in the left table with the incoming PK of > >> the > >> > > right > >> > > > > > >>>>> table. If > >> > > > > > >>>>>>>> they don't match, we just drop the event, since it > >> must be > >> > > > > > outdated. > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>> > >> > > > > > >>>>>>>> In your KIP, you gave an example in which (1: A, xyz) > >> gets > >> > > > > updated > >> > > > > > >> to > >> > > > > > >>>>>>> (1: > >> > > > > > >>>>>>>> B, xyz), ultimately yielding a conundrum about > whether > >> the > >> > > > final > >> > > > > > >>>>> state > >> > > > > > >>>>>>>> should be (1: null) or (1: joined-on-B). With the > >> > algorithm > >> > > > > above, > >> > > > > > >>>>> you > >> > > > > > >>>>>>>> would be considering (1: (B, xyz), (A, null)) vs (1: > >> (B, > >> > > xyz), > >> > > > > (B, > >> > > > > > >>>>>>>> EntityB)). It seems like this does give you enough > >> > > information > >> > > > > to > >> > > > > > >>>>> make > >> > > > > > >>>>>>> the > >> > > > > > >>>>>>>> right choice, regardless of disordering. > >> > > > > > >>>>>>> > >> > > > > > >>>>>>> Will check Adams patch, but this should work. As > >> mentioned > >> > > > often > >> > > > > I > >> > > > > > am > >> > > > > > >>>>>>> not convinced on partitioning back for the user > >> > > automatically. > >> > > > I > >> > > > > > >> think > >> > > > > > >>>>>>> this is the real performance eater ;) > >> > > > > > >>>>>>> > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> 7. Last thought... I'm a little concerned about the > >> > > > performance > >> > > > > of > >> > > > > > >>>>> the > >> > > > > > >>>>>>>> range scans when records change in the right table. > >> You've > >> > > > said > >> > > > > > that > >> > > > > > >>>>>>> you've > >> > > > > > >>>>>>>> been using the algorithm you presented in production > >> for a > >> > > > > while. > >> > > > > > >> Can > >> > > > > > >>>>>>> you > >> > > > > > >>>>>>>> give us a sense of the performance characteristics > >> you've > >> > > > > > observed? > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>> > >> > > > > > >>>>>>> Make it work, make it fast, make it beautiful. The > >> topmost > >> > > > thing > >> > > > > > here > >> > > > > > >>>>> is > >> > > > > > >>>>>>> / was correctness. In practice I do not measure the > >> > > performance > >> > > > > of > >> > > > > > >> the > >> > > > > > >>>>>>> range scan. Usual cases I run this with is emitting > >> 500k - > >> > > 1kk > >> > > > > rows > >> > > > > > >>>>>>> on a left hand side change. The range scan is just the > >> work > >> > > you > >> > > > > > gotta > >> > > > > > >>>>>>> do, also when you pack your data into different > formats, > >> > > > usually > >> > > > > > the > >> > > > > > >>>>>>> rocks performance is very tight to the size of the > data > >> and > >> > > we > >> > > > > > can't > >> > > > > > >>>>>>> really change that. It is more important for users to > >> > prevent > >> > > > > > useless > >> > > > > > >>>>>>> updates to begin with. My left hand side is guarded to > >> drop > >> > > > > changes > >> > > > > > >>>>> that > >> > > > > > >>>>>>> are not going to change my join output. > >> > > > > > >>>>>>> > >> > > > > > >>>>>>> usually it's: > >> > > > > > >>>>>>> > >> > > > > > >>>>>>> drop unused fields and then don't forward if > >> > old.equals(new) > >> > > > > > >>>>>>> > >> > > > > > >>>>>>> regarding to the performance of creating an iterator > for > >> > > > smaller > >> > > > > > >>>>>>> fanouts, users can still just do a group by first then > >> > > anyways. > >> > > > > > >>>>>>> > >> > > > > > >>>>>>> > >> > > > > > >>>>>>> > >> > > > > > >>>>>>>> I could only think of one alternative, but I'm not > >> sure if > >> > > > it's > >> > > > > > >>>>> better > >> > > > > > >>>>>>> or > >> > > > > > >>>>>>>> worse... If the first re-key only needs to preserve > the > >> > > > original > >> > > > > > >> key, > >> > > > > > >>>>>>> as I > >> > > > > > >>>>>>>> proposed in #6, then we could store a vector of keys > in > >> > the > >> > > > > value: > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> Left table: > >> > > > > > >>>>>>>> 1: A,... > >> > > > > > >>>>>>>> 2: B,... > >> > > > > > >>>>>>>> 3: A,... > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> Gets re-keyed: > >> > > > > > >>>>>>>> A: [1, 3] > >> > > > > > >>>>>>>> B: [2] > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> Then, the rhs part of the join would only need a > >> regular > >> > > > > > single-key > >> > > > > > >>>>>>> lookup. > >> > > > > > >>>>>>>> Of course we have to deal with the problem of large > >> > values, > >> > > as > >> > > > > > >>>>> there's > >> > > > > > >>>>>>> no > >> > > > > > >>>>>>>> bound on the number of lhs records that can reference > >> rhs > >> > > > > records. > >> > > > > > >>>>>>> Offhand, > >> > > > > > >>>>>>>> I'd say we could page the values, so when one row is > >> past > >> > > the > >> > > > > > >>>>>>> threshold, we > >> > > > > > >>>>>>>> append the key for the next page. Then in most cases, > >> it > >> > > would > >> > > > > be > >> > > > > > a > >> > > > > > >>>>>>> single > >> > > > > > >>>>>>>> key lookup, but for large fan-out updates, it would > be > >> one > >> > > per > >> > > > > > (max > >> > > > > > >>>>>>> value > >> > > > > > >>>>>>>> size)/(avg lhs key size). > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> This seems more complex, though... Plus, I think > >> there's > >> > > some > >> > > > > > extra > >> > > > > > >>>>>>>> tracking we'd need to do to know when to emit a > >> > retraction. > >> > > > For > >> > > > > > >>>>> example, > >> > > > > > >>>>>>>> when record 1 is deleted, the re-key table would just > >> have > >> > > (A: > >> > > > > > [3]). > >> > > > > > >>>>>>> Some > >> > > > > > >>>>>>>> kind of tombstone is needed so that the join result > >> for 1 > >> > > can > >> > > > > also > >> > > > > > >> be > >> > > > > > >>>>>>>> retracted. > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> That's all! > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> Thanks so much to both Adam and Jan for the > thoughtful > >> > KIP. > >> > > > > Sorry > >> > > > > > >> the > >> > > > > > >>>>>>>> discussion has been slow. > >> > > > > > >>>>>>>> -John > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>> On Fri, Oct 12, 2018 at 2:20 AM Jan Filipiak < > >> > > > > > >>>>> jan.filip...@trivago.com> > >> > > > > > >>>>>>>> wrote: > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>>>> Id say you can just call the vote. > >> > > > > > >>>>>>>>> > >> > > > > > >>>>>>>>> that happens all the time, and if something comes > up, > >> it > >> > > just > >> > > > > > goes > >> > > > > > >>>>> back > >> > > > > > >>>>>>>>> to discuss. > >> > > > > > >>>>>>>>> > >> > > > > > >>>>>>>>> would not expect to much attention with another > >> another > >> > > email > >> > > > > in > >> > > > > > >>>>> this > >> > > > > > >>>>>>>>> thread. > >> > > > > > >>>>>>>>> > >> > > > > > >>>>>>>>> best Jan > >> > > > > > >>>>>>>>> > >> > > > > > >>>>>>>>> On 09.10.2018 13:56, Adam Bellemare wrote: > >> > > > > > >>>>>>>>>> Hello Contributors > >> > > > > > >>>>>>>>>> > >> > > > > > >>>>>>>>>> I know that 2.1 is about to be released, but I do > >> need > >> > to > >> > > > bump > >> > > > > > >>>>> this to > >> > > > > > >>>>>>>>> keep > >> > > > > > >>>>>>>>>> visibility up. I am still intending to push this > >> through > >> > > > once > >> > > > > > >>>>>>> contributor > >> > > > > > >>>>>>>>>> feedback is given. > >> > > > > > >>>>>>>>>> > >> > > > > > >>>>>>>>>> Main points that need addressing: > >> > > > > > >>>>>>>>>> 1) Any way (or benefit) in structuring the current > >> > > singular > >> > > > > > graph > >> > > > > > >>>>> node > >> > > > > > >>>>>>>>> into > >> > > > > > >>>>>>>>>> multiple nodes? It has a whopping 25 parameters > right > >> > > now. I > >> > > > > am > >> > > > > > a > >> > > > > > >>>>> bit > >> > > > > > >>>>>>>>> fuzzy > >> > > > > > >>>>>>>>>> on how the optimizations are supposed to work, so I > >> > would > >> > > > > > >>>>> appreciate > >> > > > > > >>>>>>> any > >> > > > > > >>>>>>>>>> help on this aspect. > >> > > > > > >>>>>>>>>> > >> > > > > > >>>>>>>>>> 2) Overall strategy for joining + resolving. This > >> thread > >> > > has > >> > > > > > much > >> > > > > > >>>>>>>>> discourse > >> > > > > > >>>>>>>>>> between Jan and I between the current highwater > mark > >> > > > proposal > >> > > > > > and > >> > > > > > >> a > >> > > > > > >>>>>>>>> groupBy > >> > > > > > >>>>>>>>>> + reduce proposal. I am of the opinion that we need > >> to > >> > > > > strictly > >> > > > > > >>>>> handle > >> > > > > > >>>>>>>>> any > >> > > > > > >>>>>>>>>> chance of out-of-order data and leave none of it up > >> to > >> > the > >> > > > > > >>>>> consumer. > >> > > > > > >>>>>>> Any > >> > > > > > >>>>>>>>>> comments or suggestions here would also help. > >> > > > > > >>>>>>>>>> > >> > > > > > >>>>>>>>>> 3) Anything else that you see that would prevent > this > >> > from > >> > > > > > moving > >> > > > > > >>>>> to a > >> > > > > > >>>>>>>>> vote? > >> > > > > > >>>>>>>>>> > >> > > > > > >>>>>>>>>> Thanks > >> > > > > > >>>>>>>>>> > >> > > > > > >>>>>>>>>> Adam > >> > > > > > >>>>>>>>>> > >> > > > > > >>>>>>>>>> > >> > > > > > >>>>>>>>>> > >> > > > > > >>>>>>>>>> > >> > > > > > >>>>>>>>>> > >> > > > > > >>>>>>>>>> > >> > > > > > >>>>>>>>>> > >> > > > > > >>>>>>>>>> On Sun, Sep 30, 2018 at 10:23 AM Adam Bellemare < > >> > > > > > >>>>>>>>> adam.bellem...@gmail.com> > >> > > > > > >>>>>>>>>> wrote: > >> > > > > > >>>>>>>>>> > >> > > > > > >>>>>>>>>>> Hi Jan > >> > > > > > >>>>>>>>>>> > >> > > > > > >>>>>>>>>>> With the Stores.windowStoreBuilder and > >> > > > > > >>>>> Stores.persistentWindowStore, > >> > > > > > >>>>>>> you > >> > > > > > >>>>>>>>>>> actually only need to specify the amount of > segments > >> > you > >> > > > want > >> > > > > > and > >> > > > > > >>>>> how > >> > > > > > >>>>>>>>> large > >> > > > > > >>>>>>>>>>> they are. To the best of my understanding, what > >> happens > >> > > is > >> > > > > that > >> > > > > > >>>>> the > >> > > > > > >>>>>>>>>>> segments are automatically rolled over as new data > >> with > >> > > new > >> > > > > > >>>>>>> timestamps > >> > > > > > >>>>>>>>> are > >> > > > > > >>>>>>>>>>> created. We use this exact functionality in some > of > >> the > >> > > > work > >> > > > > > done > >> > > > > > >>>>>>>>>>> internally at my company. For reference, this is > the > >> > > > hopping > >> > > > > > >>>>> windowed > >> > > > > > >>>>>>>>> store. > >> > > > > > >>>>>>>>>>> > >> > > > > > >>>>>>>>>>> > >> > > > > > >>>>>>>>> > >> > > > > > >>>>>>> > >> > > > > > >>>>> > >> > > > > > >> > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > >> > https://kafka.apache.org/11/documentation/streams/developer-guide/dsl-api.html#id21 > >> > > > > > >>>>>>>>>>> > >> > > > > > >>>>>>>>>>> In the code that I have provided, there are going > >> to be > >> > > two > >> > > > > 24h > >> > > > > > >>>>>>>>> segments. > >> > > > > > >>>>>>>>>>> When a record is put into the windowStore, it will > >> be > >> > > > > inserted > >> > > > > > at > >> > > > > > >>>>>>> time > >> > > > > > >>>>>>>>> T in > >> > > > > > >>>>>>>>>>> both segments. The two segments will always > overlap > >> by > >> > > 12h. > >> > > > > As > >> > > > > > >>>>> time > >> > > > > > >>>>>>>>> goes on > >> > > > > > >>>>>>>>>>> and new records are added (say at time T+12h+), > the > >> > > oldest > >> > > > > > >> segment > >> > > > > > >>>>>>> will > >> > > > > > >>>>>>>>> be > >> > > > > > >>>>>>>>>>> automatically deleted and a new segment created. > The > >> > > > records > >> > > > > > are > >> > > > > > >>>>> by > >> > > > > > >>>>>>>>> default > >> > > > > > >>>>>>>>>>> inserted with the context.timestamp(), such that > it > >> is > >> > > the > >> > > > > > record > >> > > > > > >>>>>>> time, > >> > > > > > >>>>>>>>> not > >> > > > > > >>>>>>>>>>> the clock time, which is used. > >> > > > > > >>>>>>>>>>> > >> > > > > > >>>>>>>>>>> To the best of my understanding, the timestamps > are > >> > > > retained > >> > > > > > when > >> > > > > > >>>>>>>>>>> restoring from the changelog. > >> > > > > > >>>>>>>>>>> > >> > > > > > >>>>>>>>>>> Basically, this is heavy-handed way to deal with > TTL > >> > at a > >> > > > > > >>>>>>> segment-level, > >> > > > > > >>>>>>>>>>> instead of at an individual record level. > >> > > > > > >>>>>>>>>>> > >> > > > > > >>>>>>>>>>> On Tue, Sep 25, 2018 at 5:18 PM Jan Filipiak < > >> > > > > > >>>>>>> jan.filip...@trivago.com> > >> > > > > > >>>>>>>>>>> wrote: > >> > > > > > >>>>>>>>>>> > >> > > > > > >>>>>>>>>>>> Will that work? I expected it to blow up with > >> > > > > > ClassCastException > >> > > > > > >>>>> or > >> > > > > > >>>>>>>>>>>> similar. > >> > > > > > >>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>> You either would have to specify the window you > >> > > fetch/put > >> > > > or > >> > > > > > >>>>> iterate > >> > > > > > >>>>>>>>>>>> across all windows the key was found in right? > >> > > > > > >>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>> I just hope the window-store doesn't check > >> stream-time > >> > > > under > >> > > > > > the > >> > > > > > >>>>>>> hoods > >> > > > > > >>>>>>>>>>>> that would be a questionable interface. > >> > > > > > >>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>> If it does: did you see my comment on checking > all > >> the > >> > > > > windows > >> > > > > > >>>>>>> earlier? > >> > > > > > >>>>>>>>>>>> that would be needed to actually give reasonable > >> time > >> > > > > > gurantees. > >> > > > > > >>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>> Best > >> > > > > > >>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>> On 25.09.2018 13:18, Adam Bellemare wrote: > >> > > > > > >>>>>>>>>>>>> Hi Jan > >> > > > > > >>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>> Check for " highwaterMat " in the PR. I only > >> changed > >> > > the > >> > > > > > state > >> > > > > > >>>>>>> store, > >> > > > > > >>>>>>>>>>>> not > >> > > > > > >>>>>>>>>>>>> the ProcessorSupplier. > >> > > > > > >>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>> Thanks, > >> > > > > > >>>>>>>>>>>>> Adam > >> > > > > > >>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>> On Mon, Sep 24, 2018 at 2:47 PM, Jan Filipiak < > >> > > > > > >>>>>>>>> jan.filip...@trivago.com > >> > > > > > >>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>> wrote: > >> > > > > > >>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>> On 24.09.2018 16:26, Adam Bellemare wrote: > >> > > > > > >>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>> @Guozhang > >> > > > > > >>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>> Thanks for the information. This is indeed > >> > something > >> > > > that > >> > > > > > >>>>> will be > >> > > > > > >>>>>>>>>>>>>>> extremely > >> > > > > > >>>>>>>>>>>>>>> useful for this KIP. > >> > > > > > >>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>> @Jan > >> > > > > > >>>>>>>>>>>>>>> Thanks for your explanations. That being > said, I > >> > will > >> > > > not > >> > > > > > be > >> > > > > > >>>>>>> moving > >> > > > > > >>>>>>>>>>>> ahead > >> > > > > > >>>>>>>>>>>>>>> with an implementation using reshuffle/groupBy > >> > > solution > >> > > > > as > >> > > > > > >> you > >> > > > > > >>>>>>>>>>>> propose. > >> > > > > > >>>>>>>>>>>>>>> That being said, if you wish to implement it > >> > yourself > >> > > > off > >> > > > > > of > >> > > > > > >>>>> my > >> > > > > > >>>>>>>>>>>> current PR > >> > > > > > >>>>>>>>>>>>>>> and submit it as a competitive alternative, I > >> would > >> > > be > >> > > > > more > >> > > > > > >>>>> than > >> > > > > > >>>>>>>>>>>> happy to > >> > > > > > >>>>>>>>>>>>>>> help vet that as an alternate solution. As it > >> > stands > >> > > > > right > >> > > > > > >>>>> now, > >> > > > > > >>>>>>> I do > >> > > > > > >>>>>>>>>>>> not > >> > > > > > >>>>>>>>>>>>>>> really have more time to invest into > >> alternatives > >> > > > without > >> > > > > > >>>>> there > >> > > > > > >>>>>>>>> being > >> > > > > > >>>>>>>>>>>> a > >> > > > > > >>>>>>>>>>>>>>> strong indication from the binding voters > which > >> > they > >> > > > > would > >> > > > > > >>>>>>> prefer. > >> > > > > > >>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>> Hey, total no worries. I think I personally > gave > >> up > >> > on > >> > > > the > >> > > > > > >>>>> streams > >> > > > > > >>>>>>>>> DSL > >> > > > > > >>>>>>>>>>>> for > >> > > > > > >>>>>>>>>>>>>> some time already, otherwise I would have > pulled > >> > this > >> > > > KIP > >> > > > > > >>>>> through > >> > > > > > >>>>>>>>>>>> already. > >> > > > > > >>>>>>>>>>>>>> I am currently reimplementing my own DSL based > on > >> > > PAPI. > >> > > > > > >>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>> I will look at finishing up my PR with the > >> windowed > >> > > > state > >> > > > > > >>>>> store > >> > > > > > >>>>>>> in > >> > > > > > >>>>>>>>> the > >> > > > > > >>>>>>>>>>>>>>> next > >> > > > > > >>>>>>>>>>>>>>> week or so, exercising it via tests, and then > I > >> > will > >> > > > come > >> > > > > > >> back > >> > > > > > >>>>>>> for > >> > > > > > >>>>>>>>>>>> final > >> > > > > > >>>>>>>>>>>>>>> discussions. In the meantime, I hope that any > of > >> > the > >> > > > > > binding > >> > > > > > >>>>>>> voters > >> > > > > > >>>>>>>>>>>> could > >> > > > > > >>>>>>>>>>>>>>> take a look at the KIP in the wiki. I have > >> updated > >> > it > >> > > > > > >>>>> according > >> > > > > > >>>>>>> to > >> > > > > > >>>>>>>>> the > >> > > > > > >>>>>>>>>>>>>>> latest plan: > >> > > > > > >>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>> > >> > > > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-213+ > >> > > > > > >>>>>>>>>>>>>>> Support+non-key+joining+in+KTable > >> > > > > > >>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>> I have also updated the KIP PR to use a > windowed > >> > > store. > >> > > > > > This > >> > > > > > >>>>>>> could > >> > > > > > >>>>>>>>> be > >> > > > > > >>>>>>>>>>>>>>> replaced by the results of KIP-258 whenever > they > >> > are > >> > > > > > >>>>> completed. > >> > > > > > >>>>>>>>>>>>>>> https://github.com/apache/kafka/pull/5527 > >> > > > > > >>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>> Thanks, > >> > > > > > >>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>> Adam > >> > > > > > >>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>> Is the HighWatermarkResolverProccessorsupplier > >> > already > >> > > > > > updated > >> > > > > > >>>>> in > >> > > > > > >>>>>>> the > >> > > > > > >>>>>>>>>>>> PR? > >> > > > > > >>>>>>>>>>>>>> expected it to change to Windowed<K>,Long > Missing > >> > > > > something? > >> > > > > > >>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>> On Fri, Sep 14, 2018 at 2:24 PM, Guozhang > Wang < > >> > > > > > >>>>>>> wangg...@gmail.com> > >> > > > > > >>>>>>>>>>>>>>> wrote: > >> > > > > > >>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>> Correction on my previous email: KAFKA-5533 is > >> the > >> > > > wrong > >> > > > > > >> link, > >> > > > > > >>>>>>> as it > >> > > > > > >>>>>>>>>>>> is > >> > > > > > >>>>>>>>>>>>>>>> for > >> > > > > > >>>>>>>>>>>>>>>> corresponding changelog mechanisms. But as > >> part of > >> > > > > KIP-258 > >> > > > > > >>>>> we do > >> > > > > > >>>>>>>>>>>> want to > >> > > > > > >>>>>>>>>>>>>>>> have "handling out-of-order data for source > >> > KTable" > >> > > > such > >> > > > > > >> that > >> > > > > > >>>>>>>>>>>> instead of > >> > > > > > >>>>>>>>>>>>>>>> blindly apply the updates to the materialized > >> > store, > >> > > > > i.e. > >> > > > > > >>>>>>> following > >> > > > > > >>>>>>>>>>>>>>>> offset > >> > > > > > >>>>>>>>>>>>>>>> ordering, we will reject updates that are > older > >> > than > >> > > > the > >> > > > > > >>>>> current > >> > > > > > >>>>>>>>>>>> key's > >> > > > > > >>>>>>>>>>>>>>>> timestamps, i.e. following timestamp > ordering. > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>> Guozhang > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>> On Fri, Sep 14, 2018 at 11:21 AM, Guozhang > >> Wang < > >> > > > > > >>>>>>>>> wangg...@gmail.com> > >> > > > > > >>>>>>>>>>>>>>>> wrote: > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>> Hello Adam, > >> > > > > > >>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> Thanks for the explanation. Regarding the > >> final > >> > > step > >> > > > > > (i.e. > >> > > > > > >>>>> the > >> > > > > > >>>>>>>>> high > >> > > > > > >>>>>>>>>>>>>>>>> watermark store, now altered to be replaced > >> with > >> > a > >> > > > > window > >> > > > > > >>>>>>> store), > >> > > > > > >>>>>>>>> I > >> > > > > > >>>>>>>>>>>>>>>>> think > >> > > > > > >>>>>>>>>>>>>>>>> another current on-going KIP may actually > >> help: > >> > > > > > >>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> > >> > > > https://cwiki.apache.org/confluence/display/KAFKA/KIP- > >> > > > > > >>>>>>>>>>>>>>>>> > >> > 258%3A+Allow+to+Store+Record+Timestamps+in+RocksDB > >> > > > > > >>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> This is for adding the timestamp into a > >> key-value > >> > > > store > >> > > > > > >>>>> (i.e. > >> > > > > > >>>>>>> only > >> > > > > > >>>>>>>>>>>> for > >> > > > > > >>>>>>>>>>>>>>>>> non-windowed KTable), and then one of its > >> usage, > >> > as > >> > > > > > >>>>> described > >> > > > > > >>>>>>> in > >> > > > > > >>>>>>>>>>>>>>>>> > >> https://issues.apache.org/jira/browse/KAFKA-5533 > >> > , > >> > > is > >> > > > > > that > >> > > > > > >>>>> we > >> > > > > > >>>>>>> can > >> > > > > > >>>>>>>>>>>> then > >> > > > > > >>>>>>>>>>>>>>>>> "reject" updates from the source topics if > its > >> > > > > timestamp > >> > > > > > is > >> > > > > > >>>>>>>>> smaller > >> > > > > > >>>>>>>>>>>> than > >> > > > > > >>>>>>>>>>>>>>>>> the current key's latest update timestamp. I > >> > think > >> > > it > >> > > > > is > >> > > > > > >>>>> very > >> > > > > > >>>>>>>>>>>> similar to > >> > > > > > >>>>>>>>>>>>>>>>> what you have in mind for high watermark > based > >> > > > > filtering, > >> > > > > > >>>>> while > >> > > > > > >>>>>>>>> you > >> > > > > > >>>>>>>>>>>> only > >> > > > > > >>>>>>>>>>>>>>>>> need to make sure that the timestamps of the > >> > > joining > >> > > > > > >> records > >> > > > > > >>>>>>> are > >> > > > > > >>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>> correctly > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> inherited though the whole topology to the > >> final > >> > > > stage. > >> > > > > > >>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> Note that this KIP is for key-value store > and > >> > hence > >> > > > > > >>>>>>> non-windowed > >> > > > > > >>>>>>>>>>>> KTables > >> > > > > > >>>>>>>>>>>>>>>>> only, but for windowed KTables we do not > >> really > >> > > have > >> > > > a > >> > > > > > good > >> > > > > > >>>>>>>>> support > >> > > > > > >>>>>>>>>>>> for > >> > > > > > >>>>>>>>>>>>>>>>> their joins anyways ( > >> > > > > > >>>>>>>>>>>> https://issues.apache.org/jira/browse/KAFKA-7107 > ) > >> > > > > > >>>>>>>>>>>>>>>>> I > >> > > > > > >>>>>>>>>>>>>>>>> think we can just consider non-windowed > >> > > KTable-KTable > >> > > > > > >>>>> non-key > >> > > > > > >>>>>>>>> joins > >> > > > > > >>>>>>>>>>>> for > >> > > > > > >>>>>>>>>>>>>>>>> now. In which case, KIP-258 should help. > >> > > > > > >>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> Guozhang > >> > > > > > >>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> On Wed, Sep 12, 2018 at 9:20 PM, Jan > Filipiak > >> < > >> > > > > > >>>>>>>>>>>> jan.filip...@trivago.com > >> > > > > > >>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> wrote: > >> > > > > > >>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>> On 11.09.2018 18:00, Adam Bellemare wrote: > >> > > > > > >>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>> Hi Guozhang > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> Current highwater mark implementation > would > >> > grow > >> > > > > > >> endlessly > >> > > > > > >>>>>>> based > >> > > > > > >>>>>>>>>>>> on > >> > > > > > >>>>>>>>>>>>>>>>>>> primary key of original event. It is a > pair > >> of > >> > > > (<this > >> > > > > > >>>>> table > >> > > > > > >>>>>>>>>>>> primary > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>> key>, > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> <highest offset seen for that key>). This is > >> used > >> > > to > >> > > > > > >>>>>>> differentiate > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>> between > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> late arrivals and new updates. My newest > >> proposal > >> > > > would > >> > > > > > be > >> > > > > > >>>>> to > >> > > > > > >>>>>>>>>>>> replace > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>> it > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> with a Windowed state store of Duration N. > >> This > >> > > would > >> > > > > > allow > >> > > > > > >>>>> the > >> > > > > > >>>>>>>>> same > >> > > > > > >>>>>>>>>>>>>>>>>>> behaviour, but cap the size based on time. > >> This > >> > > > > should > >> > > > > > >>>>> allow > >> > > > > > >>>>>>> for > >> > > > > > >>>>>>>>>>>> all > >> > > > > > >>>>>>>>>>>>>>>>>>> late-arriving events to be processed, and > >> > should > >> > > be > >> > > > > > >>>>>>> customizable > >> > > > > > >>>>>>>>>>>> by > >> > > > > > >>>>>>>>>>>>>>>>>>> the > >> > > > > > >>>>>>>>>>>>>>>>>>> user to tailor to their own needs (ie: > >> perhaps > >> > > just > >> > > > > 10 > >> > > > > > >>>>>>> minutes > >> > > > > > >>>>>>>>> of > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>> window, > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> or perhaps 7 days...). > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> Hi Adam, using time based retention can do > >> the > >> > > > trick > >> > > > > > >> here. > >> > > > > > >>>>>>> Even > >> > > > > > >>>>>>>>>>>> if I > >> > > > > > >>>>>>>>>>>>>>>>>> would still like to see the automatic > >> > > repartitioning > >> > > > > > >>>>> optional > >> > > > > > >>>>>>>>>>>> since I > >> > > > > > >>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> would > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> just reshuffle again. With windowed store I > >> am a > >> > > > little > >> > > > > > bit > >> > > > > > >>>>>>>>>>>> sceptical > >> > > > > > >>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> about > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> how to determine the window. So esentially > one > >> > > could > >> > > > > run > >> > > > > > >>>>> into > >> > > > > > >>>>>>>>>>>> problems > >> > > > > > >>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> when > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> the rapid change happens near a window > >> border. I > >> > > will > >> > > > > > check > >> > > > > > >>>>> you > >> > > > > > >>>>>>>>>>>>>>>>>> implementation in detail, if its > >> problematic, we > >> > > > could > >> > > > > > >>>>> still > >> > > > > > >>>>>>>>> check > >> > > > > > >>>>>>>>>>>>>>>>>> _all_ > >> > > > > > >>>>>>>>>>>>>>>>>> windows on read with not to bad performance > >> > > impact I > >> > > > > > >> guess. > >> > > > > > >>>>>>> Will > >> > > > > > >>>>>>>>>>>> let > >> > > > > > >>>>>>>>>>>>>>>>>> you > >> > > > > > >>>>>>>>>>>>>>>>>> know if the implementation would be correct > >> as > >> > > is. I > >> > > > > > >>>>> wouldn't > >> > > > > > >>>>>>> not > >> > > > > > >>>>>>>>>>>> like > >> > > > > > >>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> to > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> assume that: offset(A) < offset(B) => > >> > > timestamp(A) < > >> > > > > > >>>>>>>>> timestamp(B). > >> > > > > > >>>>>>>>>>>> I > >> > > > > > >>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> think > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> we can't expect that. > >> > > > > > >>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> @Jan > >> > > > > > >>>>>>>>>>>>>>>>>>> I believe I understand what you mean now - > >> > thanks > >> > > > for > >> > > > > > the > >> > > > > > >>>>>>>>>>>> diagram, it > >> > > > > > >>>>>>>>>>>>>>>>>>> did really help. You are correct that I do > >> not > >> > > have > >> > > > > the > >> > > > > > >>>>>>> original > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>> primary > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> key available, and I can see that if it was > >> > > available > >> > > > > > then > >> > > > > > >>>>> you > >> > > > > > >>>>>>>>>>>> would be > >> > > > > > >>>>>>>>>>>>>>>>>>> able to add and remove events from the > Map. > >> > That > >> > > > > being > >> > > > > > >>>>> said, > >> > > > > > >>>>>>> I > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>> encourage > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> you to finish your diagrams / charts just > for > >> > > clarity > >> > > > > for > >> > > > > > >>>>>>> everyone > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>> else. > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> Yeah 100%, this giphy thing is just really > >> hard > >> > > > work. > >> > > > > > But > >> > > > > > >>>>> I > >> > > > > > >>>>>>>>>>>> understand > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>> the benefits for the rest. Sorry about the > >> > > original > >> > > > > > >> primary > >> > > > > > >>>>>>> key, > >> > > > > > >>>>>>>>> We > >> > > > > > >>>>>>>>>>>>>>>>>> have > >> > > > > > >>>>>>>>>>>>>>>>>> join and Group by implemented our own in > PAPI > >> > and > >> > > > > > >> basically > >> > > > > > >>>>>>> not > >> > > > > > >>>>>>>>>>>> using > >> > > > > > >>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> any > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> DSL (Just the abstraction). Completely > missed > >> > that > >> > > in > >> > > > > > >>>>> original > >> > > > > > >>>>>>> DSL > >> > > > > > >>>>>>>>>>>> its > >> > > > > > >>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> not > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> there and just assumed it. total brain mess > >> up on > >> > > my > >> > > > > end. > >> > > > > > >>>>> Will > >> > > > > > >>>>>>>>>>>> finish > >> > > > > > >>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> the > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> chart as soon as i get a quite evening this > >> week. > >> > > > > > >>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>> My follow up question for you is, won't the > >> Map > >> > > stay > >> > > > > > >> inside > >> > > > > > >>>>>>> the > >> > > > > > >>>>>>>>>>>> State > >> > > > > > >>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> Store indefinitely after all of the > changes > >> > have > >> > > > > > >>>>> propagated? > >> > > > > > >>>>>>>>> Isn't > >> > > > > > >>>>>>>>>>>>>>>>>>> this > >> > > > > > >>>>>>>>>>>>>>>>>>> effectively the same as a highwater mark > >> state > >> > > > store? > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> Thing is that if the map is empty, > >> substractor > >> > is > >> > > > > gonna > >> > > > > > >>>>>>> return > >> > > > > > >>>>>>>>>>>> `null` > >> > > > > > >>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> and > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> the key is removed from the keyspace. But > >> there > >> > is > >> > > > > going > >> > > > > > to > >> > > > > > >>>>> be > >> > > > > > >>>>>>> a > >> > > > > > >>>>>>>>>>>> store > >> > > > > > >>>>>>>>>>>>>>>>>> 100%, the good thing is that I can use this > >> > store > >> > > > > > directly > >> > > > > > >>>>> for > >> > > > > > >>>>>>>>>>>>>>>>>> materialize() / enableSendingOldValues() > is a > >> > > > regular > >> > > > > > >>>>> store, > >> > > > > > >>>>>>>>>>>> satisfying > >> > > > > > >>>>>>>>>>>>>>>>>> all gurantees needed for further groupby / > >> join. > >> > > The > >> > > > > > >>>>> Windowed > >> > > > > > >>>>>>>>>>>> store is > >> > > > > > >>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> not > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> keeping the values, so for the next > statefull > >> > > > operation > >> > > > > > we > >> > > > > > >>>>>>> would > >> > > > > > >>>>>>>>>>>>>>>>>> need to instantiate an extra store. or we > >> have > >> > the > >> > > > > > window > >> > > > > > >>>>>>> store > >> > > > > > >>>>>>>>>>>> also > >> > > > > > >>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> have > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> the values then. > >> > > > > > >>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>> Long story short. if we can flip in a > custom > >> > group > >> > > > by > >> > > > > > >>>>> before > >> > > > > > >>>>>>>>>>>>>>>>>> repartitioning to the original primary key > i > >> > think > >> > > > it > >> > > > > > >> would > >> > > > > > >>>>>>> help > >> > > > > > >>>>>>>>>>>> the > >> > > > > > >>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> users > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> big time in building efficient apps. Given > the > >> > > > original > >> > > > > > >>>>> primary > >> > > > > > >>>>>>>>> key > >> > > > > > >>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> issue I > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> understand that we do not have a solid > >> foundation > >> > > to > >> > > > > > build > >> > > > > > >>>>> on. > >> > > > > > >>>>>>>>>>>>>>>>>> Leaving primary key carry along to the > user. > >> > very > >> > > > > > >>>>>>> unfortunate. I > >> > > > > > >>>>>>>>>>>> could > >> > > > > > >>>>>>>>>>>>>>>>>> understand the decision goes like that. I > do > >> not > >> > > > think > >> > > > > > its > >> > > > > > >>>>> a > >> > > > > > >>>>>>> good > >> > > > > > >>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> decision. > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> Thanks > >> > > > > > >>>>>>>>>>>>>>>>>>> Adam > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> On Tue, Sep 11, 2018 at 10:07 AM, Prajakta > >> > > Dumbre < > >> > > > > > >>>>>>>>>>>>>>>>>>> dumbreprajakta...@gmail.com <mailto: > >> > > > > > >>>>>>> dumbreprajakta...@gmail.com > >> > > > > > >>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>> wrote: > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> please remove me from this group > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> On Tue, Sep 11, 2018 at 1:29 PM > >> Jan > >> > > > > Filipiak > >> > > > > > >>>>>>>>>>>>>>>>>>> <jan.filip...@trivago.com > >> <mailto: > >> > > > > > >>>>>>>>> jan.filip...@trivago.com > >> > > > > > >>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> wrote: > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > Hi Adam, > >> > > > > > >>>>>>>>>>>>>>>>>>> > > >> > > > > > >>>>>>>>>>>>>>>>>>> > give me some time, will make > >> such a > >> > > > > chart. > >> > > > > > >> last > >> > > > > > >>>>>>> time i > >> > > > > > >>>>>>>>>>>> didn't > >> > > > > > >>>>>>>>>>>>>>>>>>> get along > >> > > > > > >>>>>>>>>>>>>>>>>>> > well with giphy and ruined all > >> your > >> > > > > charts. > >> > > > > > >>>>>>>>>>>>>>>>>>> > Hopefully i can get it done > >> today > >> > > > > > >>>>>>>>>>>>>>>>>>> > > >> > > > > > >>>>>>>>>>>>>>>>>>> > On 08.09.2018 16:00, Adam > >> Bellemare > >> > > > > wrote: > >> > > > > > >>>>>>>>>>>>>>>>>>> > > Hi Jan > >> > > > > > >>>>>>>>>>>>>>>>>>> > > > >> > > > > > >>>>>>>>>>>>>>>>>>> > > I have included a diagram of > >> > what I > >> > > > > > >> attempted > >> > > > > > >>>>> on > >> > > > > > >>>>>>> the > >> > > > > > >>>>>>>>>>>> KIP. > >> > > > > > >>>>>>>>>>>>>>>>>>> > > > >> > > > > > >>>>>>>>>>>>>>>>>>> > > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>> > >> > > > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-213+Su > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>> > >> pport+non-key+joining+in+KTable#KIP-213Supportnon-keyjoining > >> > > > > > >>>>>>>>>>>>>>>>>>> inKTable-GroupBy+Reduce/Aggregate > >> > > > > > >>>>>>>>>>>>>>>>>>> < > >> > > > > > >>>>>>>>>>>> > >> > > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-213+S > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>> > >> upport+non-key+joining+in+KTable#KIP-213Supportnon-keyjoinin > >> > > > > > >>>>>>>>>>>>>>>>>>> ginKTable-GroupBy+Reduce/Aggregate> > >> > > > > > >>>>>>>>>>>>>>>>>>> > > > >> > > > > > >>>>>>>>>>>>>>>>>>> > > I attempted this back at the > >> > start > >> > > of > >> > > > > my > >> > > > > > own > >> > > > > > >>>>>>>>>>>> implementation > >> > > > > > >>>>>>>>>>>>>>>>>>> of > >> > > > > > >>>>>>>>>>>>>>>>>>> this > >> > > > > > >>>>>>>>>>>>>>>>>>> > > solution, and since I could > >> not > >> > get > >> > > > it > >> > > > > to > >> > > > > > >>>>> work I > >> > > > > > >>>>>>> have > >> > > > > > >>>>>>>>>>>> since > >> > > > > > >>>>>>>>>>>>>>>>>>> discarded the > >> > > > > > >>>>>>>>>>>>>>>>>>> > > code. At this point in time, > >> if > >> > you > >> > > > > wish > >> > > > > > to > >> > > > > > >>>>>>> continue > >> > > > > > >>>>>>>>>>>> pursuing > >> > > > > > >>>>>>>>>>>>>>>>>>> for your > >> > > > > > >>>>>>>>>>>>>>>>>>> > > groupBy solution, I ask that > >> you > >> > > > please > >> > > > > > >>>>> create a > >> > > > > > >>>>>>>>>>>> diagram on > >> > > > > > >>>>>>>>>>>>>>>>>>> the KIP > >> > > > > > >>>>>>>>>>>>>>>>>>> > > carefully explaining your > >> > solution. > >> > > > > > Please > >> > > > > > >>>>> feel > >> > > > > > >>>>>>> free > >> > > > > > >>>>>>>>> to > >> > > > > > >>>>>>>>>>>> use > >> > > > > > >>>>>>>>>>>>>>>>>>> the image I > >> > > > > > >>>>>>>>>>>>>>>>>>> > > just posted as a starting > >> point. > >> > I > >> > > am > >> > > > > > having > >> > > > > > >>>>>>> trouble > >> > > > > > >>>>>>>>>>>>>>>>>>> understanding your > >> > > > > > >>>>>>>>>>>>>>>>>>> > > explanations but I think > that > >> a > >> > > > > carefully > >> > > > > > >>>>>>> constructed > >> > > > > > >>>>>>>>>>>> diagram > >> > > > > > >>>>>>>>>>>>>>>>>>> will clear > >> > > > > > >>>>>>>>>>>>>>>>>>> > up > >> > > > > > >>>>>>>>>>>>>>>>>>> > > any misunderstandings. > >> > Alternately, > >> > > > > > please > >> > > > > > >>>>> post a > >> > > > > > >>>>>>>>>>>>>>>>>>> comprehensive PR with > >> > > > > > >>>>>>>>>>>>>>>>>>> > > your solution. I can only > >> guess > >> > at > >> > > > what > >> > > > > > you > >> > > > > > >>>>>>> mean, and > >> > > > > > >>>>>>>>>>>> since I > >> > > > > > >>>>>>>>>>>>>>>>>>> value my > >> > > > > > >>>>>>>>>>>>>>>>>>> > own > >> > > > > > >>>>>>>>>>>>>>>>>>> > > time as much as you value > >> yours, > >> > I > >> > > > > > believe > >> > > > > > >> it > >> > > > > > >>>>> is > >> > > > > > >>>>>>> your > >> > > > > > >>>>>>>>>>>>>>>>>>> responsibility to > >> > > > > > >>>>>>>>>>>>>>>>>>> > > provide an implementation > >> instead > >> > > of > >> > > > me > >> > > > > > >>>>> trying to > >> > > > > > >>>>>>>>> guess. > >> > > > > > >>>>>>>>>>>>>>>>>>> > > > >> > > > > > >>>>>>>>>>>>>>>>>>> > > Adam > >> > > > > > >>>>>>>>>>>>>>>>>>> > > > >> > > > > > >>>>>>>>>>>>>>>>>>> > > > >> > > > > > >>>>>>>>>>>>>>>>>>> > > > >> > > > > > >>>>>>>>>>>>>>>>>>> > > > >> > > > > > >>>>>>>>>>>>>>>>>>> > > > >> > > > > > >>>>>>>>>>>>>>>>>>> > > > >> > > > > > >>>>>>>>>>>>>>>>>>> > > > >> > > > > > >>>>>>>>>>>>>>>>>>> > > > >> > > > > > >>>>>>>>>>>>>>>>>>> > > > >> > > > > > >>>>>>>>>>>>>>>>>>> > > On Sat, Sep 8, 2018 at 8:00 > >> AM, > >> > Jan > >> > > > > > Filipiak > >> > > > > > >>>>>>>>>>>>>>>>>>> <jan.filip...@trivago.com > >> <mailto: > >> > > > > > >>>>>>>>> jan.filip...@trivago.com > >> > > > > > >>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > > wrote: > >> > > > > > >>>>>>>>>>>>>>>>>>> > > > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> Hi James, > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> nice to see you beeing > >> > interested. > >> > > > > kafka > >> > > > > > >>>>>>> streams at > >> > > > > > >>>>>>>>>>>> this > >> > > > > > >>>>>>>>>>>>>>>>>>> point supports > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> all sorts of joins as long > as > >> > both > >> > > > > > streams > >> > > > > > >>>>> have > >> > > > > > >>>>>>> the > >> > > > > > >>>>>>>>>>>> same > >> > > > > > >>>>>>>>>>>>>>>>>>> key. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> Adam is currently > >> implementing a > >> > > > join > >> > > > > > >> where a > >> > > > > > >>>>>>> KTable > >> > > > > > >>>>>>>>>>>> and a > >> > > > > > >>>>>>>>>>>>>>>>>>> KTable can > >> > > > > > >>>>>>>>>>>>>>>>>>> > have > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> a one to many relation ship > >> > (1:n). > >> > > > We > >> > > > > > >> exploit > >> > > > > > >>>>>>> that > >> > > > > > >>>>>>>>>>>> rocksdb > >> > > > > > >>>>>>>>>>>>>>>>>>> is > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>> a > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> > >> datastore that keeps data > >> sorted > >> > (At > >> > > > > least > >> > > > > > >>>>>>> exposes an > >> > > > > > >>>>>>>>>>>> API to > >> > > > > > >>>>>>>>>>>>>>>>>>> access the > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> stored data in a sorted > >> > fashion). > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> I think the technical > caveats > >> > are > >> > > > well > >> > > > > > >>>>>>> understood > >> > > > > > >>>>>>>>> now > >> > > > > > >>>>>>>>>>>> and we > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>> are > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> > basically > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> down to philosophy and API > >> > Design > >> > > ( > >> > > > > when > >> > > > > > >> Adam > >> > > > > > >>>>>>> sees > >> > > > > > >>>>>>>>> my > >> > > > > > >>>>>>>>>>>> newest > >> > > > > > >>>>>>>>>>>>>>>>>>> message). > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> I have a lengthy track > >> record of > >> > > > > loosing > >> > > > > > >>>>> those > >> > > > > > >>>>>>> kinda > >> > > > > > >>>>>>>>>>>>>>>>>>> arguments within > >> > > > > > >>>>>>>>>>>>>>>>>>> > the > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> streams community and I > have > >> no > >> > > clue > >> > > > > > why. > >> > > > > > >> So > >> > > > > > >>>>> I > >> > > > > > >>>>>>>>>>>> literally > >> > > > > > >>>>>>>>>>>>>>>>>>> can't wait for > >> > > > > > >>>>>>>>>>>>>>>>>>> > you > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> to churn through this > thread > >> and > >> > > > give > >> > > > > > you > >> > > > > > >>>>>>> opinion on > >> > > > > > >>>>>>>>>>>> how we > >> > > > > > >>>>>>>>>>>>>>>>>>> should > >> > > > > > >>>>>>>>>>>>>>>>>>> > design > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> the return type of the > >> > > oneToManyJoin > >> > > > > and > >> > > > > > >> how > >> > > > > > >>>>>>> many > >> > > > > > >>>>>>>>>>>> power we > >> > > > > > >>>>>>>>>>>>>>>>>>> want to give > >> > > > > > >>>>>>>>>>>>>>>>>>> > to > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> the user vs "simplicity" > >> (where > >> > > > > > simplicity > >> > > > > > >>>>> isn't > >> > > > > > >>>>>>>>>>>> really that > >> > > > > > >>>>>>>>>>>>>>>>>>> as users > >> > > > > > >>>>>>>>>>>>>>>>>>> > still > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> need to understand it I > >> argue) > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> waiting for you to join in > on > >> > the > >> > > > > > >> discussion > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> Best Jan > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> On 07.09.2018 15:49, James > >> Kwan > >> > > > wrote: > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>> I am new to this group > and I > >> > > found > >> > > > > this > >> > > > > > >>>>> subject > >> > > > > > >>>>>>>>>>>>>>>>>>> interesting. Sounds > >> > > > > > >>>>>>>>>>>>>>>>>>> > like > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>> you guys want to > implement a > >> > join > >> > > > > > table of > >> > > > > > >>>>> two > >> > > > > > >>>>>>>>>>>> streams? Is > >> > > > > > >>>>>>>>>>>>>>>>>>> there > >> > > > > > >>>>>>>>>>>>>>>>>>> > somewhere > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>> I can see the original > >> > > requirement > >> > > > or > >> > > > > > >>>>> proposal? > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>> On Sep 7, 2018, at 8:13 > AM, > >> Jan > >> > > > > > Filipiak > >> > > > > > >>>>>>>>>>>>>>>>>>> <jan.filip...@trivago.com > >> <mailto: > >> > > > > > >>>>>>>>> jan.filip...@trivago.com > >> > > > > > >>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> wrote: > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> On 05.09.2018 22:17, Adam > >> > > > Bellemare > >> > > > > > >> wrote: > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> I'm currently testing > >> using a > >> > > > > > Windowed > >> > > > > > >>>>> Store > >> > > > > > >>>>>>> to > >> > > > > > >>>>>>>>>>>> store the > >> > > > > > >>>>>>>>>>>>>>>>>>> highwater > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> mark. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> By all indications this > >> > should > >> > > > work > >> > > > > > >> fine, > >> > > > > > >>>>>>> with > >> > > > > > >>>>>>>>> the > >> > > > > > >>>>>>>>>>>> caveat > >> > > > > > >>>>>>>>>>>>>>>>>>> being that > >> > > > > > >>>>>>>>>>>>>>>>>>> > it > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> can > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> only resolve > out-of-order > >> > > arrival > >> > > > > > for up > >> > > > > > >>>>> to > >> > > > > > >>>>>>> the > >> > > > > > >>>>>>>>>>>> size of > >> > > > > > >>>>>>>>>>>>>>>>>>> the window > >> > > > > > >>>>>>>>>>>>>>>>>>> > (ie: > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> 24h, 72h, etc). This > would > >> > > remove > >> > > > > the > >> > > > > > >>>>>>> possibility > >> > > > > > >>>>>>>>>>>> of it > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>> being > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> > unbounded > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> in > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> size. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> With regards to Jan's > >> > > > suggestion, I > >> > > > > > >>>>> believe > >> > > > > > >>>>>>> this > >> > > > > > >>>>>>>>> is > >> > > > > > >>>>>>>>>>>> where > >> > > > > > >>>>>>>>>>>>>>>>>>> we will > >> > > > > > >>>>>>>>>>>>>>>>>>> > have > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> to > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> remain in disagreement. > >> > While I > >> > > > do > >> > > > > > not > >> > > > > > >>>>>>> disagree > >> > > > > > >>>>>>>>>>>> with your > >> > > > > > >>>>>>>>>>>>>>>>>>> statement > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> about > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> there likely to be > >> additional > >> > > > joins > >> > > > > > done > >> > > > > > >>>>> in a > >> > > > > > >>>>>>>>>>>> real-world > >> > > > > > >>>>>>>>>>>>>>>>>>> workflow, I > >> > > > > > >>>>>>>>>>>>>>>>>>> > do > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> not > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> see how you can > >> conclusively > >> > > deal > >> > > > > > with > >> > > > > > >>>>>>>>> out-of-order > >> > > > > > >>>>>>>>>>>>>>>>>>> arrival > >> > > > > > >>>>>>>>>>>>>>>>>>> of > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> foreign-key > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> changes and subsequent > >> > joins. I > >> > > > > have > >> > > > > > >>>>>>> attempted > >> > > > > > >>>>>>>>> what > >> > > > > > >>>>>>>>>>>> I > >> > > > > > >>>>>>>>>>>>>>>>>>> think you have > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> proposed (without a > >> > high-water, > >> > > > > using > >> > > > > > >>>>>>> groupBy and > >> > > > > > >>>>>>>>>>>> reduce) > >> > > > > > >>>>>>>>>>>>>>>>>>> and found > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> that if > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> the foreign key changes > >> too > >> > > > > quickly, > >> > > > > > or > >> > > > > > >>>>> the > >> > > > > > >>>>>>> load > >> > > > > > >>>>>>>>> on > >> > > > > > >>>>>>>>>>>> a > >> > > > > > >>>>>>>>>>>>>>>>>>> stream thread > >> > > > > > >>>>>>>>>>>>>>>>>>> > is > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> too > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> high, the joined > messages > >> > will > >> > > > > arrive > >> > > > > > >>>>>>>>> out-of-order > >> > > > > > >>>>>>>>>>>> and be > >> > > > > > >>>>>>>>>>>>>>>>>>> incorrectly > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> propagated, such that an > >> > > > > intermediate > >> > > > > > >>>>> event > >> > > > > > >>>>>>> is > >> > > > > > >>>>>>>>>>>>>>>>>>> represented > >> > > > > > >>>>>>>>>>>>>>>>>>> as the > >> > > > > > >>>>>>>>>>>>>>>>>>> > final > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> event. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> Can you shed some light > on > >> > your > >> > > > > > groupBy > >> > > > > > >>>>>>>>>>>> implementation. > >> > > > > > >>>>>>>>>>>>>>>>>>> There must be > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> some sort of flaw in it. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> I have a suspicion where > it > >> > is, > >> > > I > >> > > > > > would > >> > > > > > >>>>> just > >> > > > > > >>>>>>> like > >> > > > > > >>>>>>>>> to > >> > > > > > >>>>>>>>>>>>>>>>>>> confirm. The idea > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> is bullet proof and it > >> must be > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> an implementation mess > up. > >> I > >> > > would > >> > > > > > like > >> > > > > > >> to > >> > > > > > >>>>>>> clarify > >> > > > > > >>>>>>>>>>>> before > >> > > > > > >>>>>>>>>>>>>>>>>>> we draw a > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> conclusion. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> Repartitioning the > >> > scattered > >> > > > > events > >> > > > > > >>>>> back to > >> > > > > > >>>>>>>>> their > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>> original > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> > >>>>> partitions is the only > way I > >> > know > >> > > > how > >> > > > > > to > >> > > > > > >>>>>>>>> conclusively > >> > > > > > >>>>>>>>>>>> deal > >> > > > > > >>>>>>>>>>>>>>>>>>> with > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> out-of-order events in a > >> > given > >> > > > time > >> > > > > > >> frame, > >> > > > > > >>>>>>> and to > >> > > > > > >>>>>>>>>>>> ensure > >> > > > > > >>>>>>>>>>>>>>>>>>> that the > >> > > > > > >>>>>>>>>>>>>>>>>>> > data > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> is > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> eventually consistent > with > >> > the > >> > > > > input > >> > > > > > >>>>> events. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> If you have some code to > >> > share > >> > > > that > >> > > > > > >>>>>>> illustrates > >> > > > > > >>>>>>>>> your > >> > > > > > >>>>>>>>>>>>>>>>>>> approach, I > >> > > > > > >>>>>>>>>>>>>>>>>>> > would > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> be > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> very grateful as it > would > >> > > remove > >> > > > > any > >> > > > > > >>>>>>>>>>>> misunderstandings > >> > > > > > >>>>>>>>>>>>>>>>>>> that I may > >> > > > > > >>>>>>>>>>>>>>>>>>> > have. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> ah okay you were looking > >> for > >> > my > >> > > > > code. > >> > > > > > I > >> > > > > > >>>>> don't > >> > > > > > >>>>>>> have > >> > > > > > >>>>>>>>>>>>>>>>>>> something easily > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> readable here as its > >> bloated > >> > > with > >> > > > > > >>>>> OO-patterns. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> its anyhow trivial: > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> @Override > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> public T apply(K > >> aggKey, > >> > V > >> > > > > > value, T > >> > > > > > >>>>>>>>> aggregate) > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> { > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> Map<U, V> > >> > > > > currentStateAsMap = > >> > > > > > >>>>>>>>>>>> asMap(aggregate); > >> > > > > > >>>>>>>>>>>>>>>>>>> << > >> > > > > > >>>>>>>>>>>>>>>>>>> imaginary > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> U toModifyKey = > >> > > > > > >>>>> mapper.apply(value); > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> << this is > the > >> > > place > >> > > > > > where > >> > > > > > >>>>> people > >> > > > > > >>>>>>>>>>>> actually > >> > > > > > >>>>>>>>>>>>>>>>>>> gonna have > >> > > > > > >>>>>>>>>>>>>>>>>>> > issues > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> and why you probably > >> couldn't > >> > do > >> > > > it. > >> > > > > > we > >> > > > > > >>>>> would > >> > > > > > >>>>>>> need > >> > > > > > >>>>>>>>>>>> to find > >> > > > > > >>>>>>>>>>>>>>>>>>> a solution > >> > > > > > >>>>>>>>>>>>>>>>>>> > here. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> I didn't realize that > yet. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> << we > >> propagate > >> > the > >> > > > > > field in > >> > > > > > >>>>> the > >> > > > > > >>>>>>>>>>>> joiner, so > >> > > > > > >>>>>>>>>>>>>>>>>>> that we can > >> > > > > > >>>>>>>>>>>>>>>>>>> > pick > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> it up in an aggregate. > >> > Probably > >> > > > you > >> > > > > > have > >> > > > > > >>>>> not > >> > > > > > >>>>>>>>> thought > >> > > > > > >>>>>>>>>>>> of > >> > > > > > >>>>>>>>>>>>>>>>>>> this in your > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> approach right? > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> << I am very > >> open > >> > > to > >> > > > > > find a > >> > > > > > >>>>>>> generic > >> > > > > > >>>>>>>>>>>> solution > >> > > > > > >>>>>>>>>>>>>>>>>>> here. In my > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> honest opinion this is > >> broken > >> > in > >> > > > > > >>>>>>>>> KTableImpl.GroupBy > >> > > > > > >>>>>>>>>>>> that > >> > > > > > >>>>>>>>>>>>>>>>>>> it > >> > > > > > >>>>>>>>>>>>>>>>>>> looses > >> > > > > > >>>>>>>>>>>>>>>>>>> > the keys > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> and only maintains the > >> > aggregate > >> > > > > key. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> << I > >> abstracted > >> > it > >> > > > away > >> > > > > > back > >> > > > > > >>>>>>> then way > >> > > > > > >>>>>>>>>>>> before > >> > > > > > >>>>>>>>>>>>>>>>>>> i > >> > > > > > >>>>>>>>>>>>>>>>>>> was > >> > > > > > >>>>>>>>>>>>>>>>>>> > thinking > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> of oneToMany join. That > is > >> > why I > >> > > > > > didn't > >> > > > > > >>>>>>> realize > >> > > > > > >>>>>>>>> its > >> > > > > > >>>>>>>>>>>>>>>>>>> significance here. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> << Opinions? > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> for (V m : > >> current) > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> { > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> > >> > > > > currentStateAsMap.put(mapper.apply(m), > >> > > > > > >> m); > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> } > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> if (isAdder) > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> { > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> > >> > > currentStateAsMap.put(toModifyKey, > >> > > > > > >> value); > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> } > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> else > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> { > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> > >> > > > > currentStateAsMap.remove(toModifyKey); > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> > >> > if(currentStateAsMap.isEmpty()){ > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> return > >> null; > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> } > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> } > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> retrun > >> > > > > > >>>>>>> asAggregateType(currentStateAsMap) > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> } > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>> Thanks, > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> Adam > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> On Wed, Sep 5, 2018 at > >> 3:35 > >> > PM, > >> > > > Jan > >> > > > > > >>>>> Filipiak > >> > > > > > >>>>>>> < > >> > > > > > >>>>>>>>>>>>>>>>>>> > jan.filip...@trivago.com > >> <mailto: > >> > > > > > >>>>>>>>> jan.filip...@trivago.com > >> > > > > > >>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> wrote: > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>> Thanks Adam for bringing > >> > > Matthias > >> > > > > to > >> > > > > > >>>>> speed! > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>> about the differences. > I > >> > think > >> > > > > > >> re-keying > >> > > > > > >>>>>>> back > >> > > > > > >>>>>>>>>>>> should be > >> > > > > > >>>>>>>>>>>>>>>>>>> optional at > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>> best. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>> I would say we return a > >> > > > > > KScatteredTable > >> > > > > > >>>>> with > >> > > > > > >>>>>>>>>>>> reshuffle() > >> > > > > > >>>>>>>>>>>>>>>>>>> returning > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>> > >> KTable<originalKey,Joined> > >> > to > >> > > > make > >> > > > > > the > >> > > > > > >>>>>>> backwards > >> > > > > > >>>>>>>>>>>>>>>>>>> repartitioning > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>> optional. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>> I am also in a big > >> favour of > >> > > > doing > >> > > > > > the > >> > > > > > >>>>> out > >> > > > > > >>>>>>> of > >> > > > > > >>>>>>>>> order > >> > > > > > >>>>>>>>>>>>>>>>>>> processing using > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>> group > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>> by instead high water > >> mark > >> > > > > tracking. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>> Just because unbounded > >> > growth > >> > > is > >> > > > > > just > >> > > > > > >>>>> scary > >> > > > > > >>>>>>> + It > >> > > > > > >>>>>>>>>>>> saves > >> > > > > > >>>>>>>>>>>>>>>>>>> us > >> > > > > > >>>>>>>>>>>>>>>>>>> the header > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>> stuff. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>> I think the abstraction > >> of > >> > > > always > >> > > > > > >>>>>>> repartitioning > >> > > > > > >>>>>>>>>>>> back is > >> > > > > > >>>>>>>>>>>>>>>>>>> just not so > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>> strong. Like the work > has > >> > been > >> > > > > done > >> > > > > > >>>>> before > >> > > > > > >>>>>>> we > >> > > > > > >>>>>>>>>>>> partition > >> > > > > > >>>>>>>>>>>>>>>>>>> back and > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>> grouping > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>> by something else > >> afterwards > >> > > is > >> > > > > > really > >> > > > > > >>>>>>> common. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>> On 05.09.2018 13:49, > Adam > >> > > > > Bellemare > >> > > > > > >>>>> wrote: > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>> Hi Matthias > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> Thank you for your > >> > feedback, > >> > > I > >> > > > do > >> > > > > > >>>>>>> appreciate > >> > > > > > >>>>>>>>> it! > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> While name spacing > >> would be > >> > > > > > possible, > >> > > > > > >> it > >> > > > > > >>>>>>> would > >> > > > > > >>>>>>>>>>>> require > >> > > > > > >>>>>>>>>>>>>>>>>>> to > >> > > > > > >>>>>>>>>>>>>>>>>>> > deserialize > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> user headers what > >> implies > >> > a > >> > > > > > runtime > >> > > > > > >>>>>>> overhead. > >> > > > > > >>>>>>>>> I > >> > > > > > >>>>>>>>>>>> would > >> > > > > > >>>>>>>>>>>>>>>>>>> suggest to > >> > > > > > >>>>>>>>>>>>>>>>>>> > no > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> namespace for now to > >> avoid > >> > > the > >> > > > > > >>>>> overhead. > >> > > > > > >>>>>>> If > >> > > > > > >>>>>>>>> this > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>> becomes a > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> > problem in > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> the future, we can > >> still > >> > add > >> > > > > name > >> > > > > > >>>>> spacing > >> > > > > > >>>>>>>>> later > >> > > > > > >>>>>>>>>>>> on. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> Agreed. I will go > with > >> > > using a > >> > > > > > >> reserved > >> > > > > > >>>>>>> string > >> > > > > > >>>>>>>>>>>> and > >> > > > > > >>>>>>>>>>>>>>>>>>> document it. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> My main concern about > >> the > >> > > > design > >> > > > > it > >> > > > > > >> the > >> > > > > > >>>>>>> type of > >> > > > > > >>>>>>>>>>>> the > >> > > > > > >>>>>>>>>>>>>>>>>>> result KTable: > >> > > > > > >>>>>>>>>>>>>>>>>>> > If > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> I > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> understood the > proposal > >> > > > > correctly, > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> In your example, you > >> have > >> > > > table1 > >> > > > > > and > >> > > > > > >>>>> table2 > >> > > > > > >>>>>>>>>>>> swapped. > >> > > > > > >>>>>>>>>>>>>>>>>>> Here is how it > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> works > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> currently: > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> 1) table1 has the > >> records > >> > > that > >> > > > > > contain > >> > > > > > >>>>> the > >> > > > > > >>>>>>>>>>>> foreign key > >> > > > > > >>>>>>>>>>>>>>>>>>> within their > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> value. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> table1 input stream: > >> > > > > > <a,(fk=A,bar=1)>, > >> > > > > > >>>>>>>>>>>>>>>>>>> <b,(fk=A,bar=2)>, > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> <c,(fk=B,bar=3)> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> table2 input stream: > >> <A,X>, > >> > > > <B,Y> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> 2) A Value mapper is > >> > required > >> > > > to > >> > > > > > >> extract > >> > > > > > >>>>>>> the > >> > > > > > >>>>>>>>>>>> foreign > >> > > > > > >>>>>>>>>>>>>>>>>>> key. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> table1 foreign key > >> mapper: > >> > ( > >> > > > > value > >> > > > > > => > >> > > > > > >>>>>>> value.fk > >> > > > > > >>>>>>>>>>>>>>>>>>> <http://value.fk> ) > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> The mapper is applied > to > >> > each > >> > > > > > element > >> > > > > > >> in > >> > > > > > >>>>>>>>> table1, > >> > > > > > >>>>>>>>>>>> and a > >> > > > > > >>>>>>>>>>>>>>>>>>> new combined > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> key is > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> made: > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> table1 mapped: <A-a, > >> > > > > (fk=A,bar=1)>, > >> > > > > > >>>>> <A-b, > >> > > > > > >>>>>>>>>>>>>>>>>>> (fk=A,bar=2)>, > >> > > > > > >>>>>>>>>>>>>>>>>>> <B-c, > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> (fk=B,bar=3)> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> 3) The rekeyed events > >> are > >> > > > > > >> copartitioned > >> > > > > > >>>>>>> with > >> > > > > > >>>>>>>>>>>> table2: > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> a) Stream Thread with > >> > > Partition > >> > > > > 0: > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> RepartitionedTable1: > >> <A-a, > >> > > > > > >>>>> (fk=A,bar=1)>, > >> > > > > > >>>>>>> <A-b, > >> > > > > > >>>>>>>>>>>>>>>>>>> (fk=A,bar=2)> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> Table2: <A,X> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> b) Stream Thread with > >> > > Partition > >> > > > > 1: > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> RepartitionedTable1: > >> <B-c, > >> > > > > > >> (fk=B,bar=3)> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> Table2: <B,Y> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> 4) From here, they can > >> be > >> > > > joined > >> > > > > > >>>>> together > >> > > > > > >>>>>>>>> locally > >> > > > > > >>>>>>>>>>>> by > >> > > > > > >>>>>>>>>>>>>>>>>>> applying the > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> joiner > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> function. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> At this point, Jan's > >> design > >> > > and > >> > > > > my > >> > > > > > >>>>> design > >> > > > > > >>>>>>>>>>>> deviate. My > >> > > > > > >>>>>>>>>>>>>>>>>>> design goes > >> > > > > > >>>>>>>>>>>>>>>>>>> > on > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> to > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> repartition the data > >> > > post-join > >> > > > > and > >> > > > > > >>>>> resolve > >> > > > > > >>>>>>>>>>>> out-of-order > >> > > > > > >>>>>>>>>>>>>>>>>>> arrival of > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> records, > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> finally returning the > >> data > >> > > > keyed > >> > > > > > just > >> > > > > > >>>>> the > >> > > > > > >>>>>>>>>>>> original key. > >> > > > > > >>>>>>>>>>>>>>>>>>> I do not > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> expose > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> the > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> CombinedKey or any of > >> the > >> > > > > internals > >> > > > > > >>>>>>> outside of > >> > > > > > >>>>>>>>> the > >> > > > > > >>>>>>>>>>>>>>>>>>> joinOnForeignKey > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> function. This does > make > >> > for > >> > > > > larger > >> > > > > > >>>>>>> footprint, > >> > > > > > >>>>>>>>>>>> but it > >> > > > > > >>>>>>>>>>>>>>>>>>> removes all > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> agency > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> for resolving > >> out-of-order > >> > > > > arrivals > >> > > > > > >> and > >> > > > > > >>>>>>>>> handling > >> > > > > > >>>>>>>>>>>>>>>>>>> CombinedKeys from > >> > > > > > >>>>>>>>>>>>>>>>>>> > the > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> user. I believe that > >> this > >> > > makes > >> > > > > the > >> > > > > > >>>>>>> function > >> > > > > > >>>>>>>>> much > >> > > > > > >>>>>>>>>>>>>>>>>>> easier > >> > > > > > >>>>>>>>>>>>>>>>>>> to use. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> Let me know if this > >> helps > >> > > > resolve > >> > > > > > your > >> > > > > > >>>>>>>>> questions, > >> > > > > > >>>>>>>>>>>> and > >> > > > > > >>>>>>>>>>>>>>>>>>> please feel > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> free to > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> add anything else on > >> your > >> > > mind. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> Thanks again, > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> Adam > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> On Tue, Sep 4, 2018 at > >> 8:36 > >> > > PM, > >> > > > > > >>>>> Matthias J. > >> > > > > > >>>>>>>>> Sax < > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> matth...@confluent.io > >> > > <mailto: > >> > > > > > >>>>>>>>>>>> matth...@confluent.io>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> wrote: > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> Hi, > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> I am just catching up > >> on > >> > > this > >> > > > > > >> thread. I > >> > > > > > >>>>>>> did > >> > > > > > >>>>>>>>> not > >> > > > > > >>>>>>>>>>>> read > >> > > > > > >>>>>>>>>>>>>>>>>>> everything so > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> far, > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> but want to share > >> couple > >> > of > >> > > > > > initial > >> > > > > > >>>>>>> thoughts: > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> Headers: I think > there > >> is > >> > a > >> > > > > > >> fundamental > >> > > > > > >>>>>>>>>>>> difference > >> > > > > > >>>>>>>>>>>>>>>>>>> between header > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> usage > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> in this KIP and > KP-258. > >> > For > >> > > > 258, > >> > > > > > we > >> > > > > > >> add > >> > > > > > >>>>>>>>> headers > >> > > > > > >>>>>>>>>>>> to > >> > > > > > >>>>>>>>>>>>>>>>>>> changelog topic > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> that > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> are owned by Kafka > >> Streams > >> > > and > >> > > > > > nobody > >> > > > > > >>>>>>> else is > >> > > > > > >>>>>>>>>>>> supposed > >> > > > > > >>>>>>>>>>>>>>>>>>> to write > >> > > > > > >>>>>>>>>>>>>>>>>>> > into > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> them. In fact, no > user > >> > > header > >> > > > > are > >> > > > > > >>>>> written > >> > > > > > >>>>>>> into > >> > > > > > >>>>>>>>>>>> the > >> > > > > > >>>>>>>>>>>>>>>>>>> changelog topic > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> and > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> thus, there are not > >> > > conflicts. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> Nevertheless, I don't > >> see > >> > a > >> > > > big > >> > > > > > issue > >> > > > > > >>>>> with > >> > > > > > >>>>>>>>> using > >> > > > > > >>>>>>>>>>>>>>>>>>> headers within > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> Streams. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> As long as we > document > >> it, > >> > > we > >> > > > > can > >> > > > > > >> have > >> > > > > > >>>>>>> some > >> > > > > > >>>>>>>>>>>> "reserved" > >> > > > > > >>>>>>>>>>>>>>>>>>> header keys > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> and > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> users are not allowed > >> to > >> > use > >> > > > > when > >> > > > > > >>>>>>> processing > >> > > > > > >>>>>>>>>>>> data with > >> > > > > > >>>>>>>>>>>>>>>>>>> Kafka > >> > > > > > >>>>>>>>>>>>>>>>>>> > Streams. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> IMHO, this should be > >> ok. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> I think there is a > safe > >> > way > >> > > to > >> > > > > > avoid > >> > > > > > >>>>>>>>> conflicts, > >> > > > > > >>>>>>>>>>>> since > >> > > > > > >>>>>>>>>>>>>>>>>>> these > >> > > > > > >>>>>>>>>>>>>>>>>>> > headers > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> are > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>> only needed in > >> internal > >> > > > topics > >> > > > > (I > >> > > > > > >>>>> think): > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>> For internal and > >> > changelog > >> > > > > > topics, > >> > > > > > >> we > >> > > > > > >>>>> can > >> > > > > > >>>>>>>>>>>> namespace > >> > > > > > >>>>>>>>>>>>>>>>>>> all headers: > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>> * user-defined > headers > >> > are > >> > > > > > >> namespaced > >> > > > > > >>>>> as > >> > > > > > >>>>>>>>>>>> "external." > >> > > > > > >>>>>>>>>>>>>>>>>>> + > >> > > > > > >>>>>>>>>>>>>>>>>>> headerKey > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>> * internal headers > are > >> > > > > > namespaced as > >> > > > > > >>>>>>>>>>>> "internal." + > >> > > > > > >>>>>>>>>>>>>>>>>>> headerKey > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>> While name spacing > >> would > >> > be > >> > > > > > >> possible, > >> > > > > > >>>>> it > >> > > > > > >>>>>>>>> would > >> > > > > > >>>>>>>>>>>>>>>>>>> require > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>> to > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> > >>>>>>>> deserialize > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> user headers what > >> implies > >> > a > >> > > > > > runtime > >> > > > > > >>>>>>> overhead. > >> > > > > > >>>>>>>>> I > >> > > > > > >>>>>>>>>>>> would > >> > > > > > >>>>>>>>>>>>>>>>>>> suggest to > >> > > > > > >>>>>>>>>>>>>>>>>>> > no > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> namespace for now to > >> avoid > >> > > the > >> > > > > > >>>>> overhead. > >> > > > > > >>>>>>> If > >> > > > > > >>>>>>>>> this > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>> becomes a > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> > problem in > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> the future, we can > >> still > >> > add > >> > > > > name > >> > > > > > >>>>> spacing > >> > > > > > >>>>>>>>> later > >> > > > > > >>>>>>>>>>>> on. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> My main concern about > >> the > >> > > > design > >> > > > > > it > >> > > > > > >> the > >> > > > > > >>>>>>> type > >> > > > > > >>>>>>>>> of > >> > > > > > >>>>>>>>>>>> the > >> > > > > > >>>>>>>>>>>>>>>>>>> result KTable: > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> If I > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> understood the > proposal > >> > > > > correctly, > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> KTable<K1,V1> table1 > = > >> ... > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> KTable<K2,V2> table2 > = > >> ... > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> KTable<K1,V3> > >> joinedTable > >> > = > >> > > > > > >>>>>>>>>>>> table1.join(table2,...); > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> implies that the > >> > > `joinedTable` > >> > > > > has > >> > > > > > >> the > >> > > > > > >>>>>>> same > >> > > > > > >>>>>>>>> key > >> > > > > > >>>>>>>>>>>> as the > >> > > > > > >>>>>>>>>>>>>>>>>>> left input > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> table. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> IMHO, this does not > >> work > >> > > > because > >> > > > > > if > >> > > > > > >>>>> table2 > >> > > > > > >>>>>>>>>>>> contains > >> > > > > > >>>>>>>>>>>>>>>>>>> multiple rows > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> that > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> join with a record in > >> > table1 > >> > > > > > (what is > >> > > > > > >>>>> the > >> > > > > > >>>>>>> main > >> > > > > > >>>>>>>>>>>> purpose > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>> of > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> a > >> > > > > > >>>>>>>>>>>>>>>>>>> > foreign > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> key > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> join), the result > table > >> > > would > >> > > > > only > >> > > > > > >>>>>>> contain a > >> > > > > > >>>>>>>>>>>> single > >> > > > > > >>>>>>>>>>>>>>>>>>> join result, > >> > > > > > >>>>>>>>>>>>>>>>>>> > but > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> not > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> multiple. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> Example: > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> table1 input stream: > >> <A,X> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> table2 input stream: > >> > > > <a,(A,1)>, > >> > > > > > >>>>> <b,(A,2)> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> We use table2 value a > >> > > foreign > >> > > > > key > >> > > > > > to > >> > > > > > >>>>>>> table1 > >> > > > > > >>>>>>>>> key > >> > > > > > >>>>>>>>>>>> (ie, > >> > > > > > >>>>>>>>>>>>>>>>>>> "A" joins). > >> > > > > > >>>>>>>>>>>>>>>>>>> > If > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> the > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> result key is the > same > >> key > >> > > as > >> > > > > key > >> > > > > > of > >> > > > > > >>>>>>> table1, > >> > > > > > >>>>>>>>> this > >> > > > > > >>>>>>>>>>>>>>>>>>> implies that the > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> result can either be > >> <A, > >> > > > > > join(X,1)> > >> > > > > > >> or > >> > > > > > >>>>> <A, > >> > > > > > >>>>>>>>>>>> join(X,2)> > >> > > > > > >>>>>>>>>>>>>>>>>>> but not > >> > > > > > >>>>>>>>>>>>>>>>>>> > both. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> Because the share the > >> same > >> > > > key, > >> > > > > > >>>>> whatever > >> > > > > > >>>>>>>>> result > >> > > > > > >>>>>>>>>>>> record > >> > > > > > >>>>>>>>>>>>>>>>>>> we emit > >> > > > > > >>>>>>>>>>>>>>>>>>> > later, > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> overwrite the > previous > >> > > result. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> This is the reason > why > >> Jan > >> > > > > > originally > >> > > > > > >>>>>>> proposed > >> > > > > > >>>>>>>>>>>> to use > >> > > > > > >>>>>>>>>>>>>>>>>>> a > >> > > > > > >>>>>>>>>>>>>>>>>>> > combination > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> of > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> both primary keys of > >> the > >> > > input > >> > > > > > tables > >> > > > > > >>>>> as > >> > > > > > >>>>>>> key > >> > > > > > >>>>>>>>> of > >> > > > > > >>>>>>>>>>>> the > >> > > > > > >>>>>>>>>>>>>>>>>>> output table. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> This > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> makes the keys of the > >> > output > >> > > > > table > >> > > > > > >>>>> unique > >> > > > > > >>>>>>> and > >> > > > > > >>>>>>>>> we > >> > > > > > >>>>>>>>>>>> can > >> > > > > > >>>>>>>>>>>>>>>>>>> store both in > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> the > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> output table: > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> Result would be <A-a, > >> > > > > join(X,1)>, > >> > > > > > >> <A-b, > >> > > > > > >>>>>>>>>>>> join(X,2)> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> Thoughts? > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> -Matthias > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> On 9/4/18 1:36 PM, > Jan > >> > > > Filipiak > >> > > > > > >> wrote: > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>> Just on remark here. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>> The high-watermark > >> could > >> > be > >> > > > > > >>>>> disregarded. > >> > > > > > >>>>>>> The > >> > > > > > >>>>>>>>>>>> decision > >> > > > > > >>>>>>>>>>>>>>>>>>> about the > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>> forward > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>> depends on the size > of > >> > the > >> > > > > > >> aggregated > >> > > > > > >>>>>>> map. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>> Only 1 element long > >> maps > >> > > > would > >> > > > > be > >> > > > > > >>>>>>> unpacked > >> > > > > > >>>>>>>>> and > >> > > > > > >>>>>>>>>>>>>>>>>>> forwarded. 0 > >> > > > > > >>>>>>>>>>>>>>>>>>> > element > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>> maps > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>> would be published > as > >> > > delete. > >> > > > > Any > >> > > > > > >>>>> other > >> > > > > > >>>>>>> count > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>> of map entries is in > >> > > "waiting > >> > > > > for > >> > > > > > >>>>> correct > >> > > > > > >>>>>>>>>>>> deletes to > >> > > > > > >>>>>>>>>>>>>>>>>>> > arrive"-state. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>> On 04.09.2018 21:29, > >> Adam > >> > > > > > Bellemare > >> > > > > > >>>>>>> wrote: > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>> It does look like I > >> could > >> > > > > replace > >> > > > > > >> the > >> > > > > > >>>>>>> second > >> > > > > > >>>>>>>>>>>>>>>>>>> repartition store > >> > > > > > >>>>>>>>>>>>>>>>>>> > and > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>> highwater store > with > >> a > >> > > > groupBy > >> > > > > > and > >> > > > > > >>>>>>> reduce. > >> > > > > > >>>>>>>>>>>> However, > >> > > > > > >>>>>>>>>>>>>>>>>>> it looks > >> > > > > > >>>>>>>>>>>>>>>>>>> > like > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>> I > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>> would > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>> still need to store > >> the > >> > > > > > highwater > >> > > > > > >>>>> value > >> > > > > > >>>>>>>>> within > >> > > > > > >>>>>>>>>>>> the > >> > > > > > >>>>>>>>>>>>>>>>>>> materialized > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>> store, > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>> to > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>> compare the arrival > of > >> > > > > > out-of-order > >> > > > > > >>>>>>> records > >> > > > > > >>>>>>>>>>>> (assuming > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>> my > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> > >>>>>>>>> understanding > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>> of > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>> THIS is correct...). > >> This > >> > > in > >> > > > > > effect > >> > > > > > >> is > >> > > > > > >>>>>>> the > >> > > > > > >>>>>>>>> same > >> > > > > > >>>>>>>>>>>> as > >> > > > > > >>>>>>>>>>>>>>>>>>> the > >> > > > > > >>>>>>>>>>>>>>>>>>> design I > >> > > > > > >>>>>>>>>>>>>>>>>>> > have > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>> now, > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>> just with the two > >> tables > >> > > > merged > >> > > > > > >>>>> together. > >> > > > > > >>>>>>>>>>>>>>>>>>> > >>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > > >> > > > > > >>>>>>>>>>>>>>>>>>> > > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> -- > >> > > > > > >>>>>>>>>>>>>>>>> -- Guozhang > >> > > > > > >>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>> -- > >> > > > > > >>>>>>>>>>>>>>>> -- Guozhang > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>>> > >> > > > > > >>>>>>>>>>>> > >> > > > > > >>>>>>>>>>> > >> > > > > > >>>>>>>>>> > >> > > > > > >>>>>>>>> > >> > > > > > >>>>>>>> > >> > > > > > >>>>>>> > >> > > > > > >>>>>> > >> > > > > > >>>>> > >> > > > > > >>>> > >> > > > > > >>> > >> > > > > > >> > >> > > > > > > > >> > > > > > > > >> > > > > > > >> > > > > > >> > > > > >> > > > >> > > >> > > >> > -- > >> > -- Guozhang > >> > > >> > > > > -- > -- Guozhang >
