Hi Till, When I run the example code that you posted, the order of the three messages (window started, contents of window and window ended) is non-deterministic. This is surprising to me, as setParallelism(1) has been used in the pipeline - I assumed this should eliminate any form of race conditions for printing. What's more is that if I *remove* setParallelism(1) from the code, the output is deterministic and correct (i.e. windowStarted -> windowContents -> windowEnded).
Clearly, something is wrong with my understanding. What is it? On Fri, Feb 28, 2020 at 1:58 PM Till Rohrmann <trohrm...@apache.org> wrote: > Great to hear that you solved the problem. Let us know if you run into any > other issues. > > Cheers, > Till > > On Fri, Feb 28, 2020 at 8:08 AM Manas Kale <manaskal...@gmail.com> wrote: > >> Hi, >> This problem is solved[1]. The issue was that the BroadcastStream did not >> contain any watermark, which prevented watermarks for any downstream >> operators from advancing. >> I appreciate all the help. >> [1] >> https://stackoverflow.com/questions/60430520/how-do-i-fire-downstream-oneventtime-method-when-using-broadcaststate-pattern >> >> Thanks, >> Manas >> >> On Thu, Feb 27, 2020 at 4:28 PM Manas Kale <manaskal...@gmail.com> wrote: >> >>> Hi Rafi and Till, >>> Thank you for pointing out that edge case, Rafi. >>> >>> Till, I am trying to get this example working with the BroadcastState >>> pattern upstream to the window operator[1]. The problem is that introducing >>> the BroadcastState makes the onEventTime() *never* fire. Is the >>> BroadcastState somehow eating up the watermark? Do I need to generate the >>> watermark again in the KeyedBroadcastProcessFunction? >>> >>> [1] https://gist.github.com/manasIU/1777c9c99e195a409441815559094b49 >>> >>> Thanks, >>> Manas >>> >>> On Fri, Feb 21, 2020 at 8:55 PM Till Rohrmann <trohrm...@apache.org> >>> wrote: >>> >>>> Hi Manas and Rafi, >>>> >>>> you are right that when using merging windows as event time session >>>> windows are, then Flink requires that any state the Trigger keeps is of >>>> type MergingState. This constraint allows that the state can be merged >>>> whenever two windows get merged. >>>> >>>> Rafi, you are right. With the current implementation it might happen >>>> that you send a wrong started window message. I think it depends on the >>>> MIN_WINDOW_SIZE and the distribution of your timestamps and, hence, also >>>> your watermark. If you want to be on the safe side, then I would recommend >>>> to use the ProcessFunction to implement the required logic. The >>>> ProcessFunction [1] is Flink's low level API and gives you access to state >>>> and timers. In it, you would need to buffer the elements and to sessionize >>>> them yourself, though. However, it would give you access to the >>>> watermark which in turn would allow you to properly handle your described >>>> edge case. >>>> >>>> [1] >>>> https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/operators/process_function.html >>>> >>>> Cheers, >>>> Till >>>> >>>> Cheers, >>>> Till >>>> >>>> On Thu, Feb 20, 2020 at 12:25 PM Rafi Aroch <rafi.ar...@gmail.com> >>>> wrote: >>>> >>>>> I think one "edge" case which is not handled would be that the first >>>>> event (by event-time) arrives late, then a wrong "started-window" would be >>>>> reported. >>>>> >>>>> Rafi >>>>> >>>>> >>>>> On Thu, Feb 20, 2020 at 12:36 PM Manas Kale <manaskal...@gmail.com> >>>>> wrote: >>>>> >>>>>> Is the reason ValueState cannot be use because session windows are >>>>>> always formed by merging proto-windows of single elements, therefore a >>>>>> state store is needed that can handle merging. ValueState does not >>>>>> provide >>>>>> this functionality, but a ReducingState does? >>>>>> >>>>>> On Thu, Feb 20, 2020 at 4:01 PM Manas Kale <manaskal...@gmail.com> >>>>>> wrote: >>>>>> >>>>>>> Hi Till, >>>>>>> Thanks for your answer! You also answered the next question that I >>>>>>> was about to ask "Can we share state between a Trigger and a Window?" >>>>>>> Currently the only (convoluted) way to share state between two >>>>>>> operators is >>>>>>> through the broadcast state pattern, right? >>>>>>> Also, in your example, why can't we use a >>>>>>> ValueStateDescriptor<Boolean> in the Trigger? I tried using it in my own >>>>>>> example but it I am not able to call the mergePartitionedState() >>>>>>> method >>>>>>> on a ValueStateDescriptor. >>>>>>> >>>>>>> Regards, >>>>>>> Manas >>>>>>> >>>>>>> >>>>>>> >>>>>>> On Tue, Feb 18, 2020 at 7:20 PM Till Rohrmann <trohrm...@apache.org> >>>>>>> wrote: >>>>>>> >>>>>>>> Hi Manas, >>>>>>>> >>>>>>>> you can implement something like this with a bit of trigger magic. >>>>>>>> What you need to do is to define your own trigger implementation which >>>>>>>> keeps state to remember whether it has triggered the "started window" >>>>>>>> message or not. In the stateful window function you would need to do >>>>>>>> something similar. The first call could trigger the output of "window >>>>>>>> started" and any subsequent call will trigger the evaluation of the >>>>>>>> window. >>>>>>>> It would have been a bit easier if the trigger and the window process >>>>>>>> function could share its internal state. Unfortunately, this is not >>>>>>>> possible at the moment. >>>>>>>> >>>>>>>> I've drafted a potential solution which you can find here [1]. >>>>>>>> >>>>>>>> [1] >>>>>>>> https://gist.github.com/tillrohrmann/5251f6d62e256b60947eea7b553519ef >>>>>>>> >>>>>>>> Cheers, >>>>>>>> Till >>>>>>>> >>>>>>>> On Mon, Feb 17, 2020 at 8:09 AM Manas Kale <manaskal...@gmail.com> >>>>>>>> wrote: >>>>>>>> >>>>>>>>> Hi, >>>>>>>>> I want to achieve the following using event time session windows: >>>>>>>>> >>>>>>>>> 1. When the window.getStart() and last event timestamp in the >>>>>>>>> window is greater than MIN_WINDOW_SIZE milliseconds, I want to >>>>>>>>> emit a >>>>>>>>> message "Window started @ timestamp". >>>>>>>>> 2. When the session window ends, i.e. the watermark passes >>>>>>>>> lasteventTimestamp + inactivityPeriod, I want to emit a message >>>>>>>>> "Window >>>>>>>>> ended @ timestamp". >>>>>>>>> >>>>>>>>> It is guaranteed that all events are on time and no lateness is >>>>>>>>> allowed. I am having difficulty implementing both 1 and 2 >>>>>>>>> simultaneously. >>>>>>>>> I am able to implement point 1 using a custom trigger, which >>>>>>>>> checks if (lastEventTimestamp - window.getStart()) > MIN_WINDOW_SIZE >>>>>>>>> and >>>>>>>>> triggers a customProcessWindowFunction(). >>>>>>>>> However, with this architecture I can't detect the end of the >>>>>>>>> window. >>>>>>>>> >>>>>>>>> Is my approach correct or is there a completely different method >>>>>>>>> to achieve this? >>>>>>>>> >>>>>>>>> Thanks, >>>>>>>>> Manas Kale >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>>
