[FLINK-36395][runtime] Drain async state processing when operator finish

Zakelly · Sep 29, 2024 · 2ad98ca · 2ad98ca
1 parent 3f2fd23
commit 2ad98ca
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 0 deletions.
diff --git a/.../org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java b/.../org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java
@@ -331,6 +331,12 @@ RecordContext getCurrentProcessingContext() {
         return currentProcessingContext;
     }
 
+    @Override
+    public void finish() throws Exception {
+        super.finish();
+        asyncExecutionController.drainInflightRecords(0);
+    }
+
     @Override
     public void close() throws Exception {
         super.close();

diff --git a/...rg/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperatorV2.java b/...rg/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperatorV2.java
@@ -268,6 +268,12 @@ public RecordContext getCurrentProcessingContext() {
         return currentProcessingContext;
     }
 
+    @Override
+    public void finish() throws Exception {
+        super.finish();
+        asyncExecutionController.drainInflightRecords(0);
+    }
+
     @Override
     public void close() throws Exception {
         super.close();