Hi Kristoff, > On Jun 23, 2020, at 6:52 AM, KristoffSC <krzysiek.chmielew...@gmail.com> > wrote: > > Hi all, > I'm using Flink 1.9.2 and I would like to ask about my use case and approach > I've took to meet it. > > The use case: > I have a keyed stream, where I have to buffer messages with logic: > 1. Buffering should start only when message arrives. > 2. The max buffer time should not be longer than 3 seconds > 3. Each new message should NOT prolong the buffer time. > 4. If particular business condition will be meet, buffering should stop and > all messages should be let through further processing.
As an alternate solution that does not use custom triggers, 1-4 can be handled with a Tumbling window, and a ProcessWindowFunction. When you override the "process" method, you'll have Iterable list of events that happened in that window. There you can check your business condition specified in step 4. > > The business logic in point 4 is taking under the consideration data from > previously buffered messages in this time buffer session. > > My setup for this is > 1.keyedStream with ProcessingTimeSessionWindow (I dont need EventTime for > this). > 2. Custom Trigger > > The custom trigger: > 1. keeps some data in its state under AggregatingStateDescriptor allowing me > to override "merge" method from Trigger class. > > 2. In onElement method, for the first call I execute > ctx.registerEventTimeTimer(window.maxTimestamp()); > Additionally in this method I added the busioenss logic which returns > TriggerResult.FIRE or TriggerResult.CONTINUE > > 3. The onProcessingTime methods returns TriggerResult.FIRE > > 3. all other methods are returning TriggerResult.CONTINUE > > > > As a result, I can observe that my window is fired two times. One from > onElement method where the busienss condition is meet and second time from > onProcessingTime method. > > What is the best way to prevent this? > > Regards, > Krzysztof > > > > -- > Sent from: > http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/
