Moved this KIP into status "inactive". Feel free to resume and any time.
-Matthias On 1/9/19 10:18 PM, Guozhang Wang wrote: > Hello Lei, > > Just checking what's the current status of this KIP. We have a KIP deadline > for 2.2 on 24th and wondering if this one may be able to make it. > > > Guozhang > > On Sat, Dec 15, 2018 at 1:01 PM Lei Chen <ley...@gmail.com> wrote: > >> Sorry for the late reply Matthias. Have been busy with other work recently. >> I'll restart the discussion and update the KIP accordingly. >> >> Lei >> >> On Tue, Nov 6, 2018 at 3:11 PM Matthias J. Sax <matth...@confluent.io> >> wrote: >> >>> Any update on this KIP? >>> >>> On 9/20/18 3:30 PM, Matthias J. Sax wrote: >>>> Thanks for following up. Very nice examples! >>>> >>>> I think, that the window definition for Flink is semantically >>>> questionable. If there is only a single record, why is the window >>>> defined as [ts, ts+gap]? To me, this definition is not sound and seems >>>> to be arbitrary. To define the windows as [ts-gap,ts+gap] as you >> mention >>>> would be semantically more useful -- still, I think that defining the >>>> window as [ts,ts] as we do currently in Kafka Streams is semantically >>>> the best. >>>> >>>> I have the impression, that Flink only defines them differently, >> because >>>> it solves the issues in the implementation. (Ie, an implementation >>>> details leaks into the semantics, what is usually not desired.) >>>> >>>> However, I believe that we could change the implementation accordingly. >>>> We could store the windowed keys, as [ts-gap,ts+gap] (or [ts,ts+gap]) >> in >>>> RocksDB, but at API level we return [ts,ts]. This way, we can still >> find >>>> all windows we need and provide the same deterministic behavior and >> keep >>>> the current window boundaries on the semantic level (there is no need >> to >>>> store the window start and/or end time). With this technique, we can >>>> also implement dynamic session gaps. I think, we would need to store >> the >>>> used "gap" for each window, too. But again, this would be an >>>> implementation detail. >>>> >>>> Let's see what others think. >>>> >>>> One tricky question we would need to address is, how we can be backward >>>> compatible. I am currently working on KIP-258 that should help to >>>> address this backward compatibility issue though. >>>> >>>> >>>> -Matthias >>>> >>>> >>>> >>>> On 9/19/18 5:17 PM, Lei Chen wrote: >>>>> Thanks Matthias. That makes sense. >>>>> >>>>> You're right that symmetric merge is necessary to ensure consistency. >> On >>>>> the other hand, I kinda feel it defeats the purpose of dynamic gap, >>> which >>>>> is to update the gap from old value to new value. The symmetric merge >>>>> always honor the larger gap in both direction, rather than honor the >> gap >>>>> carried by record with larger timestamp. I wasn't able to find any >>> semantic >>>>> definitions w.r.t this particular aspect online, but spent some time >>>>> looking into other streaming engines like Apache Flink. >>>>> >>>>> Apache Flink defines the window differently, that uses (start time, >>> start >>>>> time + gap). >>>>> >>>>> so our previous example (10, 10), (19,5),(15,3) in Flink's case will >> be: >>>>> [10,20] >>>>> [19,24] => merged to [10,24] >>>>> [15,18] => merged to [10,24] >>>>> >>>>> while example (15,3)(19,5)(10,10) will be >>>>> [15,18] >>>>> [19,24] => no merge >>>>> [10,20] => merged to [10,24] >>>>> >>>>> however, since it only records gap in future direction, not past, a >> late >>>>> record might not trigger any merge where in symmetric merge it would. >>>>> (7,2),(10, 10), (19,5),(15,3) >>>>> [7,9] >>>>> [10,20] >>>>> [19,24] => merged to [10,24] >>>>> [15,18] => merged to [10,24] >>>>> so at the end >>>>> two windows [7,9][10,24] are there. >>>>> >>>>> As you can see, in Flink, the gap semantic is more toward to the way >>> that, >>>>> a gap carried by one record only affects how this record merges with >>> future >>>>> records. e.g. a later event (T2, G2) will only be merged with (T1, G1) >>> is >>>>> T2 is less than T1+G1, but not when T1 is less than T2 - G2. Let's >> call >>>>> this "forward-merge" way of handling this. I just went thought some >>> source >>>>> code and if my understanding is incorrect about Flink's >> implementation, >>>>> please correct me. >>>>> >>>>> On the other hand, if we want to do symmetric merge in Kafka Streams, >> we >>>>> can change the window definition to [start time - gap, start time + >>> gap]. >>>>> This way the example (7,2),(10, 10), (19,5),(15,3) will be >>>>> [5,9] >>>>> [0,20] => merged to [0,20] >>>>> [14,24] => merged to [0,24] >>>>> [12,18] => merged to [0,24] >>>>> >>>>> (19,5),(15,3)(7,2),(10, 10) will generate same result >>>>> [14,24] >>>>> [12,18] => merged to [12,24] >>>>> [5,9] => no merge >>>>> [0,20] => merged to [0,24] >>>>> >>>>> Note that symmetric-merge would require us to change the way how Kafka >>>>> Steams fetch windows now, instead of fetching range from timestamp-gap >>> to >>>>> timestamp+gap, we will need to fetch all windows that are not expired >>> yet. >>>>> On the other hand, I'm not sure how this will impact the current logic >>> of >>>>> how a window is considered as closed, because the window doesn't carry >>> end >>>>> timestamp anymore, but end timestamp + gap. >>>>> >>>>> So do you guys think forward-merge approach used by Flink makes more >>> sense >>>>> in Kafka Streams, or symmetric-merge makes more sense? Both of them >>> seems >>>>> to me can give deterministic result. >>>>> >>>>> BTW I'll add the use case into original KIP. >>>>> >>>>> Lei >>>>> >>>>> >>>>> On Tue, Sep 11, 2018 at 5:45 PM Matthias J. Sax < >> matth...@confluent.io> >>>>> wrote: >>>>> >>>>>> Thanks for explaining your understanding. And thanks for providing >> more >>>>>> details about the use-case. Maybe you can add this to the KIP? >>>>>> >>>>>> >>>>>> First one general comment. I guess that my and Guozhangs >> understanding >>>>>> about gap/close/gracePeriod is the same as yours -- we might not have >>>>>> use the term precisely correct in previous email. >>>>>> >>>>>> >>>>>> To you semantics of gap in detail: >>>>>> >>>>>>> I thought when (15,3) is received, kafka streams look up for >> neighbor >>>>>>> record/window that is within the gap >>>>>>> of [15-3, 15+3], and merge if any. Previous record (10, 10) created >>> its >>>>>> own >>>>>>> window [10, 10], which is >>>>>>> out of the gap, so nothing will be found and no merge occurs. Hence >> we >>>>>> have >>>>>>> two windows now in session store, >>>>>>> [10, 10] and [15, 15] respectively. >>>>>> >>>>>> If you have record (10,10), we currently create a window of size >>>>>> [10,10]. When record (15,3) arrives, your observation that the gap 3 >> is >>>>>> too small to be merged into [10,10] window -- however, merging is a >>>>>> symmetric operation and the existing window of [10,10] has a gap of >> 10 >>>>>> defined: thus, 15 is close enough to fall into the gap, and (15,3) is >>>>>> merged into the existing window resulting in window [10,15]. >>>>>> >>>>>> If we don't respect the gap both ways, we end up with inconsistencies >>> if >>>>>> data is out-of-order. For example, if we use the same input record >>>>>> (10,10) and (15,3) from above, and it happens that (15,3) is >> processed >>>>>> first, when processing out-of-order record (10,10) we would want to >>>>>> merge both into a single window, too? >>>>>> >>>>>> Does this make sense? >>>>>> >>>>>> Now the question remains, if two records with different gap parameter >>>>>> are merged, which gap should we apply for processing/merging future >>>>>> records into the window? It seems, that we should use the gap >> parameter >>>>>> from the record with this larges timestamp. In the example above >>> (15,3). >>>>>> We would use gap 3 after merging independent of the order of >>> processing. >>>>>> >>>>>> >>>>>>> Also another thing worth mentioning is that, the session window >> object >>>>>>> created in current kafka streams >>>>>>> implementation doesn't have gap info, it has start and end, which is >>> the >>>>>>> earliest and latest event timestamp >>>>>>> in that window interval, i.e for (10,10), the session window gets >>> created >>>>>>> is [10,10], rather than [10,20]. Just to clarify >>>>>>> so that it's clear why (10,10) cannot be fetched when looking for >> gap >>> of >>>>>>> (15,3), it's because the end boundary 10 of >>>>>>> [10,10] is smaller than search boundary [12,18]. >>>>>> >>>>>> We don't need to store the gap, because the gap is know from the >> window >>>>>> definition. The created window size depends on the data that is >>>>>> contained in the window. I guess one could define it differently, >> too, >>>>>> ie, for the (10,10) record, we create a window [0,20] -- not sure if >> it >>>>>> makes a big difference in practice though. Note, that creating window >>>>>> [10,20] would not be correct, because the gap must be applied in both >>>>>> directions, not just into the future. >>>>>> >>>>>> About the second part: the search would not be applied from (15,3) in >>>>>> range [12,18], but from existing window [10,10] into range [0,20] and >>> 15 >>>>>> is contained there. This example also shows, that we would need to >> come >>>>>> up with a clever way, to identify window [10,10] when processing >> (15,3) >>>>>> -- not sure atm how to do this. However, only consider (15,3) would >>>>>> result in inconsistencies for out-of-order data as pointed out above >>> and >>>>>> would not be sufficient. >>>>>> >>>>>> >>>>>> Does this make sense? >>>>>> >>>>>> >>>>>> Or is there another way to define dynamic session gap semantics in a >>>>>> deterministic way with regard to out-of-order data? >>>>>> >>>>>> >>>>>> >>>>>> -Matthias >>>>>> >>>>>> >>>>>> On 9/11/18 4:28 PM, Lei Chen wrote: >>>>>>> Thanks Matthias and Guozhang for the response. >>>>>>> >>>>>>> Seems like our understanding mainly differs in the semantics of gap >> in >>>>>>> session windows. >>>>>>> >>>>>>> My understanding is that gap is used to merge nearby records >> together >>>>>> such >>>>>>> that no record >>>>>>> in the merged window has distance later than gap. In Kafka Streams's >>>>>>> implementation it's >>>>>>> mainly used to find neighbor records/windows in session store so >> that >>>>>>> nearby records can >>>>>>> be merge. It is NOT used to determine when a window should be >> closed, >>>>>> which >>>>>>> is in >>>>>>> fact determined by window's grace period. >>>>>>> >>>>>>> Guozhang you said "b. When later we received (15, 3), it means that >>> this >>>>>>> record ** changed ** >>>>>>> the window gap interval from 10 to 3, and hence we received a new >>> record >>>>>> at >>>>>>> 15, with the new window gap of 3, it means that by timestamp 18 (15 >> + >>> 3) >>>>>> if >>>>>>> we have not received any new data, the window should be closed, i.e. >>> the >>>>>>> window is now [10, 18) which includes two records at 10 and 15." >>>>>>> >>>>>>> This is different from what i thought will happen. >>>>>>> >>>>>>> I thought when (15,3) is received, kafka streams look up for >> neighbor >>>>>>> record/window that is within the gap >>>>>>> of [15-3, 15+3], and merge if any. Previous record (10, 10) created >>> its >>>>>> own >>>>>>> window [10, 10], which is >>>>>>> out of the gap, so nothing will be found and no merge occurs. Hence >> we >>>>>> have >>>>>>> two windows now in session store, >>>>>>> [10, 10] and [15, 15] respectively. >>>>>>> >>>>>>> Also another thing worth mentioning is that, the session window >> object >>>>>>> created in current kafka streams >>>>>>> implementation doesn't have gap info, it has start and end, which is >>> the >>>>>>> earliest and latest event timestamp >>>>>>> in that window interval, i.e for (10,10), the session window gets >>> created >>>>>>> is [10,10], rather than [10,20]. Just to clarify >>>>>>> so that it's clear why (10,10) cannot be fetched when looking for >> gap >>> of >>>>>>> (15,3), it's because the end boundary 10 of >>>>>>> [10,10] is smaller than search boundary [12,18]. >>>>>>> >>>>>>> Please correct me if my understanding is wrong here. >>>>>>> >>>>>>> @Matthias, to answer your use case question, we have an use case >> where >>>>>>> asynchronous time series data >>>>>>> are received in the stream, from different contributors, with >>> different >>>>>>> quality and at different pace. >>>>>>> Inside Kafka Streams, we use state to maintain statistic >> aggregations >>> and >>>>>>> other mathematics model to track >>>>>>> the liquidity and calculate time decay rate and dynamic gap, so that >>> at >>>>>>> runtime, for each contributor we can >>>>>>> 1. determine how many historical records we should maintain in >> state. >>>>>>> 2. for each incoming record, output a record using aggregations from >>>>>>> *nearby* records from that contributor. >>>>>>> Why fixed gap session window doesn't work here? Because the >>> definition of >>>>>>> "nearby" here is determined by >>>>>>> several very dynamic factors in our case, it changes not only with >>>>>>> different hours in a day, but also related to >>>>>>> other contributors. >>>>>>> The purpose of this KIP is to suggest a dynamic session window >>>>>>> implementation so that we can embed such >>>>>>> dynamic "nearby" calculation capability into kafka streams session >>>>>> windows >>>>>>> semantics. Hope it makes sense to you. >>>>>>> >>>>>>> Lei >>>>>>> >>>>>>> >>>>>>> >>>>>>> >>>>>>> On Mon, Sep 10, 2018 at 5:27 PM Guozhang Wang <wangg...@gmail.com> >>>>>> wrote: >>>>>>> >>>>>>>> Hello Lei, >>>>>>>> >>>>>>>> As Matthias mentioned, the key question here is that because of the >>> late >>>>>>>> arrivals of records which may indicate a shorter session gap >>> interval, >>>>>> some >>>>>>>> session windows may be "mistakenly" merged and hence need to be >>> undone >>>>>> the >>>>>>>> merge, i.e. to split them again. >>>>>>>> >>>>>>>> Back to my example, you are right that the processing result of >>>>>>>> >>>>>>>> (10, 10), (19, 5), (15, 3) .. >>>>>>>> >>>>>>>> should be the same as the processing result of >>>>>>>> >>>>>>>> (10, 10), (15, 3), (19, 5) .. >>>>>>>> >>>>>>>> Note that the second value is NOT the window end time, but the >>> extracted >>>>>>>> window gap interval, as you suggested in the KIP this value can be >>>>>>>> dynamically changed >>>>>>>> >>>>>>>> a. If you take a look at the second ordering, when we receive (10, >>> 10) >>>>>> it >>>>>>>> means a window starting at 10 is created, and its gap interval is >> 10, >>>>>> which >>>>>>>> means that if by the timestamp of 20 we do not receive any new >> data, >>>>>> then >>>>>>>> the window should be closed, i.e. the window [10, 20). >>>>>>>> >>>>>>>> b. When later we received (15, 3), it means that this record ** >>> changed >>>>>> ** >>>>>>>> the window gap interval from 10 to 3, and hence we received a new >>>>>> record at >>>>>>>> 15, with the new window gap of 3, it means that by timestamp 18 >> (15 + >>>>>> 3) if >>>>>>>> we have not received any new data, the window should be closed, >> i.e. >>> the >>>>>>>> window is now [10, 18) which includes two records at 10 and 15. >>>>>>>> >>>>>>>> c. The third record is received at 19, which is after the window >>> close >>>>>> time >>>>>>>> 18, it means that we should now start a new window starting at 19, >>> i.e. >>>>>> the >>>>>>>> window is [19, 24), >>>>>>>> >>>>>>>> >>>>>>>> BUT, because of the out of ordering, we did not receive (15, 3) in >>> time, >>>>>>>> but received (19, 5), it will cause us to mistakenly merge the >>> window of >>>>>>>> [10, 20) with [19, 24) to [10, 24), and only when later we received >>>>>> (15, 3) >>>>>>>> we realized that the previous window should have been ended at 18. >>>>>>>> >>>>>>>> Does that make sense to you? >>>>>>>> >>>>>>>> >>>>>>>> Guozhang >>>>>>>> >>>>>>>> >>>>>>>> On Thu, Sep 6, 2018 at 9:50 PM, Matthias J. Sax < >>> matth...@confluent.io> >>>>>>>> wrote: >>>>>>>> >>>>>>>>> I cannot follow the example: >>>>>>>>> >>>>>>>>>>> (10, 10), (15, 3), (19, 5) ... >>>>>>>>> >>>>>>>>> First, [10,10] is created, second the window is extended to >> [10,15], >>>>>> and >>>>>>>>> third [19,19] is created. Why would there be a [15,15]? And why >>> would >>>>>>>>> (19,5) be merged into [15,15] -- the gap was set to 3 via (15,3) >> and >>>>>>>>> thus [19,19] should be its own window? >>>>>>>>> >>>>>>>>>> Take a look at another example, >>>>>>>>>> (13, 3), (19, 5), (15, 3) ... >>>>>>>>>> >>>>>>>>>> in this case when (15, 3) is received, [13,13] should be >> retrieved >>> and >>>>>>>>>> merged to a new window [13, 15], then [19,19] should be updated >> to >>>>>> [13, >>>>>>>>>> 19]. Correct? >>>>>>>>> >>>>>>>>> This example makes sense. However, Guozhang's example was >> different. >>>>>> The >>>>>>>>> late even, _reduces_ the gap and this can lead to a window split. >>>>>>>>> Guozhang's example was >>>>>>>>> >>>>>>>>>>>> (10, 10), (19, 5), (15, 3) ... >>>>>>>>> >>>>>>>>> First [10,10] is created, second [10,19] is create (gap is 10, so >> 10 >>>>>> and >>>>>>>>> 19 merge). Last, (15,3) reduced the gap from 10 to 3, thus [10,15] >>> and >>>>>>>>> [19,19] must be two windows, ie, original window [10,19] must be >>> split. >>>>>>>>> >>>>>>>>> >>>>>>>>> Or maybe you have different semantic about gaps are dynamically >>>>>> modified >>>>>>>>> in mind? It's a little unclear for the KIP itself what semantics >>>>>> dynamic >>>>>>>>> sessions windows should have. >>>>>>>>> >>>>>>>>> >>>>>>>>> What is also unclear to me atm is, what use cases you have in >> mind? >>> The >>>>>>>>> KIP only says >>>>>>>>> >>>>>>>>>> the statistical aggregation result, liquidity of the records, >>>>>>>>> >>>>>>>>> >>>>>>>>> I am not sure what this means. Can you elaborate? >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> -Matthias >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> On 8/30/18 3:32 PM, Lei Chen wrote: >>>>>>>>>> Hi Guozhang, >>>>>>>>>> >>>>>>>>>> Thanks for reviewing the proposal. I didn't think of out of order >>>>>>>> events >>>>>>>>>> and glad that you brought it up. >>>>>>>>>> >>>>>>>>>> In the example you gave, >>>>>>>>>> >>>>>>>>>> (10, 10), (19, 5), (15, 3) ... >>>>>>>>>> >>>>>>>>>> my understanding is that the correct result window should be the >>> same >>>>>>>> as >>>>>>>>> in >>>>>>>>>> order events >>>>>>>>>> >>>>>>>>>> (10, 10), (15, 3), (19, 5) ... >>>>>>>>>> >>>>>>>>>> when (15, 3) is received, [15,15] is creatd >>>>>>>>>> when (19, 5) is received, [15, 15] and [19, 19] are merged and >> [15, >>>>>> 19] >>>>>>>>> is >>>>>>>>>> created, meanwhile [15,15] is removed >>>>>>>>>> >>>>>>>>>> back to out of order case, >>>>>>>>>> >>>>>>>>>> when (19 ,5) is received, [19, 19] is created >>>>>>>>>> when (15, 3) is received, in order to generate the same result, >>>>>>>>>> 1. if late event is later than retention period, it will be >> dropped >>>>>>>>>> 2. otherwise, adjacent session windows within gap should be >>> retrieved >>>>>>>> and >>>>>>>>>> merged accordingly, in this case [19, 19], and create a new >> session >>>>>>>> [15, >>>>>>>>> 19] >>>>>>>>>> I'm little confused when you said "the window [15, 15] SHOULD >>> actually >>>>>>>> be >>>>>>>>>> expired at 18 and hence the next record (19, 5) should be for a >> new >>>>>>>>> session >>>>>>>>>> already.". If i understand it correctly, the expiration of the >>> window >>>>>>>> is >>>>>>>>>> only checked when next event (19,5) comes and then it should be >>> merged >>>>>>>> to >>>>>>>>>> it. [15, 15] will then be closed. Is that also what you meant? >>>>>>>>>> I cannot think of a case where a window will be split by a late >>> event, >>>>>>>>>> because if event A and C fall into the same session window, a >> late >>>>>>>> event >>>>>>>>> B >>>>>>>>>> in middle will definitely fall into C's gap as well. IOW, late >>> event >>>>>>>> will >>>>>>>>>> only cause window extension, not split. >>>>>>>>>> >>>>>>>>>> Take a look at another example, >>>>>>>>>> (13, 3), (19, 5), (15, 3) ... >>>>>>>>>> >>>>>>>>>> in this case when (15, 3) is received, [13,13] should be >> retrieved >>> and >>>>>>>>>> merged to a new window [13, 15], then [19,19] should be updated >> to >>>>>> [13, >>>>>>>>>> 19]. Correct? >>>>>>>>>> >>>>>>>>>> To be able to achieve that, like you said, the gap needs to be >>> stored >>>>>>>> for >>>>>>>>>> sessions. We don't need to save the gap with each event, but only >>> for >>>>>>>>> each >>>>>>>>>> session window. To avoid upgrading existing session window, how >>> about >>>>>>>>>> create a new Window type extended from SessionWindow along with a >>> new >>>>>>>>>> KeySchema? >>>>>>>>>> >>>>>>>>>> What do you think? >>>>>>>>>> >>>>>>>>>> Lei >>>>>>>>>> >>>>>>>>>> >>>>>>>>>> On Fri, Aug 24, 2018 at 9:42 AM Guozhang Wang < >> wangg...@gmail.com> >>>>>>>>> wrote: >>>>>>>>>> >>>>>>>>>>> Hello Lei, >>>>>>>>>>> >>>>>>>>>>> Thanks for the proposal. I've just made a quick pass over it and >>>>>> there >>>>>>>>> is a >>>>>>>>>>> question I have: >>>>>>>>>>> >>>>>>>>>>> The session windows are defined per key, i.e. does that mean >> that >>>>>> each >>>>>>>>>>> incoming record of the key can dynamically change the gap of the >>>>>>>> window? >>>>>>>>>>> For example, say you have the following record for the same key >>>>>> coming >>>>>>>>> in >>>>>>>>>>> order, where the first time is the timestamp of the record, and >>> the >>>>>>>>> second >>>>>>>>>>> value is the extracted gap value: >>>>>>>>>>> >>>>>>>>>>> (10, 10), (19, 5), ... >>>>>>>>>>> >>>>>>>>>>> >>>>>>>>>>> When we receive the first record at time 10, the gap is >> extracted >>> as >>>>>>>> 10, >>>>>>>>>>> and hence the window will be expired at 20 if no other record is >>>>>>>>> received. >>>>>>>>>>> When we receive the second record at time 19, the gap is >> modified >>> to >>>>>>>> 5, >>>>>>>>> and >>>>>>>>>>> hence the window will be expired at 24 if no other record is >>>>>> received. >>>>>>>>>>> >>>>>>>>>>> >>>>>>>>>>> If that's the case, I'm wondering how out-of-order data can be >>>>>> handled >>>>>>>>>>> then, consider this stream: >>>>>>>>>>> >>>>>>>>>>> (10, 10), (19, 5), (15, 3) ... >>>>>>>>>>> >>>>>>>>>>> I.e. you received a late record indicating at timestamp 15, >> which >>>>>>>>> shorten >>>>>>>>>>> the gap to 3. It means that the window SHOULD actually be >> expired >>> at >>>>>>>> 18, >>>>>>>>>>> and hence the next record (19, 5) should be for a new session >>>>>> already. >>>>>>>>>>> Today Streams session window implementation does not do "window >>>>>>>> split", >>>>>>>>> so >>>>>>>>>>> have you thought about how this can be extended? >>>>>>>>>>> >>>>>>>>>>> Also since in your proposal each session window's gap value >> would >>> be >>>>>>>>>>> different, we need to store this value along with each record >>> then, >>>>>>>> how >>>>>>>>>>> would we store it, and what would be the upgrade path if it is >>> not a >>>>>>>>>>> compatible change on disk storage etc? >>>>>>>>>>> >>>>>>>>>>> >>>>>>>>>>> >>>>>>>>>>> Guozhang >>>>>>>>>>> >>>>>>>>>>> >>>>>>>>>>> >>>>>>>>>>> On Wed, Aug 22, 2018 at 10:05 AM, Lei Chen <ley...@gmail.com> >>> wrote: >>>>>>>>>>> >>>>>>>>>>>> Hi All, >>>>>>>>>>>> >>>>>>>>>>>> I created a KIP to add dynamic gap session window support to >>> Kafka >>>>>>>>>>> Streams >>>>>>>>>>>> DSL. >>>>>>>>>>>> >>>>>>>>>>>> https://cwiki.apache.org/confluence/display/KAFKA/KIP- >>>>>>>>>>>> 362%3A+Support+dynamic+gap+session+window >>>>>>>>>>>> >>>>>>>>>>>> Please take a look, >>>>>>>>>>>> >>>>>>>>>>>> Thanks, >>>>>>>>>>>> Lei >>>>>>>>>>>> >>>>>>>>>>> >>>>>>>>>>> >>>>>>>>>>> >>>>>>>>>>> -- >>>>>>>>>>> -- Guozhang >>>>>>>>>>> >>>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>> >>>>>>>> >>>>>>>> -- >>>>>>>> -- Guozhang >>>>>>>> >>>>>>> >>>>>> >>>>>> >>>>> >>>> >>> >>> >> > >
signature.asc
Description: OpenPGP digital signature
