You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2022/03/04 09:20:10 UTC
[flink] branch release-1.14 updated (2b32f3c1 -> 39afe94)
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a change to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 2b32f3c1 [FLINK-26407][end-to-end-tests] Increase timeouts for MetricsAvailabilityITCase.
new 970932b [FLINK-26049][checkpoint] initialize CheckpointLocation after create PendingCheckpoint
new 27f1dfd [FLINK-26049][checkpoint] Moving checkpoint failure log and report failed checkpoint to CheckpointFailureManager
new 39afe94 [FLINK-26049][checkpoint] Adding CheckpointStatsTracker logic without pending checkpoint
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../runtime/checkpoint/CheckpointCoordinator.java | 125 +++++++++------------
.../checkpoint/CheckpointFailureManager.java | 51 +++++++--
.../runtime/checkpoint/CheckpointStatsCounts.java | 9 ++
.../runtime/checkpoint/CheckpointStatsTracker.java | 15 +++
.../runtime/checkpoint/PendingCheckpoint.java | 33 ++----
.../checkpoint/CheckpointCoordinatorTest.java | 63 ++++++++++-
.../CheckpointCoordinatorTestingUtils.java | 45 +++++---
.../checkpoint/CheckpointFailureManagerTest.java | 42 +++++++
.../checkpoint/CheckpointStatsCountsTest.java | 7 ++
.../checkpoint/CheckpointStatsTrackerTest.java | 9 ++
.../runtime/checkpoint/PendingCheckpointTest.java | 111 +++++-------------
11 files changed, 308 insertions(+), 202 deletions(-)
[flink] 03/03: [FLINK-26049][checkpoint] Adding CheckpointStatsTracker logic without pending checkpoint
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 39afe944840e37a4d0290d99e91849d20e60e1c3
Author: fanrui <19...@gmail.com>
AuthorDate: Mon Feb 28 18:12:34 2022 +0800
[FLINK-26049][checkpoint] Adding CheckpointStatsTracker logic without pending checkpoint
---
.../runtime/checkpoint/CheckpointCoordinator.java | 5 ++-
.../checkpoint/CheckpointFailureManager.java | 9 +++-
.../runtime/checkpoint/CheckpointStatsCounts.java | 9 ++++
.../runtime/checkpoint/CheckpointStatsTracker.java | 15 +++++++
.../checkpoint/CheckpointCoordinatorTest.java | 49 ++++++++++++++++++++++
.../checkpoint/CheckpointFailureManagerTest.java | 22 ++++++++--
.../checkpoint/CheckpointStatsCountsTest.java | 7 ++++
.../checkpoint/CheckpointStatsTrackerTest.java | 9 ++++
8 files changed, 118 insertions(+), 7 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 3d79824..e74ad24 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
@@ -911,7 +911,7 @@ public class CheckpointCoordinator {
}
} else {
failureManager.handleCheckpointException(
- checkpoint, checkpointProperties, cause, null, job, null);
+ checkpoint, checkpointProperties, cause, null, job, null, statsTracker);
}
} finally {
isTriggering = false;
@@ -1942,7 +1942,8 @@ public class CheckpointCoordinator {
exception,
executionAttemptID,
job,
- getStatsCallback(pendingCheckpoint));
+ getStatsCallback(pendingCheckpoint),
+ statsTracker);
} finally {
sendAbortedMessages(
pendingCheckpoint.getCheckpointPlan().getTasksToCommitTo(),
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointFailureManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointFailureManager.java
index 7b30034..f77d1a6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointFailureManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointFailureManager.java
@@ -88,6 +88,7 @@ public class CheckpointFailureManager {
* @param executionAttemptID the execution attempt id, as a safe guard.
* @param job the JobID.
* @param pendingCheckpointStats the pending checkpoint statistics.
+ * @param statsTracker the tracker for checkpoint statistics.
*/
public void handleCheckpointException(
@Nullable PendingCheckpoint pendingCheckpoint,
@@ -95,12 +96,13 @@ public class CheckpointFailureManager {
CheckpointException exception,
@Nullable ExecutionAttemptID executionAttemptID,
JobID job,
- @Nullable PendingCheckpointStats pendingCheckpointStats) {
+ @Nullable PendingCheckpointStats pendingCheckpointStats,
+ CheckpointStatsTracker statsTracker) {
long checkpointId =
pendingCheckpoint == null
? UNKNOWN_CHECKPOINT_ID
: pendingCheckpoint.getCheckpointID();
- updateStatsAfterCheckpointFailed(pendingCheckpointStats, exception);
+ updateStatsAfterCheckpointFailed(pendingCheckpointStats, statsTracker, exception);
LOG.warn(
"Failed to trigger checkpoint {} for job {}. ({} consecutive failed attempts so far)",
@@ -124,10 +126,13 @@ public class CheckpointFailureManager {
*/
private void updateStatsAfterCheckpointFailed(
@Nullable PendingCheckpointStats pendingCheckpointStats,
+ CheckpointStatsTracker statsTracker,
CheckpointException exception) {
if (pendingCheckpointStats != null) {
long failureTimestamp = System.currentTimeMillis();
pendingCheckpointStats.reportFailedCheckpoint(failureTimestamp, exception);
+ } else {
+ statsTracker.reportFailedCheckpointsWithoutInProgress();
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointStatsCounts.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointStatsCounts.java
index 8d06a1b..7265bb3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointStatsCounts.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointStatsCounts.java
@@ -163,6 +163,15 @@ public class CheckpointStatsCounts implements Serializable {
}
/**
+ * Increments the number of failed checkpoints without in progress checkpoint. For example, it
+ * should be callback when triggering checkpoint failure before creating PendingCheckpoint.
+ */
+ void incrementFailedCheckpointsWithoutInProgress() {
+ numFailedCheckpoints++;
+ numTotalCheckpoints++;
+ }
+
+ /**
* Creates a snapshot of the current state.
*
* @return Snapshot of the current state.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointStatsTracker.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointStatsTracker.java
index 8c4ac97..afcd9e0 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointStatsTracker.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointStatsTracker.java
@@ -255,6 +255,21 @@ public class CheckpointStatsTracker {
}
}
+ /**
+ * Callback when a checkpoint failure without in progress checkpoint. For example, it should be
+ * callback when triggering checkpoint failure before creating PendingCheckpoint.
+ */
+ public void reportFailedCheckpointsWithoutInProgress() {
+ statsReadWriteLock.lock();
+ try {
+ counts.incrementFailedCheckpointsWithoutInProgress();
+
+ dirty = true;
+ } finally {
+ statsReadWriteLock.unlock();
+ }
+ }
+
public PendingCheckpointStats getPendingCheckpointStats(long checkpointId) {
statsReadWriteLock.lock();
try {
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 f6f50b9..f8a0005 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
@@ -120,6 +120,7 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.ArgumentMatchers.anyLong;
@@ -3068,6 +3069,46 @@ public class CheckpointCoordinatorTest extends TestLogger {
testTriggerCheckpoint(checkpointCoordinator, PERIODIC_SCHEDULER_SHUTDOWN);
}
+ /** Tests that do not trigger checkpoint when CheckpointIDCounter IOException occurred. */
+ @Test
+ public void testTriggerCheckpointWithCounterIOException() throws Exception {
+ // given: Checkpoint coordinator which fails on getCheckpointId.
+ IOExceptionCheckpointIDCounter testingCounter = new IOExceptionCheckpointIDCounter();
+ TestFailJobCallback failureCallback = new TestFailJobCallback();
+
+ CheckpointStatsTracker statsTracker =
+ new CheckpointStatsTracker(
+ Integer.MAX_VALUE,
+ mock(CheckpointCoordinatorConfiguration.class),
+ new UnregisteredMetricsGroup());
+
+ CheckpointCoordinator checkpointCoordinator =
+ new CheckpointCoordinatorBuilder()
+ .setCheckpointIDCounter(testingCounter)
+ .setFailureManager(new CheckpointFailureManager(0, failureCallback))
+ .setTimer(manuallyTriggeredScheduledExecutor)
+ .setCheckpointStatsTracker(statsTracker)
+ .build();
+ testingCounter.setOwner(checkpointCoordinator);
+
+ // when: The checkpoint is triggered.
+ testTriggerCheckpoint(checkpointCoordinator, IO_EXCEPTION);
+
+ // then: Failure manager should fail the job.
+ assertEquals(1, failureCallback.getInvokeCounter());
+
+ // then: The NumberOfFailedCheckpoints and TotalNumberOfCheckpoints should be 1.
+ CheckpointStatsCounts counts = statsTracker.createSnapshot().getCounts();
+ assertEquals(0, counts.getNumberOfRestoredCheckpoints());
+ assertEquals(1, counts.getTotalNumberOfCheckpoints());
+ assertEquals(0, counts.getNumberOfInProgressCheckpoints());
+ assertEquals(0, counts.getNumberOfCompletedCheckpoints());
+ assertEquals(1, counts.getNumberOfFailedCheckpoints());
+
+ // then: The PendingCheckpoint shouldn't be created.
+ assertNull(statsTracker.getPendingCheckpointStats(1));
+ }
+
private void testTriggerCheckpoint(
CheckpointCoordinator checkpointCoordinator,
CheckpointFailureReason expectedFailureReason)
@@ -3776,6 +3817,14 @@ public class CheckpointCoordinatorTest extends TestLogger {
}
}
+ private static class IOExceptionCheckpointIDCounter extends CheckpointIDCounterWithOwner {
+ @Override
+ public long getAndIncrement() throws Exception {
+ checkNotNull(owner);
+ throw new IOException("disk is error!");
+ }
+ }
+
private static class IOExceptionCheckpointStorage extends JobManagerCheckpointStorage {
@Override
public CheckpointStorageAccess createCheckpointStorage(JobID jobId) throws IOException {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointFailureManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointFailureManagerTest.java
index e027ed9..243e3e1 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointFailureManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointFailureManagerTest.java
@@ -167,7 +167,8 @@ public class CheckpointFailureManagerTest extends TestLogger {
new CheckpointFailureManager(2, new TestFailJobCallback());
PendingCheckpoint pendingCheckpoint = mock(PendingCheckpoint.class);
- PendingCheckpointStats callback = mock(PendingCheckpointStats.class);
+ PendingCheckpointStats pendingCheckpointCallback = mock(PendingCheckpointStats.class);
+ CheckpointStatsTracker statsTracker = mock(CheckpointStatsTracker.class);
failureManager.handleCheckpointException(
pendingCheckpoint,
@@ -175,8 +176,23 @@ public class CheckpointFailureManagerTest extends TestLogger {
new CheckpointException(CheckpointFailureReason.CHECKPOINT_SUBSUMED, null),
null,
new JobID(),
- callback);
- verify(callback, times(1)).reportFailedCheckpoint(anyLong(), any(Exception.class));
+ pendingCheckpointCallback,
+ statsTracker);
+ verify(pendingCheckpointCallback, times(1))
+ .reportFailedCheckpoint(anyLong(), any(Exception.class));
+ verify(statsTracker, times(0)).reportFailedCheckpointsWithoutInProgress();
+
+ failureManager.handleCheckpointException(
+ null,
+ checkpointProperties,
+ new CheckpointException(CheckpointFailureReason.CHECKPOINT_SUBSUMED, null),
+ null,
+ new JobID(),
+ null,
+ statsTracker);
+ verify(pendingCheckpointCallback, times(1))
+ .reportFailedCheckpoint(anyLong(), any(Exception.class));
+ verify(statsTracker, times(1)).reportFailedCheckpointsWithoutInProgress();
}
/** A failure handler callback for testing. */
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointStatsCountsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointStatsCountsTest.java
index c3ffd6d..5d60810 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointStatsCountsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointStatsCountsTest.java
@@ -72,6 +72,13 @@ public class CheckpointStatsCountsTest {
assertEquals(0, counts.getNumberOfInProgressCheckpoints());
assertEquals(1, counts.getNumberOfCompletedCheckpoints());
assertEquals(1, counts.getNumberOfFailedCheckpoints());
+
+ counts.incrementFailedCheckpointsWithoutInProgress();
+ assertEquals(1, counts.getNumberOfRestoredCheckpoints());
+ assertEquals(3, counts.getTotalNumberOfCheckpoints());
+ assertEquals(0, counts.getNumberOfInProgressCheckpoints());
+ assertEquals(1, counts.getNumberOfCompletedCheckpoints());
+ assertEquals(2, counts.getNumberOfFailedCheckpoints());
}
/**
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointStatsTrackerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointStatsTrackerTest.java
index 0e97958..433d4bc 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointStatsTrackerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointStatsTrackerTest.java
@@ -205,6 +205,15 @@ public class CheckpointStatsTrackerTest {
assertEquals(2, counts.getNumberOfCompletedCheckpoints());
assertEquals(1, counts.getNumberOfFailedCheckpoints());
+ tracker.reportFailedCheckpointsWithoutInProgress();
+
+ CheckpointStatsSnapshot snapshot1 = tracker.createSnapshot();
+ counts = snapshot1.getCounts();
+ assertEquals(5, counts.getTotalNumberOfCheckpoints());
+ assertEquals(1, counts.getNumberOfInProgressCheckpoints());
+ assertEquals(2, counts.getNumberOfCompletedCheckpoints());
+ assertEquals(2, counts.getNumberOfFailedCheckpoints());
+
// Summary stats
CompletedCheckpointStatsSummarySnapshot summary = snapshot.getSummaryStats();
assertEquals(2, summary.getStateSizeStats().getCount());
[flink] 01/03: [FLINK-26049][checkpoint] initialize CheckpointLocation after create PendingCheckpoint
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 970932bfe207617be45568ea6ad5c6357a412fc7
Author: fanrui <19...@gmail.com>
AuthorDate: Mon Feb 28 11:35:09 2022 +0800
[FLINK-26049][checkpoint] initialize CheckpointLocation after create PendingCheckpoint
---
.../runtime/checkpoint/CheckpointCoordinator.java | 95 +++++++++++-----------
.../runtime/checkpoint/PendingCheckpoint.java | 16 ++--
.../checkpoint/CheckpointCoordinatorTest.java | 14 +++-
.../CheckpointCoordinatorTestingUtils.java | 45 ++++++----
.../runtime/checkpoint/PendingCheckpointTest.java | 22 ++---
5 files changed, 113 insertions(+), 79 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 a2bff9d..384e047 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
@@ -541,13 +541,12 @@ public class CheckpointCoordinator {
.thenApplyAsync(
plan -> {
try {
- CheckpointIdAndStorageLocation
- checkpointIdAndStorageLocation =
- initializeCheckpoint(
- request.props,
- request.externalSavepointLocation);
- return new Tuple2<>(
- plan, checkpointIdAndStorageLocation);
+ // this must happen outside the coordinator-wide lock,
+ // because it communicates with external services
+ // (in HA mode) and may block for a while.
+ long checkpointID =
+ checkpointIdCounter.getAndIncrement();
+ return new Tuple2<>(plan, checkpointID);
} catch (Throwable e) {
throw new CompletionException(e);
}
@@ -560,20 +559,41 @@ public class CheckpointCoordinator {
request.props,
checkpointInfo.f0,
request.isPeriodic,
- checkpointInfo.f1.checkpointId,
- checkpointInfo.f1.checkpointStorageLocation,
+ checkpointInfo.f1,
request.getOnCompletionFuture()),
timer);
final CompletableFuture<?> coordinatorCheckpointsComplete =
- pendingCheckpointCompletableFuture.thenComposeAsync(
- (pendingCheckpoint) ->
- OperatorCoordinatorCheckpoints
- .triggerAndAcknowledgeAllCoordinatorCheckpointsWithCompletion(
- coordinatorsToCheckpoint,
- pendingCheckpoint,
- timer),
- timer);
+ pendingCheckpointCompletableFuture
+ .thenApplyAsync(
+ pendingCheckpoint -> {
+ try {
+ CheckpointStorageLocation checkpointStorageLocation =
+ initializeCheckpointLocation(
+ pendingCheckpoint.getCheckpointID(),
+ request.props,
+ request.externalSavepointLocation);
+ return Tuple2.of(
+ pendingCheckpoint, checkpointStorageLocation);
+ } catch (Throwable e) {
+ throw new CompletionException(e);
+ }
+ },
+ executor)
+ .thenComposeAsync(
+ (checkpointInfo) -> {
+ PendingCheckpoint pendingCheckpoint = checkpointInfo.f0;
+ synchronized (lock) {
+ pendingCheckpoint.setCheckpointTargetLocation(
+ checkpointInfo.f1);
+ }
+ return OperatorCoordinatorCheckpoints
+ .triggerAndAcknowledgeAllCoordinatorCheckpointsWithCompletion(
+ coordinatorsToCheckpoint,
+ pendingCheckpoint,
+ timer);
+ },
+ timer);
// We have to take the snapshot of the master hooks after the coordinator checkpoints
// has completed.
@@ -714,29 +734,24 @@ public class CheckpointCoordinator {
}
/**
- * Initialize the checkpoint trigger asynchronously. It will expected to be executed in io
+ * Initialize the checkpoint location asynchronously. It will expected to be executed in io
* thread due to it might be time-consuming.
*
+ * @param checkpointID checkpoint id
* @param props checkpoint properties
* @param externalSavepointLocation the external savepoint location, it might be null
- * @return the initialized result, checkpoint id and checkpoint location
+ * @return the checkpoint location
*/
- private CheckpointIdAndStorageLocation initializeCheckpoint(
- CheckpointProperties props, @Nullable String externalSavepointLocation)
+ private CheckpointStorageLocation initializeCheckpointLocation(
+ long checkpointID,
+ CheckpointProperties props,
+ @Nullable String externalSavepointLocation)
throws Exception {
- // this must happen outside the coordinator-wide lock, because it
- // communicates
- // with external services (in HA mode) and may block for a while.
- long checkpointID = checkpointIdCounter.getAndIncrement();
-
- CheckpointStorageLocation checkpointStorageLocation =
- props.isSavepoint()
- ? checkpointStorageView.initializeLocationForSavepoint(
- checkpointID, externalSavepointLocation)
- : checkpointStorageView.initializeLocationForCheckpoint(checkpointID);
-
- return new CheckpointIdAndStorageLocation(checkpointID, checkpointStorageLocation);
+ return props.isSavepoint()
+ ? checkpointStorageView.initializeLocationForSavepoint(
+ checkpointID, externalSavepointLocation)
+ : checkpointStorageView.initializeLocationForCheckpoint(checkpointID);
}
private PendingCheckpoint createPendingCheckpoint(
@@ -745,7 +760,6 @@ public class CheckpointCoordinator {
CheckpointPlan checkpointPlan,
boolean isPeriodic,
long checkpointID,
- CheckpointStorageLocation checkpointStorageLocation,
CompletableFuture<CompletedCheckpoint> onCompletionPromise) {
synchronized (lock) {
@@ -767,7 +781,6 @@ public class CheckpointCoordinator {
OperatorInfo.getIds(coordinatorsToCheckpoint),
masterHooks.keySet(),
props,
- checkpointStorageLocation,
onCompletionPromise);
trackPendingCheckpointStats(checkpoint);
@@ -2025,18 +2038,6 @@ public class CheckpointCoordinator {
}
}
- private static class CheckpointIdAndStorageLocation {
- private final long checkpointId;
- private final CheckpointStorageLocation checkpointStorageLocation;
-
- CheckpointIdAndStorageLocation(
- long checkpointId, CheckpointStorageLocation checkpointStorageLocation) {
-
- this.checkpointId = checkpointId;
- this.checkpointStorageLocation = checkNotNull(checkpointStorageLocation);
- }
- }
-
static class CheckpointTriggerRequest {
final long timestamp;
final CheckpointProperties props;
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 f64af16..47e7a32 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
@@ -105,12 +105,12 @@ public class PendingCheckpoint implements Checkpoint {
/** The checkpoint properties. */
private final CheckpointProperties props;
- /** Target storage location to persist the checkpoint metadata to. */
- private final CheckpointStorageLocation targetLocation;
-
/** The promise to fulfill once the checkpoint has been completed. */
private final CompletableFuture<CompletedCheckpoint> onCompletionPromise;
+ /** Target storage location to persist the checkpoint metadata to. */
+ @Nullable private CheckpointStorageLocation targetLocation;
+
private int numAcknowledgedTasks;
private boolean disposed;
@@ -131,7 +131,6 @@ public class PendingCheckpoint implements Checkpoint {
Collection<OperatorID> operatorCoordinatorsToConfirm,
Collection<String> masterStateIdentifiers,
CheckpointProperties props,
- CheckpointStorageLocation targetLocation,
CompletableFuture<CompletedCheckpoint> onCompletionPromise) {
checkArgument(
@@ -149,7 +148,6 @@ public class PendingCheckpoint implements Checkpoint {
}
this.props = checkNotNull(props);
- this.targetLocation = checkNotNull(targetLocation);
this.operatorStates = new HashMap<>();
this.masterStates = new ArrayList<>(masterStateIdentifiers.size());
@@ -186,6 +184,10 @@ public class PendingCheckpoint implements Checkpoint {
return checkpointId;
}
+ public void setCheckpointTargetLocation(CheckpointStorageLocation targetLocation) {
+ this.targetLocation = targetLocation;
+ }
+
public CheckpointStorageLocation getCheckpointStorageLocation() {
return targetLocation;
}
@@ -599,7 +601,9 @@ public class PendingCheckpoint implements Checkpoint {
// unregistered shared states are still considered private at this point.
try {
StateUtil.bestEffortDiscardAllStateObjects(operatorStates.values());
- targetLocation.disposeOnFailure();
+ if (targetLocation != null) {
+ targetLocation.disposeOnFailure();
+ }
} catch (Throwable t) {
LOG.warn(
"Could not properly dispose the private states in the pending checkpoint {} of job {}.",
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 8c716db..f6f50b9 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
@@ -671,20 +671,30 @@ public class CheckpointCoordinatorTest extends TestLogger {
/** Tests that do not trigger checkpoint when IOException occurred. */
@Test
- public void testTriggerCheckpointAfterIOException() throws Exception {
- // given: Checkpoint coordinator which fails on initializeLocationForCheckpoint.
+ public void testTriggerCheckpointAfterCheckpointStorageIOException() throws Exception {
+ // given: Checkpoint coordinator which fails on initializeCheckpointLocation.
TestFailJobCallback failureCallback = new TestFailJobCallback();
+ CheckpointStatsTracker statsTracker =
+ new CheckpointStatsTracker(
+ Integer.MAX_VALUE,
+ mock(CheckpointCoordinatorConfiguration.class),
+ new UnregisteredMetricsGroup());
CheckpointCoordinator checkpointCoordinator =
new CheckpointCoordinatorBuilder()
.setFailureManager(new CheckpointFailureManager(0, failureCallback))
.setCheckpointStorage(new IOExceptionCheckpointStorage())
.setTimer(manuallyTriggeredScheduledExecutor)
+ .setCheckpointStatsTracker(statsTracker)
.build();
+
// when: The checkpoint is triggered.
testTriggerCheckpoint(checkpointCoordinator, IO_EXCEPTION);
// then: Failure manager should fail the job.
assertEquals(1, failureCallback.getInvokeCounter());
+
+ // then: Should created PendingCheckpoint
+ assertNotNull(statsTracker.getPendingCheckpointStats(1));
}
@Test
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTestingUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTestingUtils.java
index a178a4c..4ba15fe 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTestingUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTestingUtils.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.checkpoint;
import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.eventtime.WatermarkStrategyTest;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.fs.FSDataInputStream;
import org.apache.flink.core.fs.Path;
@@ -724,6 +725,12 @@ public class CheckpointCoordinatorTestingUtils {
private boolean allowCheckpointsAfterTasksFinished;
+ private CheckpointStatsTracker checkpointStatsTracker =
+ new CheckpointStatsTracker(
+ 1,
+ CheckpointCoordinatorConfiguration.builder().build(),
+ new WatermarkStrategyTest.DummyMetricGroup());
+
public CheckpointCoordinatorBuilder setCheckpointCoordinatorConfiguration(
CheckpointCoordinatorConfiguration checkpointCoordinatorConfiguration) {
this.checkpointCoordinatorConfiguration = checkpointCoordinatorConfiguration;
@@ -798,6 +805,12 @@ public class CheckpointCoordinatorTestingUtils {
return this;
}
+ public CheckpointCoordinatorBuilder setCheckpointStatsTracker(
+ CheckpointStatsTracker checkpointStatsTracker) {
+ this.checkpointStatsTracker = checkpointStatsTracker;
+ return this;
+ }
+
public CheckpointCoordinator build() throws Exception {
if (executionGraph == null) {
executionGraph =
@@ -813,20 +826,24 @@ public class CheckpointCoordinatorTestingUtils {
executionGraph.getVerticesTopologically(),
allowCheckpointsAfterTasksFinished);
- return new CheckpointCoordinator(
- executionGraph.getJobID(),
- checkpointCoordinatorConfiguration,
- coordinatorsToCheckpoint,
- checkpointIDCounter,
- completedCheckpointStore,
- checkpointStorage,
- ioExecutor,
- checkpointsCleaner,
- timer,
- sharedStateRegistryFactory,
- failureManager,
- checkpointPlanCalculator,
- new ExecutionAttemptMappingProvider(executionGraph.getAllExecutionVertices()));
+ CheckpointCoordinator checkpointCoordinator =
+ new CheckpointCoordinator(
+ executionGraph.getJobID(),
+ checkpointCoordinatorConfiguration,
+ coordinatorsToCheckpoint,
+ checkpointIDCounter,
+ completedCheckpointStore,
+ checkpointStorage,
+ ioExecutor,
+ checkpointsCleaner,
+ timer,
+ sharedStateRegistryFactory,
+ failureManager,
+ checkpointPlanCalculator,
+ new ExecutionAttemptMappingProvider(
+ executionGraph.getAllExecutionVertices()));
+ checkpointCoordinator.setCheckpointStatsTracker(checkpointStatsTracker);
+ return checkpointCoordinator;
}
}
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 7c2091f..3174e2b 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
@@ -685,16 +685,18 @@ public class PendingCheckpointTest {
1024,
4096);
- return new PendingCheckpoint(
- new JobID(),
- 0,
- 1,
- checkpointPlan,
- operatorCoordinators,
- masterStateIdentifiers,
- props,
- location,
- new CompletableFuture<>());
+ PendingCheckpoint pendingCheckpoint =
+ new PendingCheckpoint(
+ new JobID(),
+ 0,
+ 1,
+ checkpointPlan,
+ operatorCoordinators,
+ masterStateIdentifiers,
+ props,
+ new CompletableFuture<>());
+ pendingCheckpoint.setCheckpointTargetLocation(location);
+ return pendingCheckpoint;
}
@SuppressWarnings("unchecked")
[flink] 02/03: [FLINK-26049][checkpoint] Moving checkpoint failure log and report failed checkpoint to CheckpointFailureManager
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 27f1dfdf72da19ea8cb20a353aed40323fc83c41
Author: fanrui <19...@gmail.com>
AuthorDate: Mon Feb 28 17:45:12 2022 +0800
[FLINK-26049][checkpoint] Moving checkpoint failure log and report failed checkpoint to CheckpointFailureManager
---
.../runtime/checkpoint/CheckpointCoordinator.java | 29 ++-----
.../checkpoint/CheckpointFailureManager.java | 46 +++++++++--
.../runtime/checkpoint/PendingCheckpoint.java | 17 +----
.../checkpoint/CheckpointFailureManagerTest.java | 26 +++++++
.../runtime/checkpoint/PendingCheckpointTest.java | 89 ++++------------------
5 files changed, 84 insertions(+), 123 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 384e047..3d79824 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
@@ -80,7 +80,6 @@ import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Predicate;
import java.util.stream.Stream;
@@ -177,9 +176,6 @@ public class CheckpointCoordinator {
/** Actor that receives status updates from the execution graph this coordinator works for. */
private JobStatusListener jobStatusListener;
- /** The number of consecutive failed trigger attempts. */
- private final AtomicInteger numUnsuccessfulCheckpointsTriggers = new AtomicInteger(0);
-
/** A handle to the current periodic trigger, to cancel it when necessary. */
private ScheduledFuture<?> currentPeriodicTrigger;
@@ -862,7 +858,6 @@ public class CheckpointCoordinator {
/** Trigger request is successful. NOTE, it must be invoked if trigger request is successful. */
private void onTriggerSuccess() {
isTriggering = false;
- numUnsuccessfulCheckpointsTriggers.set(0);
executeQueuedRequest();
}
@@ -911,25 +906,12 @@ public class CheckpointCoordinator {
CheckpointFailureReason.TRIGGER_CHECKPOINT_FAILURE, throwable);
if (checkpoint != null && !checkpoint.isDisposed()) {
- int numUnsuccessful = numUnsuccessfulCheckpointsTriggers.incrementAndGet();
- LOG.warn(
- "Failed to trigger checkpoint {} for job {}. ({} consecutive failed attempts so far)",
- checkpoint.getCheckpointId(),
- job,
- numUnsuccessful,
- throwable);
-
synchronized (lock) {
abortPendingCheckpoint(checkpoint, cause);
}
} else {
- LOG.info(
- "Failed to trigger checkpoint for job {} because {}",
- job,
- throwable.getMessage());
-
failureManager.handleCheckpointException(
- checkpoint, checkpointProperties, cause, null);
+ checkpoint, checkpointProperties, cause, null, job, null);
}
} finally {
isTriggering = false;
@@ -1778,8 +1760,6 @@ public class CheckpointCoordinator {
final CheckpointException reason =
new CheckpointException(CheckpointFailureReason.CHECKPOINT_COORDINATOR_SUSPEND);
abortPendingAndQueuedCheckpoints(reason);
-
- numUnsuccessfulCheckpointsTriggers.set(0);
}
}
@@ -1954,14 +1934,15 @@ public class CheckpointCoordinator {
exception.getCause(),
checkpointsCleaner,
this::scheduleTriggerRequest,
- executor,
- getStatsCallback(pendingCheckpoint));
+ executor);
failureManager.handleCheckpointException(
pendingCheckpoint,
pendingCheckpoint.getProps(),
exception,
- executionAttemptID);
+ executionAttemptID,
+ job,
+ getStatsCallback(pendingCheckpoint));
} finally {
sendAbortedMessages(
pendingCheckpoint.getCheckpointPlan().getTasksToCommitTo(),
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointFailureManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointFailureManager.java
index 8ccccc6..7b30034 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointFailureManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointFailureManager.java
@@ -17,10 +17,14 @@
package org.apache.flink.runtime.checkpoint;
+import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkRuntimeException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import javax.annotation.Nullable;
import java.util.Set;
@@ -34,6 +38,8 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
/** The checkpoint failure manager which centralized manage checkpoint failure processing logic. */
public class CheckpointFailureManager {
+ private static final Logger LOG = LoggerFactory.getLogger(CheckpointFailureManager.class);
+
public static final int UNLIMITED_TOLERABLE_FAILURE_NUMBER = Integer.MAX_VALUE;
public static final String EXCEEDED_CHECKPOINT_TOLERABLE_FAILURE_MESSAGE =
"Exceeded checkpoint tolerable failure threshold.";
@@ -80,25 +86,51 @@ public class CheckpointFailureManager {
* strategy can be used.
* @param exception the checkpoint exception.
* @param executionAttemptID the execution attempt id, as a safe guard.
+ * @param job the JobID.
+ * @param pendingCheckpointStats the pending checkpoint statistics.
*/
public void handleCheckpointException(
@Nullable PendingCheckpoint pendingCheckpoint,
CheckpointProperties checkpointProperties,
CheckpointException exception,
- @Nullable ExecutionAttemptID executionAttemptID) {
+ @Nullable ExecutionAttemptID executionAttemptID,
+ JobID job,
+ @Nullable PendingCheckpointStats pendingCheckpointStats) {
+ long checkpointId =
+ pendingCheckpoint == null
+ ? UNKNOWN_CHECKPOINT_ID
+ : pendingCheckpoint.getCheckpointID();
+ updateStatsAfterCheckpointFailed(pendingCheckpointStats, exception);
+
+ LOG.warn(
+ "Failed to trigger checkpoint {} for job {}. ({} consecutive failed attempts so far)",
+ checkpointId == UNKNOWN_CHECKPOINT_ID ? "UNKNOWN_CHECKPOINT_ID" : checkpointId,
+ job,
+ continuousFailureCounter.get(),
+ exception);
if (isJobManagerFailure(exception, executionAttemptID)) {
- handleJobLevelCheckpointException(
- checkpointProperties,
- exception,
- pendingCheckpoint == null
- ? UNKNOWN_CHECKPOINT_ID
- : pendingCheckpoint.getCheckpointID());
+ handleJobLevelCheckpointException(checkpointProperties, exception, checkpointId);
} else {
handleTaskLevelCheckpointException(
checkNotNull(pendingCheckpoint), exception, checkNotNull(executionAttemptID));
}
}
+ /**
+ * Updating checkpoint statistics after checkpoint failed.
+ *
+ * @param pendingCheckpointStats the pending checkpoint statistics.
+ * @param exception the checkpoint exception.
+ */
+ private void updateStatsAfterCheckpointFailed(
+ @Nullable PendingCheckpointStats pendingCheckpointStats,
+ CheckpointException exception) {
+ if (pendingCheckpointStats != null) {
+ long failureTimestamp = System.currentTimeMillis();
+ pendingCheckpointStats.reportFailedCheckpoint(failureTimestamp, exception);
+ }
+ }
+
private boolean isJobManagerFailure(
CheckpointException exception, @Nullable ExecutionAttemptID executionAttemptID) {
// TODO: Try to get rid of checking nullability of executionAttemptID because false value of
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 47e7a32..94ae442 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
@@ -542,12 +542,10 @@ public class PendingCheckpoint implements Checkpoint {
@Nullable Throwable cause,
CheckpointsCleaner checkpointsCleaner,
Runnable postCleanup,
- Executor executor,
- PendingCheckpointStats statsCallback) {
+ Executor executor) {
try {
failureCause = new CheckpointException(reason, cause);
onCompletionPromise.completeExceptionally(failureCause);
- reportFailedCheckpoint(failureCause, statsCallback);
assertAbortSubsumedForced(reason);
} finally {
dispose(true, checkpointsCleaner, postCleanup, executor);
@@ -627,19 +625,6 @@ public class PendingCheckpoint implements Checkpoint {
}
}
- /**
- * Reports a failed checkpoint with the given optional cause.
- *
- * @param cause The failure cause or <code>null</code>.
- */
- private void reportFailedCheckpoint(Exception cause, PendingCheckpointStats statsCallback) {
- // to prevent null-pointers from concurrent modification, copy reference onto stack
- if (statsCallback != null) {
- long failureTimestamp = System.currentTimeMillis();
- statsCallback.reportFailedCheckpoint(failureTimestamp, cause);
- }
- }
-
// ------------------------------------------------------------------------
// Utilities
// ------------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointFailureManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointFailureManagerTest.java
index 8ed9b03..e027ed9 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointFailureManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointFailureManagerTest.java
@@ -17,6 +17,7 @@
package org.apache.flink.runtime.checkpoint;
+import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.JobException;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.util.TestLogger;
@@ -29,6 +30,11 @@ import static org.apache.flink.runtime.checkpoint.CheckpointFailureReason.CHECKP
import static org.apache.flink.runtime.checkpoint.CheckpointProperties.forCheckpoint;
import static org.apache.flink.runtime.checkpoint.CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION;
import static org.junit.Assert.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
/** Tests for the checkpoint failure manager. */
public class CheckpointFailureManagerTest extends TestLogger {
@@ -153,6 +159,26 @@ public class CheckpointFailureManagerTest extends TestLogger {
assertEquals(0, callback.getInvokeCounter());
}
+ /** Tests that the stats callbacks happen if the callback is registered. */
+ @Test
+ public void testPendingCheckpointStatsCallbacks() throws Exception {
+ CheckpointProperties checkpointProperties = forCheckpoint(NEVER_RETAIN_AFTER_TERMINATION);
+ CheckpointFailureManager failureManager =
+ new CheckpointFailureManager(2, new TestFailJobCallback());
+
+ PendingCheckpoint pendingCheckpoint = mock(PendingCheckpoint.class);
+ PendingCheckpointStats callback = mock(PendingCheckpointStats.class);
+
+ failureManager.handleCheckpointException(
+ pendingCheckpoint,
+ checkpointProperties,
+ new CheckpointException(CheckpointFailureReason.CHECKPOINT_SUBSUMED, null),
+ null,
+ new JobID(),
+ callback);
+ verify(callback, times(1)).reportFailedCheckpoint(anyLong(), any(Exception.class));
+ }
+
/** A failure handler callback for testing. */
private static class TestFailJobCallback implements CheckpointFailureManager.FailJobCallback {
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 3174e2b..e7d3872 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
@@ -73,7 +73,6 @@ import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.nullable;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.mock;
@@ -271,70 +270,20 @@ public class PendingCheckpointTest {
/** Tests that the stats callbacks happen if the callback is registered. */
@Test
public void testPendingCheckpointStatsCallbacks() throws Exception {
- {
- // Complete successfully
- PendingCheckpointStats callback = mock(PendingCheckpointStats.class);
- PendingCheckpoint pending =
- createPendingCheckpoint(
- CheckpointProperties.forCheckpoint(
- CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION));
-
- pending.acknowledgeTask(ATTEMPT_ID, null, new CheckpointMetrics(), callback);
- verify(callback, times(1))
- .reportSubtaskStats(nullable(JobVertexID.class), any(SubtaskStateStats.class));
-
- pending.finalizeCheckpoint(
- new CheckpointsCleaner(), () -> {}, Executors.directExecutor(), callback);
- verify(callback, times(1)).reportCompletedCheckpoint(any(String.class));
- }
-
- {
- // Fail subsumed
- PendingCheckpointStats callback = mock(PendingCheckpointStats.class);
- PendingCheckpoint pending =
- createPendingCheckpoint(
- CheckpointProperties.forCheckpoint(
- CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION));
-
- abort(pending, CheckpointFailureReason.CHECKPOINT_SUBSUMED, callback);
- verify(callback, times(1)).reportFailedCheckpoint(anyLong(), any(Exception.class));
- }
-
- {
- // Fail subsumed
- PendingCheckpointStats callback = mock(PendingCheckpointStats.class);
- PendingCheckpoint pending =
- createPendingCheckpoint(
- CheckpointProperties.forCheckpoint(
- CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION));
-
- abort(pending, CheckpointFailureReason.CHECKPOINT_DECLINED, callback);
- verify(callback, times(1)).reportFailedCheckpoint(anyLong(), any(Exception.class));
- }
-
- {
- // Fail subsumed
- PendingCheckpointStats callback = mock(PendingCheckpointStats.class);
- PendingCheckpoint pending =
- createPendingCheckpoint(
- CheckpointProperties.forCheckpoint(
- CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION));
-
- abort(pending, CheckpointFailureReason.CHECKPOINT_SUBSUMED, callback);
- verify(callback, times(1)).reportFailedCheckpoint(anyLong(), any(Exception.class));
- }
+ // Complete successfully
+ PendingCheckpointStats callback = mock(PendingCheckpointStats.class);
+ PendingCheckpoint pending =
+ createPendingCheckpoint(
+ CheckpointProperties.forCheckpoint(
+ CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION));
- {
- // Fail subsumed
- PendingCheckpointStats callback = mock(PendingCheckpointStats.class);
- PendingCheckpoint pending =
- createPendingCheckpoint(
- CheckpointProperties.forCheckpoint(
- CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION));
+ pending.acknowledgeTask(ATTEMPT_ID, null, new CheckpointMetrics(), callback);
+ verify(callback, times(1))
+ .reportSubtaskStats(nullable(JobVertexID.class), any(SubtaskStateStats.class));
- abort(pending, CheckpointFailureReason.CHECKPOINT_EXPIRED, callback);
- verify(callback, times(1)).reportFailedCheckpoint(anyLong(), any(Exception.class));
- }
+ pending.finalizeCheckpoint(
+ new CheckpointsCleaner(), () -> {}, Executors.directExecutor(), callback);
+ verify(callback, times(1)).reportCompletedCheckpoint(any(String.class));
}
/**
@@ -711,20 +660,8 @@ public class PendingCheckpointTest {
}
private void abort(PendingCheckpoint checkpoint, CheckpointFailureReason reason) {
- abort(checkpoint, reason, null);
- }
-
- private void abort(
- PendingCheckpoint checkpoint,
- CheckpointFailureReason reason,
- PendingCheckpointStats statsCallback) {
checkpoint.abort(
- reason,
- null,
- new CheckpointsCleaner(),
- () -> {},
- Executors.directExecutor(),
- statsCallback);
+ reason, null, new CheckpointsCleaner(), () -> {}, Executors.directExecutor());
}
private static final class QueueExecutor implements Executor {