This is an automated email from the ASF dual-hosted git repository.
jqin pushed a commit to branch release-1.15
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.15 by this push:
new 9fc89a05f12 [FLINK-26394][checkpoint] Cancel the checkpoint
completable future when checkpoint is aborting.
9fc89a05f12 is described below
commit 9fc89a05f128ab645b73687f240fb14b57790fc6
Author: Gen Luo <[email protected]>
AuthorDate: Wed Apr 6 16:38:39 2022 +0800
[FLINK-26394][checkpoint] Cancel the checkpoint completable future when
checkpoint is aborting.
---
.../runtime/checkpoint/CheckpointCoordinator.java | 17 +++++++--
.../runtime/checkpoint/PendingCheckpoint.java | 7 +++-
.../checkpoint/CheckpointCoordinatorTest.java | 44 ++++++++++++++++++++++
.../runtime/checkpoint/PendingCheckpointTest.java | 3 +-
4 files changed, 65 insertions(+), 6 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
index 72a6b7032de..2d47d79f063 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
@@ -536,6 +536,8 @@ public class CheckpointCoordinator {
boolean initializeBaseLocations =
!baseLocationsForCheckpointInitialized;
baseLocationsForCheckpointInitialized = true;
+ CompletableFuture<Void> masterTriggerCompletionPromise = new
CompletableFuture<>();
+
final CompletableFuture<PendingCheckpoint>
pendingCheckpointCompletableFuture =
checkpointPlanFuture
.thenApplyAsync(
@@ -560,7 +562,8 @@ public class CheckpointCoordinator {
checkpointInfo.f0,
request.isPeriodic,
checkpointInfo.f1,
-
request.getOnCompletionFuture()),
+
request.getOnCompletionFuture(),
+
masterTriggerCompletionPromise),
timer);
final CompletableFuture<?> coordinatorCheckpointsComplete =
@@ -615,8 +618,12 @@ public class CheckpointCoordinator {
},
timer);
+ FutureUtils.forward(
+ CompletableFuture.allOf(masterStatesComplete,
coordinatorCheckpointsComplete),
+ masterTriggerCompletionPromise);
+
FutureUtils.assertNoException(
- CompletableFuture.allOf(masterStatesComplete,
coordinatorCheckpointsComplete)
+ masterTriggerCompletionPromise
.handleAsync(
(ignored, throwable) -> {
final PendingCheckpoint checkpoint =
@@ -778,7 +785,8 @@ public class CheckpointCoordinator {
CheckpointPlan checkpointPlan,
boolean isPeriodic,
long checkpointID,
- CompletableFuture<CompletedCheckpoint> onCompletionPromise) {
+ CompletableFuture<CompletedCheckpoint> onCompletionPromise,
+ CompletableFuture<Void> masterTriggerCompletionPromise) {
synchronized (lock) {
try {
@@ -803,7 +811,8 @@ public class CheckpointCoordinator {
masterHooks.keySet(),
props,
onCompletionPromise,
- pendingCheckpointStats);
+ pendingCheckpointStats,
+ masterTriggerCompletionPromise);
synchronized (lock) {
pendingCheckpoints.put(checkpointID, checkpoint);
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
index b4bd8eacc5e..8ca6e22020e 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
@@ -112,6 +112,8 @@ public class PendingCheckpoint implements Checkpoint {
@Nullable private final PendingCheckpointStats pendingCheckpointStats;
+ private final CompletableFuture<Void> masterTriggerCompletionPromise;
+
/** Target storage location to persist the checkpoint metadata to. */
@Nullable private CheckpointStorageLocation targetLocation;
@@ -136,7 +138,8 @@ public class PendingCheckpoint implements Checkpoint {
Collection<String> masterStateIdentifiers,
CheckpointProperties props,
CompletableFuture<CompletedCheckpoint> onCompletionPromise,
- @Nullable PendingCheckpointStats pendingCheckpointStats) {
+ @Nullable PendingCheckpointStats pendingCheckpointStats,
+ CompletableFuture<Void> masterTriggerCompletionPromise) {
checkArgument(
checkpointPlan.getTasksToWaitFor().size() > 0,
"Checkpoint needs at least one vertex that commits the
checkpoint");
@@ -166,6 +169,7 @@ public class PendingCheckpoint implements Checkpoint {
this.acknowledgedTasks = new
HashSet<>(checkpointPlan.getTasksToWaitFor().size());
this.onCompletionPromise = checkNotNull(onCompletionPromise);
this.pendingCheckpointStats = pendingCheckpointStats;
+ this.masterTriggerCompletionPromise =
checkNotNull(masterTriggerCompletionPromise);
}
//
--------------------------------------------------------------------------------------------
@@ -544,6 +548,7 @@ public class PendingCheckpoint implements Checkpoint {
try {
failureCause = new CheckpointException(reason, cause);
onCompletionPromise.completeExceptionally(failureCause);
+ masterTriggerCompletionPromise.completeExceptionally(failureCause);
assertAbortSubsumedForced(reason);
} finally {
dispose(true, checkpointsCleaner, postCleanup, executor);
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
index 9d3a87ddd84..234765b298f 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
@@ -3658,6 +3658,50 @@ public class CheckpointCoordinatorTest extends
TestLogger {
}
}
+ @Test
+ public void testTimeoutWhileCheckpointOperatorCoordinatorNotFinishing()
throws Exception {
+ JobVertexID jobVertexID = new JobVertexID();
+ ExecutionGraph graph =
+ new
CheckpointCoordinatorTestingUtils.CheckpointExecutionGraphBuilder()
+ .addJobVertex(jobVertexID)
+ .build();
+
+
CheckpointCoordinatorTestingUtils.MockOperatorCoordinatorCheckpointContext
context =
+ new CheckpointCoordinatorTestingUtils
+
.MockOperatorCheckpointCoordinatorContextBuilder()
+ .setOperatorID(new OperatorID())
+ .setOnCallingCheckpointCoordinator(
+ (ignored, future) -> {
+ // Never complete
+ })
+ .build();
+
+ ScheduledExecutorService executorService =
Executors.newSingleThreadScheduledExecutor();
+ CheckpointCoordinator checkpointCoordinator =
+ new CheckpointCoordinatorBuilder()
+ .setExecutionGraph(graph)
+ .setCheckpointCoordinatorConfiguration(
+ CheckpointCoordinatorConfiguration.builder()
+ .setCheckpointTimeout(10)
+ .build())
+ .setTimer(manuallyTriggeredScheduledExecutor)
+
.setCoordinatorsToCheckpoint(Collections.singleton(context))
+ .build();
+ try {
+ checkpointCoordinator.triggerCheckpoint(false);
+ manuallyTriggeredScheduledExecutor.triggerAll();
+ Assert.assertTrue(checkpointCoordinator.isTriggering());
+
+
manuallyTriggeredScheduledExecutor.triggerNonPeriodicScheduledTasks();
+ manuallyTriggeredScheduledExecutor.triggerAll();
+
+ Assert.assertFalse(checkpointCoordinator.isTriggering());
+ } finally {
+ checkpointCoordinator.shutdown();
+ executorService.shutdownNow();
+ }
+ }
+
@Test
public void testReportLatestCompletedCheckpointIdWithAbort() throws
Exception {
JobVertexID jobVertexID = new JobVertexID();
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/PendingCheckpointTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/PendingCheckpointTest.java
index 392bd0ef67c..fc8d57f9f93 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/PendingCheckpointTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/PendingCheckpointTest.java
@@ -644,7 +644,8 @@ public class PendingCheckpointTest {
masterStateIdentifiers,
props,
new CompletableFuture<>(),
- null);
+ null,
+ new CompletableFuture<>());
pendingCheckpoint.setCheckpointTargetLocation(location);
return pendingCheckpoint;
}
