Repository: flink Updated Branches: refs/heads/master 16ee4a5ce -> 2029c14eb
[FLINK-5149] let ContinuousEventTimeTrigger fire at the end of the window This changes the ContinuousEventTimeTrigger to behave like the EventTimeTrigger in the sense that it also triggers at the end of the window. This prevents the trigger from not firing at all in case the first trigger interval is after the window end. This closes #2860. [typo] fix toString() of ContinuousEventTimeTrigger This closes #2854. Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/2029c14e Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/2029c14e Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/2029c14e Branch: refs/heads/master Commit: 2029c14eb7d5b6788eb9297c271dd16a0d819fbe Parents: 16ee4a5 Author: Maximilian Michels <[email protected]> Authored: Wed Nov 23 16:01:35 2016 +0100 Committer: Maximilian Michels <[email protected]> Committed: Thu Nov 24 11:33:14 2016 +0100 ---------------------------------------------------------------------- .../triggers/ContinuousEventTimeTrigger.java | 22 +++++++++++++------- 1 file changed, 15 insertions(+), 7 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flink/blob/2029c14e/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousEventTimeTrigger.java ---------------------------------------------------------------------- diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousEventTimeTrigger.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousEventTimeTrigger.java index fa5bb2f..02c2a42 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousEventTimeTrigger.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousEventTimeTrigger.java @@ -52,31 +52,39 @@ public class ContinuousEventTimeTrigger<W extends Window> extends Trigger<Object @Override public TriggerResult onElement(Object element, long timestamp, W window, TriggerContext ctx) throws Exception { - ReducingState<Long> fireTimestamp = ctx.getPartitionedState(stateDesc); + if (window.maxTimestamp() <= ctx.getCurrentWatermark()) { + // if the watermark is already past the window fire immediately + return TriggerResult.FIRE; + } else { + ctx.registerEventTimeTimer(window.maxTimestamp()); + } + ReducingState<Long> fireTimestamp = ctx.getPartitionedState(stateDesc); if (fireTimestamp.get() == null) { long start = timestamp - (timestamp % interval); long nextFireTimestamp = start + interval; - ctx.registerEventTimeTimer(nextFireTimestamp); - fireTimestamp.add(nextFireTimestamp); - return TriggerResult.CONTINUE; } + return TriggerResult.CONTINUE; } @Override public TriggerResult onEventTime(long time, W window, TriggerContext ctx) throws Exception { - ReducingState<Long> fireTimestamp = ctx.getPartitionedState(stateDesc); + if (time == window.maxTimestamp()){ + return TriggerResult.FIRE; + } + + ReducingState<Long> fireTimestamp = ctx.getPartitionedState(stateDesc); if (fireTimestamp.get().equals(time)) { fireTimestamp.clear(); fireTimestamp.add(time + interval); ctx.registerEventTimeTimer(time + interval); return TriggerResult.FIRE; - } + return TriggerResult.CONTINUE; } @@ -112,7 +120,7 @@ public class ContinuousEventTimeTrigger<W extends Window> extends Trigger<Object @Override public String toString() { - return "ContinuousProcessingTimeTrigger(" + interval + ")"; + return "ContinuousEventTimeTrigger(" + interval + ")"; } @VisibleForTesting
