[FLINK-34371][runtime] Set OutputOnlyAfterEndOfStream on WindowOperat…

…or and StreamSortOperator This closes #24272
apache · Feb 28, 2024 · 94b5f03 · 94b5f03
1 parent 5916aa4
commit 94b5f03
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 0 deletions.
diff --git a/.../src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java b/.../src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java
@@ -57,8 +57,11 @@
 import org.apache.flink.streaming.api.operators.InternalTimer;
 import org.apache.flink.streaming.api.operators.InternalTimerService;
 import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.OperatorAttributes;
+import org.apache.flink.streaming.api.operators.OperatorAttributesBuilder;
 import org.apache.flink.streaming.api.operators.TimestampedCollector;
 import org.apache.flink.streaming.api.operators.Triggerable;
+import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
 import org.apache.flink.streaming.api.windowing.assigners.MergingWindowAssigner;
 import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
 import org.apache.flink.streaming.api.windowing.triggers.Trigger;
@@ -527,6 +530,16 @@ && isCleanupTime(triggerContext.window, timer.getTimestamp())) {
         }
     }
 
+    @Override
+    public OperatorAttributes getOperatorAttributes() {
+        boolean isOutputOnlyAfterEndOfStream =
+                windowAssigner instanceof GlobalWindows
+                        && trigger instanceof GlobalWindows.EndOfStreamTrigger;
+        return new OperatorAttributesBuilder()
+                .setOutputOnlyAfterEndOfStream(isOutputOnlyAfterEndOfStream)
+                .build();
+    }
+
     /**
      * Drops all state for the given window and calls {@link Trigger#clear(Window,
      * Trigger.TriggerContext)}.

diff --git a/...ntime/src/main/java/org/apache/flink/table/runtime/operators/sort/StreamSortOperator.java b/...ntime/src/main/java/org/apache/flink/table/runtime/operators/sort/StreamSortOperator.java
@@ -26,6 +26,8 @@
 import org.apache.flink.runtime.state.StateInitializationContext;
 import org.apache.flink.runtime.state.StateSnapshotContext;
 import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.OperatorAttributes;
+import org.apache.flink.streaming.api.operators.OperatorAttributesBuilder;
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.data.binary.BinaryRowData;
@@ -153,4 +155,9 @@ public void finish() throws Exception {
         }
         super.finish();
     }
+
+    @Override
+    public OperatorAttributes getOperatorAttributes() {
+        return new OperatorAttributesBuilder().setOutputOnlyAfterEndOfStream(true).build();
+    }
 }