You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2021/08/10 06:35:28 UTC
[flink] 08/11: [hotfix] Fix CollectResultBuffer
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit cdb783761b459bc42b56d44463106ba0b3cf48fe
Author: Dawid Wysakowicz <dw...@apache.org>
AuthorDate: Sun Jul 25 23:31:51 2021 +0200
[hotfix] Fix CollectResultBuffer
---
.../collect/AbstractCollectResultBuffer.java | 3 +++
.../operators/collect/CollectResultBufferTest.java | 21 ++++++++++++++++++++-
2 files changed, 23 insertions(+), 1 deletion(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/collect/AbstractCollectResultBuffer.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/collect/AbstractCollectResultBuffer.java
index bc50a30..835d0b5 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/collect/AbstractCollectResultBuffer.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/collect/AbstractCollectResultBuffer.java
@@ -128,6 +128,9 @@ public abstract class AbstractCollectResultBuffer<T> {
if (!results.isEmpty()) {
// response contains some data, add them to buffer
int addStart = (int) (offset - responseOffset);
+ if (addStart > results.size()) {
+ return;
+ }
List<T> addedResults = results.subList(addStart, results.size());
buffer.addAll(addedResults);
offset += addedResults.size();
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/collect/CollectResultBufferTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/collect/CollectResultBufferTest.java
index 7070a63..4fa2b83 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/collect/CollectResultBufferTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/collect/CollectResultBufferTest.java
@@ -159,8 +159,27 @@ public class CollectResultBufferTest {
Assert.assertEquals(expectedValue, buffer.next());
}
+ // send some uncommitted data
+ response =
+ new CollectCoordinationResponse(
+ version, 6, createSerializedResults(Arrays.asList(8, 9, 10)));
+ buffer.dealWithResponse(response, 7);
+ // send some committed data, but less than uncommitted data we've already sent
+ expected = Arrays.asList(7);
+ response =
+ new CollectCoordinationResponse(
+ version, 7, createSerializedResults(Arrays.asList(8, 9)));
+ buffer.dealWithResponse(response, 7);
+ // results before checkpoint can be seen now
+ for (Integer expectedValue : expected) {
+ Assert.assertEquals(expectedValue, buffer.next());
+ }
+
buffer.complete();
- Assert.assertEquals((Integer) 7, buffer.next());
+ expected = Arrays.asList(8, 9, 10);
+ for (Integer expectedValue : expected) {
+ Assert.assertEquals(expectedValue, buffer.next());
+ }
Assert.assertNull(buffer.next());
}