Hi,
I want to perform some processing on events only when the watermark is updated. Otherwise, for all other events, I want to keep buffering them till the watermark arrives.
The main motivation behind doing this is that I have several operators that emit events/messages to a downstream operator. Since the order in which events arrive at the downstream operator is not guaranteed to be in chronological event time, I want to manually sort events when the watermark arrives and only then proceed.
Specifically, I want to first combine multiple streams and then do the above. Something like :
stream1.union(stream2, steream3)...
One solution I am exploring is using a global window with a trigger that will fire only when the watermark updates.
stream1.union(stream2, steream3).
keyBy(...).
window(GlobalWindows.create()).
trigger(new OnWatermarkUpdateTrigger()).
process(...)
I will store the latest watermark in the trigger's state store. In the onElement() method, I will FIRE if the current watermark is different than the stored one.
Is this the best way to implement the functionality described above?