Hi Richard, Thanks for the update!
I read it over, and overall it looks good! I have only a minor concern about the rate metric definition: > The rate option indicates the ratio of records dropped to actual volume of > records passing through the task That's not the definition of a "rate". It should be something more like "the average number of dropped idempotent updates per second". Incidentally, I mentioned this KIP to Guozhang, and he brought up an interesting concern I'd like to share with the list. He noted that if we filter out idempotent table updates, stream time will not advance with every input event anymore. He asked if this would have a negative impact on operations that depend on stream time. I think this is a valid concern. For example, you can use Suppress to buffer KTable updates until a certain amount of stream time passes. Specifically, the Suppress processor itself advances stream time as it receives new records to its `process` method. In the pathological case, all updates are idempotent, get dropped, and the Suppress operator never emits anything, even though to an outside observer, stream time should have advanced. Example scenario: > inputTable = builder.table(input) > suppressed = inputTable.suppress(untilTimeLimit(10)) > > input: key=A, timestamp=0, value=X > inputTable: key=A, timestamp=0, value=X > suppressed buffer: key=A, timestamp=0, value=X (observed stream time = 0) > output: (nothing) > > input: key=A, timestamp=11, value=X > // update is idempotent, so it gets dropped > inputTable: key=A, timestamp=0, value=X > suppressed buffer: key=A, timestamp=0, value=X (observed stream time = 0) > output: (nothing) Note that even though stream time has technically advanced beyond the suppression config of 10, the suppression buffer doesn't see it because the KTable source already dropped the idempotent update. I'm thinking that this situation is indeed concerning, and we should be aware and make a note of it. However, I don't think that it should change our proposal. To understand why, I'll enumerate all the usages of stream time in Kafka Streams: windowed operations ----------------------------- KStreamSessionWindowAggregate & KStreamWindowAggregate: - used to determine if an incoming record belongs to a window that is already closed Suppressed.untilWindowCloses(): - used to flush out results for windows, once they are closed AbstractRocksDBSegmentedBytesStore & InMemorySessionStore & InMemoryWindowStore: - used to create new segments and drop old ones that are out of retention non-windowed operations ----------------------------------- Suppressed.untilTimeLimit - used to flush out prior updates that have passed the configured age, in stream time Note that most of the usages are in windowed operations. One interesting thing about this context is that records with higher timestamps belong to different windows. In order to advance stream time far enough to close a window or push it out of retention, the new records must have timestamps that in later windows, which means that they are updates to _new_ keys, which means they would not be suppressed as idempotent. Updates within a window could still be suppressed, though: For example, if the window size is 10, and the grace period is 5, and we get updates all for the same key with timestamps 0, 11, and 16, today, we would emit the record for the [0,10) window as soon as we got the update at time 16 (since stream time is now past the window end + grace period time of 15). But if we drop the time=16 update, we wouldn't emit the [0,10) window results until we get a record with time >= 20. You can see that the maximum amount of (stream) time that dropping idempotent updates could delay updates is one window. This might be surprising, but it doesn't seem too bad, especially since Suppress explicitly does _not_ promise to emit the results at the earliest possible time, just at some time after the window closes. Even better, though, all the windowed aggregations are _stream_ aggregations that _produce_ a KTable, and we already decided that (at least for now), we would include the timestamp in the idempotence check for stream aggregation results. With this strategy, we would actually not suppress _any_ update to a stream aggregation (windowed or otherwise) that advances stream time. So there's no problem at all with windowed operations. This only leaves non-windowed operations, of which there's only one. I have to admit that I regret Suppressed.untilTimeLimit. It turns out that everyone I've heard of who used this API actually wanted it to be wall-clock based, not stream-time based. So, I'm not sure in practice if this sharp edge will actually cut anyone. I think a "proper" fix would be to introduce some kind of control message to advance stream time independently of records. We've already talked about this a little in KAFKA-8769, and it would also be necessary for global consistency markers. Basically, once a record is dropped as an idempotent update, we could still forward a time marker through the topology, which could trigger suppressions, as well as window boundaries, but wouldn't result in any computations. But practically speaking, I'm really not confident that there's anyone really using the untilTimeLimit suppression, and even if they are, if they would really see consecutive idempotent updates for long enough to really have an observable impact on what gets emitted from the suppression buffer. In fact, if some hypothetical person did find themselves on the wrong end of my assumption, they could _remove_ the suppression, and rely instead on idempotence-dropping to get the same level of traffic reduction that they enjoy from the suppression. Anyway, long story short, I'm advocating to continue with the current proposal and just make a note of the situations I've outlined above. Thanks for your time, -John On Thu, Feb 27, 2020, at 14:33, Richard Yu wrote: > Hi all, > > I might've made a minor mistake. The processor node level is level 3, not > level 1. > I will correct the KIP accordingly. > > After looking over things, I decided to start the voting thread this > afternoon. > > Cheers, > Richard > > On Thu, Feb 27, 2020 at 12:29 PM Richard Yu <yohan.richard...@gmail.com> > wrote: > > > Hi Bruno, Hi John, > > > > Thanks for your comments! I updated the KIP accordingly, and it looks like > > for quite a few points. I was doing some beating around the bush which > > could've been avoided. > > > > Looks like we can reduce the metric to Level 1 (per processor node) then. > > > > I've cleaned up most of the unnecessary info, and we should be fairly > > close. > > I will start working on a PR soon for this KIP. (although we might split > > that up into stages) > > > > Cheers, > > Richard > > > > On Thu, Feb 27, 2020 at 6:06 AM Bruno Cadonna <br...@confluent.io> wrote: > > > >> Hi John, > >> > >> I agree with you. It is better to measure the metric on processor node > >> level. The users can do the rollup to task-level by themselves. > >> > >> Best, > >> Bruno > >> > >> On Thu, Feb 27, 2020 at 12:09 AM John Roesler <vvcep...@apache.org> > >> wrote: > >> > > >> > Hi Richard, > >> > > >> > I've been making a final pass over the KIP. > >> > > >> > Re: Proposed Behavior Change: > >> > > >> > I think this point is controversial and probably doesn't need to be > >> there at all: > >> > > 2.b. In certain situations where there is a high volume of idempotent > >> > > updates throughout the Streams DAG, it will be recommended practice > >> > > to materialize all operations to reduce traffic overall across the > >> entire > >> > > network of nodes. > >> > > >> > Re-reading all the points, it seems like we can sum them up in a way > >> that's > >> > a little more straight to the point, and gives us the right amount of > >> flexibility: > >> > > >> > > Proposed Behavior Changes > >> > > > >> > > Definition: "idempotent update" is one in which the new result and > >> prior > >> > > result, when serialized, are identical byte arrays > >> > > > >> > > Note: an "update" is a concept that only applies to Table operations, > >> so > >> > > the concept of an "idempotent update" also only applies to Table > >> operations. > >> > > See > >> https://kafka.apache.org/documentation/streams/developer-guide/dsl-api.html#streams_concepts_ktable > >> > > for more information. > >> > > > >> > > Given that definition, we propose for Streams to drop idempotent > >> updates > >> > > in any situation where it's possible and convenient to do so. For > >> example, > >> > > any time we already have both the prior and new results serialized, we > >> > > may compare them, and drop the update if it is idempotent. > >> > > > >> > > Note that under this proposal, we can implement idempotence checking > >> > > in the following situations: > >> > > 1. Any aggregation (for example, KGroupedStream, KGroupedTable, > >> > > TimeWindowedKStream, and SessionWindowedKStream operations) > >> > > 2. Any Materialized KTable operation > >> > > 3. Repartition operations, when we need to send both prior and new > >> results > >> > > >> > Notice that in my proposed wording, we neither limit ourselves to just > >> the > >> > situations enumerated, nor promise to implement the optimization in > >> every > >> > possible situation. IMHO, this is the best way to propose such a > >> feature. > >> > That way, we have the flexibility to implement it in stages, and also > >> to add > >> > on to the implementation in the future. > >> > > >> > > >> > Re: Metrics > >> > > >> > I agree with Bruno, although, I think it might just be a confusing > >> statement. > >> > It might be clearer to drop all the "discussion", and just say: "We > >> will add a > >> > metric to count the number of idempotent updates that we have dropped". > >> > > >> > Also, with respect to the metric, I'm wondering if the metric should be > >> task- > >> > level or processor-node-level. Since the interesting action takes place > >> inside > >> > individual processor nodes, I _think_ it would be higher leverage to > >> just > >> > measure it at the node level. WDYT? > >> > > >> > Re: Design Reasoning > >> > > >> > This section seems to be a little bit outdated. I also just noticed a > >> "surprise" > >> > configuration "timestamp.aggregation.selection.policy" hidden in point > >> 1.a. > >> > Is that part of the proposal? We haven't discussed it, and I think we > >> were > >> > talking about this KIP being "configuration free". > >> > > >> > There is also some discussion of discarded alternative in the Design > >> Reasoning > >> > section, which is confusing. Finally, there was a point there I didn't > >> understand > >> > at all, about stateless operators not being intended to load prior > >> results. > >> > This statement doesn't seem to be true, but it also doesn't seem to be > >> relevant, > >> > so maybe we can just drop it. > >> > > >> > Overall, it might help if you make a pass on this section, and just > >> discuss as > >> > briefly as possible the justification for the proposed behavior change, > >> and > >> > not adding a configuration. Try to avoid talking about things that we > >> are not > >> > proposing, since that will just lead to confusion. > >> > > >> > Similarly, I'd just completely remove the "Implementation [discarded]" > >> section. > >> > It was good to have this as part of the discussion initially, but as we > >> move > >> > toward a vote, it's better to just streamline the KIP document as much > >> as > >> > possible. Keeping a "discarded" section in the document will just make > >> it > >> > harder for new people to understand the proposal. We did the same thing > >> > with KIP-441, where there were two prior drafts included at the end of > >> the > >> > document, and we just deleted them for clarity. > >> > > >> > I liked the "Compatibility" and "Rejected Alternatives" section. Very > >> clear > >> > and to the point. > >> > > >> > Thanks again for the contribution! I think once the KIP document is > >> cleaned > >> > up, we'll be in good shape to finalize the discussion. > >> > -John > >> > > >> > > >> > On Wed, Feb 26, 2020, at 07:27, Bruno Cadonna wrote: > >> > > Hi Richard, > >> > > > >> > > 1. Could you change "idempotent update operations will only be dropped > >> > > from KTables, not from other classes." -> idempotent update operations > >> > > will only be dropped from materialized KTables? For non-materialized > >> > > KTables -- as they can occur after optimization of the topology -- we > >> > > cannot drop idempotent updates. > >> > > > >> > > 2. I cannot completely follow the metrics section. Do you want to > >> > > record all idempotent updates or only the dropped ones? In particular, > >> > > I do not understand the following sentences: > >> > > "For that matter, even if we don't drop idempotent updates, we should > >> > > at the very least record the number of idempotent updates that has > >> > > been seen go through a particular processor." > >> > > "Therefore, we should add some metrics which will count the number of > >> > > idempotent updates that each node has seen." > >> > > I do not see how we can record idempotent updates that we do not drop. > >> > > If we see them, we should drop them. If we do not see them, we cannot > >> > > drop them and we cannot record them. > >> > > > >> > > Best, > >> > > Bruno > >> > > > >> > > On Wed, Feb 26, 2020 at 4:57 AM Richard Yu < > >> yohan.richard...@gmail.com> wrote: > >> > > > > >> > > > 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 > >> > > > > > >>> > > > > > > >> > > > > > >>> > > > > >> > > > > > >>> > > >> > > > > > >>> > >> > > > > > >> > >> > > > > > > >> > > > > > >> > > > >> > > > >> > > >
