You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by xt...@apache.org on 2023/12/19 11:00:47 UTC
(flink) 01/09: [hotfix][runtime] CollectionDataInput handles WatermarkStatus
This is an automated email from the ASF dual-hosted git repository.
xtsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 72a0ed61085dfe5131fef88f2233e0f5ad9d323b
Author: sxnan <su...@gmail.com>
AuthorDate: Mon Dec 11 17:00:13 2023 +0800
[hotfix][runtime] CollectionDataInput handles WatermarkStatus
---
.../apache/flink/streaming/api/operators/sort/CollectionDataInput.java | 2 ++
1 file changed, 2 insertions(+)
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/CollectionDataInput.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/CollectionDataInput.java
index 2926c3acd05..63be46754b7 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/CollectionDataInput.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/CollectionDataInput.java
@@ -53,6 +53,8 @@ final class CollectionDataInput<E> implements StreamTaskInput<E> {
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);
}