Hi John, Sounds goods. It looks like we are close to wrapping things up. If there isn't any other revisions which needs to be made. (If so, please comment in the thread) I will start the voting process this Thursday (Pacific Standard Time).
Cheers, Richard On Tue, Feb 25, 2020 at 11:59 AM John Roesler <vvcep...@apache.org> wrote: > Hi Richard, > > Sorry for the slow reply. I actually think we should avoid checking > equals() for now. Your reasoning is good, but the truth is that > depending on the implementation of equals() is non-trivial, > semantically, and (though I proposed it before), I'm not convinced > it's worth the risk. Much better to start with exactly one kind of > "idempotence detection". > > Even if someone does update their serdes, we know that the new > serde would still be able to _de_serialize the old format, or the whole > app would break. The situation is that the new result gets encoded > in the new binary format, which means we don't detect an idempotent > update for what it is. In this case, we'd write the new binary format to > disk and the changelog, and forward it downstream. However, we only > do this once. Now that the binary format for that record has been updated, > we would correctly detect idempotence of any subsequent updates. > > Plus, we would still be able to filter out idempotent updates in > repartition > sinks, since for those, we use the new serde to serialize both the "old" > and > "new" result. > > It's certainly a good observation, but I think we can just make a note of > it > in "rejected alternatives" for now, and plan to refine it later, if it does > pose a big performance problem. > > Thanks! > -John > > On Sat, Feb 22, 2020, at 18:14, Richard Yu wrote: > > Hi all, > > > > Updated the KIP. > > > > Just a question: do you think it would be a good idea if we check for > both > > Object#equals() and binary equality? > > Because there might be some subtle changes in the serialization (for > > example, if the user decides to upgrade their serialization procedure to > a > > new one), but the underlying values of the result might be the same. > > (therefore equals() might return true) > > > > Do you think this would be plausible? > > > > Cheers, > > Richard > > > > On Fri, Feb 21, 2020 at 2:37 PM Richard Yu <yohan.richard...@gmail.com> > > wrote: > > > > > Hello, > > > > > > Just to make some updates. I changed the name of the metric so that it > was > > > more in line with usual Kafka naming conventions for metrics / sensors. > > > Below is the updated description of the metric: > > > > > > dropped-idempotent-updates : (Level 2 - Per Task) DEBUG (rate | total) > > > > > > Description: This metric will record the number of updates that have > been > > > dropped since they are essentially re-performing an earlier operation. > > > > > > Note: > > > > > > - The rate option indicates the ratio of records dropped to actual > > > volume of records passing through the task. > > > - The total option will just give a raw count of the number of > records > > > dropped. > > > > > > > > > I hope that this is more on point. > > > > > > Best, > > > Richard > > > > > > On Fri, Feb 21, 2020 at 2:20 PM Richard Yu <yohan.richard...@gmail.com > > > > > wrote: > > > > > >> Hi all, > > >> > > >> Thanks for the clarification. I was just confused a little on what was > > >> going on. > > >> > > >> So I guess then that for the actual proposal. We got the following: > > >> > > >> 1. We check for binary equality, and perform no extra look ups. > > >> 2. Emphasize that this applies only to materialized tables. > > >> 3. We drop aggregation updates if key, value and timestamp is the > same. > > >> > > >> Then that settles the behavior changes. So it looks like the Metric > that > > >> is the only thing that is left. In this case, I think the metric > would be > > >> named the following: IdempotentUpdateMetric. This is mostly off the > top of > > >> my head. So if you think that we change it, feel free to say so. > > >> The metric will report the number of dropped operations inherently. > > >> > > >> It will probably be added as a Sensor, similar to the dropped records > > >> sensor we already have. > > >> > > >> If there isn't anything else, I will probably start the voting process > > >> next week! > > >> > > >> Cheers, > > >> Richard > > >> > > >> > > >> On Fri, Feb 21, 2020 at 11:23 AM John Roesler <vvcep...@apache.org> > > >> wrote: > > >> > > >>> Hi Bruno, > > >>> > > >>> Thanks for the clarification. Indeed, I was thinking two things: > > >>> 1. For the initial implementation, we can just avoid adding any extra > > >>> lookups, but only do the comparison when we already happen to have > > >>> the prior value. > > >>> 2. I think, as a result of the timestamp semantics, we actually _do_ > look > > >>> up the prior value approximately all the time, so the idempotence > check > > >>> should be quite effective. > > >>> > > >>> I think that second point is the same thing you're referring to > > >>> potentially > > >>> being unnecessary. It does mean that we do fetch the whole value in a > > >>> lot of cases where we really only need the timestamp, so it could > > >>> certainly > > >>> be optimized in the future. In that future, we would need to weigh > that > > >>> optimization against losing the idempotence check. But, that's a > problem > > >>> for tomorrow :) > > >>> > > >>> I'm 100% on board with scrutinizing the performance as we implement > > >>> this feature. > > >>> > > >>> Thanks again, > > >>> -John > > >>> > > >>> On Thu, Feb 20, 2020, at 03:25, Bruno Cadonna wrote: > > >>> > Hi John, > > >>> > > > >>> > I am glad to help you with your imagination. With overhead, I > mainly > > >>> > meant the additional lookup into the state store to get the current > > >>> > value, but I see now in the code that we do that lookup anyways > > >>> > (although I think we could avoid that in the cases where we do not > > >>> > need the old value). With or without config, we need to evaluate > the > > >>> > performance benefits of this change, in any case. > > >>> > > > >>> > Best, > > >>> > Bruno > > >>> > > > >>> > On Wed, Feb 19, 2020 at 7:48 PM John Roesler <vvcep...@apache.org> > > >>> wrote: > > >>> > > > > >>> > > Thanks for your remarks, Bruno! > > >>> > > > > >>> > > I'm in favor of standardizing on terminology like "not forwarding > > >>> > > idempotent updates" or "dropping idempotent updates". Maybe we > > >>> > > should make a pass on the KIP and just convert everything to this > > >>> > > phrasing. In retrospect, even the term "emit-on-change" has too > much > > >>> > > semantic baggage, since it implies the semantics from the SECRET > > >>> > > paper, which we don't really want to imply here. > > >>> > > > > >>> > > I'm also in favor of the metric as you propose. > > >>> > > > > >>> > > Likewise with stream aggregations, I was also under the > impression > > >>> > > that we agreed on dropping idempotent updates to the aggregation > > >>> > > result, any time we find that our "new" (key, value, timestamp) > > >>> result > > >>> > > is identical to the prior one. > > >>> > > > > >>> > > Also, I'm +1 on all your recommendations for updating the KIP > > >>> document > > >>> > > for clarity. > > >>> > > > > >>> > > Regarding the opt-out config. Perhaps I'm suffering from a > failure of > > >>> > > imagination, but I don't see how the current proposal could > really > > >>> have > > >>> > > a measurable impact on latency. If all we do is make a single > extra > > >>> pass > > >>> > > to compare two byte arrays for equality, only in the cases where > we > > >>> already > > >>> > > have the byte arrays available, it seems unlikely to measurably > > >>> affect the > > >>> > > processing of non-idempotent updates. It seems guaranteed to > > >>> _decrease_ > > >>> > > the latency of processing idempotent updates, since we get to > skip a > > >>> > > store#put, at least one producer#send, and also all downstream > > >>> processing, > > >>> > > including all the disk and network operations associated with > > >>> downstream > > >>> > > operations. > > >>> > > > > >>> > > It seems like if we're pretty sure this change would only > _help_, we > > >>> shouldn't > > >>> > > introduce the operational burden of an extra configuration. If we > > >>> want to > > >>> > > be more aggressive about dropping idempotent operations in the > > >>> future, > > >>> > > such as depending on equals() or adding a ChangeDetector > interface, > > >>> then > > >>> > > we should consider adding a configuration as part of that future > > >>> work. In > > >>> > > fact, if we add a simple "opt-in/opt-out" switch right now, we > might > > >>> find > > >>> > > that it's actually insufficient for whatever future feature we > might > > >>> propose, > > >>> > > then we have a mess of deprecating the opt-out config and > replacing > > >>> it. > > >>> > > > > >>> > > What do you think? > > >>> > > -John > > >>> > > > > >>> > > On Wed, Feb 19, 2020, at 09:50, Bruno Cadonna wrote: > > >>> > > > Hi all, > > >>> > > > > > >>> > > > Sorry for the late reply! > > >>> > > > > > >>> > > > I am also in favour of baby steps. > > >>> > > > > > >>> > > > I am undecided whether the KIP should contain a opt-out config > or > > >>> not. > > >>> > > > The overhead of emit-on-change might affect latency. For > > >>> applications > > >>> > > > where low latency is crucial and there are not too many > idempotent > > >>> > > > updates, it would be better to fall back to emit-on-update. > > >>> However, > > >>> > > > we do not know how much emit-on-change impacts latency. We > would > > >>> first > > >>> > > > need to benchmark that before we can decide about the > > >>> opt-out-config. > > >>> > > > > > >>> > > > A metric of dropped idempotent updates seems useful to me to be > > >>> > > > informed about potential upstream applications or upstream > > >>> operators > > >>> > > > that produce too many idempotent updates. The KIP should state > the > > >>> > > > name of the metric, its group, its tags, and its recording > level > > >>> (see > > >>> > > > KIP-444 or KIP-471 for examples). I propose DEBUG as reporting > > >>> level. > > >>> > > > > > >>> > > > Richard, what competing proposals for emit-on-change for > > >>> aggregations > > >>> > > > do you mean? I have the feeling that we agreed to get rid of > > >>> > > > idempotent updates if the aggregate is updated with the same > key, > > >>> > > > value, AND timestamp. I am also fine if we do not include this > into > > >>> > > > this KIP (remember: baby steps). > > >>> > > > > > >>> > > > You write that "emit-on-change is more correct". Since we > agreed > > >>> that > > >>> > > > this is an optimization, IMO you cannot argue this way. > > >>> > > > > > >>> > > > Please put "Alternative Approaches" under "Rejected > Alternatives", > > >>> so > > >>> > > > that it becomes clear that we are not going to implement them. > In > > >>> > > > general, I think the KIP needs a bit of clean-up (probably, you > > >>> > > > already planned for it). "Design Reasoning" is a bit of > behavior > > >>> > > > changes, rejected alternatives and duplicates a bit the > content in > > >>> > > > those sections. > > >>> > > > > > >>> > > > I do not like the name "no-op operations" or "no-ops", because > they > > >>> > > > are rather generic. I like more "idempotent updates". > > >>> > > > > > >>> > > > Best, > > >>> > > > Bruno > > >>> > > > > > >>> > > > > > >>> > > > On Tue, Feb 18, 2020 at 7:25 PM Richard Yu < > > >>> yohan.richard...@gmail.com> wrote: > > >>> > > > > > > >>> > > > > Hi all, > > >>> > > > > > > >>> > > > > We are definitely making progress! > > >>> > > > > > > >>> > > > > @John should I emphasize in the proposed behavior changes > that > > >>> we are only > > >>> > > > > doing binary equality checks for stateful operators? > > >>> > > > > It looks like we have come close to finalizing this part of > the > > >>> KIP. (I > > >>> > > > > will note in the KIP that this proposal is intended for > > >>> optimization, not > > >>> > > > > semantics correctness) > > >>> > > > > > > >>> > > > > I do think maybe we still have one other detail we need to > > >>> discuss. So far, > > >>> > > > > there has been quite a bit of back and forth about what the > > >>> behavior of > > >>> > > > > aggregations should look like in emit on change. I have seen > > >>> > > > > multiple competing proposals, so I am not completely certain > > >>> which one we > > >>> > > > > should go with, or how we will be able to compromise in > between > > >>> them. > > >>> > > > > > > >>> > > > > Let me know what your thoughts are on this matter, since we > are > > >>> probably > > >>> > > > > close to wrapping up most other stuff. > > >>> > > > > @Matthias J. Sax <matth...@confluent.io> and @Bruno, see > what > > >>> you think > > >>> > > > > about this. > > >>> > > > > > > >>> > > > > Best, > > >>> > > > > Richard > > >>> > > > > > > >>> > > > > > > >>> > > > > > > >>> > > > > On Tue, Feb 18, 2020 at 9:06 AM John Roesler < > > >>> vvcep...@apache.org> wrote: > > >>> > > > > > > >>> > > > > > Thanks, Matthias! > > >>> > > > > > > > >>> > > > > > Regarding numbers, it would be hard to know how many > > >>> applications > > >>> > > > > > would benefit, since we don't know how many applications > there > > >>> are, > > >>> > > > > > or anything about their data sets or topologies. We could > do a > > >>> survey, > > >>> > > > > > but it seems overkill if we take the conservative approach. > > >>> > > > > > > > >>> > > > > > I have my own practical stream processing experience that > > >>> tells me this > > >>> > > > > > is absolutely critical for any moderate-to-large relational > > >>> stream > > >>> > > > > > processing use cases. I'll leave it to you to decide if you > > >>> find that > > >>> > > > > > convincing, but it's definitely not an _assumption_. I've > also > > >>> heard from > > >>> > > > > > a few Streams users who have already had to implement > their own > > >>> > > > > > noop-suppression transformers in order to get to production > > >>> scale. > > >>> > > > > > > > >>> > > > > > Regardless, it sounds like we can agree on taking an > > >>> opportunistic approach > > >>> > > > > > and targeting the optimization just to use a > binary-equality > > >>> check at > > >>> > > > > > stateful operators. (I'd also suggest in sink nodes, when > we > > >>> are about to > > >>> > > > > > send old and new values, since they are also already > present > > >>> and serialized > > >>> > > > > > at that point.) We could make the KIP even more vague, and > > >>> just say that > > >>> > > > > > we'll drop no-op updates "when possible". > > >>> > > > > > > > >>> > > > > > I'm curious what Bruno and the others think about this. If > it > > >>> seems like > > >>> > > > > > a good starting point, perhaps we could move to a vote soon > > >>> and get to > > >>> > > > > > work on the implementation! > > >>> > > > > > > > >>> > > > > > Thanks, > > >>> > > > > > -John > > >>> > > > > > > > >>> > > > > > On Mon, Feb 17, 2020, at 20:54, Matthias J. Sax wrote: > > >>> > > > > > > Talking about optimizations and reducing downstream load: > > >>> > > > > > > > > >>> > > > > > > Do we actually have any numbers? I have the impression > that > > >>> this KIP is > > >>> > > > > > > more or less build on the _assumption_ that there is a > > >>> problem. Yes, > > >>> > > > > > > there are some use cases that would benefit from this; > But > > >>> how many > > >>> > > > > > > applications would actually benefit? And how much load > > >>> reduction would > > >>> > > > > > > they get? > > >>> > > > > > > > > >>> > > > > > > The simplest approach (following John idea to make baby > > >>> steps) would be > > >>> > > > > > > to apply the emit-on-change pattern only if there is a > > >>> store. For this > > >>> > > > > > > case we need to serialize old and new result anyway and > thus > > >>> a simple > > >>> > > > > > > byte-array comparison is no overhead. > > >>> > > > > > > > > >>> > > > > > > Sending `oldValues` by default would become expensive > > >>> because we would > > >>> > > > > > > need to serialize the recomputed old result, as well as > the > > >>> new result, > > >>> > > > > > > to make the comparison (and we now the serialization is > not > > >>> cheap). We > > >>> > > > > > > are facing a trade-off between CPU overhead and > downstream > > >>> load and I am > > >>> > > > > > > not sure if we should hard code this. My original > argument > > >>> for sending > > >>> > > > > > > `oldValues` was about semantics; but for an > optimization, I > > >>> am not sure > > >>> > > > > > > if this would be the right choice. > > >>> > > > > > > > > >>> > > > > > > For now, users who want to opt-in can force a > > >>> materialization. A > > >>> > > > > > > materialization may be expensive and if we see future > > >>> demand, we could > > >>> > > > > > > still add an option to send `oldValues` instead of > > >>> materialization (this > > >>> > > > > > > would at least save the store overhead). As we consider > the > > >>> KIP an > > >>> > > > > > > optimization, a "config" seems to make sense. > > >>> > > > > > > > > >>> > > > > > > > > >>> > > > > > > -Matthias > > >>> > > > > > > > > >>> > > > > > > > > >>> > > > > > > On 2/17/20 5:21 PM, Richard Yu wrote: > > >>> > > > > > > > Hi John! > > >>> > > > > > > > > > >>> > > > > > > > Thanks for the reply. > > >>> > > > > > > > > > >>> > > > > > > > About the changes we have discussed so far. I think > upon > > >>> further > > >>> > > > > > > > consideration, we have been mostly talking about this > from > > >>> the > > >>> > > > > > perspective > > >>> > > > > > > > that no stop-gap effort is acceptable. However, in > recent > > >>> discussion, > > >>> > > > > > > if we > > >>> > > > > > > > consider optimization, then it appears that the > > >>> perspective I > > >>> > > > > > mentioned no > > >>> > > > > > > > longer applies. After all, we are no longer concerned > so > > >>> much about > > >>> > > > > > > > semantics correctness, then reducing traffic as much as > > >>> possible > > >>> > > > > > without > > >>> > > > > > > > performance tradeoffs. > > >>> > > > > > > > > > >>> > > > > > > > In this case, I think a cache would be a good idea for > > >>> stateless > > >>> > > > > > > > operations. This cache will not be backed by a store > > >>> obviously. We can > > >>> > > > > > > > probably use Kafka's ThreadCache. We should be able to > > >>> catch a large > > >>> > > > > > > > portion of the no-ops if we at least store some > results in > > >>> the cache. > > >>> > > > > > Not > > >>> > > > > > > > all will be caught, but I think the impact will be > > >>> significant. > > >>> > > > > > > > > > >>> > > > > > > > On another note, I think that we should implement > > >>> competing proposals > > >>> > > > > > i.e. > > >>> > > > > > > > one where we forward both old and new values with a > > >>> reasonable > > >>> > > > > > proportion > > >>> > > > > > > > of artificial no-ops (we do not necessarily have to > rely > > >>> on equals so > > >>> > > > > > much > > >>> > > > > > > > as comparing the serialized binary data after the > > >>> operation), and in > > >>> > > > > > > > another scenario, the cache for stateless ops. It > would be > > >>> > > > > > unreasonable if > > >>> > > > > > > > we completely disregard either approach, since they > both > > >>> have merit. > > >>> > > > > > The > > >>> > > > > > > > reason for implementing both is to perform benchmark > tests > > >>> on them, and > > >>> > > > > > > > compare them with the original. This way, we can more > > >>> clearly see what > > >>> > > > > > is > > >>> > > > > > > > the drawbacks and the gains. So far, we have been > > >>> discussing only > > >>> > > > > > > > hypotheticals, and if we continue to do so, I think it > is > > >>> likely no > > >>> > > > > > ground > > >>> > > > > > > > will be gained. > > >>> > > > > > > > > > >>> > > > > > > > After all, what we seek is optimization, and > performance > > >>> benchmarks > > >>> > > > > > > will be > > >>> > > > > > > > mandatory for a KIP of this nature. > > >>> > > > > > > > > > >>> > > > > > > > Hope this helps, > > >>> > > > > > > > Richard > > >>> > > > > > > > > > >>> > > > > > > > > > >>> > > > > > > > > > >>> > > > > > > > > > >>> > > > > > > > > > >>> > > > > > > > > > >>> > > > > > > > On Mon, Feb 17, 2020 at 2:12 PM John Roesler < > > >>> vvcep...@apache.org> > > >>> > > > > > wrote: > > >>> > > > > > > > > > >>> > > > > > > >> Hi again, all, > > >>> > > > > > > >> > > >>> > > > > > > >> Sorry on my part for my silence. > > >>> > > > > > > >> > > >>> > > > > > > >> I've just taken another look over the recent history > of > > >>> this > > >>> > > > > > > discussion. It > > >>> > > > > > > >> seems like the #1 point to clarify (because it affect > > >>> everything > > >>> > > > > > else) is > > >>> > > > > > > >> that, > > >>> > > > > > > >> yes, I was 100% envisioning this as an _optimization_. > > >>> > > > > > > >> > > >>> > > > > > > >> As a consequence, I don't think it's critical to make > any > > >>> hard > > >>> > > > > > guarantees > > >>> > > > > > > >> about > > >>> > > > > > > >> what results get forwarded and what (no-op updates) > get > > >>> dropped. I'd > > >>> > > > > > > >> initially > > >>> > > > > > > >> just been thinking about doing this opportunistically > in > > >>> cases where > > >>> > > > > > we > > >>> > > > > > > >> already > > >>> > > > > > > >> had the "old" and "new" result in memory, thanks to a > > >>> request to > > >>> > > > > > > "emit old > > >>> > > > > > > >> values", or to the implementation of timestamp > semantics. > > >>> > > > > > > >> > > >>> > > > > > > >> However, whether or not it's semantically critical, I > do > > >>> think that > > >>> > > > > > > >> Matthias's > > >>> > > > > > > >> idea to use the change-forwarding mechanism to check > for > > >>> no-ops even > > >>> > > > > > on > > >>> > > > > > > >> stateless operations is pretty interesting. > Specifically, > > >>> this would > > >>> > > > > > > >> _really_ > > >>> > > > > > > >> let you pare down useless updates by using mapValues > to > > >>> strip down > > >>> > > > > > > records > > >>> > > > > > > >> only > > >>> > > > > > > >> to what you really need. However, the dependence on > the > > >>> > > > > > implementation of > > >>> > > > > > > >> equals() is troubling. > > >>> > > > > > > >> > > >>> > > > > > > >> It might make sense to table this idea, as well as my > > >>> complicated > > >>> > > > > > no-op > > >>> > > > > > > >> detection algorithm, and initially propose just a > > >>> nonconfigurable > > >>> > > > > > feature > > >>> > > > > > > >> to > > >>> > > > > > > >> check "old" and "new" results for binary equality > before > > >>> forwarding. > > >>> > > > > > > I.e., > > >>> > > > > > > >> if > > >>> > > > > > > >> any operation determines that the old and new results > are > > >>> > > > > > > >> binary-identical, we > > >>> > > > > > > >> would not forward. > > >>> > > > > > > >> > > >>> > > > > > > >> I'll admit that this doesn't serve Tommy's use case > very > > >>> well, but it > > >>> > > > > > > >> might be > > >>> > > > > > > >> better to take baby steps with an optimization like > this > > >>> and not risk > > >>> > > > > > > >> over-reaching in a way that actually harms > performance or > > >>> > > > > > correctness. We > > >>> > > > > > > >> could > > >>> > > > > > > >> always expand the feature to use equals() or some > kind of > > >>> > > > > > ChangeDetector > > >>> > > > > > > >> later > > >>> > > > > > > >> on, in a more focused discussion. > > >>> > > > > > > >> > > >>> > > > > > > >> Regarding metrics or debug logs, I guess I don't feel > > >>> strongly, but it > > >>> > > > > > > >> feels > > >>> > > > > > > >> like two things will happen that make it nicer to add > > >>> them: > > >>> > > > > > > >> > > >>> > > > > > > >> 1. This feature is going to surprise/annoy _somebody_, > > >>> and it would be > > >>> > > > > > > >> nice to > > >>> > > > > > > >> be able to definitively say the reason that updates > are > > >>> dropped is > > >>> > > > > > that > > >>> > > > > > > >> they > > >>> > > > > > > >> were no-ops. The easiest smoking gun is if there are > > >>> debug-logs that > > >>> > > > > > > can be > > >>> > > > > > > >> enabled. This person might just be looking at the > > >>> dashboards, > > >>> > > > > > > wondering why > > >>> > > > > > > >> there are 100K updates per second going into their > app, > > >>> but only 1K > > >>> > > > > > > >> results per > > >>> > > > > > > >> second coming out. Having the metric there makes the > > >>> accounting > > >>> > > > > > easier. > > >>> > > > > > > >> > > >>> > > > > > > >> 2. Somebody is going to struggle with high-volume > > >>> updates, and it > > >>> > > > > > > would be > > >>> > > > > > > >> nice > > >>> > > > > > > >> for them to know that this feature is saving them > > >>> X-thousand updates > > >>> > > > > > per > > >>> > > > > > > >> second, > > >>> > > > > > > >> etc. > > >>> > > > > > > >> > > >>> > > > > > > >> What does everyone think about this? Note, as I read > it, > > >>> what I've > > >>> > > > > > said > > >>> > > > > > > >> above is > > >>> > > > > > > >> already reflected in the text of the KIP. > > >>> > > > > > > >> > > >>> > > > > > > >> Thanks, > > >>> > > > > > > >> -John > > >>> > > > > > > >> > > >>> > > > > > > >> > > >>> > > > > > > >> On Tue, Feb 11, 2020, at 18:27, Richard Yu wrote: > > >>> > > > > > > >>> Hi all, > > >>> > > > > > > >>> > > >>> > > > > > > >>> Bumping this. If you feel that this KIP is not too > > >>> urgent. Then let > > >>> > > > > > me > > >>> > > > > > > >>> know. :) > > >>> > > > > > > >>> > > >>> > > > > > > >>> Cheers, > > >>> > > > > > > >>> Richard > > >>> > > > > > > >>> > > >>> > > > > > > >>> On Thu, Feb 6, 2020 at 4:55 PM Richard Yu < > > >>> > > > > > yohan.richard...@gmail.com> > > >>> > > > > > > >>> wrote: > > >>> > > > > > > >>> > > >>> > > > > > > >>>> Hi all, > > >>> > > > > > > >>>> > > >>> > > > > > > >>>> I've had just a few thoughts regarding the > forwarding > > >>> of <key, > > >>> > > > > > > >>>> change<old_value, new_value>>. As Matthias already > > >>> mentioned, there > > >>> > > > > > > >> are two > > >>> > > > > > > >>>> separate priorities by which we can judge this KIP: > > >>> > > > > > > >>>> > > >>> > > > > > > >>>> 1. A optimization perspective: In this case, the > user > > >>> would prefer > > >>> > > > > > the > > >>> > > > > > > >>>> impact of this KIP to be as minimal as possible. By > > >>> such logic, if > > >>> > > > > > > >>>> stateless operations are performed twice, that could > > >>> prove > > >>> > > > > > > >> unacceptable for > > >>> > > > > > > >>>> them. (since operations can prove expensive) > > >>> > > > > > > >>>> > > >>> > > > > > > >>>> 2. Semantics correctness perspective: Unlike the > > >>> optimization > > >>> > > > > > > >> approach, we > > >>> > > > > > > >>>> are more concerned with all KTable operations > obeying > > >>> the same > > >>> > > > > > emission > > >>> > > > > > > >>>> policy. i.e. emit on change. In this case, a > > >>> discrepancy would not > > >>> > > > > > be > > >>> > > > > > > >>>> tolerated, even though an extra performance cost > will > > >>> be incurred. > > >>> > > > > > > >>>> Therefore, we will follow Matthias's approach, and > then > > >>> perform the > > >>> > > > > > > >>>> operation once on the old value, and once on the > new. > > >>> > > > > > > >>>> > > >>> > > > > > > >>>> The issue here I think is more black and white than > in > > >>> between. The > > >>> > > > > > > >> second > > >>> > > > > > > >>>> option in particular would be favorable for users > with > > >>> inexpensive > > >>> > > > > > > >>>> stateless operations, while for the former option, > we > > >>> are probably > > >>> > > > > > > >> dealing > > >>> > > > > > > >>>> with more expensive ones. So the simplest solution > is > > >>> probably to > > >>> > > > > > > >> allow the > > >>> > > > > > > >>>> user to choose one of the behaviors, and have a > config > > >>> which can > > >>> > > > > > > >> switch in > > >>> > > > > > > >>>> between them. > > >>> > > > > > > >>>> > > >>> > > > > > > >>>> Its the simplest compromise I can come up with at > the > > >>> moment, but if > > >>> > > > > > > >> you > > >>> > > > > > > >>>> think you have a better plan which could better > balance > > >>> tradeoffs. > > >>> > > > > > Then > > >>> > > > > > > >>>> please let us know. :) > > >>> > > > > > > >>>> > > >>> > > > > > > >>>> Best, > > >>> > > > > > > >>>> Richard > > >>> > > > > > > >>>> > > >>> > > > > > > >>>> On Wed, Feb 5, 2020 at 5:12 PM John Roesler < > > >>> vvcep...@apache.org> > > >>> > > > > > > >> wrote: > > >>> > > > > > > >>>> > > >>> > > > > > > >>>>> Hi all, > > >>> > > > > > > >>>>> > > >>> > > > > > > >>>>> Thanks for the thoughtful comments! > > >>> > > > > > > >>>>> > > >>> > > > > > > >>>>> I need more time to reflect on your thoughts, but > just > > >>> wanted to > > >>> > > > > > offer > > >>> > > > > > > >>>>> a quick clarification about equals(). > > >>> > > > > > > >>>>> > > >>> > > > > > > >>>>> I only meant that we can't be sure if a class's > > >>> equals() > > >>> > > > > > > >> implementation > > >>> > > > > > > >>>>> returns true for two semantically identical > instances. > > >>> I.e., if a > > >>> > > > > > > >> class > > >>> > > > > > > >>>>> doesn't > > >>> > > > > > > >>>>> override the default equals() implementation, then > we > > >>> would see > > >>> > > > > > > >> behavior > > >>> > > > > > > >>>>> like: > > >>> > > > > > > >>>>> > > >>> > > > > > > >>>>> new MyPair("A", 1).equals(new MyPair("A", 1)) > returns > > >>> false > > >>> > > > > > > >>>>> > > >>> > > > > > > >>>>> In that case, I would still like to catch no-op > > >>> updates by > > >>> > > > > > comparing > > >>> > > > > > > >> the > > >>> > > > > > > >>>>> serialized form of the records when we happen to > have > > >>> it serialized > > >>> > > > > > > >> anyway > > >>> > > > > > > >>>>> (such as when the operation is stateful, or when > we're > > >>> sending to a > > >>> > > > > > > >>>>> repartition topic and we have both the "new" and > "old" > > >>> value from > > >>> > > > > > > >>>>> upstream). > > >>> > > > > > > >>>>> > > >>> > > > > > > >>>>> I didn't mean to suggest we'd try to use > reflection to > > >>> detect > > >>> > > > > > whether > > >>> > > > > > > >>>>> equals > > >>> > > > > > > >>>>> is implemented, although that is a neat trick. I > was > > >>> thinking more > > >>> > > > > > of > > >>> > > > > > > >> a > > >>> > > > > > > >>>>> belt-and-suspenders algorithm where we do the check > > >>> for no-ops > > >>> > > > > > based > > >>> > > > > > > >> on > > >>> > > > > > > >>>>> equals() and then _also_ check the serialized bytes > > >>> for equality. > > >>> > > > > > > >>>>> > > >>> > > > > > > >>>>> Thanks, > > >>> > > > > > > >>>>> -John > > >>> > > > > > > >>>>> > > >>> > > > > > > >>>>> On Wed, Feb 5, 2020, at 15:31, Ted Yu wrote: > > >>> > > > > > > >>>>>> Thanks for the comments, Matthias. > > >>> > > > > > > >>>>>> > > >>> > > > > > > >>>>>> w.r.t. requirement of an `equals()` > implementation, > > >>> each template > > >>> > > > > > > >> type > > >>> > > > > > > >>>>>> would have an equals() method. We can use the > > >>> following code to > > >>> > > > > > know > > >>> > > > > > > >>>>>> whether it is provided by JVM or provided by user. > > >>> > > > > > > >>>>>> > > >>> > > > > > > >>>>>> boolean customEquals = false; > > >>> > > > > > > >>>>>> try { > > >>> > > > > > > >>>>>> Class cls = > value.getClass().getMethod("equals", > > >>> > > > > > > >>>>>> Object.class).getDeclaringClass(); > > >>> > > > > > > >>>>>> if (!Object.class.equals(cls)) { > > >>> > > > > > > >>>>>> customEquals = true; > > >>> > > > > > > >>>>>> } > > >>> > > > > > > >>>>>> } catch (NoSuchMethodException nsme) { > > >>> > > > > > > >>>>>> // equals is always defined, this wouldn't hit > > >>> > > > > > > >>>>>> } > > >>> > > > > > > >>>>>> > > >>> > > > > > > >>>>>> The next question is: what if the user doesn't > > >>> provide equals() > > >>> > > > > > > >> method ? > > >>> > > > > > > >>>>>> Would we automatically fall back to > emit-on-update ? > > >>> > > > > > > >>>>>> > > >>> > > > > > > >>>>>> Cheers > > >>> > > > > > > >>>>>> > > >>> > > > > > > >>>>>> On Tue, Feb 4, 2020 at 1:37 PM Matthias J. Sax < > > >>> mj...@apache.org> > > >>> > > > > > > >>>>> wrote: > > >>> > > > > > > >>>>>> > > >>> > > > > > > > First a high level comment: > > >>> > > > > > > > > > >>> > > > > > > > Overall, I would like to make one step back, and make > sure > > >>> we are > > >>> > > > > > > > discussion on the same level. Originally, I understood > > >>> this KIP > > >>> > > > > > > >>> as a > > >>> > > > > > > > proposed change of _semantics_, however, given the > latest > > >>> > > > > > > >>> discussion > > >>> > > > > > > > it seems it's actually not -- it's more an > _optimization_ > > >>> > > > > > > >>> proposal. > > >>> > > > > > > > Hence, we only need to make sure that this optimization > > >>> does not > > >>> > > > > > > >>> break > > >>> > > > > > > > existing semantics. It this the right way to think > about > > >>> it? > > >>> > > > > > > > > > >>> > > > > > > > If yes, than it might actually be ok to have different > > >>> behavior > > >>> > > > > > > > depending if there is a materialized KTable or not. So > > >>> far, we > > >>> > > > > > > >>> never > > >>> > > > > > > > defined a public contract about our emit strategy and > it > > >>> seems > > >>> > > > > > > >>> this > > >>> > > > > > > > KIP does not define one either. > > >>> > > > > > > > > > >>> > > > > > > > Hence, I don't have as strong of an opinion about > sending > > >>> > > > > > > >>> oldValues > > >>> > > > > > > > for example any longer. I guess the question is really, > > >>> what can > > >>> > > > > > > >>> we > > >>> > > > > > > > implement in a reasonable way. > > >>> > > > > > > > > > >>> > > > > > > > > > >>> > > > > > > > > > >>> > > > > > > > Other comments: > > >>> > > > > > > > > > >>> > > > > > > > > > >>> > > > > > > > @Richard: > > >>> > > > > > > > > > >>> > > > > > > > Can you please add the KIP to the KIP overview table: > It's > > >>> missing > > >>> > > > > > > > ( > > >>> > > > > > > >>>>>> > > >>> > > > > > > >>> > > >>> > > > > > > > >>> > https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Improvement+Pro > > >>> > > > > > > > posals). > > >>> > > > > > > > > > >>> > > > > > > > > > >>> > > > > > > > @Bruno: > > >>> > > > > > > > > > >>> > > > > > > > You mentioned caching. I think it's irrelevant > > >>> (orthogonal) and > > >>> > > > > > > >>> we can > > >>> > > > > > > > discuss this KIP without considering it. > > >>> > > > > > > > > > >>> > > > > > > > > > >>> > > > > > > > @John: > > >>> > > > > > > > > > >>> > > > > > > >>>>>>>>> Even in the source table, we forward the > updated > > >>> record with > > >>> > > > > > the > > >>> > > > > > > >>>>>>>>> higher of the two timestamps. So the example is > > >>> more like: > > >>> > > > > > > > > > >>> > > > > > > > That is not correct. Currently, we forward with the > smaller > > >>> > > > > > > > out-of-order timestamp (changing the timestamp would > > >>> corrupt the > > >>> > > > > > > >>> data > > >>> > > > > > > > -- we don't know, because we don't check, if the value > is > > >>> the > > >>> > > > > > > >>> same > > >>> > > > > > > >>>>>> or > > >>> > > > > > > > a different one, hence, we must emit the out-of-order > > >>> record > > >>> > > > > > > >>> as-is). > > >>> > > > > > > > > > >>> > > > > > > > If we start to do emit-on-change, we also need to emit > a > > >>> new > > >>> > > > > > > >>> record if > > >>> > > > > > > > the timestamp changes due to out-of-order data, hence, > we > > >>> would > > >>> > > > > > > >>> still > > >>> > > > > > > > need to emit <K,V,T1> because that give us correct > > >>> semantics: > > >>> > > > > > > >>> assume > > >>> > > > > > > > you have a filter() and afterward use the filter > KTable in > > >>> a > > >>> > > > > > > > stream-table join -- the lower T1 timestamp must be > > >>> propagated to > > >>> > > > > > > >>> the > > >>> > > > > > > > filtered KTable to ensure that that the stream-table > join > > >>> compute > > >>> > > > > > > >>> the > > >>> > > > > > > > correct result. > > >>> > > > > > > > > > >>> > > > > > > > > > >>> > > > > > > > > > >>> > > > > > > > Your point about requiring an `equals()` > implementation is > > >>> > > > > > > >>> actually a > > >>> > > > > > > > quite interesting one and boils down to my statement > from > > >>> above > > >>> > > > > > > >>> about > > >>> > > > > > > > "what can we actually implement". What I don't > understand > > >>> is: > > >>> > > > > > > > > > >>> > > > > > > >>>>>>>>> This way, we still don't have to rely on the > > >>> existence of an > > >>> > > > > > > >>>>>>>>> equals() method, but if it is there, we can > > >>> benefit from it. > > >>> > > > > > > > > > >>> > > > > > > > Your bullet point (2) says it uses `equals()` -- > hence, it > > >>> seems > > >>> > > > > > > >>> we > > >>> > > > > > > > actually to rely on it? Also, how can we detect if > there > > >>> is an > > >>> > > > > > > > `equals()` method to do the comparison? Would be fail > if > > >>> we don't > > >>> > > > > > > >>> have > > >>> > > > > > > > `equals()` nor corresponding serializes to do the > > >>> comparison? > > >>> > > > > > > > > > >>> > > > > > > > > > >>> > > > > > > > > > >>> > > > > > > >>>>>>>>> Wow, really good catch! Yes, we absolutely need > > >>> metrics and > > >>> > > > > > > >>> logs if > > >>> > > > > > > >>>>>>>>> we're going to drop any records. And, yes, we > > >>> should propose > > >>> > > > > > > >>>>>>>>> metrics and logs that are similar to the > existing > > >>> ones when we > > >>> > > > > > > >>> drop > > >>> > > > > > > >>>>>>>>> records for other reasons. > > >>> > > > > > > > > > >>> > > > > > > > I am not sure about this point. In fact, we have > already > > >>> some > > >>> > > > > > > >>> no-ops > > >>> > > > > > > > in Kafka Streams in our join-operators and don't report > > >>> any of > > >>> > > > > > > >>> those > > >>> > > > > > > > either. Emit-on-change is operator semantics and I > don't > > >>> see why > > >>> > > > > > > >>> we > > >>> > > > > > > > would need to have a metric for it? It seems to be > quite > > >>> different > > >>> > > > > > > > compared to dropping late or malformed records. > > >>> > > > > > > > > > >>> > > > > > > > > > >>> > > > > > > > -Matthias > > >>> > > > > > > > > > >>> > > > > > > > > > >>> > > > > > > > > > >>> > > > > > > > On 2/4/20 7:13 AM, Thomas Becker wrote: > > >>> > > > > > > >>>>>>>>> Thanks John for your thoughtful reply. Some > > >>> comments inline. > > >>> > > > > > > >>>>>>>>> > > >>> > > > > > > >>>>>>>>> > > >>> > > > > > > >>>>>>>>> On Mon, 2020-02-03 at 11:51 -0600, John Roesler > > >>> wrote: > > >>> > > > > > > >>>>>>>>>> [EXTERNAL EMAIL] Attention: This email was > sent > > >>> from outside > > >>> > > > > > > >>>>>>>>>> TiVo. DO NOT CLICK any links or attachments > > >>> unless you > > >>> > > > > > expected > > >>> > > > > > > >>>>>>>>>> them. ________________________________ > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> Hi Tommy, > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> Thanks for the context. I can see the > attraction > > >>> of > > >>> > > > > > considering > > >>> > > > > > > >>>>>>>>>> these use cases together. > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> To answer your question, if a part of the > record > > >>> is not > > >>> > > > > > > >>> relevant > > >>> > > > > > > >>>>>>>>>> to downstream consumers, I was thinking you > could > > >>> just use a > > >>> > > > > > > >>>>>>>>>> mapValue to remove it. > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> E.g., suppose you wanted to do a join between > two > > >>> tables. > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> employeeInfo.join( employeePayroll, (info, > > >>> payroll) -> new > > >>> > > > > > > >>>>>>>>>> Result(info.name(), payroll.salary()) ) > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> We only care about one attribute from the Info > > >>> table (name), > > >>> > > > > > > >>> and > > >>> > > > > > > >>>>>>>>>> one from the Payroll table (salary), and these > > >>> attributes > > >>> > > > > > > >>> change > > >>> > > > > > > >>>>>>>>>> rarely. On the other hand, there might be many > > >>> other > > >>> > > > > > attributes > > >>> > > > > > > >>>>>>>>>> that change frequently of these tables. We can > > >>> avoid > > >>> > > > > > triggering > > >>> > > > > > > >>>>>>>>>> the join unnecessarily by mapping the input > > >>> tables to drop the > > >>> > > > > > > >>>>>>>>>> unnecessary information before the join: > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> names = employeeInfo.mapValues(info -> > info.name()) > > >>> salaries > > >>> > > > > > = > > >>> > > > > > > >>>>>>>>>> employeePayroll.mapValues(payroll -> > > >>> payroll.salary()) > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> names.join( salaries, (name, salary) -> new > > >>> Result(name, > > >>> > > > > > > >>> salary) > > >>> > > > > > > >>>>>>>>>> ) > > >>> > > > > > > >>>>>>>>> > > >>> > > > > > > >>>>>>>>> Ahh yes I see. This works, but in the case > where > > >>> you're using > > >>> > > > > > > >>>>>>>>> schemas as we are (e.g. Avro), it seems like > this > > >>> approach > > >>> > > > > > could > > >>> > > > > > > >>>>>>>>> lead to a proliferation of "skinny" record > types > > >>> that just drop > > >>> > > > > > > >>>>>>>>> various fields. > > >>> > > > > > > >>>>>>>>> > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> Especially if we take Matthias's idea to drop > > >>> non-changes even > > >>> > > > > > > >>>>>>>>>> for stateless operations, this would be quite > > >>> efficient and is > > >>> > > > > > > >>>>>>>>>> also a very straightforward optimization to > > >>> understand once > > >>> > > > > > you > > >>> > > > > > > >>>>>>>>>> know that Streams provides emit-on-change. > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> From the context that you provided, it seems > like > > >>> a slightly > > >>> > > > > > > >>>>>>>>>> different situation, though. Reading between > the > > >>> lines a > > >>> > > > > > > >>> little, > > >>> > > > > > > >>>>>>>>>> it sounds like: in contrast to the example > above, > > >>> in which we > > >>> > > > > > > >>> are > > >>> > > > > > > >>>>>>>>>> filtering out extra _data_, you have some > extra > > >>> _metadata_ > > >>> > > > > > that > > >>> > > > > > > >>>>>>>>>> you still wish to pass down with the data when > > >>> there is a > > >>> > > > > > > >>> "real" > > >>> > > > > > > >>>>>>>>>> update, but you don't want the metadata > itself to > > >>> cause an > > >>> > > > > > > >>>>>>>>>> update. > > >>> > > > > > > >>>>>>>>> > > >>> > > > > > > >>>>>>>>> Despite my lack of clarity, yes you've got it > > >>> right ;) This > > >>> > > > > > > >>>>>>>>> particular processor is the first stop for this > > >>> data after > > >>> > > > > > > >>> coming > > >>> > > > > > > >>>>>>>>> in from external users, who often simply post > the > > >>> same content > > >>> > > > > > > >>> each > > >>> > > > > > > >>>>>>>>> time and we're trying to shield downstream > > >>> consumers from > > >>> > > > > > > >>>>>>>>> unnecessary churn. > > >>> > > > > > > >>>>>>>>> > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> It does seem handy to be able to plug in a > custom > > >>> > > > > > > >>> ChangeDetector > > >>> > > > > > > >>>>>>>>>> for this purpose, but I worry about the API > > >>> complexity. Maybe > > >>> > > > > > > >>> you > > >>> > > > > > > >>>>>>>>>> can help think though how to provide the same > > >>> benefit while > > >>> > > > > > > >>>>>>>>>> limiting user-facing complexity. > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> Here's some extra context to consider: > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> We currently don't make any extra requirements > > >>> about the > > >>> > > > > > nature > > >>> > > > > > > >>>>>>>>>> of data that you can use in Streams. For > example, > > >>> you don't > > >>> > > > > > > >>> have > > >>> > > > > > > >>>>>>>>>> to implement hashCode and equals, or > compareTo, > > >>> etc. With the > > >>> > > > > > > >>>>>>>>>> current proposal, we can do an airtight > > >>> comparison based only > > >>> > > > > > > >>> on > > >>> > > > > > > >>>>>>>>>> the serialized form of the values, and we > > >>> actually don't have > > >>> > > > > > > >>> to > > >>> > > > > > > >>>>>>>>>> deserialize the "prior" value at all for a > large > > >>> number of > > >>> > > > > > > >>>>>>>>>> operations. Admitedly, if we extend the > proposal > > >>> to include > > >>> > > > > > > >>> no-op > > >>> > > > > > > >>>>>>>>>> detection for stateless operations, we'd > probably > > >>> need to rely > > >>> > > > > > > >>> on > > >>> > > > > > > >>>>>>>>>> equals() for no-op checking, otherwise we'd > wind > > >>> up requiring > > >>> > > > > > > >>>>>>>>>> serdes for stateless operations as well. > > >>> Actually, I'd > > >>> > > > > > probably > > >>> > > > > > > >>>>>>>>>> argue for doing exactly that: > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> 1. In stateful operations, drop if the > serialized > > >>> byte[]s are > > >>> > > > > > > >>> the > > >>> > > > > > > >>>>>>>>>> same. After deserializing, also drop if the > > >>> objects are equal > > >>> > > > > > > >>>>>>>>>> according to Object#equals(). > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> 2. In stateless operations, compare the "new" > and > > >>> "old" values > > >>> > > > > > > >>>>>>>>>> (if "old" is available) based on > Object#equals(). > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> 3. As a final optimization, after serializing > and > > >>> before > > >>> > > > > > > >>> sending > > >>> > > > > > > >>>>>>>>>> repartition records, compare the serialized > data > > >>> and drop > > >>> > > > > > > >>>>>>>>>> no-ops. > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> This way, we still don't have to rely on the > > >>> existence of an > > >>> > > > > > > >>>>>>>>>> equals() method, but if it is there, we can > > >>> benefit from it. > > >>> > > > > > > >>>>>>>>>> Also, we don't require a serde in any new > > >>> situations, but we > > >>> > > > > > > >>> can > > >>> > > > > > > >>>>>>>>>> still leverage it when it is available. > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> For clarity, in my example above, even if the > > >>> employeeInfo and > > >>> > > > > > > >>>>>>>>>> employeePayroll and Result records all have > > >>> serdes, we need > > >>> > > > > > the > > >>> > > > > > > >>>>>>>>>> "name" field (presumably String) and the > "salary" > > >>> field > > >>> > > > > > > >>>>>>>>>> (presumable a Double) to have serdes as well > in > > >>> the naive > > >>> > > > > > > >>>>>>>>>> implementation. But if we can leverage > equals(), > > >>> then the > > >>> > > > > > > >>> "right > > >>> > > > > > > >>>>>>>>>> thing" happens automatically. > > >>> > > > > > > >>>>>>>>> > > >>> > > > > > > >>>>>>>>> I still don't totally follow why the individual > > >>> components > > >>> > > > > > > >>> (name, > > >>> > > > > > > >>>>>>>>> salary) would have to have serdes here. If > Result > > >>> has one, we > > >>> > > > > > > >>>>>>>>> compare bytes, and if Result additionally has > an > > >>> equals() > > >>> > > > > > method > > >>> > > > > > > >>>>>>>>> (which presumably includes equals comparisons > on > > >>> the > > >>> > > > > > constituent > > >>> > > > > > > >>>>>>>>> fields), have we not covered our bases? > > >>> > > > > > > >>>>>>>>> > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> This dovetails in with my primary UX concern; > > >>> where would the > > >>> > > > > > > >>>>>>>>>> ChangeDetector actually be registered? None of > > >>> the operators > > >>> > > > > > in > > >>> > > > > > > >>>>>>>>>> my example have names or topics or any other > > >>> identifiable > > >>> > > > > > > >>>>>>>>>> characteristic that could be passed to a > > >>> ChangeDetector class > > >>> > > > > > > >>>>>>>>>> registered via config. You could say that we > make > > >>> > > > > > > >>> ChangeDetector > > >>> > > > > > > >>>>>>>>>> an optional parameter to every operation in > > >>> Streams, but this > > >>> > > > > > > >>>>>>>>>> seems to carry quite a bit of mental burden > with > > >>> it. People > > >>> > > > > > > >>> will > > >>> > > > > > > >>>>>>>>>> wonder what it's for and whether or not they > > >>> should be using > > >>> > > > > > > >>> it. > > >>> > > > > > > >>>>>>>>>> There would almost certainly be a > misconception > > >>> that it's > > >>> > > > > > > >>>>>>>>>> preferable to implement it always, which > would be > > >>> unfortunate. > > >>> > > > > > > >>>>>>>>>> Plus, to actually implment metadata flowing > > >>> through the > > >>> > > > > > > >>> topology > > >>> > > > > > > >>>>>>>>>> as in your use case, you'd have to do two > things: > > >>> 1. make sure > > >>> > > > > > > >>>>>>>>>> that all operations actually preserve the > > >>> metadata alongside > > >>> > > > > > > >>> the > > >>> > > > > > > >>>>>>>>>> data (e.g., don't accidentally add a mapValues > > >>> like I did, or > > >>> > > > > > > >>> you > > >>> > > > > > > >>>>>>>>>> drop the metadata). 2. implement a > ChangeDetector > > >>> for every > > >>> > > > > > > >>>>>>>>>> single operation in the topology, or you don't > > >>> get the benefit > > >>> > > > > > > >>> of > > >>> > > > > > > >>>>>>>>>> dropping non-changes internally 2b. > > >>> Alternatively, you could > > >>> > > > > > > >>> just > > >>> > > > > > > >>>>>>>>>> add the ChangeDetector to one operation toward > > >>> the end of the > > >>> > > > > > > >>>>>>>>>> topology. This would not drop redundant > > >>> computation > > >>> > > > > > internally, > > >>> > > > > > > >>>>>>>>>> but only drop redundant _outputs_. But this is > > >>> just about the > > >>> > > > > > > >>>>>>>>>> same as your current solution. > > >>> > > > > > > >>>>>>>>> > > >>> > > > > > > >>>>>>>>> I definitely see your point regarding > > >>> configuration. I was > > >>> > > > > > > >>>>>>>>> originally thinking about this when the > > >>> deduplication was going > > >>> > > > > > > >>> to > > >>> > > > > > > >>>>>>>>> be opt-in, and it seemed very natural to say > > >>> something like: > > >>> > > > > > > >>>>>>>>> > > >>> > > > > > > >>>>>>>>> employeeInfo.join(employeePayroll, (info, > payroll) > > >>> -> new > > >>> > > > > > > >>>>>>>>> Result(info.name(), payroll.salary())) > > >>> > > > > > > >>>>>>>>> > > >>> .suppress(duplicatesAccordingTo(someChangeDetector)) > > >>> > > > > > > >>>>>>>>> > > >>> > > > > > > >>>>>>>>> Alternatively you can imagine a similar method > > >>> being on > > >>> > > > > > > >>>>>>>>> Materialized, though obviously this makes less > > >>> sense if we > > >>> > > > > > don't > > >>> > > > > > > >>>>>>>>> want to require materialization. If we're now > > >>> talking about > > >>> > > > > > > >>>>>>>>> changing the default behavior and not having > any > > >>> configuration > > >>> > > > > > > >>>>>>>>> options, it's harder to find a place for this. > > >>> > > > > > > >>>>>>>>> > > >>> > > > > > > >>>>>>>>> > > >>> > > > > > > >>>>>>>>> > > >>> > > > > > > >>>>>>>>>> A final thought; if it really is a metadata > > >>> question, can we > > >>> > > > > > > >>> just > > >>> > > > > > > >>>>>>>>>> plan to finish up the support for headers in > > >>> Streams? I.e., > > >>> > > > > > > >>> give > > >>> > > > > > > >>>>>>>>>> you a way to control the way that headers flow > > >>> through the > > >>> > > > > > > >>>>>>>>>> topology? Then, we could treat headers the > same > > >>> way we treat > > >>> > > > > > > >>>>>>>>>> timestamps in the no-op checking... We > completely > > >>> ignore them > > >>> > > > > > > >>>>>>>>>> for the sake of comparison. Thus, neither the > > >>> timestamp nor > > >>> > > > > > the > > >>> > > > > > > >>>>>>>>>> headers would get updated in internal state > or in > > >>> downstream > > >>> > > > > > > >>>>>>>>>> views as long as the value itself doesn't > change. > > >>> This seems > > >>> > > > > > to > > >>> > > > > > > >>>>>>>>>> give us a way to support your use case without > > >>> adding to the > > >>> > > > > > > >>>>>>>>>> mental overhead of using Streams for simple > > >>> things. > > >>> > > > > > > >>>>>>>>> > > >>> > > > > > > >>>>>>>>> Agree headers could be a decent fit for this > > >>> particular case > > >>> > > > > > > >>>>>>>>> because it's mostly metadata, though to be > honest > > >>> we haven't > > >>> > > > > > > >>> looked > > >>> > > > > > > >>>>>>>>> at headers much (mostly because, and to your > > >>> point, support > > >>> > > > > > > >>> seems > > >>> > > > > > > >>>>>>>>> to be lacking). I feel like there would be > other > > >>> cases where > > >>> > > > > > > >>> this > > >>> > > > > > > >>>>>>>>> feature could be valuable, but I admit I can't > > >>> come up with > > >>> > > > > > > >>>>>>>>> anything right this second. Perhaps yuzhihong > had > > >>> an example in > > >>> > > > > > > >>>>>>>>> mind? > > >>> > > > > > > >>>>>>>>> > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> I.e., simple things should be easy, and > complex > > >>> things should > > >>> > > > > > > >>> be > > >>> > > > > > > >>>>>>>>>> possible. > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> What are your thoughts? Thanks, -John > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> On Mon, Feb 3, 2020, at 07:19, Thomas Becker > > >>> wrote: > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> Hi John, Can you describe how you'd use > > >>> filtering/mapping to > > >>> > > > > > > >>>>>>>>>> deduplicate records? To give some background > on > > >>> my suggestion > > >>> > > > > > > >>> we > > >>> > > > > > > >>>>>>>>>> currently have a small stream processor that > > >>> exists solely to > > >>> > > > > > > >>>>>>>>>> deduplicate, which we do using a process that > I > > >>> assume would > > >>> > > > > > be > > >>> > > > > > > >>>>>>>>>> similar to what would be done here (with a > store > > >>> of keys and > > >>> > > > > > > >>> hash > > >>> > > > > > > >>>>>>>>>> values). But the records we are deduplicating > > >>> have some > > >>> > > > > > > >>> metadata > > >>> > > > > > > >>>>>>>>>> fields (such as timestamps of when the record > was > > >>> posted) that > > >>> > > > > > > >>> we > > >>> > > > > > > >>>>>>>>>> don't consider semantically meaningful for > > >>> downstream > > >>> > > > > > > >>> consumers, > > >>> > > > > > > >>>>>>>>>> and therefore we also suppress updates that > only > > >>> touch those > > >>> > > > > > > >>>>>>>>>> fields. > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> -Tommy > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> On Fri, 2020-01-31 at 19:30 -0600, John > Roesler > > >>> wrote: > > >>> > > > > > > >>> [EXTERNAL > > >>> > > > > > > >>>>>>>>>> EMAIL] Attention: This email was sent from > > >>> outside TiVo. DO > > >>> > > > > > NOT > > >>> > > > > > > >>>>>>>>>> CLICK any links or attachments unless you > > >>> expected them. > > >>> > > > > > > >>>>>>>>>> ________________________________ > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> Hi Thomas and yuzhihong, That’s an interesting > > >>> idea. Can you > > >>> > > > > > > >>> help > > >>> > > > > > > >>>>>>>>>> think of a use case that isn’t also served by > > >>> filtering or > > >>> > > > > > > >>>>>>>>>> mapping beforehand? Thanks for helping to > design > > >>> this feature! > > >>> > > > > > > >>>>>>>>>> -John On Fri, Jan 31, 2020, at 18:56, > > >>> yuzhih...@gmail.com > > >>> > > > > > > >>>>>>>>>> <mailto:yuzhih...@gmail.com> wrote: I think > this > > >>> is good > > >>> > > > > > > >>> idea. On > > >>> > > > > > > >>>>>>>>>> Jan 31, 2020, at 4:49 PM, Thomas Becker < > > >>> > > > > > > >>> thomas.bec...@tivo.com > > >>> > > > > > > >>>>>>>>>> <mailto:thomas.bec...@tivo.com>> wrote: How > do > > >>> folks feel > > >>> > > > > > > >>> about > > >>> > > > > > > >>>>>>>>>> allowing the mechanism by which no-ops are > > >>> detected to be > > >>> > > > > > > >>>>>>>>>> pluggable? Meaning use something like a hash > by > > >>> default, but > > >>> > > > > > > >>> you > > >>> > > > > > > >>>>>>>>>> could optionally provide an implementation of > > >>> something to use > > >>> > > > > > > >>>>>>>>>> instead, like a ChangeDetector. This could be > > >>> useful for > > >>> > > > > > > >>> example > > >>> > > > > > > >>>>>>>>>> to ignore changes to certain fields, which may > > >>> not be relevant > > >>> > > > > > > >>> to > > >>> > > > > > > >>>>>>>>>> the operation being performed. > > >>> > > > > > ________________________________ > > >>> > > > > > > >>>>>>>>>> From: John Roesler <vvcep...@apache.org > > >>> > > > > > > >>>>>>>>>> <mailto:vvcep...@apache.org>> Sent: Friday, > > >>> January 31, 2020 > > >>> > > > > > > >>> 4:51 > > >>> > > > > > > >>>>>>>>>> PM To: dev@kafka.apache.org <mailto: > > >>> dev@kafka.apache.org> > > >>> > > > > > > >>>>>>>>>> <dev@kafka.apache.org <mailto: > > >>> dev@kafka.apache.org>> Subject: > > >>> > > > > > > >>> Re: > > >>> > > > > > > >>>>>>>>>> [KAFKA-557] Add emit on change support for > Kafka > > >>> Streams > > >>> > > > > > > >>>>>>>>>> [EXTERNAL EMAIL] Attention: This email was > sent > > >>> from outside > > >>> > > > > > > >>>>>>>>>> TiVo. DO NOT CLICK any links or attachments > > >>> unless you > > >>> > > > > > expected > > >>> > > > > > > >>>>>>>>>> them. ________________________________ > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> Hello all, Sorry for my silence. It seems > like we > > >>> are getting > > >>> > > > > > > >>>>>>>>>> close to consensus. Hopefully, we could move > to a > > >>> vote soon! > > >>> > > > > > > >>> All > > >>> > > > > > > >>>>>>>>>> of the reasoning from Matthias and Bruno > around > > >>> timestamp is > > >>> > > > > > > >>>>>>>>>> compelling. I would be strongly in favor of > > >>> stating a few > > >>> > > > > > > >>> things > > >>> > > > > > > >>>>>>>>>> very clearly in the KIP: 1. Streams will drop > > >>> no-op updates > > >>> > > > > > > >>> only > > >>> > > > > > > >>>>>>>>>> for KTable operations. That is, we won't make > any > > >>> changes to > > >>> > > > > > > >>>>>>>>>> KStream aggregations at the moment. It does > seem > > >>> like we can > > >>> > > > > > > >>>>>>>>>> potentially revisit the time semantics of that > > >>> operation in > > >>> > > > > > the > > >>> > > > > > > >>>>>>>>>> future, but we don't need to do it now. On the > > >>> other hand, the > > >>> > > > > > > >>>>>>>>>> proposed semantics for KTable timestamps > (marking > > >>> the > > >>> > > > > > beginning > > >>> > > > > > > >>>>>>>>>> of the validity of that record) makes sense to > > >>> me. 2. Streams > > >>> > > > > > > >>>>>>>>>> will only drop no-op updates for _stateful_ > > >>> KTable operations. > > >>> > > > > > > >>> We > > >>> > > > > > > >>>>>>>>>> don't want to add a hard guarantee that > Streams > > >>> will _never_ > > >>> > > > > > > >>>>>>>>>> emit a no-op table update because it would > > >>> require adding > > >>> > > > > > state > > >>> > > > > > > >>>>>>>>>> to otherwise stateless operations. If someone > is > > >>> really > > >>> > > > > > > >>> concerned > > >>> > > > > > > >>>>>>>>>> about a particular stateless operation > producing > > >>> a lot of > > >>> > > > > > no-op > > >>> > > > > > > >>>>>>>>>> results, all they have to do is materialize > it, > > >>> and Streams > > >>> > > > > > > >>> would > > >>> > > > > > > >>>>>>>>>> automatically drop the no-ops. Additionally, > I'm > > >>> +1 on not > > >>> > > > > > > >>> adding > > >>> > > > > > > >>>>>>>>>> an opt-out at this time. Regarding the KIP > > >>> itself, I would > > >>> > > > > > > >>> clean > > >>> > > > > > > >>>>>>>>>> it up a bit before calling for a vote. There > is a > > >>> lot of > > >>> > > > > > > >>>>>>>>>> "discussion"-type language there, which is > very > > >>> natural to > > >>> > > > > > > >>> read, > > >>> > > > > > > >>>>>>>>>> but makes it a bit hard to see what _exactly_ > the > > >>> kip is > > >>> > > > > > > >>>>>>>>>> proposing. Richard, would you mind just making > > >>> the "proposed > > >>> > > > > > > >>>>>>>>>> behavior change" a simple and succinct list of > > >>> bullet points? > > >>> > > > > > > >>>>>>>>>> I.e., please drop glue phrases like "there has > > >>> been some > > >>> > > > > > > >>>>>>>>>> discussion" or "possibly we could do X". For > the > > >>> final version > > >>> > > > > > > >>> of > > >>> > > > > > > >>>>>>>>>> the KIP, it should just say, "Streams will do > X, > > >>> Streams will > > >>> > > > > > > >>> do > > >>> > > > > > > >>>>>>>>>> Y". Feel free to add an elaboration section to > > >>> explain more > > >>> > > > > > > >>> about > > >>> > > > > > > >>>>>>>>>> what X and Y mean, but we don't need to talk > about > > >>> > > > > > > >>> possibilities > > >>> > > > > > > >>>>>>>>>> or alternatives except in the "rejected > > >>> alternatives" section. > > >>> > > > > > > >>>>>>>>>> Accordingly, can you also move the options you > > >>> presented in > > >>> > > > > > the > > >>> > > > > > > >>>>>>>>>> intro to the "rejected alternatives" section > and > > >>> only mention > > >>> > > > > > > >>> the > > >>> > > > > > > >>>>>>>>>> final proposal itself? This just really helps > > >>> reviewers to > > >>> > > > > > know > > >>> > > > > > > >>>>>>>>>> what they are voting for, and it helps > everyone > > >>> after the fact > > >>> > > > > > > >>>>>>>>>> when they are trying to get clarity on what > > >>> exactly the > > >>> > > > > > > >>> proposal > > >>> > > > > > > >>>>>>>>>> is, versus all the things it could have been. > > >>> Thanks, -John > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> On Mon, Jan 27, 2020, at 18:14, Richard Yu > wrote: > > >>> Hello to > > >>> > > > > > all, > > >>> > > > > > > >>>>>>>>>> I've finished making some initial > modifications > > >>> to the KIP. I > > >>> > > > > > > >>>>>>>>>> have decided to keep the implementation > section > > >>> in the KIP for > > >>> > > > > > > >>>>>>>>>> record-keeping purposes. For now, we should > focus > > >>> on only the > > >>> > > > > > > >>>>>>>>>> proposed behavior changes instead. See if you > > >>> have any > > >>> > > > > > > >>> comments! > > >>> > > > > > > >>>>>>>>>> Cheers, Richard On Sat, Jan 25, 2020 at 11:12 > AM > > >>> Richard Yu > > >>> > > > > > > >>>>>>>>>> <yohan.richard...@gmail.com <mailto: > > >>> > > > > > yohan.richard...@gmail.com > > >>> > > > > > > >>>>> > > >>> > > > > > > >>>>>>>>>> wrote: Hi all, Thanks for all the discussion! > > >>> @John and @Bruno > > >>> > > > > > > >>> I > > >>> > > > > > > >>>>>>>>>> will survey other possible systems and see > what I > > >>> can do. Just > > >>> > > > > > > >>> a > > >>> > > > > > > >>>>>>>>>> question, by systems, I suppose you would mean > > >>> the pros and > > >>> > > > > > > >>> cons > > >>> > > > > > > >>>>>>>>>> of different reporting strategies? I'm not > > >>> completely certain > > >>> > > > > > > >>> on > > >>> > > > > > > >>>>>>>>>> this point, so it would be great if you can > > >>> clarify on that. > > >>> > > > > > So > > >>> > > > > > > >>>>>>>>>> here's what I got from all the discussion so > far: > > >>> - Since both > > >>> > > > > > > >>>>>>>>>> Matthias and John seems to have come to a > > >>> consensus on this, > > >>> > > > > > > >>> then > > >>> > > > > > > >>>>>>>>>> we will go for an all-round behavorial change > for > > >>> KTables. > > >>> > > > > > > >>> After > > >>> > > > > > > >>>>>>>>>> some thought, I decided that for now, an > opt-out > > >>> config will > > >>> > > > > > > >>> not > > >>> > > > > > > >>>>>>>>>> be added. As John have pointed out, no-op > changes > > >>> tend to > > >>> > > > > > > >>>>>>>>>> explode further down the topology as they are > > >>> forwarded to > > >>> > > > > > more > > >>> > > > > > > >>>>>>>>>> and more processor nodes downstream. - About > > >>> using hash codes, > > >>> > > > > > > >>>>>>>>>> after some explanation from John, it looks > like > > >>> hash codes > > >>> > > > > > > >>> might > > >>> > > > > > > >>>>>>>>>> not be as ideal (for implementation). For > now, we > > >>> will omit > > >>> > > > > > > >>> that > > >>> > > > > > > >>>>>>>>>> detail, and save it for the PR. - @Bruno You > do > > >>> have valid > > >>> > > > > > > >>>>>>>>>> concerns. Though, I am not completely certain > if > > >>> we want to do > > >>> > > > > > > >>>>>>>>>> emit-on-change only for materialized KTables. > I > > >>> will put it > > >>> > > > > > > >>> down > > >>> > > > > > > >>>>>>>>>> in the KIP regardless. I will do my best to > > >>> address all points > > >>> > > > > > > >>>>>>>>>> raised so far on the discussion. Hope we could > > >>> keep this > > >>> > > > > > going! > > >>> > > > > > > >>>>>>>>>> Best, Richard On Fri, Jan 24, 2020 at 6:07 PM > > >>> Bruno Cadonna > > >>> > > > > > > >>>>>>>>>> <br...@confluent.io <mailto: > br...@confluent.io>> > > >>> wrote: Thank > > >>> > > > > > > >>> you > > >>> > > > > > > >>>>>>>>>> Matthias for the use cases! Looking at both > use > > >>> cases, I think > > >>> > > > > > > >>>>>>>>>> you need to elaborate on them in the KIP, > > >>> Richard. Emit from > > >>> > > > > > > >>>>>>>>>> plain KTable: I agree with Matthias that the > > >>> lower timestamp > > >>> > > > > > > >>>>>>>>>> makes sense because it marks the start of the > > >>> validity of the > > >>> > > > > > > >>>>>>>>>> record. Idempotent records with a higher > > >>> timestamp can be > > >>> > > > > > > >>> safely > > >>> > > > > > > >>>>>>>>>> ignored. A corner case that I discussed with > > >>> Matthias offline > > >>> > > > > > > >>> is > > >>> > > > > > > >>>>>>>>>> when we do not materialize a KTable due to > > >>> optimization. Then > > >>> > > > > > > >>> we > > >>> > > > > > > >>>>>>>>>> cannot avoid the idempotent records because > we do > > >>> not keep the > > >>> > > > > > > >>>>>>>>>> first record with the lower timestamp to > compare > > >>> to. Emit from > > >>> > > > > > > >>>>>>>>>> KTable with aggregations: If we specify that > an > > >>> aggregation > > >>> > > > > > > >>>>>>>>>> result should have the highest timestamp of > the > > >>> records that > > >>> > > > > > > >>>>>>>>>> participated in the aggregation, we cannot > ignore > > >>> any > > >>> > > > > > > >>> idempotent > > >>> > > > > > > >>>>>>>>>> records. Admittedly, the result of an > aggregation > > >>> usually > > >>> > > > > > > >>>>>>>>>> changes, but there are aggregations where the > > >>> result may not > > >>> > > > > > > >>>>>>>>>> change like min and max, or sum when the > incoming > > >>> records have > > >>> > > > > > > >>> a > > >>> > > > > > > >>>>>>>>>> value of zero. In those cases, we could > benefit > > >>> of the emit on > > >>> > > > > > > >>>>>>>>>> change, but only if we define the semantics > of the > > >>> > > > > > aggregations > > >>> > > > > > > >>>>>>>>>> to not use the highest timestamp of the > > >>> participating records > > >>> > > > > > > >>> for > > >>> > > > > > > >>>>>>>>>> the result. In Kafka Streams, we do not have > min, > > >>> max, and sum > > >>> > > > > > > >>> as > > >>> > > > > > > >>>>>>>>>> explicit aggregations, but we need to provide > an > > >>> API to define > > >>> > > > > > > >>>>>>>>>> what timestamp should be used for the result > of > > >>> an aggregation > > >>> > > > > > > >>> if > > >>> > > > > > > >>>>>>>>>> we want to go down this path. All of this does > > >>> not block this > > >>> > > > > > > >>> KIP > > >>> > > > > > > >>>>>>>>>> and I just wanted to put this aspects up for > > >>> discussion. The > > >>> > > > > > > >>> KIP > > >>> > > > > > > >>>>>>>>>> can limit itself to emit from materialized > > >>> KTables. However, > > >>> > > > > > > >>> the > > >>> > > > > > > >>>>>>>>>> limits should be explicitly stated in the KIP. > > >>> Best, Bruno > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> On Fri, Jan 24, 2020 at 10:58 AM Matthias J. > Sax > > >>> > > > > > > >>>>>>>>>> <matth...@confluent.io <mailto: > > >>> matth...@confluent.io>> wrote: > > >>> > > > > > > >>>>>>>>>> IMHO, the question about semantics depends on > the > > >>> use case, in > > >>> > > > > > > >>>>>>>>>> particular on the origin of a KTable. If > there is > > >>> a changlog > > >>> > > > > > > >>>>>>>>>> topic that one reads directly into a KTable, > > >>> emit-on-change > > >>> > > > > > > >>> does > > >>> > > > > > > >>>>>>>>>> actually make sense, because the timestamp > > >>> indicates _when_ > > >>> > > > > > the > > >>> > > > > > > >>>>>>>>>> update was _effective_. For this case, it is > > >>> semantically > > >>> > > > > > sound > > >>> > > > > > > >>>>>>>>>> to _not_ update the timestamp in the store, > > >>> because the second > > >>> > > > > > > >>>>>>>>>> update is actually idempotent and advancing > the > > >>> timestamp is > > >>> > > > > > > >>> not > > >>> > > > > > > >>>>>>>>>> ideal (one could even consider it to be wrong > to > > >>> advance the > > >>> > > > > > > >>>>>>>>>> timestamp) because the "valid time" of the > record > > >>> pair did not > > >>> > > > > > > >>>>>>>>>> change. This reasoning also applies to > > >>> KTable-KTable joins. > > >>> > > > > > > >>>>>>>>>> However, if the KTable is the result of an > > >>> aggregation, I > > >>> > > > > > think > > >>> > > > > > > >>>>>>>>>> emit-on-update is more natural, because the > > >>> timestamp reflects > > >>> > > > > > > >>>>>>>>>> the _last_ time (ie, highest timestamp) of all > > >>> input records > > >>> > > > > > > >>> the > > >>> > > > > > > >>>>>>>>>> contributed to the result. Hence, updating the > > >>> timestamp and > > >>> > > > > > > >>>>>>>>>> emitting a new record actually sounds correct > to > > >>> me. This > > >>> > > > > > > >>> applies > > >>> > > > > > > >>>>>>>>>> to windowed and non-windowed aggregations > IMHO. > > >>> However, > > >>> > > > > > > >>>>>>>>>> considering the argument that the timestamp > > >>> should not be > > >>> > > > > > > >>> update > > >>> > > > > > > >>>>>>>>>> in the first case in the store to begin with, > > >>> both cases are > > >>> > > > > > > >>>>>>>>>> actually the same, and both can be modeled as > > >>> emit-on-change: > > >>> > > > > > > >>> if > > >>> > > > > > > >>>>>>>>>> a `table()` operator does not update the > > >>> timestamp if the > > >>> > > > > > value > > >>> > > > > > > >>>>>>>>>> does not change, there is _no_ change and thus > > >>> nothing is > > >>> > > > > > > >>>>>>>>>> emitted. At the same time, if an aggregation > > >>> operator does > > >>> > > > > > > >>> update > > >>> > > > > > > >>>>>>>>>> the timestamp (even if the value does not > change) > > >>> there _is_ a > > >>> > > > > > > >>>>>>>>>> change and we emit. Note that handling > > >>> out-of-order data for > > >>> > > > > > > >>>>>>>>>> aggregations would also work seamlessly with > this > > >>> approach -- > > >>> > > > > > > >>> for > > >>> > > > > > > >>>>>>>>>> out-of-order records, the timestamp does never > > >>> change, and > > >>> > > > > > > >>> thus, > > >>> > > > > > > >>>>>>>>>> we only emit if the result itself changes. > > >>> Therefore, I would > > >>> > > > > > > >>>>>>>>>> argue that we might not even need any config, > > >>> because the > > >>> > > > > > > >>>>>>>>>> emit-on-change behavior is just correct and > > >>> reduced the > > >>> > > > > > > >>>>>>>>>> downstream load, while our current behavior is > > >>> not ideal (even > > >>> > > > > > > >>> if > > >>> > > > > > > >>>>>>>>>> it's also correct). Thoughts? -Matthias On > > >>> 1/24/20 9:37 AM, > > >>> > > > > > > >>> John > > >>> > > > > > > >>>>>>>>>> Roesler wrote: Hi Bruno, Thanks for that > idea. I > > >>> hadn't > > >>> > > > > > > >>>>>>>>>> considered that option before, and it does > seem > > >>> like that > > >>> > > > > > would > > >>> > > > > > > >>>>>>>>>> be the right place to put it if we think it > might > > >>> be > > >>> > > > > > > >>> semantically > > >>> > > > > > > >>>>>>>>>> important to control on a table-by-table > basis. I > > >>> had been > > >>> > > > > > > >>>>>>>>>> thinking of it less semantically and more > > >>> practically. In the > > >>> > > > > > > >>>>>>>>>> context of a large topology, or more > generally, a > > >>> large > > >>> > > > > > > >>> software > > >>> > > > > > > >>>>>>>>>> system that contains many topologies and other > > >>> event-driven > > >>> > > > > > > >>>>>>>>>> systems, each no-op result becomes an input > that > > >>> is destined > > >>> > > > > > to > > >>> > > > > > > >>>>>>>>>> itself become a no-op result, and so on, all > the > > >>> way through > > >>> > > > > > > >>> the > > >>> > > > > > > >>>>>>>>>> system. Thus, a single pointless processing > > >>> result becomes > > >>> > > > > > > >>>>>>>>>> amplified into a large number of pointless > > >>> computations, cache > > >>> > > > > > > >>>>>>>>>> perturbations, and network and disk I/O > > >>> operations. If you > > >>> > > > > > also > > >>> > > > > > > >>>>>>>>>> consider operations with fan-out implications, > > >>> like branching > > >>> > > > > > > >>> or > > >>> > > > > > > >>>>>>>>>> foreign-key joins, the wasted resources are > > >>> amplified not just > > >>> > > > > > > >>> in > > >>> > > > > > > >>>>>>>>>> proportion to the size of the system, but the > > >>> size of the > > >>> > > > > > > >>> system > > >>> > > > > > > >>>>>>>>>> times the average fan-out (to the power of the > > >>> number of > > >>> > > > > > > >>> fan-out > > >>> > > > > > > >>>>>>>>>> operations on the path(s) through the > system). In > > >>> my time > > >>> > > > > > > >>>>>>>>>> operating such systems, I've observed these > > >>> effects to be very > > >>> > > > > > > >>>>>>>>>> real, and actually, the system and use case > > >>> doesn't have to be > > >>> > > > > > > >>>>>>>>>> very large before the amplification poses an > > >>> existential > > >>> > > > > > threat > > >>> > > > > > > >>>>>>>>>> to the system as a whole. This is the basis > of my > > >>> advocating > > >>> > > > > > > >>> for > > >>> > > > > > > >>>>>>>>>> a simple behavior change, rather than an > opt-in > > >>> config of any > > >>> > > > > > > >>>>>>>>>> kind. It seems like Streams should "do the > right > > >>> thing" for > > >>> > > > > > the > > >>> > > > > > > >>>>>>>>>> majority use case. My theory (which may be > wrong) > > >>> is that the > > >>> > > > > > > >>>>>>>>>> majority use case is more like "relational > > >>> queries" than "CEP > > >>> > > > > > > >>>>>>>>>> queries". Even if you were doing some > > >>> event-sensitive > > >>> > > > > > > >>>>>>>>>> computation, wouldn't you do them as Stream > > >>> operations (where > > >>> > > > > > > >>>>>>>>>> this feature is inapplicable anyway)? In > keeping > > >>> with the > > >>> > > > > > > >>>>>>>>>> "practical" perspective, I suggested the > opt-out > > >>> config only > > >>> > > > > > in > > >>> > > > > > > >>>>>>>>>> the (I think unlikely) event that filtering > out > > >>> pointless > > >>> > > > > > > >>> updates > > >>> > > > > > > >>>>>>>>>> actually harms performance. I'd also be > perfectly > > >>> fine without > > >>> > > > > > > >>>>>>>>>> the opt-out config. I really think that > (because > > >>> of the > > >>> > > > > > > >>> timestamp > > >>> > > > > > > >>>>>>>>>> semantics work already underway), we're > already > > >>> pre-fetching > > >>> > > > > > > >>> the > > >>> > > > > > > >>>>>>>>>> prior result most of the time, so there would > > >>> actually be very > > >>> > > > > > > >>>>>>>>>> little extra I/O involved in implementing > > >>> emit-on-change. > > >>> > > > > > > >>>>>>>>>> However, we should consider whether my > experience > > >>> is likely to > > >>> > > > > > > >>>>>>>>>> be general. Do you have some use case in mind > for > > >>> which you'd > > >>> > > > > > > >>>>>>>>>> actually want some KTable results to be > > >>> emit-on-update for > > >>> > > > > > > >>>>>>>>>> semantic reasons? Thanks, -John > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> On Fri, Jan 24, 2020, at 11:02, Bruno Cadonna > > >>> wrote: Hi > > >>> > > > > > > >>> Richard, > > >>> > > > > > > >>>>>>>>>> Thank you for the KIP. I agree with John that > we > > >>> should focus > > >>> > > > > > > >>> on > > >>> > > > > > > >>>>>>>>>> the interface and behavior change in a KIP. We > > >>> can discuss the > > >>> > > > > > > >>>>>>>>>> implementation later. I am also +1 for the > > >>> survey. I had a > > >>> > > > > > > >>>>>>>>>> thought about this. Couldn't we consider > > >>> emit-on-change to be > > >>> > > > > > > >>> one > > >>> > > > > > > >>>>>>>>>> config of suppress (like `untilWindowCloses`)? > > >>> What you > > >>> > > > > > > >>>>>>>>>> basically propose is to suppress updates if > they > > >>> do not change > > >>> > > > > > > >>>>>>>>>> the result. Considering emit on change as a > > >>> flavour of > > >>> > > > > > suppress > > >>> > > > > > > >>>>>>>>>> would be more flexible because it would > specify > > >>> the behavior > > >>> > > > > > > >>>>>>>>>> locally for a KTable instead of globally for > all > > >>> KTables. > > >>> > > > > > > >>>>>>>>>> Additionally, specifying the behavior in one > > >>> place instead of > > >>> > > > > > > >>>>>>>>>> multiple places feels more intuitive and > > >>> consistent to me. > > >>> > > > > > > >>> Best, > > >>> > > > > > > >>>>>>>>>> Bruno On Fri, Jan 24, 2020 at 7:49 AM John > Roesler > > >>> > > > > > > >>>>>>>>>> <vvcep...@apache.org <mailto: > vvcep...@apache.org>> > > >>> wrote: Hi > > >>> > > > > > > >>>>>>>>>> Richard, Thanks for picking this up! I know > of at > > >>> least one > > >>> > > > > > > >>> large > > >>> > > > > > > >>>>>>>>>> community member for which this feature is > > >>> absolutely > > >>> > > > > > > >>> essential. > > >>> > > > > > > >>>>>>>>>> If I understand your two options, it seems > like > > >>> the proposal > > >>> > > > > > is > > >>> > > > > > > >>>>>>>>>> to implement it as a behavior change > regardless, > > >>> and the > > >>> > > > > > > >>> question > > >>> > > > > > > >>>>>>>>>> is whether to provide an opt-out config or > not. > > >>> Given that any > > >>> > > > > > > >>>>>>>>>> implementation of this feature would have some > > >>> performance > > >>> > > > > > > >>> impact > > >>> > > > > > > >>>>>>>>>> under some workloads, and also that we don't > know > > >>> if anyone > > >>> > > > > > > >>>>>>>>>> really depends on emit-on-update time > semantics, > > >>> it seems like > > >>> > > > > > > >>> we > > >>> > > > > > > >>>>>>>>>> should propose to add an opt-out config. Can > you > > >>> update the > > >>> > > > > > KIP > > >>> > > > > > > >>>>>>>>>> to mention the exact config key and value(s) > > >>> you'd propose? > > >>> > > > > > > >>> Just > > >>> > > > > > > >>>>>>>>>> to move the discussion forward, maybe > something > > >>> like: emit.on > > >>> > > > > > > >>> := > > >>> > > > > > > >>>>>>>>>> change|update with the new default being > "change" > > >>> Thanks for > > >>> > > > > > > >>>>>>>>>> pointing out the timestamp issue in > particular. I > > >>> agree that > > >>> > > > > > if > > >>> > > > > > > >>>>>>>>>> we discard the latter update as a no-op, then > we > > >>> also have to > > >>> > > > > > > >>>>>>>>>> discard its timestamp (obviously, we don't > > >>> forward the > > >>> > > > > > > >>> timestamp > > >>> > > > > > > >>>>>>>>>> update, as that's the whole point, but we also > > >>> can't update > > >>> > > > > > the > > >>> > > > > > > >>>>>>>>>> timestamp in the store, as the store must > remain > > >>> consistent > > >>> > > > > > > >>> with > > >>> > > > > > > >>>>>>>>>> what has been emitted). I have to confess > that I > > >>> disagree with > > >>> > > > > > > >>>>>>>>>> your implementation proposal, but it's also > not > > >>> necessary to > > >>> > > > > > > >>>>>>>>>> discuss implementation in the KIP. Maybe it > would > > >>> be less > > >>> > > > > > > >>>>>>>>>> controversial if you just drop that section > for > > >>> now, so that > > >>> > > > > > > >>> the > > >>> > > > > > > >>>>>>>>>> KIP discussion can focus on the behavior > change > > >>> and config. > > >>> > > > > > > >>> Just > > >>> > > > > > > >>>>>>>>>> for reference, there is some research into > this > > >>> domain. For > > >>> > > > > > > >>>>>>>>>> example, see the "Report" section (3.2.3) of > the > > >>> SECRET paper: > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>> > > >>> > > > > > > > >>> > https://nam04.safelinks.protection.outlook.com/?url=http%3A%2F%2Fpeop > > >>> > > > > > > > le.csail.mit.edu > > >>> > > > > > > >>>>>> > > >>> %2Ftatbul%2Fpublications%2Fmaxstream_vldb10.pdf&data > > >>> > > > > > > > =02%7C01%7CThomas.Becker%40tivo.com > > >>> > > > > > > >>>>>> %7Ce0235483b1eb4f259c5c08d7a8d1c16b%7 > > >>> > > > > > > > > > >>> > > > > > > >>>>>> > > >>> > > > > > > >>> > > >>> > > > > > > > >>> > Cd05b7c6912014c0db45d7f1dcc227e4d%7C1%7C1%7C637163491160859282&sdata > > >>> > > > > > > > > > >>> =4dSGIS8jNPAPP7B48r9e%2BUgFh3WdmzVyXhyT63eP8dI%3D&reserved=0 > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > > It might help to round out the proposal if you take a > brief > > >>> > > > > > > >>> survey of > > >>> > > > > > > >>>>>>>>>> the behaviors of other systems, along with > pros > > >>> and cons if > > >>> > > > > > any > > >>> > > > > > > >>>>>>>>>> are reported. Thanks, -John > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> On Fri, Jan 10, 2020, at 22:27, Richard Yu > wrote: > > >>> Hi > > >>> > > > > > everybody! > > >>> > > > > > > >>>>>>>>>> I'd like to propose a change that we probably > > >>> should've added > > >>> > > > > > > >>> for > > >>> > > > > > > >>>>>>>>>> a long time now. The key benefit of this KIP > > >>> would be reduced > > >>> > > > > > > >>>>>>>>>> traffic in Kafka Streams since a lot of no-op > > >>> results would no > > >>> > > > > > > >>>>>>>>>> longer be sent downstream. Here is the KIP for > > >>> reference. > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>> > > >>> > > > > > > > >>> > https://nam04.safelinks.protection.outlook.com/?url=https%3A%2F%2Fcwi > > >>> > > > > > > > ki.apache.org > > >>> > > > > > > >>>>>> > > >>> %2Fconfluence%2Fdisplay%2FKAFKA%2FKIP-557%253A%2BAdd%2Bemit > > >>> > > > > > > > > > >>> > > > > > > >>>>>> > > >>> > > > > > > >>> > > >>> > > > > > > > >>> > %2Bon%2Bchange%2Bsupport%2Bfor%2BKafka%2BStreams&data=02%7C01%7CThom > > >>> > > > > > > > as.Becker%40tivo.com > > >>> > > > > > > >>>>>> > %7Ce0235483b1eb4f259c5c08d7a8d1c16b%7Cd05b7c6912014c > > >>> > > > > > > > > > >>> > > > > > > >>>>>> > > >>> > > > > > > >>> > > >>> > > > > > > > >>> > 0db45d7f1dcc227e4d%7C1%7C1%7C637163491160869277&sdata=zYpCSFOsyN4%2B > > >>> > > > > > > > 4rKRZBQ%2FZvcGQ4EINR9Qm6PLsB7EKrc%3D&reserved=0 > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > > Currently, I seek to formalize our approach for this > KIP > > >>> first > > >>> > > > > > > >>> before > > >>> > > > > > > >>>>>>>>>> we determine concrete API additions / > > >>> configurations. Some > > >>> > > > > > > >>>>>>>>>> configs might warrant adding, whiles others > are > > >>> not necessary > > >>> > > > > > > >>>>>>>>>> since adding them would only increase > complexity > > >>> of Kafka > > >>> > > > > > > >>>>>>>>>> Streams. Cheers, Richard > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> ________________________________ This email > and > > >>> any > > >>> > > > > > attachments > > >>> > > > > > > >>>>>>>>>> may contain confidential and privileged > material > > >>> for the sole > > >>> > > > > > > >>> use > > >>> > > > > > > >>>>>>>>>> of the intended recipient. Any review, > copying, or > > >>> > > > > > distribution > > >>> > > > > > > >>>>>>>>>> of this email (or any attachments) by others > is > > >>> prohibited. If > > >>> > > > > > > >>>>>>>>>> you are not the intended recipient, please > > >>> contact the sender > > >>> > > > > > > >>>>>>>>>> immediately and permanently delete this email > and > > >>> any > > >>> > > > > > > >>>>>>>>>> attachments. No employee or agent of TiVo is > > >>> authorized to > > >>> > > > > > > >>>>>>>>>> conclude any binding agreement on behalf of > TiVo > > >>> by email. > > >>> > > > > > > >>>>>>>>>> Binding agreements with TiVo may only be made > by > > >>> a signed > > >>> > > > > > > >>> written > > >>> > > > > > > >>>>>>>>>> agreement. -- *Tommy Becker* *Principal > Engineer * > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> *Personalized Content Discovery* > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> *O* +1 919.460.4747 *tivo.com* < > > >>> http://www.tivo.com/> > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> > > >>> > > > > > > >>>>>>>>>> This email and any attachments may contain > > >>> confidential and > > >>> > > > > > > >>>>>>>>>> privileged material for the sole use of the > > >>> intended > > >>> > > > > > recipient. > > >>> > > > > > > >>>>>>>>>> Any review, copying, or distribution of this > > >>> email (or any > > >>> > > > > > > >>>>>>>>>> attachments) by others is prohibited. If you > are > > >>> not the > > >>> > > > > > > >>> intended > > >>> > > > > > > >>>>>>>>>> recipient, please contact the sender > immediately > > >>> and > > >>> > > > > > > >>> permanently > > >>> > > > > > > >>>>>>>>>> delete this email and any attachments. No > > >>> employee or agent of > > >>> > > > > > > >>>>>>>>>> TiVo is authorized to conclude any binding > > >>> agreement on behalf > > >>> > > > > > > >>> of > > >>> > > > > > > >>>>>>>>>> TiVo by email. Binding agreements with TiVo > may > > >>> only be made > > >>> > > > > > > >>> by a > > >>> > > > > > > >>>>>>>>>> signed written agreement. > > >>> > > > > > > >>>>>>>>> > > >>> > > > > > > >>>>>>>>> -- > > >>> > > > > > > >>>>>>>>> > > >>> > > > > > > >>>>>>>>> *Tommy Becker* /Principal Engineer / > > >>> > > > > > > >>>>>>>>> > > >>> > > > > > > >>>>>>>>> /Personalized Content Discovery/ > > >>> > > > > > > >>>>>>>>> > > >>> > > > > > > >>>>>>>>> *O* +1 919.460.4747 *tivo.com* < > > >>> http://www.tivo.com/> > > >>> > > > > > > >>>>>>>>> > > >>> > > > > > > >>>>>>>>> > > >>> > > > > > > >>>>>>>>> > > >>> > > > > > > >>>>>> > > >>> > > > > > > > >>> > ---------------------------------------------------------------------- > > >>> > > > > > > >>>>>>> > > >>> > > > > > > >>>>>> > > >>> > > > > > > >>>>> > > >>> > > > > > > >>>> > > >>> > > > > > > >>> > > >>> > > > > > > >> > > >>> > > > > > > > > > >>> > > > > > > > > >>> > > > > > > > > >>> > > > > > > Attachments: > > >>> > > > > > > * signature.asc > > >>> > > > > > > > >>> > > > > > >>> > > > >>> > > >> > > >