Right now, when a Trigger returns TriggerResult.PURGE from any of the on*() methods the WindowOperator will clear all state of that window (window contents, merging window set) and call Trigger.clear() so that the Trigger can clean up its state/timers.
This is problematic in some cases. For example, with merging windows (session windows) this means that a late-arriving element will not be put into the session that was previously built up but will be put into a completely new session that only contains this one element.
The proposed new behaviour is this:
- Only clean window contents on PURGE
- Register cleanup timer for any window, don't delete this on PURGE
- When the cleanup timer fires: clean window state, clean merging window set, call Trigger.clear() to allow it to clean state/timers
FLINK-4415 this would give the trigger the power to purge state and window contents separately.