[FLINK-20443][API/DataStream] ContinuousEventTimeTrigger optimization

silenceland · Nov 1, 2021 · 3b163c7 · 3b163c7
1 parent a7e1526
commit 3b163c7
Showing 1 changed file with 13 additions and 8 deletions.
diff --git a/...in/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousEventTimeTrigger.java b/...in/java/org/apache/flink/streaming/api/windowing/triggers/ContinuousEventTimeTrigger.java
@@ -59,12 +59,10 @@ public TriggerResult onElement(Object element, long timestamp, W window, Trigger
             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);
+        ReducingState<Long> fireTimestampState = ctx.getPartitionedState(stateDesc);
+        if (fireTimestampState.get() == null) {
+            registerNextFireTimestamp(
+                    timestamp - (timestamp % interval), window, ctx, fireTimestampState);
         }
 
         return TriggerResult.CONTINUE;
@@ -83,8 +81,7 @@ public TriggerResult onEventTime(long time, W window, TriggerContext ctx) throws
 
         if (fireTimestamp != null && fireTimestamp == time) {
             fireTimestampState.clear();
-            fireTimestampState.add(time + interval);
-            ctx.registerEventTimeTimer(time + interval);
+            registerNextFireTimestamp(time, window, ctx, fireTimestampState);
             return TriggerResult.FIRE;
         }
 
@@ -149,4 +146,12 @@ public Long reduce(Long value1, Long value2) throws Exception {
             return Math.min(value1, value2);
         }
     }
+
+    private void registerNextFireTimestamp(
+            long time, W window, TriggerContext ctx, ReducingState<Long> fireTimestampState)
+            throws Exception {
+        long nextFireTimestamp = Math.min(time + interval, window.maxTimestamp());
+        fireTimestampState.add(nextFireTimestamp);
+        ctx.registerEventTimeTimer(nextFireTimestamp);
+    }
 }