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 0c718666476 Revert "[FLINK-26394][checkpoint] Cancel the checkpoint
completable future when checkpoint is aborting."
0c718666476 is described below
commit 0c718666476ae469fb825e8fdf362470d7af2488
Author: Jiangjie (Becket) Qin <[email protected]>
AuthorDate: Thu Apr 14 17:56:59 2022 +0800
Revert "[FLINK-26394][checkpoint] Cancel the checkpoint completable future
when checkpoint is aborting."
This reverts commit 9fc89a05f128ab645b73687f240fb14b57790fc6 due to a
FLINK-27148.
---
.../runtime/checkpoint/CheckpointCoordinator.java | 17 ++-------
.../runtime/checkpoint/PendingCheckpoint.java | 7 +---
.../checkpoint/CheckpointCoordinatorTest.java | 44 ----------------------
.../runtime/checkpoint/PendingCheckpointTest.java | 3 +-
4 files changed, 6 insertions(+), 65 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 2d47d79f063..72a6b7032de 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,8 +536,6 @@ public class CheckpointCoordinator {
boolean initializeBaseLocations =
!baseLocationsForCheckpointInitialized;
baseLocationsForCheckpointInitialized = true;
- CompletableFuture<Void> masterTriggerCompletionPromise = new
CompletableFuture<>();
-
final CompletableFuture<PendingCheckpoint>
pendingCheckpointCompletableFuture =
checkpointPlanFuture
.thenApplyAsync(
@@ -562,8 +560,7 @@ public class CheckpointCoordinator {
checkpointInfo.f0,
request.isPeriodic,
checkpointInfo.f1,
-
request.getOnCompletionFuture(),
-
masterTriggerCompletionPromise),
+
request.getOnCompletionFuture()),
timer);
final CompletableFuture<?> coordinatorCheckpointsComplete =
@@ -618,12 +615,8 @@ public class CheckpointCoordinator {
},
timer);
- FutureUtils.forward(
- CompletableFuture.allOf(masterStatesComplete,
coordinatorCheckpointsComplete),
- masterTriggerCompletionPromise);
-
FutureUtils.assertNoException(
- masterTriggerCompletionPromise
+ CompletableFuture.allOf(masterStatesComplete,
coordinatorCheckpointsComplete)
.handleAsync(
(ignored, throwable) -> {
final PendingCheckpoint checkpoint =
@@ -785,8 +778,7 @@ public class CheckpointCoordinator {
CheckpointPlan checkpointPlan,
boolean isPeriodic,
long checkpointID,
- CompletableFuture<CompletedCheckpoint> onCompletionPromise,
- CompletableFuture<Void> masterTriggerCompletionPromise) {
+ CompletableFuture<CompletedCheckpoint> onCompletionPromise) {
synchronized (lock) {
try {
@@ -811,8 +803,7 @@ public class CheckpointCoordinator {
masterHooks.keySet(),
props,
onCompletionPromise,
- pendingCheckpointStats,
- masterTriggerCompletionPromise);
+ pendingCheckpointStats);
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 8ca6e22020e..b4bd8eacc5e 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,8 +112,6 @@ 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;
@@ -138,8 +136,7 @@ public class PendingCheckpoint implements Checkpoint {
Collection<String> masterStateIdentifiers,
CheckpointProperties props,
CompletableFuture<CompletedCheckpoint> onCompletionPromise,
- @Nullable PendingCheckpointStats pendingCheckpointStats,
- CompletableFuture<Void> masterTriggerCompletionPromise) {
+ @Nullable PendingCheckpointStats pendingCheckpointStats) {
checkArgument(
checkpointPlan.getTasksToWaitFor().size() > 0,
"Checkpoint needs at least one vertex that commits the
checkpoint");
@@ -169,7 +166,6 @@ public class PendingCheckpoint implements Checkpoint {
this.acknowledgedTasks = new
HashSet<>(checkpointPlan.getTasksToWaitFor().size());
this.onCompletionPromise = checkNotNull(onCompletionPromise);
this.pendingCheckpointStats = pendingCheckpointStats;
- this.masterTriggerCompletionPromise =
checkNotNull(masterTriggerCompletionPromise);
}
//
--------------------------------------------------------------------------------------------
@@ -548,7 +544,6 @@ 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 234765b298f..9d3a87ddd84 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,50 +3658,6 @@ 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 fc8d57f9f93..392bd0ef67c 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,8 +644,7 @@ public class PendingCheckpointTest {
masterStateIdentifiers,
props,
new CompletableFuture<>(),
- null,
- new CompletableFuture<>());
+ null);
pendingCheckpoint.setCheckpointTargetLocation(location);
return pendingCheckpoint;
}
