Happy to name it BatchWindows. Will give some people time to chime in and then change the name.
- Almog On Tue, Feb 4, 2025 at 11:10 PM Sophie Blee-Goldman <sop...@responsive.dev> wrote: > One minor suggestion: use BatchWindows instead of BatchedWindows. The > version without the "ed" matches up with the established naming pattern and > grammar used by other Windows classes: eg TimeWindows, SessionWindows, > SlidingWindows > > Not a big deal though, won't redact my +1 on the voting thread if you > prefer to keep it as BatchedWindows > > On Tue, Feb 4, 2025 at 10:51 AM Almog Gavra <almog.ga...@gmail.com> wrote: > > > Thanks for the discussion everyone! I've updated the Wiki with the > > following changes: > > > > - Renamed to BatchedWindows > > - Add a note in rejected alternatives about more general purpose > > (micro-)batching functionality since the scope of that is much wider. > > > > Since it looks like we've stabilized the discussion I'm going to go ahead > > and open up the vote! Definitely feel free to take another look and leave > > any additional thoughts. > > > > > > On Thu, Jan 30, 2025 at 12:28 PM Matthias J. Sax <mj...@apache.org> > wrote: > > > > > > batch window with max N records, > > > >> and then also specifying a BufferConfig.maxRecords() > > > > > > That's actually two different and independent dimensions. "N records" > > > would be the number of records in the window, but `maxRecords` is the > > > number of unique keys/row in the buffer before it's flushed. > > > > > > > > > > > > > the current proposal > > > > that only includes stream time based batches is no worse than > existing > > > > windows which also have that potential indeterminism in this scenario > > > > except for that it's more likely since grace period is always 0 > > > > > > Guess this becomes a philosophical question. For TimeWindows, even if > > > grace=0, it's still deterministic into which window each record goes -- > > > of course, the "cut off point" when we start to drop late records is > > > subject to "noise" due to repartitioning, as stream-time advances "non > > > deterministically". > > > > > > So if we drop different records on a "re-run" on the same input data, > we > > > might still get different windows (as different records would have been > > > dropped). But that is exactly why we have a grace-period to begin with > > > (to avoid it, and to make a re-run deterministic -- I think of a "too > > > short" grace period as a misconfiguration -- or an informed decision to > > > sacrifice determinism for something else)... > > > > > > And as the new window type, by definition, does not need/want a > > > grace-period, IMHO, the new window type would be "worse" (for the lack > > > of a better word...); I don't think its really worse, it just > inherently > > > non-deterministic, and that's fine. > > > > > > Guess we are overall on the same page and share common understanding of > > > the trade-off. So I think we are good :) > > > > > > > > > -Matthias > > > > > > > > > On 1/30/25 9:07 AM, Almog Gavra wrote: > > > > I'm not opposed to "BatchedWindows" - I think I like that the most so > > > far. > > > > I'll let that sit on the discussion thread for a while, and change > the > > > KIP > > > > to match if no concerns. > > > > > > > >> What I don't understand is, why the relationship to > > > > suppress()/emitStrategy() is relevant? Can you elaborate a little > bit? > > > > > > > > The existing proposal has no impact on suppression/emitStrategy, but > > I'm > > > > uncertain it's unrelated if you start to introduce window constraints > > > that > > > > aren't just stream time. Imagine having a batch window with max N > > > records, > > > > and then also specifying a BufferConfig.maxRecords() suppression... > > what > > > > happens in that case? With the current scope of the KIP, we don't > need > > to > > > > worry about any of that so it's pretty well contained. > > > > > > > >> if data is auto-repartitioned before a `BatchWindows` step, > > > > repartitioning introduces non-deterministic order in the repartition > > > topic > > > > > > > > This is a good point, I did not think about it (my original comment > was > > > > specifically about wall clock time). I guess, though, the current > > > proposal > > > > that only includes stream time based batches is no worse than > existing > > > > windows which also have that potential indeterminism in this scenario > > > > except for that it's more likely since grace period is always 0. > > > > > > > >> I don't see any relationship to EOS or ALOS. Can you explain what > you > > > > mean? > > > > > > > > Hmm, I'm not totally sure what I was thinking... we can ignore that > > part. > > > > > > > > - Almog > > > > > > > > > > > > On Tue, Jan 28, 2025 at 3:40 PM Matthias J. Sax <mj...@apache.org> > > > wrote: > > > > > > > >> Interesting thoughts. So maybe we could go with `BatchWindows` as a > > > >> name? Again, only spit-balling... > > > >> > > > >> If we really put "(micro-)batching" in the center of this idea, I > > think > > > >> both count-based and time-based (and time could actually be either > > > >> stream-time or wall-clock-time), or any combination of these > > dimensions > > > >> could all make sense. > > > >> > > > >> Having said this: I don't think we need to support all dimensions > > > >> initially, but if we add something like `BatchWindows` we can extent > > the > > > >> supported specification incrementally. > > > >> > > > >> > > > >> > > > >> What I don't understand is, why the relationship to > > > >> suppress()/emitStrategy() is relevant? Can you elaborate a little > bit? > > > >> > > > >> > > > >> > > > >>> how to make it deterministic with regards to time > > > >> > > > >> I think, in any case, we are leaving determinism-land (at least to > > some > > > >> extent). I guess, as long as the `BatchWindows` are applied directly > > to > > > >> an input topic, we get determinism for both stream-time size, as > well > > as > > > >> count-size windows (not for wall-clock0time, of course). > > > >> > > > >> However, if data is auto-repartitioned before a `BatchWindows` step, > > > >> repartitioning introduces non-deterministic order in the repartition > > > >> topic due to interleaved writes, and thus, also stream-time based > > > >> windows would become non-deterministic (as there is no grace-period > by > > > >> design to "fix" the non-deterministic order, in contrast to full > > > >> event-time based windows we support so far). > > > >> > > > >> So I don't think that is an actual difference between stream-time or > > > >> count-based `BatchWindows` with regard to determinism. > > > >> > > > >> > > > >> > > > >>> which is important for EOS and even ALOS > > > >> > > > >> I don't see any relationship to EOS or ALOS. Can you explain what > you > > > mean? > > > >> > > > >> > > > >> > > > >> -Matthias > > > >> > > > >> > > > >> On 1/28/25 10:43 AM, Almog Gavra wrote: > > > >>> Thanks for the feedback Lucas and Bruno! > > > >>> > > > >>> L0. "Given the motivation section, it sounds we actually want > > something > > > >>> that I'd call "batching" rather than "windowing"." > > > >>> > > > >>> You are right here, and I think ultimately introducing more > flexible > > > and > > > >>> controlled micro-batching will be useful for Kafka Streams, but > that > > > >>> expands the scope a little more than I'd want. We'd have to rethink > > the > > > >>> implications of suppression as well as emitting and how to make it > > > >>> deterministic with regards to time (which is important for EOS and > > even > > > >>> ALOS). The proposal here is a halfway point that I think is easy > > enough > > > >> to > > > >>> reason about within the existing semantics of Kafka Streams and > gets > > us > > > >> 80% > > > >>> of the benefit. (See the second part below since these questions > are > > > >>> related) > > > >>> > > > >>> L2/B1. "why are we specifically emitting based on stream-time?" / > > > >> "Wouldn't > > > >>> it make more sense to have similarly sized batches?" > > > >>> > > > >>> There are two motivations for this: (a) most of the use cases for > > this > > > >> that > > > >>> we've encountered are still time sensitive in that they don't want > to > > > >> batch > > > >>> N records, but rather batch over N seconds (there may be an uneven > > > >>> distribution of keys so that some keys hit N records very quickly > and > > > >>> others are one and done, we'll never see that key again). (b) this > is > > > >> much > > > >>> easier to implement, and is a step forward, given this fits really > > well > > > >>> into the existing windowing semantics. > > > >>> > > > >>> Ultimately, as you suggest, I'd want to be able to control these > > > >> "batches" > > > >>> with multiple emit strategies. It would be nice to specify a > > condition > > > >> like > > > >>> "emit after N records, or N elapsed seconds, or N bytes > accumulated" > > - > > > >> but > > > >>> again (as mentioned above) modeling this with existing Kafka > Streams > > > >>> semantics/operators is a big stretch and I don't want to expand the > > > scope > > > >>> of this KIP too much for that. > > > >>> > > > >>> Let me know if that makes sense! Basically I view this as a good > > middle > > > >>> ground that doesn't compromise semantics but probably addresses > most > > > real > > > >>> (or near-real) time streaming use cases. > > > >>> > > > >>> - Almog > > > >>> > > > >>> On Tue, Jan 28, 2025 at 7:15 AM Bruno Cadonna <cado...@apache.org> > > > >> wrote: > > > >>> > > > >>>> Hi Almog, > > > >>>> > > > >>>> I had similar thoughts as Lucas. When I read the KIP, I asked > myself > > > why > > > >>>> are the windows not specified on number of records instead of time > > if > > > we > > > >>>> do not care about whether the event time of the records is in the > > time > > > >>>> range of the window? > > > >>>> > > > >>>> In your motivation, you write that users might collect small > batches > > > of > > > >>>> records to be passed to a consumer that can handle batched > messages > > > more > > > >>>> effectively than individual messages. Wouldn't it make more sense > to > > > >>>> have similarly sized batches? > > > >>>> You could also consider to do something like the Kafka producer > that > > > has > > > >>>> a batch size and a linger time. > > > >>>> > > > >>>> Best, > > > >>>> Bruno > > > >>>> > > > >>>> On 28.01.25 15:44, Lucas Brutschy wrote: > > > >>>>> Hi Almog, > > > >>>>> > > > >>>>> this seems useful to me. I don't see anything wrong with the > > details > > > >>>>> of the proposal. > > > >>>>> > > > >>>>> More generally, I'd like to hear your thoughts on this vs. > > batching. > > > >>>>> Given the motivation section, it sounds we actually want > something > > > >>>>> that I'd call "batching" rather than "windowing". If you do not > > > really > > > >>>>> care about the including events that fall into different time > > > windows, > > > >>>>> and the order of events does not matter much, why are we > > specifically > > > >>>>> emitting based on stream-time? Would you expect this mechanism to > > > >>>>> extend also to emitting batches based on number of records, > > byte-size > > > >>>>> of batch, wall-clock time? > > > >>>>> > > > >>>>> Cheers, > > > >>>>> Lucas > > > >>>>> > > > >>>>> On Thu, Jan 23, 2025 at 7:59 PM Matthias J. Sax < > mj...@apache.org> > > > >>>> wrote: > > > >>>>>> > > > >>>>>> Thanks, Almog. > > > >>>>>> > > > >>>>>> Good call out about `TimeWindows` vs `TimeWindow` (yes, I am > aware > > > and > > > >>>>>> was actually re-reading my previous email before sending it a > few > > > >> times > > > >>>>>> to make sure I use the right one; it's very subtle.) > > > >>>>>> > > > >>>>>> For `TimeWindows` semantics are certainly well defined, and > there > > is > > > >>>>>> nothing to be discussed. > > > >>>>>> > > > >>>>>> For `TimeWindow`, I am not sure as I said, and your > interpretation > > > as > > > >>>>>> "just a container" might be fine, too. I agree to the problem, > > that > > > if > > > >>>>>> we add something new, we might just leak it again, and thus not > > gain > > > >>>>>> much, so the lesser evil might be to just re-use `TimeWindow` as > > you > > > >>>>>> propose. I just wanted to point out this question to sanity > check, > > > and > > > >>>>>> collect feedback about it. > > > >>>>>> > > > >>>>>> > > > >>>>>> -Matthias > > > >>>>>> > > > >>>>>> On 1/23/25 8:58 AM, Almog Gavra wrote: > > > >>>>>>> Thanks Matthias for the quick and detailed feedback! > > > >>>>>>> > > > >>>>>>>> Nit: it seems you are mixing the terms "out-of-order" and > "late" > > > and > > > >>>>>>> using them as synonymous, what we usually not do. > > > >>>>>>> > > > >>>>>>> M1. Ah, in my mind "late arriving" was after the window closed > > but > > > >>>>>>> potentially before grace (and "out of order" was just anything > > that > > > >> is > > > >>>> out > > > >>>>>>> of order). Do we have a specific word for "after the window > > closes > > > >> but > > > >>>>>>> before grace"? Maybe we should have "fashionably late" and "too > > > late" > > > >>>>>>> (haha, just kidding). > > > >>>>>>> > > > >>>>>>> I'll clear up the terminology in the KIP. > > > >>>>>>> > > > >>>>>>>>> Today the only way to implement this semantic is using a > > > >> punctuation > > > >>>>>>> and manually storing events in an aggregation (which is > > inefficient > > > >> for > > > >>>>>>> many reasons and can cause severe bottlenecks) > > > >>>>>>>> Not sure if I follow this argument? I don't see any > relationship > > > to > > > >>>>>>> punctuations. Can you elaborate? > > > >>>>>>> > > > >>>>>>> M2. The way we've seen people implement the desired behavior in > > > this > > > >>>> KIP is > > > >>>>>>> to use a non-windowed aggregation on a stream, and then use a > > > >>>> punctuation > > > >>>>>>> every N seconds to scan the table and emit and delete the > records > > > to > > > >>>>>>> simulate a window closing. It's certainly suboptimal, but it > gets > > > you > > > >>>> very > > > >>>>>>> similar semantics to what I describe in the KIP. > > > >>>>>>> > > > >>>>>>> M3. Re: extends Windows<TimeWindow> > > > >>>>>>> > > > >>>>>>> (note for other readers, I think you get this Matthias) > > TimeWindow > > > is > > > >>>>>>> distinct from TimeWindows (with an s at the end). TimeWindows > > (with > > > >> an > > > >>>> s) > > > >>>>>>> implies exactly what you suggest. I do think, however, that > > > >> TimeWindow > > > >>>>>>> (without an s) is a perfect abstraction to leverage here. That > > > class > > > >>>>>>> doesn't do anything except to indicate the bounds of a window > (it > > > >> has a > > > >>>>>>> start time, an end time and a definition of whether two windows > > > >>>> overlap). > > > >>>>>>> The javadoc for TimeWindow (without an s) doesn't even need to > > > change > > > >>>> to be > > > >>>>>>> used in the way this KIP suggests. > > > >>>>>>> > > > >>>>>>> As suggested, the StreamTimeWindow still uses time as its > bounds > > - > > > it > > > >>>> just > > > >>>>>>> doesn't use the event time to determine which time window the > > event > > > >> is > > > >>>>>>> placed in. > > > >>>>>>> > > > >>>>>>> M4. Re: I just realized that `Window` is in an `internal` > package > > > but > > > >>>>>>> `TimeWindows#windowsFor` does return `Map<Long, TimeWindow>` > > > >>>>>>> > > > >>>>>>> Yeah, I was wondering about that as well. I also agree its a > bit > > > >>>> orthogonal > > > >>>>>>> to the KIP and I think if we introduce a new "extends Window" > > type > > > >>>> we'll > > > >>>>>>> just leak that type as well, so I don't really see a benefit > from > > > >>>> that. If > > > >>>>>>> we do decide to tackle the leaky abstraction we might as well > > just > > > >> have > > > >>>>>>> TimeWindow to handle :) > > > >>>>>>> > > > >>>>>>> M5. Specifying abstract methods > > > >>>>>>> > > > >>>>>>> Can do! > > > >>>>>>> > > > >>>>>>> M6. Naming > > > >>>>>>> > > > >>>>>>> I like "OffsetOrderedWindows", I will wait for some other > > feedback > > > >> and > > > >>>> if > > > >>>>>>> no one has something better I'll update the KIP to that. > > > >>>>>>> > > > >>>>>>> Re: `extends Window<TimeWindow>` I do think the fact that it > > > defines > > > >>>> the > > > >>>>>>> window size with time is important, we're not saying the window > > > >> closes > > > >>>>>>> after N offsets have passed but rather the stream time has > > passed N > > > >>>> seconds > > > >>>>>>> (or ms or whatever). I agree that it doesn't need to be part of > > the > > > >>>> main > > > >>>>>>> name, however. > > > >>>>>>> > > > >>>>>>> > > > >>>>>>> On Thu, Jan 23, 2025 at 12:14 AM Matthias J. Sax < > > mj...@apache.org > > > > > > > >>>> wrote: > > > >>>>>>> > > > >>>>>>>> Interesting KIP. It's a known problem, and the proposed > solution > > > >> make > > > >>>>>>>> sense to me. > > > >>>>>>>> > > > >>>>>>>> Nit: it seems you are mixing the terms "out-of-order" and > "late" > > > and > > > >>>>>>>> using them as synonymous, what we usually not do. > > > >>>>>>>> > > > >>>>>>>> "Out-of-order" is the more generic term, while "late" means > > after > > > >> the > > > >>>>>>>> grace period (hence, late arriving data is still out-of-order, > > but > > > >>>> it's > > > >>>>>>>> a subset of out-of-order records... all late records are > > > >> out-of-order, > > > >>>>>>>> but not the other way around). I think the KIP could gain > > clarity > > > >> and > > > >>>>>>>> avoid potential confusion to use both terms not a synonymous. > > > >>>>>>>> > > > >>>>>>>> In the end, "late" data cannot be handled at all right now, > > based > > > on > > > >>>> the > > > >>>>>>>> definition of "late". "late" data is always dropped. > > > >>>>>>>> > > > >>>>>>>> > > > >>>>>>>> > > > >>>>>>>>> Today the only way to implement this semantic is using a > > > >> punctuation > > > >>>> and > > > >>>>>>>> manually storing events in an aggregation (which is > inefficient > > > for > > > >>>> many > > > >>>>>>>> reasons and can cause severe bottlenecks) > > > >>>>>>>> > > > >>>>>>>> Not sure if I follow this argument? I don't see any > relationship > > > to > > > >>>>>>>> punctuations. Can you elaborate? > > > >>>>>>>> > > > >>>>>>>> It's also not clear to me, why "storing events in an > > aggregation" > > > is > > > >>>>>>>> problematic, and/or how this new window type would avoid it? > > > >>>>>>>> > > > >>>>>>>> > > > >>>>>>>> > > > >>>>>>>>> extends Windows<TimeWindow> > > > >>>>>>>> > > > >>>>>>>> Not sure if this is the right thing to do it? I understand > your > > > >>>> intent, > > > >>>>>>>> but to me, `TimeWindow` has certain semantics, implying we > only > > > put > > > >>>> data > > > >>>>>>>> with record-ts between the window bounds into the window. Not > > sure > > > >> if > > > >>>> my > > > >>>>>>>> interpretation is off, and re-using could be ok? It's just > > > something > > > >>>> we > > > >>>>>>>> should figure out I guess. > > > >>>>>>>> > > > >>>>>>>> To be fair: it's not 100% clearly specific if these semantics > > > apply > > > >> or > > > >>>>>>>> not. If we believe they don't apply, happy to pivot on this, > but > > > we > > > >>>>>>>> might want to update the JavaDocs, if we only want to use it > as > > > >>>>>>>> "container class" (for the lack of better word). > > > >>>>>>>> > > > >>>>>>>> What actually raises a different issue: I just realized that > > > >> `Window` > > > >>>> is > > > >>>>>>>> in an `internal` package, but `TimeWindows#windowsFor` does > > return > > > >>>>>>>> `Map<Long, TimeWindow>`, thus making it effectively kinda > > > public... > > > >> It > > > >>>>>>>> seems we should we do something about this? End users do not > > > really > > > >>>> need > > > >>>>>>>> to use `TimeWindow` -- it's really internal, and it seems > > > >> `windowsFor` > > > >>>>>>>> is effective a leaky abstraction... > > > >>>>>>>> > > > >>>>>>>> If we keep `TimeWindow` internal, I think we can easily be > more > > > >>>> relaxed > > > >>>>>>>> with applied semantics. But it's kinda leaking right now, what > > > does > > > >>>> not > > > >>>>>>>> make me happy... (Of course, I also don't want to derail and > > > hijack > > > >>>> this > > > >>>>>>>> KIP for some related, but orthogonal issue in the API...) > > > >>>>>>>> > > > >>>>>>>> On other nit I am wondering about is, if it might be good to > > > specify > > > >>>> the > > > >>>>>>>> abstract methods inherited from `Windows` on the KIP for > > clarity? > > > >> Even > > > >>>>>>>> they are implicitly well specified it might be beneficial to > > spell > > > >> it > > > >>>>>>>> out. After all, `StreamTimeWindow` needs to implement there > > > methods. > > > >>>>>>>> > > > >>>>>>>> > > > >>>>>>>> > > > >>>>>>>> > > > >>>>>>>> And of course > > > >>>>>>>> > > > >>>>>>>>> Note: I really don't like the name I cam up with > > > >>>> StreamTimeWindows... if > > > >>>>>>>> anyone has a better name in mind please suggest one! > > > >>>>>>>> > > > >>>>>>>> I am not a fan of the name either. But not really sure what a > > > better > > > >>>>>>>> name could be. In the end, it's more of an "offset ordered > > window" > > > >>>>>>>> rather than a "time window", because it does not really take > the > > > >>>> record > > > >>>>>>>> timestamps into account to figure out into what window a > record > > > goes > > > >>>>>>>> into, but it really uses the offset order. > > > >>>>>>>> > > > >>>>>>>> I believe that the size of window itself does not matter too > > much > > > >> for > > > >>>>>>>> naming, so even if the size is defined in time, I don't think > > that > > > >> the > > > >>>>>>>> term "time" must be part of the window name. (That's also why > I > > am > > > >>>>>>>> wondering if we should go with `extends<TimeWindow>` or add > > > >> something > > > >>>>>>>> new, which we hopefully can keep internal...) > > > >>>>>>>> > > > >>>>>>>> Not sure if we ever intent to make `StreamTimeWindows` > > > overlapping? > > > >>>>>>>> Given the described use-cases it seems very unlikely? So > > > effectively > > > >>>>>>>> it's a "tumbling window" . Of course, we should not just > > overload > > > >> this > > > >>>>>>>> term, but there is no `TumblingWindows` in the API -- only the > > > term > > > >>>>>>>> Tumbling Window in the docs to describe the special case of > > > >>>>>>>> non-overlapping `TimeWindows`. > > > >>>>>>>> > > > >>>>>>>> So maybe `OffsetTumblingWindows` could work? Or > > > >>>> `OffsetOrderedWindows`? > > > >>>>>>>> Just spitballing here, to get a discussion going... > > > >>>>>>>> > > > >>>>>>>> > > > >>>>>>>> > > > >>>>>>>> -Matthias > > > >>>>>>>> > > > >>>>>>>> > > > >>>>>>>> > > > >>>>>>>> On 1/22/25 8:26 PM, Almog Gavra wrote: > > > >>>>>>>>> Hello! > > > >>>>>>>>> > > > >>>>>>>>> I'd like to initiate a discussion thread on KIP-1127: > > > >>>>>>>>> > > > >>>>>>>> > > > >>>> > > > >> > > > > > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-1127+Flexible+Windows+for+Late+Arriving+Data > > > >>>>>>>>> > > > >>>>>>>>> This KIP aims to make it easier to specify windowing > semantics > > > that > > > >>>> are > > > >>>>>>>>> more tolerable to late arriving data, particularly with > > > >> suppression. > > > >>>>>>>>> > > > >>>>>>>>> Would appreciate any and all feedback. > > > >>>>>>>>> > > > >>>>>>>>> Cheers, > > > >>>>>>>>> Almog > > > >>>>>>>>> > > > >>>>>>>> > > > >>>>>>>> > > > >>>>>>> > > > >>>>>> > > > >>>> > > > >>>> > > > >>> > > > >> > > > >> > > > > > > > > > > > > >
