[FLINK-20217][task] Enable splittable timers in CepOperator and Table…

…StreamOperator
yhjhoo · Jun 12, 2024 · f26ee47 · f26ee47
1 parent f93cac3
commit f26ee47
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 0 deletions.
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CepOperator.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CepOperator.java
@@ -160,6 +160,11 @@ public CepOperator(
         }
     }
 
+    @Override
+    public boolean useSplittableTimers() {
+        return true;
+    }
+
     @Override
     public void setup(
             StreamTask<?, ?> containingTask,

diff --git a/...e-runtime/src/main/java/org/apache/flink/table/runtime/operators/TableStreamOperator.java b/...e-runtime/src/main/java/org/apache/flink/table/runtime/operators/TableStreamOperator.java
@@ -52,6 +52,11 @@ public void open() throws Exception {
         this.ctx = new ContextImpl(getProcessingTimeService());
     }
 
+    @Override
+    public boolean useSplittableTimers() {
+        return true;
+    }
+
     /** Compute memory size from memory faction. */
     public long computeMemorySize() {
         final Environment environment = getContainingTask().getEnvironment();
@@ -71,6 +76,7 @@ public void processWatermark(Watermark mark) throws Exception {
         currentWatermark = mark.getTimestamp();
         super.processWatermark(mark);
     }
+
     /** Information available in an invocation of processElement. */
     protected class ContextImpl implements TimerService {