This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 64ca88ac989ee7525cb821670f293404b7b30d2d
Author: Zhijiang <wangzhijiang...@aliyun.com>
AuthorDate: Fri May 29 11:57:33 2020 +0200
[FLINK-17994][checkpointing] Fix the race condition between
CheckpointBarrierUnaligner#processBarrier and #notifyBarrierReceived
The race condition happens as following:
1. CheckpointBarrierUnaligner#notifyBarrierReceived triggers an async
checkpoint(ch1) in mailbox by netty thread.
2. CheckpointBarrierUnaligner#processBarrier also triggers a sync
checkpoint(ch2) by task thread and executes immediately.
3. When ch1 is taken from mailbox by task thread to execute, it will cause
illegal argument exception because it is smaller than the previous executed ch2.
For async checkpoint action, before it is actual executing, we can compare
its id with previous executed checkpoint id. If it is not larger than the
previous
one, we should ignore it to exit directly.
This closes #12406.
---
.../runtime/io/CheckpointBarrierUnaligner.java | 26 ++++--
.../flink/streaming/runtime/tasks/StreamTask.java | 3 +-
.../runtime/io/CheckpointBarrierUnalignerTest.java | 100 ++++++++++++++++++++-
3 files changed, 119 insertions(+), 10 deletions(-)
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java
index 01b1219..1d4bf82 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java
@@ -137,8 +137,9 @@ public class CheckpointBarrierUnaligner extends
CheckpointBarrierHandler {
}
/**
- * We still need to trigger checkpoint while reading the first barrier
from one channel, because
- * this might happen earlier than the previous async trigger via
mailbox by netty thread.
+ * We still need to trigger checkpoint via {@link
ThreadSafeUnaligner#notifyBarrierReceived(CheckpointBarrier, InputChannelInfo)}
+ * while reading the first barrier from one channel, because this might
happen
+ * earlier than the previous async trigger via mailbox by netty thread.
*
* <p>Note this is also suitable for the trigger case of local input
channel.
*/
@@ -256,8 +257,20 @@ public class CheckpointBarrierUnaligner extends
CheckpointBarrierHandler {
return threadSafeUnaligner.getNumOpenChannels();
}
+ @VisibleForTesting
+ ThreadSafeUnaligner getThreadSafeUnaligner() {
+ return threadSafeUnaligner;
+ }
+
+ private void notifyCheckpoint(CheckpointBarrier barrier) throws
IOException {
+ // ignore the previous triggered checkpoint by netty thread if
it was already canceled or aborted before.
+ if (barrier.getId() >=
threadSafeUnaligner.getCurrentCheckpointId()) {
+ super.notifyCheckpoint(barrier, 0);
+ }
+ }
+
@ThreadSafe
- private static class ThreadSafeUnaligner implements
BufferReceivedListener, Closeable {
+ static class ThreadSafeUnaligner implements BufferReceivedListener,
Closeable {
/**
* Tag the state of which input channel has not received the
barrier, such that newly arriving buffers need
@@ -280,7 +293,6 @@ public class CheckpointBarrierUnaligner extends
CheckpointBarrierHandler {
*/
private long currentReceivedCheckpointId = -1L;
- /** The number of open channels. */
private int numOpenChannels;
private final ChannelStateWriter channelStateWriter;
@@ -300,7 +312,7 @@ public class CheckpointBarrierUnaligner extends
CheckpointBarrierHandler {
if (currentReceivedCheckpointId < barrierId) {
handleNewCheckpoint(barrier);
- handler.executeInTaskThread(() ->
handler.notifyCheckpoint(barrier, 0), "notifyCheckpoint");
+ handler.executeInTaskThread(() ->
handler.notifyCheckpoint(barrier), "notifyCheckpoint");
}
int channelIndex =
handler.getFlattenedChannelIndex(channelInfo);
@@ -396,5 +408,9 @@ public class CheckpointBarrierUnaligner extends
CheckpointBarrierHandler {
synchronized int getNumOpenChannels() {
return numOpenChannels;
}
+
+ synchronized long getCurrentCheckpointId() {
+ return currentReceivedCheckpointId;
+ }
}
}
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
index 7c08712..7f0011f 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
@@ -545,7 +545,8 @@ public abstract class StreamTask<OUT, OP extends
StreamOperator<OUT>>
cleanUpInvoke();
}
- protected boolean runMailboxStep() throws Exception {
+ @VisibleForTesting
+ public boolean runMailboxStep() throws Exception {
return mailboxProcessor.runMailboxStep();
}
diff --git
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnalignerTest.java
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnalignerTest.java
index 7244539..18a60ec 100644
---
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnalignerTest.java
+++
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnalignerTest.java
@@ -20,6 +20,8 @@ package org.apache.flink.streaming.runtime.io;
import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter;
+import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo;
import org.apache.flink.runtime.checkpoint.channel.RecordingChannelStateWriter;
import org.apache.flink.runtime.io.network.NettyShuffleEnvironment;
import org.apache.flink.runtime.io.network.NettyShuffleEnvironmentBuilder;
@@ -37,6 +39,9 @@ import
org.apache.flink.runtime.io.network.partition.consumer.SingleInputGateBui
import org.apache.flink.runtime.io.network.util.TestBufferFactory;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
import org.apache.flink.runtime.operators.testutils.DummyEnvironment;
+import
org.apache.flink.streaming.runtime.io.CheckpointBarrierUnaligner.ThreadSafeUnaligner;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.runtime.tasks.mailbox.MailboxDefaultAction;
import org.apache.flink.util.function.ThrowingRunnable;
import org.junit.After;
@@ -48,6 +53,9 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
@@ -79,11 +87,15 @@ public class CheckpointBarrierUnalignerTest {
@After
public void ensureEmpty() throws Exception {
- assertFalse(inputGate.pollNext().isPresent());
- assertTrue(inputGate.isFinished());
+ if (inputGate != null) {
+ assertFalse(inputGate.pollNext().isPresent());
+ assertTrue(inputGate.isFinished());
+ inputGate.close();
+ }
- channelStateWriter.close();
- inputGate.close();
+ if (channelStateWriter != null) {
+ channelStateWriter.close();
+ }
}
//
------------------------------------------------------------------------
@@ -463,6 +475,43 @@ public class CheckpointBarrierUnalignerTest {
assertInflightData();
}
+ /**
+ * Tests the race condition between {@link
CheckpointBarrierUnaligner#processBarrier(CheckpointBarrier, int)}
+ * and {@link
ThreadSafeUnaligner#notifyBarrierReceived(CheckpointBarrier,
InputChannelInfo)}. The barrier
+ * notification will trigger an async checkpoint (ch1) via mailbox, and
meanwhile the barrier processing will
+ * execute the next checkpoint (ch2) directly in advance. When the ch1
action is taken from mailbox to execute,
+ * it should be exit because it is smaller than the finished ch2.
+ */
+ @Test
+ public void testConcurrentProcessBarrierAndNotifyBarrierReceived()
throws Exception {
+ final ValidatingCheckpointInvokable invokable = new
ValidatingCheckpointInvokable();
+ final CheckpointBarrierUnaligner handler = new
CheckpointBarrierUnaligner(new int[] { 1 }, ChannelStateWriter.NO_OP, "test",
invokable);
+ final InputChannelInfo channelInfo = new InputChannelInfo(0, 0);
+ final ExecutorService executor =
Executors.newFixedThreadPool(1);
+
+ try {
+ // Enqueue the checkpoint (ch0) action into the mailbox
of invokable because it is triggered by other thread.
+ Callable<Void> notifyTask = () -> {
+
handler.getThreadSafeUnaligner().notifyBarrierReceived(buildCheckpointBarrier(0),
channelInfo);
+ return null;
+ };
+ Future<Void> result = executor.submit(notifyTask);
+ result.get();
+
+ // Execute the checkpoint (ch1) directly because it is
triggered by main thread.
+ handler.processBarrier(buildCheckpointBarrier(1), 0);
+
+ // Run the previous queued mailbox action to execute
ch0.
+ invokable.runMailboxStep();
+
+ // ch0 will not be executed finally because it is
smaller than the previously executed ch1.
+ assertEquals(1, invokable.getTriggeredCheckpointId());
+ assertEquals(1,
invokable.getTotalTriggeredCheckpoints());
+ } finally {
+ executor.shutdown();
+ }
+ }
+
//
------------------------------------------------------------------------
// Utils
//
------------------------------------------------------------------------
@@ -571,6 +620,10 @@ public class CheckpointBarrierUnalignerTest {
.collect(Collectors.toList());
}
+ private CheckpointBarrier buildCheckpointBarrier(int id) {
+ return new CheckpointBarrier(id, 0,
CheckpointOptions.forCheckpointWithDefaultLocation());
+ }
+
//
------------------------------------------------------------------------
// Testing Mocks
//
------------------------------------------------------------------------
@@ -639,4 +692,43 @@ public class CheckpointBarrierUnalignerTest {
return lastCanceledCheckpointId;
}
}
+
+ /**
+ * Specific {@link AbstractInvokable} implementation to record and
validate which checkpoint
+ * id is executed and how many checkpoints are executed.
+ */
+ private static final class ValidatingCheckpointInvokable extends
StreamTask {
+
+ private long expectedCheckpointId;
+
+ private int totalNumCheckpoints;
+
+ ValidatingCheckpointInvokable() throws Exception {
+ super(new DummyEnvironment("test", 1, 0));
+ }
+
+ @Override
+ public void init() {
+ }
+
+ @Override
+ protected void processInput(MailboxDefaultAction.Controller
controller) {
+ }
+
+ public void triggerCheckpointOnBarrier(
+ CheckpointMetaData checkpointMetaData,
+ CheckpointOptions checkpointOptions,
+ CheckpointMetrics checkpointMetrics) {
+ expectedCheckpointId =
checkpointMetaData.getCheckpointId();
+ totalNumCheckpoints++;
+ }
+
+ long getTriggeredCheckpointId() {
+ return expectedCheckpointId;
+ }
+
+ int getTotalTriggeredCheckpoints() {
+ return totalNumCheckpoints;
+ }
+ }
}
