[hotfix][runtime] CollectionDataInput handles WatermarkStatus

Gerrrr · Dec 19, 2023 · 72a0ed6 · 72a0ed6
1 parent 4cc24c1
commit 72a0ed6
Showing 1 changed file with 2 additions and 0 deletions.
diff --git a/...java/src/test/java/org/apache/flink/streaming/api/operators/sort/CollectionDataInput.java b/...java/src/test/java/org/apache/flink/streaming/api/operators/sort/CollectionDataInput.java
@@ -53,6 +53,8 @@ public DataInputStatus emitNext(DataOutput<E> output) throws Exception {
                 output.emitRecord(streamElement.asRecord());
             } else if (streamElement instanceof Watermark) {
                 output.emitWatermark(streamElement.asWatermark());
+            } else if (streamElement.isWatermarkStatus()) {
+                output.emitWatermarkStatus(streamElement.asWatermarkStatus());
             } else {
                 throw new IllegalStateException("Unsupported element type: " + streamElement);
             }