You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ga...@apache.org on 2021/08/27 02:42:00 UTC
[flink] branch master updated: [FLINK-23884][checkpoint] Fix the
concurrent problem between triggering savepoint with drain and task
finishing
This is an automated email from the ASF dual-hosted git repository.
gaoyunhaii pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 475d9ca [FLINK-23884][checkpoint] Fix the concurrent problem between triggering savepoint with drain and task finishing
475d9ca is described below
commit 475d9cace988fc7a8b163df64b19bbd84c360bfc
Author: Yun Gao <ga...@gmail.com>
AuthorDate: Wed Aug 25 23:59:20 2021 +0800
[FLINK-23884][checkpoint] Fix the concurrent problem between triggering savepoint with drain and task finishing
This closes #16905.
---
.../runtime/tasks/MultipleInputStreamTask.java | 35 +++++++----
.../runtime/tasks/SourceOperatorStreamTask.java | 20 +++++--
.../flink/streaming/runtime/tasks/StreamTask.java | 9 +--
.../runtime/tasks/MultipleInputStreamTaskTest.java | 66 +++++++++++++++++++++
.../tasks/SourceOperatorStreamTaskTest.java | 58 ++++++++++++++++++
.../runtime/tasks/SourceStreamTaskTest.java | 69 ++++++++++++++++++++++
6 files changed, 237 insertions(+), 20 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTask.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTask.java
index 92a6500..89a5434 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTask.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTask.java
@@ -196,16 +196,7 @@ public class MultipleInputStreamTask<OUT>
// unfinished network inputs, and the checkpoint would be triggered
// after received all the EndOfPartitionEvent.
if (options.getCheckpointType().shouldDrain()) {
- CompletableFuture<Void> sourcesStopped =
- FutureUtils.waitForAll(
- operatorChain.getSourceTaskInputs().stream()
- .map(s -> s.getOperator().stop())
- .collect(Collectors.toList()));
-
- return assertTriggeringCheckpointExceptions(
- sourcesStopped.thenCompose(
- ignore -> triggerSourcesCheckpointAsync(metadata, options)),
- metadata.getCheckpointId());
+ return triggerStopWithSavepointWithDrainAsync(metadata, options);
} else {
return triggerSourcesCheckpointAsync(metadata, options);
}
@@ -244,6 +235,30 @@ public class MultipleInputStreamTask<OUT>
return resultFuture;
}
+ private CompletableFuture<Boolean> triggerStopWithSavepointWithDrainAsync(
+ CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions) {
+
+ CompletableFuture<Void> sourcesStopped = new CompletableFuture<>();
+ mainMailboxExecutor.execute(
+ () -> {
+ setSynchronousSavepoint(checkpointMetaData.getCheckpointId(), true);
+ FutureUtils.forward(
+ FutureUtils.waitForAll(
+ operatorChain.getSourceTaskInputs().stream()
+ .map(s -> s.getOperator().stop())
+ .collect(Collectors.toList())),
+ sourcesStopped);
+ },
+ "stop chained Flip-27 source for stop-with-savepoint --drain");
+
+ return assertTriggeringCheckpointExceptions(
+ sourcesStopped.thenCompose(
+ ignore ->
+ triggerSourcesCheckpointAsync(
+ checkpointMetaData, checkpointOptions)),
+ checkpointMetaData.getCheckpointId());
+ }
+
private void checkPendingCheckpointCompletedFuturesSize() {
if (pendingCheckpointCompletedFutures.size() > MAX_TRACKED_CHECKPOINTS) {
ArrayList<Long> pendingCheckpointIds =
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java
index d2ac97d..a2658ae 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTask.java
@@ -40,6 +40,7 @@ import org.apache.flink.streaming.runtime.metrics.WatermarkGauge;
import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
+import org.apache.flink.util.concurrent.FutureUtils;
import javax.annotation.Nullable;
@@ -127,13 +128,20 @@ public class SourceOperatorStreamTask<T> extends StreamTask<T, SourceOperator<T,
private CompletableFuture<Boolean> triggerStopWithSavepointWithDrainAsync(
CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions) {
+
+ CompletableFuture<Void> operatorFinished = new CompletableFuture<>();
+ mainMailboxExecutor.execute(
+ () -> {
+ setSynchronousSavepoint(checkpointMetaData.getCheckpointId(), true);
+ FutureUtils.forward(mainOperator.stop(), operatorFinished);
+ },
+ "stop Flip-27 source for stop-with-savepoint --drain");
+
return assertTriggeringCheckpointExceptions(
- mainOperator
- .stop()
- .thenCompose(
- (ignore) ->
- super.triggerCheckpointAsync(
- checkpointMetaData, checkpointOptions)),
+ operatorFinished.thenCompose(
+ (ignore) ->
+ super.triggerCheckpointAsync(
+ checkpointMetaData, checkpointOptions)),
checkpointMetaData.getCheckpointId());
}
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 59fb033..b2ba02b 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
@@ -1199,18 +1199,19 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
CompletableFuture<Boolean> checkpointTriggered =
triggerFuture.exceptionally(
error -> {
- if (error instanceof RejectedExecutionException) {
+ if (ExceptionUtils.findThrowable(
+ error, RejectedExecutionException.class)
+ .isPresent()) {
// This may happen if the mailbox is closed. It means that
- // the task is shutting
- // down, so we just ignore it.
+ // the task is shutting down, so we just ignore it.
LOG.debug(
"Triggering checkpoint {} for {} was rejected by the mailbox",
checkpointId,
getTaskNameWithSubtaskAndId());
+ return false;
} else {
throw new WrappingRuntimeException(error);
}
- return null;
});
FutureUtils.assertNoException(checkpointTriggered);
return checkpointTriggered;
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java
index e586ef2..1fa0549 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java
@@ -19,6 +19,7 @@
package org.apache.flink.streaming.runtime.tasks;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.eventtime.TimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;
@@ -41,9 +42,12 @@ import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.Gauge;
import org.apache.flink.metrics.Metric;
import org.apache.flink.metrics.groups.OperatorMetricGroup;
+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.CheckpointType;
import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.io.network.api.CancelCheckpointMarker;
import org.apache.flink.runtime.io.network.api.CheckpointBarrier;
import org.apache.flink.runtime.io.network.api.EndOfData;
@@ -63,6 +67,8 @@ import org.apache.flink.runtime.metrics.util.InterceptingTaskMetricGroup;
import org.apache.flink.runtime.source.event.AddSplitEvent;
import org.apache.flink.runtime.source.event.NoMoreSplitsEvent;
import org.apache.flink.runtime.state.CheckpointStorageLocationReference;
+import org.apache.flink.runtime.taskmanager.CheckpointResponder;
+import org.apache.flink.runtime.taskmanager.TestCheckpointResponder;
import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.operators.AbstractInput;
@@ -114,6 +120,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
+import static org.apache.flink.runtime.checkpoint.CheckpointType.SAVEPOINT_TERMINATE;
import static org.apache.flink.streaming.runtime.tasks.StreamTaskFinalCheckpointsTest.processMailTillCheckpointSucceeds;
import static org.apache.flink.streaming.runtime.tasks.StreamTaskFinalCheckpointsTest.triggerCheckpoint;
import static org.apache.flink.util.Preconditions.checkArgument;
@@ -1054,6 +1061,65 @@ public class MultipleInputStreamTaskTest {
}
}
+ @Test
+ public void testTriggeringStopWithSavepointWithDrain() throws Exception {
+ SourceOperatorFactory<Integer> sourceOperatorFactory =
+ new SourceOperatorFactory<>(
+ new MockSource(Boundedness.CONTINUOUS_UNBOUNDED, 2),
+ WatermarkStrategy.noWatermarks());
+
+ CompletableFuture<Boolean> checkpointCompleted = new CompletableFuture<>();
+ CheckpointResponder checkpointResponder =
+ new TestCheckpointResponder() {
+ @Override
+ public void acknowledgeCheckpoint(
+ JobID jobID,
+ ExecutionAttemptID executionAttemptID,
+ long checkpointId,
+ CheckpointMetrics checkpointMetrics,
+ TaskStateSnapshot subtaskState) {
+ super.acknowledgeCheckpoint(
+ jobID,
+ executionAttemptID,
+ checkpointId,
+ checkpointMetrics,
+ subtaskState);
+ checkpointCompleted.complete(null);
+ }
+ };
+
+ try (StreamTaskMailboxTestHarness<String> testHarness =
+ new StreamTaskMailboxTestHarnessBuilder<>(
+ MultipleInputStreamTask::new, BasicTypeInfo.STRING_TYPE_INFO)
+ .setCollectNetworkEvents()
+ .modifyStreamConfig(config -> config.setCheckpointingEnabled(true))
+ .modifyExecutionConfig(applyObjectReuse(objectReuse))
+ .addInput(BasicTypeInfo.INT_TYPE_INFO)
+ .addInput(BasicTypeInfo.INT_TYPE_INFO)
+ .addInput(BasicTypeInfo.INT_TYPE_INFO)
+ .setTaskStateSnapshot(1, TaskStateSnapshot.FINISHED_ON_RESTORE)
+ .setupOperatorChain(new LifeCycleMonitorMultipleInputOperatorFactory())
+ .finishForSingletonOperatorChain(StringSerializer.INSTANCE)
+ .setCheckpointResponder(checkpointResponder)
+ .build()) {
+ CompletableFuture<Boolean> triggerResult =
+ testHarness.streamTask.triggerCheckpointAsync(
+ new CheckpointMetaData(1, 1),
+ CheckpointOptions.alignedNoTimeout(
+ SAVEPOINT_TERMINATE,
+ CheckpointStorageLocationReference.getDefault()));
+ checkpointCompleted.whenComplete(
+ (ignored, exception) ->
+ testHarness.streamTask.notifyCheckpointCompleteAsync(1));
+ testHarness.waitForTaskCompletion();
+ testHarness.finishProcessing();
+
+ assertTrue(triggerResult.isDone());
+ assertTrue(triggerResult.get());
+ assertTrue(checkpointCompleted.isDone());
+ }
+ }
+
/** Test implementation of {@link MultipleInputStreamOperator}. */
protected static class MapToStringMultipleInputOperator extends AbstractStreamOperatorV2<String>
implements MultipleInputStreamOperator<String> {
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTaskTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTaskTest.java
index 6f188fe..a91be4f 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTaskTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTaskTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.streaming.runtime.tasks;
+import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.eventtime.TimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
@@ -33,15 +34,19 @@ import org.apache.flink.api.connector.source.mocks.MockSourceSplit;
import org.apache.flink.api.connector.source.mocks.MockSourceSplitSerializer;
import org.apache.flink.core.io.InputStatus;
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.CheckpointType;
import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.io.network.api.CheckpointBarrier;
import org.apache.flink.runtime.io.network.api.EndOfData;
import org.apache.flink.runtime.io.network.api.writer.RecordOrEventCollectingResultPartitionWriter;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.source.event.AddSplitEvent;
import org.apache.flink.runtime.state.CheckpointStorageLocationReference;
+import org.apache.flink.runtime.taskmanager.CheckpointResponder;
+import org.apache.flink.runtime.taskmanager.TestCheckpointResponder;
import org.apache.flink.streaming.api.operators.SourceOperator;
import org.apache.flink.streaming.api.operators.SourceOperatorFactory;
import org.apache.flink.streaming.api.watermark.Watermark;
@@ -66,6 +71,7 @@ import java.util.concurrent.Future;
import java.util.function.Supplier;
import java.util.stream.IntStream;
+import static org.apache.flink.runtime.checkpoint.CheckpointType.SAVEPOINT_TERMINATE;
import static org.apache.flink.streaming.util.TestHarnessUtil.assertOutputEquals;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.MatcherAssert.assertThat;
@@ -215,6 +221,58 @@ public class SourceOperatorStreamTaskTest extends SourceStreamTaskTestBase {
}
}
+ @Test
+ public void testTriggeringStopWithSavepointWithDrain() throws Exception {
+ SourceOperatorFactory<Integer> sourceOperatorFactory =
+ new SourceOperatorFactory<>(
+ new MockSource(Boundedness.CONTINUOUS_UNBOUNDED, 2),
+ WatermarkStrategy.noWatermarks());
+
+ CompletableFuture<Boolean> checkpointCompleted = new CompletableFuture<>();
+ CheckpointResponder checkpointResponder =
+ new TestCheckpointResponder() {
+ @Override
+ public void acknowledgeCheckpoint(
+ JobID jobID,
+ ExecutionAttemptID executionAttemptID,
+ long checkpointId,
+ CheckpointMetrics checkpointMetrics,
+ TaskStateSnapshot subtaskState) {
+ super.acknowledgeCheckpoint(
+ jobID,
+ executionAttemptID,
+ checkpointId,
+ checkpointMetrics,
+ subtaskState);
+ checkpointCompleted.complete(null);
+ }
+ };
+
+ try (StreamTaskMailboxTestHarness<Integer> testHarness =
+ new StreamTaskMailboxTestHarnessBuilder<>(
+ SourceOperatorStreamTask::new, BasicTypeInfo.INT_TYPE_INFO)
+ .setupOutputForSingletonOperatorChain(sourceOperatorFactory)
+ .setCheckpointResponder(checkpointResponder)
+ .build()) {
+
+ CompletableFuture<Boolean> triggerResult =
+ testHarness.streamTask.triggerCheckpointAsync(
+ new CheckpointMetaData(1, 1),
+ CheckpointOptions.alignedNoTimeout(
+ SAVEPOINT_TERMINATE,
+ CheckpointStorageLocationReference.getDefault()));
+ checkpointCompleted.whenComplete(
+ (ignored, exception) ->
+ testHarness.streamTask.notifyCheckpointCompleteAsync(1));
+ testHarness.waitForTaskCompletion();
+ testHarness.finishProcessing();
+
+ assertTrue(triggerResult.isDone());
+ assertTrue(triggerResult.get());
+ assertTrue(checkpointCompleted.isDone());
+ }
+ }
+
private TaskStateSnapshot executeAndWaitForCheckpoint(
long checkpointId, TaskStateSnapshot initialSnapshot, IntStream expectedRecords)
throws Exception {
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTest.java
index 26a3156..dd91678 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTest.java
@@ -45,6 +45,8 @@ import org.apache.flink.runtime.io.network.partition.ResultPartition;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.operators.testutils.ExpectedTestException;
+import org.apache.flink.runtime.state.CheckpointStorageLocationReference;
+import org.apache.flink.runtime.taskmanager.CheckpointResponder;
import org.apache.flink.runtime.taskmanager.TestCheckpointResponder;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.checkpoint.ListCheckpointed;
@@ -93,6 +95,7 @@ import java.util.concurrent.atomic.AtomicLong;
import static org.apache.flink.api.common.typeinfo.BasicTypeInfo.INT_TYPE_INFO;
import static org.apache.flink.api.common.typeinfo.BasicTypeInfo.STRING_TYPE_INFO;
import static org.apache.flink.runtime.checkpoint.CheckpointType.SAVEPOINT_SUSPEND;
+import static org.apache.flink.runtime.checkpoint.CheckpointType.SAVEPOINT_TERMINATE;
import static org.apache.flink.runtime.state.CheckpointStorageLocationReference.getDefault;
import static org.apache.flink.streaming.runtime.tasks.StreamTaskFinalCheckpointsTest.triggerCheckpoint;
import static org.apache.flink.util.Preconditions.checkState;
@@ -733,6 +736,55 @@ public class SourceStreamTaskTest extends SourceStreamTaskTestBase {
}
}
+ @Test
+ public void testTriggeringStopWithSavepointWithDrain() throws Exception {
+ SourceFunction<String> testSource = new EmptySource();
+
+ CompletableFuture<Boolean> checkpointCompleted = new CompletableFuture<>();
+ CheckpointResponder checkpointResponder =
+ new TestCheckpointResponder() {
+ @Override
+ public void acknowledgeCheckpoint(
+ JobID jobID,
+ ExecutionAttemptID executionAttemptID,
+ long checkpointId,
+ CheckpointMetrics checkpointMetrics,
+ TaskStateSnapshot subtaskState) {
+ super.acknowledgeCheckpoint(
+ jobID,
+ executionAttemptID,
+ checkpointId,
+ checkpointMetrics,
+ subtaskState);
+ checkpointCompleted.complete(null);
+ }
+ };
+
+ try (StreamTaskMailboxTestHarness<String> harness =
+ new StreamTaskMailboxTestHarnessBuilder<>(SourceStreamTask::new, STRING_TYPE_INFO)
+ .setTaskStateSnapshot(1, TaskStateSnapshot.FINISHED_ON_RESTORE)
+ .setCheckpointResponder(checkpointResponder)
+ .setupOutputForSingletonOperatorChain(new StreamSource<>(testSource))
+ .build()) {
+ CompletableFuture<Boolean> triggerResult =
+ harness.streamTask.triggerCheckpointAsync(
+ new CheckpointMetaData(1, 1),
+ CheckpointOptions.alignedNoTimeout(
+ SAVEPOINT_TERMINATE,
+ CheckpointStorageLocationReference.getDefault()));
+ checkpointCompleted.whenComplete(
+ (ignored, exception) -> harness.streamTask.notifyCheckpointCompleteAsync(1));
+
+ // Run mailbox till the source thread finished and suspend the mailbox
+ harness.streamTask.runMailboxLoop();
+ harness.finishProcessing();
+
+ assertTrue(triggerResult.isDone());
+ assertTrue(triggerResult.get());
+ assertTrue(checkpointCompleted.isDone());
+ }
+ }
+
private static class LifeCycleMonitorSource extends RichParallelSourceFunction<String> {
private final LifeCycleMonitor lifeCycleMonitor = new LifeCycleMonitor();
@@ -1085,4 +1137,21 @@ public class SourceStreamTaskTest extends SourceStreamTaskTestBase {
ALLOW_EXIT.trigger();
}
}
+
+ private static class EmptySource implements SourceFunction<String> {
+
+ private volatile boolean isCanceled;
+
+ @Override
+ public void run(SourceContext<String> ctx) throws Exception {
+ while (!isCanceled) {
+ Thread.sleep(100);
+ }
+ }
+
+ @Override
+ public void cancel() {
+ isCanceled = true;
+ }
+ }
}