This is an automated email from the ASF dual-hosted git repository.
fpaul pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.14 by this push:
new 8a168e1 [FLINK-26304][streaming] Do not forward failed committables
8a168e1 is described below
commit 8a168e147ef889116a3db9872ded73fa8ce7ea7d
Author: Fabian Paul <[email protected]>
AuthorDate: Tue Feb 22 17:19:08 2022 +0100
[FLINK-26304][streaming] Do not forward failed committables
With the introduction of retries of failed committables we should not
simply forward all committables but check their status. Only
committables that have been committed successfully should be forwarded.
---
.../sink/AbstractStreamingCommitterHandler.java | 16 ++++++++++++++--
.../operators/sink/StreamingCommitterHandlerTest.java | 7 +++++++
2 files changed, 21 insertions(+), 2 deletions(-)
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/AbstractStreamingCommitterHandler.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/AbstractStreamingCommitterHandler.java
index 2efd949..ba147b2 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/AbstractStreamingCommitterHandler.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/AbstractStreamingCommitterHandler.java
@@ -33,11 +33,14 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.IdentityHashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
+import java.util.Set;
import java.util.TreeMap;
+import java.util.stream.Collectors;
/**
* Abstract base class for operators that work with a {@link Committer} or a
{@link
@@ -140,8 +143,17 @@ abstract class AbstractStreamingCommitterHandler<InputT,
CommT>
}
LOG.info("Committing the state for checkpoint {}", checkpointId);
- recoveredCommittables(commit(readyCommittables));
- return readyCommittables;
+ final List<CommT> failedCommittables = commit(readyCommittables);
+ recoveredCommittables(failedCommittables);
+
+ // Do not forward failed committables
+ final Set<CommT> failedCommittableSet = Collections.newSetFromMap(new
IdentityHashMap<>());
+ failedCommittableSet.addAll(failedCommittables);
+
+ // Retain ordering of the committables
+ return readyCommittables.stream()
+ .filter(c -> !failedCommittableSet.contains(c))
+ .collect(Collectors.toList());
}
@Override
diff --git
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/StreamingCommitterHandlerTest.java
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/StreamingCommitterHandlerTest.java
index 86e2300..02774d4 100644
---
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/StreamingCommitterHandlerTest.java
+++
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/StreamingCommitterHandlerTest.java
@@ -39,6 +39,8 @@ import static
org.apache.flink.streaming.util.TestHarnessUtil.buildSubtaskState;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasSize;
+import static org.junit.Assert.assertTrue;
/** Test the {@link StreamingCommitterHandler}. */
public class StreamingCommitterHandlerTest extends TestLogger {
@@ -65,9 +67,14 @@ public class StreamingCommitterHandlerTest extends
TestLogger {
testHarness.prepareSnapshotPreBarrier(1);
testHarness.snapshot(1L, 1L);
testHarness.notifyOfCompletedCheckpoint(1L);
+
+ assertTrue(testHarness.getOutput().isEmpty());
+
testHarness.snapshot(2L, 2L);
testHarness.notifyOfCompletedCheckpoint(2L);
+ assertThat(testHarness.getOutput(), hasSize(2));
+
testHarness.close();
// committedData has the format (<input>, null, <number>)
