You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ta...@apache.org on 2021/01/29 01:55:36 UTC
[flink] branch release-1.12 updated (d4d51eb -> 4d6b952)
This is an automated email from the ASF dual-hosted git repository.
tangyun pushed a change to branch release-1.12
in repository https://gitbox.apache.org/repos/asf/flink.git.
from d4d51eb [FLINK-21163][python] Fix the issue that Python dependencies specified via CLI override the dependencies specified in configuration
new 333d626 [FLINK-16443][checkpointing] Make sure that CheckpointException are also serialized in DeclineCheckpoint.
new d1f8f41 [FLINK-20675][checkpointing] Only decline async checkpoint failure when task is still running
new 4d6b952 [FLINK-20675][checkpointing] Ensure asynchronous checkpoint failure could fail the job by default
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 | 15 +-
.../checkpoint/CheckpointFailureManager.java | 7 +-
.../messages/checkpoint/DeclineCheckpoint.java | 37 ++--
.../checkpoint/SerializedCheckpointException.java | 53 +++++
.../taskexecutor/rpc/RpcCheckpointResponder.java | 12 +-
.../checkpoint/CheckpointCoordinatorTest.java | 105 ++++++++--
.../CheckpointCoordinatorTestingUtils.java | 6 +
.../checkpoint/CheckpointFailureManagerTest.java | 3 +-
.../flink/runtime/jobmaster/JobMasterTest.java | 20 +-
.../runtime/tasks/AsyncCheckpointRunnable.java | 40 ++--
.../flink/streaming/runtime/tasks/StreamTask.java | 2 +-
.../tasks/SubtaskCheckpointCoordinator.java | 2 +-
.../tasks/SubtaskCheckpointCoordinatorImpl.java | 23 +--
.../runtime/tasks/AsyncCheckpointRunnableTest.java | 27 ++-
.../runtime/tasks/LocalStateForwardingTest.java | 3 +-
.../tasks/StreamTaskExecutionDecorationTest.java | 7 +-
.../tasks/SubtaskCheckpointCoordinatorTest.java | 12 +-
.../tasks/TestSubtaskCheckpointCoordinator.java | 2 +-
.../CheckpointFailureManagerITCase.java | 214 +++++++++++++++++++++
19 files changed, 497 insertions(+), 93 deletions(-)
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/messages/checkpoint/SerializedCheckpointException.java
create mode 100644 flink-tests/src/test/java/org/apache/flink/test/checkpointing/CheckpointFailureManagerITCase.java
[flink] 01/03: [FLINK-16443][checkpointing] Make sure that
CheckpointException are also serialized in DeclineCheckpoint.
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tangyun pushed a commit to branch release-1.12
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 333d62619f498805a7df3fec0f348ee686d5520d
Author: Arvid Heise <ar...@ververica.com>
AuthorDate: Mon Nov 23 13:56:02 2020 +0100
[FLINK-16443][checkpointing] Make sure that CheckpointException are also serialized in DeclineCheckpoint.
---
.../messages/checkpoint/DeclineCheckpoint.java | 20 ++++++++++----------
1 file changed, 10 insertions(+), 10 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/messages/checkpoint/DeclineCheckpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/messages/checkpoint/DeclineCheckpoint.java
index fe1dd09..c76c921 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/messages/checkpoint/DeclineCheckpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/messages/checkpoint/DeclineCheckpoint.java
@@ -19,10 +19,11 @@
package org.apache.flink.runtime.messages.checkpoint;
import org.apache.flink.api.common.JobID;
-import org.apache.flink.runtime.checkpoint.CheckpointException;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.util.SerializedThrowable;
+import javax.annotation.Nullable;
+
/**
* This message is sent from the {@link org.apache.flink.runtime.taskexecutor.TaskExecutor} to the
* {@link org.apache.flink.runtime.jobmaster.JobMaster} to tell the checkpoint coordinator that a
@@ -34,22 +35,21 @@ public class DeclineCheckpoint extends AbstractCheckpointMessage implements java
private static final long serialVersionUID = 2094094662279578953L;
/** The reason why the checkpoint was declined. */
- private final Throwable reason;
+ @Nullable private final SerializedThrowable reason;
public DeclineCheckpoint(JobID job, ExecutionAttemptID taskExecutionId, long checkpointId) {
this(job, taskExecutionId, checkpointId, null);
}
public DeclineCheckpoint(
- JobID job, ExecutionAttemptID taskExecutionId, long checkpointId, Throwable reason) {
+ JobID job,
+ ExecutionAttemptID taskExecutionId,
+ long checkpointId,
+ @Nullable Throwable reason) {
super(job, taskExecutionId, checkpointId);
- if (reason == null || reason instanceof CheckpointException) {
- this.reason = reason;
- } else {
- // some other exception. replace with a serialized throwable, to be on the safe side
- this.reason = new SerializedThrowable(reason);
- }
+ // some other exception. replace with a serialized throwable, to be on the safe side
+ this.reason = reason == null ? null : new SerializedThrowable(reason);
}
// --------------------------------------------------------------------------------------------
@@ -59,7 +59,7 @@ public class DeclineCheckpoint extends AbstractCheckpointMessage implements java
*
* @return The reason why the checkpoint was declined
*/
- public Throwable getReason() {
+ public SerializedThrowable getReason() {
return reason;
}
[flink] 03/03: [FLINK-20675][checkpointing] Ensure asynchronous
checkpoint failure could fail the job by default
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tangyun pushed a commit to branch release-1.12
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 4d6b952a82facc838d5ea73bc8fe20e6dc66e1cf
Author: Yun Tang <my...@live.com>
AuthorDate: Mon Jan 25 00:58:11 2021 +0800
[FLINK-20675][checkpointing] Ensure asynchronous checkpoint failure could fail the job by default
---
.../runtime/checkpoint/CheckpointCoordinator.java | 15 +-
.../checkpoint/CheckpointFailureManager.java | 7 +-
.../messages/checkpoint/DeclineCheckpoint.java | 31 +--
.../checkpoint/SerializedCheckpointException.java | 53 +++++
.../taskexecutor/rpc/RpcCheckpointResponder.java | 12 +-
.../checkpoint/CheckpointCoordinatorTest.java | 105 ++++++++--
.../CheckpointCoordinatorTestingUtils.java | 6 +
.../checkpoint/CheckpointFailureManagerTest.java | 3 +-
.../flink/runtime/jobmaster/JobMasterTest.java | 20 +-
.../tasks/StreamTaskExecutionDecorationTest.java | 7 +-
.../CheckpointFailureManagerITCase.java | 214 +++++++++++++++++++++
11 files changed, 423 insertions(+), 50 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 3c3b008..be8a749 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
@@ -980,7 +980,9 @@ public class CheckpointCoordinator {
}
final long checkpointId = message.getCheckpointId();
- final String reason = (message.getReason() != null ? message.getReason().getMessage() : "");
+ final CheckpointException checkpointException =
+ message.getSerializedCheckpointException().unwrap();
+ final String reason = checkpointException.getMessage();
PendingCheckpoint checkpoint;
@@ -1004,16 +1006,7 @@ public class CheckpointCoordinator {
message.getTaskExecutionId(),
job,
taskManagerLocationInfo,
- message.getReason());
- final CheckpointException checkpointException;
- if (message.getReason() == null) {
- checkpointException =
- new CheckpointException(CheckpointFailureReason.CHECKPOINT_DECLINED);
- } else {
- checkpointException =
- getCheckpointException(
- CheckpointFailureReason.JOB_FAILURE, message.getReason());
- }
+ checkpointException.getCause());
abortPendingCheckpoint(
checkpoint, checkpointException, message.getTaskExecutionId());
} else if (LOG.isDebugEnabled()) {
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 40f213a..44dc411 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
@@ -33,6 +33,8 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
public class CheckpointFailureManager {
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.";
private final int tolerableCpFailureNumber;
private final FailJobCallback failureCallback;
@@ -94,8 +96,7 @@ public class CheckpointFailureManager {
if (continuousFailureCounter.get() > tolerableCpFailureNumber) {
clearCount();
errorHandler.accept(
- new FlinkRuntimeException(
- "Exceeded checkpoint tolerable failure threshold."));
+ new FlinkRuntimeException(EXCEEDED_CHECKPOINT_TOLERABLE_FAILURE_MESSAGE));
}
}
}
@@ -127,7 +128,6 @@ public class CheckpointFailureManager {
case CHECKPOINT_DECLINED_INPUT_END_OF_STREAM:
case EXCEPTION:
- case CHECKPOINT_ASYNC_EXCEPTION:
case TASK_FAILURE:
case TASK_CHECKPOINT_FAILURE:
case UNKNOWN_TASK_CHECKPOINT_NOTIFICATION_FAILURE:
@@ -136,6 +136,7 @@ public class CheckpointFailureManager {
// ignore
break;
+ case CHECKPOINT_ASYNC_EXCEPTION:
case CHECKPOINT_DECLINED:
case CHECKPOINT_EXPIRED:
// we should make sure one checkpoint only be counted once
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/messages/checkpoint/DeclineCheckpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/messages/checkpoint/DeclineCheckpoint.java
index c76c921..0122d9a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/messages/checkpoint/DeclineCheckpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/messages/checkpoint/DeclineCheckpoint.java
@@ -19,10 +19,11 @@
package org.apache.flink.runtime.messages.checkpoint;
import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
-import org.apache.flink.util.SerializedThrowable;
+import org.apache.flink.util.Preconditions;
-import javax.annotation.Nullable;
+import javax.annotation.Nonnull;
/**
* This message is sent from the {@link org.apache.flink.runtime.taskexecutor.TaskExecutor} to the
@@ -34,22 +35,20 @@ public class DeclineCheckpoint extends AbstractCheckpointMessage implements java
private static final long serialVersionUID = 2094094662279578953L;
- /** The reason why the checkpoint was declined. */
- @Nullable private final SerializedThrowable reason;
-
- public DeclineCheckpoint(JobID job, ExecutionAttemptID taskExecutionId, long checkpointId) {
- this(job, taskExecutionId, checkpointId, null);
- }
+ /** The serialized reason why the checkpoint was declined. */
+ private final SerializedCheckpointException serializedCheckpointException;
public DeclineCheckpoint(
JobID job,
ExecutionAttemptID taskExecutionId,
long checkpointId,
- @Nullable Throwable reason) {
+ CheckpointException checkpointException) {
super(job, taskExecutionId, checkpointId);
- // some other exception. replace with a serialized throwable, to be on the safe side
- this.reason = reason == null ? null : new SerializedThrowable(reason);
+ Preconditions.checkNotNull(
+ checkpointException, "The checkpoint exception must not be null.");
+ // replace with a serialized throwable, to be on the safe side
+ this.serializedCheckpointException = new SerializedCheckpointException(checkpointException);
}
// --------------------------------------------------------------------------------------------
@@ -59,8 +58,9 @@ public class DeclineCheckpoint extends AbstractCheckpointMessage implements java
*
* @return The reason why the checkpoint was declined
*/
- public SerializedThrowable getReason() {
- return reason;
+ @Nonnull
+ public SerializedCheckpointException getSerializedCheckpointException() {
+ return serializedCheckpointException;
}
// --------------------------------------------------------------------------------------------
@@ -69,6 +69,9 @@ public class DeclineCheckpoint extends AbstractCheckpointMessage implements java
public String toString() {
return String.format(
"Declined Checkpoint %d for (%s/%s): %s",
- getCheckpointId(), getJob(), getTaskExecutionId(), reason);
+ getCheckpointId(),
+ getJob(),
+ getTaskExecutionId(),
+ serializedCheckpointException.getCheckpointFailureReason());
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/messages/checkpoint/SerializedCheckpointException.java b/flink-runtime/src/main/java/org/apache/flink/runtime/messages/checkpoint/SerializedCheckpointException.java
new file mode 100644
index 0000000..a18ce3d
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/messages/checkpoint/SerializedCheckpointException.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.messages.checkpoint;
+
+import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.checkpoint.CheckpointFailureReason;
+import org.apache.flink.util.SerializedThrowable;
+
+import java.io.Serializable;
+
+/**
+ * Serialized checkpoint exception which wraps the checkpoint failure reason and its serialized
+ * throwable.
+ */
+public class SerializedCheckpointException implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final CheckpointFailureReason checkpointFailureReason;
+ private final SerializedThrowable serializedThrowable;
+
+ public SerializedCheckpointException(CheckpointException checkpointException) {
+ this.checkpointFailureReason = checkpointException.getCheckpointFailureReason();
+ this.serializedThrowable = new SerializedThrowable(checkpointException);
+ }
+
+ public CheckpointFailureReason getCheckpointFailureReason() {
+ return checkpointFailureReason;
+ }
+
+ public SerializedThrowable getSerializedThrowable() {
+ return serializedThrowable;
+ }
+
+ public CheckpointException unwrap() {
+ return new CheckpointException(checkpointFailureReason, serializedThrowable);
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/rpc/RpcCheckpointResponder.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/rpc/RpcCheckpointResponder.java
index b445941..d9788f4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/rpc/RpcCheckpointResponder.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/rpc/RpcCheckpointResponder.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.taskexecutor.rpc;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.checkpoint.CheckpointCoordinatorGateway;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
@@ -55,7 +56,16 @@ public class RpcCheckpointResponder implements CheckpointResponder {
long checkpointId,
Throwable cause) {
+ // TODO the passed parameter 'cause' is actually always instance of CheckpointException,
+ // we should change the interfaces to narrow all declined checkpoint's throwable to
+ // CheckpointException.
+ Preconditions.checkArgument(
+ cause instanceof CheckpointException,
+ "The given cause is "
+ + cause.getClass()
+ + " instead of expected CheckpointException.");
checkpointCoordinatorGateway.declineCheckpoint(
- new DeclineCheckpoint(jobID, executionAttemptID, checkpointId, cause));
+ new DeclineCheckpoint(
+ jobID, executionAttemptID, checkpointId, (CheckpointException) cause));
}
}
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 d088606..dd48389 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
@@ -99,6 +99,8 @@ import java.util.concurrent.atomic.AtomicReference;
import static org.apache.flink.runtime.checkpoint.CheckpointCoordinatorTestingUtils.mockExecutionJobVertex;
import static org.apache.flink.runtime.checkpoint.CheckpointCoordinatorTestingUtils.mockExecutionVertex;
+import static org.apache.flink.runtime.checkpoint.CheckpointFailureReason.CHECKPOINT_ASYNC_EXCEPTION;
+import static org.apache.flink.runtime.checkpoint.CheckpointFailureReason.CHECKPOINT_DECLINED;
import static org.apache.flink.runtime.checkpoint.CheckpointFailureReason.CHECKPOINT_EXPIRED;
import static org.apache.flink.util.Preconditions.checkNotNull;
import static org.junit.Assert.assertEquals;
@@ -315,7 +317,11 @@ public class CheckpointCoordinatorTest extends TestLogger {
// decline checkpoint from the other task
checkpointCoordinator.receiveDeclineMessage(
- new DeclineCheckpoint(jobId, attemptID1, checkpointId),
+ new DeclineCheckpoint(
+ jobId,
+ attemptID1,
+ checkpointId,
+ new CheckpointException(CHECKPOINT_DECLINED)),
TASK_MANAGER_LOCATION_INFO);
fail("Test failed.");
@@ -361,14 +367,26 @@ public class CheckpointCoordinatorTest extends TestLogger {
}
}
+ @Test
+ public void testTriggerAndDeclineSyncCheckpointFailureSimple() {
+ testTriggerAndDeclineCheckpointSimple(CHECKPOINT_DECLINED);
+ }
+
+ @Test
+ public void testTriggerAndDeclineAsyncCheckpointFailureSimple() {
+ testTriggerAndDeclineCheckpointSimple(CHECKPOINT_ASYNC_EXCEPTION);
+ }
+
/**
* This test triggers a checkpoint and then sends a decline checkpoint message from one of the
* tasks. The expected behaviour is that said checkpoint is discarded and a new checkpoint is
* triggered.
*/
- @Test
- public void testTriggerAndDeclineCheckpointSimple() {
+ private void testTriggerAndDeclineCheckpointSimple(
+ CheckpointFailureReason checkpointFailureReason) {
try {
+ final CheckpointException checkpointException =
+ new CheckpointException(checkpointFailureReason);
final JobID jobId = new JobID();
// create some mock Execution vertices that receive the checkpoint trigger messages
@@ -377,9 +395,21 @@ public class CheckpointCoordinatorTest extends TestLogger {
ExecutionVertex vertex1 = mockExecutionVertex(attemptID1);
ExecutionVertex vertex2 = mockExecutionVertex(attemptID2);
+ TestFailJobCallback failJobCallback = new TestFailJobCallback();
// set up the coordinator and validate the initial state
CheckpointCoordinator checkpointCoordinator =
- getCheckpointCoordinator(jobId, vertex1, vertex2);
+ new CheckpointCoordinatorBuilder()
+ .setJobId(jobId)
+ .setTasks(new ExecutionVertex[] {vertex1, vertex2})
+ .setCheckpointCoordinatorConfiguration(
+ CheckpointCoordinatorConfiguration.builder()
+ .setAlignmentTimeout(Long.MAX_VALUE)
+ .setMaxConcurrentCheckpoints(Integer.MAX_VALUE)
+ .build())
+ .setTimer(manuallyTriggeredScheduledExecutor)
+ .setCheckpointFailureManager(
+ new CheckpointFailureManager(0, failJobCallback))
+ .build();
assertEquals(0, checkpointCoordinator.getNumberOfPendingCheckpoints());
assertEquals(0, checkpointCoordinator.getNumberOfRetainedSuccessfulCheckpoints());
@@ -445,7 +475,7 @@ public class CheckpointCoordinatorTest extends TestLogger {
// decline checkpoint from the other task, this should cancel the checkpoint
// and trigger a new one
checkpointCoordinator.receiveDeclineMessage(
- new DeclineCheckpoint(jobId, attemptID1, checkpointId),
+ new DeclineCheckpoint(jobId, attemptID1, checkpointId, checkpointException),
TASK_MANAGER_LOCATION_INFO);
assertTrue(checkpoint.isDisposed());
@@ -459,12 +489,13 @@ public class CheckpointCoordinatorTest extends TestLogger {
// decline again, nothing should happen
// decline from the other task, nothing should happen
checkpointCoordinator.receiveDeclineMessage(
- new DeclineCheckpoint(jobId, attemptID1, checkpointId),
+ new DeclineCheckpoint(jobId, attemptID1, checkpointId, checkpointException),
TASK_MANAGER_LOCATION_INFO);
checkpointCoordinator.receiveDeclineMessage(
- new DeclineCheckpoint(jobId, attemptID2, checkpointId),
+ new DeclineCheckpoint(jobId, attemptID2, checkpointId, checkpointException),
TASK_MANAGER_LOCATION_INFO);
assertTrue(checkpoint.isDisposed());
+ assertEquals(1, failJobCallback.getInvokeCounter());
checkpointCoordinator.shutdown(JobStatus.FINISHED);
} catch (Exception e) {
@@ -563,7 +594,11 @@ public class CheckpointCoordinatorTest extends TestLogger {
// decline checkpoint from one of the tasks, this should cancel the checkpoint
checkpointCoordinator.receiveDeclineMessage(
- new DeclineCheckpoint(jobId, attemptID1, checkpoint1Id),
+ new DeclineCheckpoint(
+ jobId,
+ attemptID1,
+ checkpoint1Id,
+ new CheckpointException(CHECKPOINT_DECLINED)),
TASK_MANAGER_LOCATION_INFO);
verify(vertex1.getCurrentExecutionAttempt(), times(1))
.notifyCheckpointAborted(eq(checkpoint1Id), any(Long.class));
@@ -602,10 +637,18 @@ public class CheckpointCoordinatorTest extends TestLogger {
// decline again, nothing should happen
// decline from the other task, nothing should happen
checkpointCoordinator.receiveDeclineMessage(
- new DeclineCheckpoint(jobId, attemptID1, checkpoint1Id),
+ new DeclineCheckpoint(
+ jobId,
+ attemptID1,
+ checkpoint1Id,
+ new CheckpointException(CHECKPOINT_DECLINED)),
TASK_MANAGER_LOCATION_INFO);
checkpointCoordinator.receiveDeclineMessage(
- new DeclineCheckpoint(jobId, attemptID2, checkpoint1Id),
+ new DeclineCheckpoint(
+ jobId,
+ attemptID2,
+ checkpoint1Id,
+ new CheckpointException(CHECKPOINT_DECLINED)),
TASK_MANAGER_LOCATION_INFO);
assertTrue(checkpoint1.isDisposed());
@@ -1445,7 +1488,11 @@ public class CheckpointCoordinatorTest extends TestLogger {
// let the checkpoint fail at the first ack vertex
reset(subtaskStateTrigger);
checkpointCoordinator.receiveDeclineMessage(
- new DeclineCheckpoint(jobId, ackAttemptId1, checkpointId),
+ new DeclineCheckpoint(
+ jobId,
+ ackAttemptId1,
+ checkpointId,
+ new CheckpointException(CHECKPOINT_DECLINED)),
TASK_MANAGER_LOCATION_INFO);
assertTrue(pendingCheckpoint.isDisposed());
@@ -2624,7 +2671,7 @@ public class CheckpointCoordinatorTest extends TestLogger {
} catch (ExecutionException e) {
final Throwable cause = ExceptionUtils.stripExecutionException(e);
assertTrue(cause instanceof CheckpointException);
- assertEquals(expectedRootCause.getMessage(), cause.getCause().getMessage());
+ assertEquals(expectedRootCause.getMessage(), cause.getCause().getCause().getMessage());
}
assertEquals(1L, invocationCounterAndException.f0.intValue());
@@ -2633,6 +2680,7 @@ public class CheckpointCoordinatorTest extends TestLogger {
&& invocationCounterAndException
.f1
.getCause()
+ .getCause()
.getMessage()
.equals(expectedRootCause.getMessage()));
@@ -2711,7 +2759,12 @@ public class CheckpointCoordinatorTest extends TestLogger {
coordinator.getNumQueuedRequests());
coordinator.receiveDeclineMessage(
- new DeclineCheckpoint(jobId, new ExecutionAttemptID(), 1L), "none");
+ new DeclineCheckpoint(
+ jobId,
+ new ExecutionAttemptID(),
+ 1L,
+ new CheckpointException(CHECKPOINT_DECLINED)),
+ "none");
manuallyTriggeredScheduledExecutor.triggerAll();
activeRequests--; // savepoint triggered
@@ -3209,7 +3262,11 @@ public class CheckpointCoordinatorTest extends TestLogger {
coordinator.getPendingCheckpoints().entrySet().iterator().next().getKey();
final PendingCheckpoint checkpoint = coordinator.getPendingCheckpoints().get(checkpointId);
coordinator.receiveDeclineMessage(
- new DeclineCheckpoint(jobId, attemptID, checkpointId, reason),
+ new DeclineCheckpoint(
+ jobId,
+ attemptID,
+ checkpointId,
+ new CheckpointException(CHECKPOINT_DECLINED, reason)),
TASK_MANAGER_LOCATION_INFO);
return checkpoint;
}
@@ -3306,4 +3363,24 @@ public class CheckpointCoordinatorTest extends TestLogger {
this.owner = checkNotNull(coordinator);
}
}
+
+ private static class TestFailJobCallback implements CheckpointFailureManager.FailJobCallback {
+
+ private int invokeCounter = 0;
+
+ @Override
+ public void failJob(Throwable cause) {
+ invokeCounter++;
+ }
+
+ @Override
+ public void failJobDueToTaskFailure(
+ final Throwable cause, final ExecutionAttemptID executionAttemptID) {
+ invokeCounter++;
+ }
+
+ public int getInvokeCounter() {
+ return invokeCounter;
+ }
+ }
}
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 a2c9bb4..d7469bd 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
@@ -713,6 +713,12 @@ public class CheckpointCoordinatorTestingUtils {
return this;
}
+ public CheckpointCoordinatorBuilder setCheckpointFailureManager(
+ CheckpointFailureManager checkpointFailureManager) {
+ this.failureManager = checkpointFailureManager;
+ return this;
+ }
+
public CheckpointCoordinatorBuilder setCompletedCheckpointStore(
CompletedCheckpointStore completedCheckpointStore) {
this.completedCheckpointStore = completedCheckpointStore;
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 a73bf18..66a137a 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
@@ -95,7 +95,8 @@ public class CheckpointFailureManagerTest extends TestLogger {
failureManager.handleJobLevelCheckpointException(new CheckpointException(reason), -1);
}
- assertEquals(2, callback.getInvokeCounter());
+ // CHECKPOINT_DECLINED, CHECKPOINT_EXPIRED and CHECKPOINT_ASYNC_EXCEPTION
+ assertEquals(3, callback.getInvokeCounter());
}
@Test
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
index 8b9862c..f3524a0 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
@@ -38,6 +38,8 @@ import org.apache.flink.core.io.InputSplitSource;
import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.queryablestate.KvStateID;
import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.checkpoint.CheckpointFailureReason;
import org.apache.flink.runtime.checkpoint.CheckpointProperties;
import org.apache.flink.runtime.checkpoint.CheckpointRecoveryFactory;
import org.apache.flink.runtime.checkpoint.CheckpointRetentionPolicy;
@@ -133,7 +135,6 @@ import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.InstantiationUtil;
-import org.apache.flink.util.SerializedThrowable;
import org.apache.flink.util.TestLogger;
import akka.actor.ActorSystem;
@@ -410,7 +411,8 @@ public class JobMasterTest extends TestLogger {
System.currentTimeMillis()) {
@Override
public void declineCheckpoint(DeclineCheckpoint declineCheckpoint) {
- declineCheckpointMessageFuture.complete(declineCheckpoint.getReason());
+ declineCheckpointMessageFuture.complete(
+ declineCheckpoint.getSerializedCheckpointException().unwrap());
}
};
@@ -428,6 +430,10 @@ public class JobMasterTest extends TestLogger {
Throwable userException =
(Throwable) Class.forName(className, false, userClassLoader).newInstance();
+ CheckpointException checkpointException =
+ new CheckpointException(
+ CheckpointFailureReason.CHECKPOINT_DECLINED, userException);
+
JobMasterGateway jobMasterGateway =
rpcService2
.connect(
@@ -439,13 +445,17 @@ public class JobMasterTest extends TestLogger {
RpcCheckpointResponder rpcCheckpointResponder =
new RpcCheckpointResponder(jobMasterGateway);
rpcCheckpointResponder.declineCheckpoint(
- jobGraph.getJobID(), new ExecutionAttemptID(), 1, userException);
+ jobGraph.getJobID(), new ExecutionAttemptID(), 1, checkpointException);
Throwable throwable =
declineCheckpointMessageFuture.get(
testingTimeout.toMilliseconds(), TimeUnit.MILLISECONDS);
- assertThat(throwable, instanceOf(SerializedThrowable.class));
- assertThat(throwable.getMessage(), equalTo(userException.getMessage()));
+ assertThat(throwable, instanceOf(CheckpointException.class));
+ Optional<Throwable> throwableWithMessage =
+ ExceptionUtils.findThrowableWithMessage(throwable, userException.getMessage());
+ assertTrue(throwableWithMessage.isPresent());
+ assertThat(
+ throwableWithMessage.get().getMessage(), equalTo(userException.getMessage()));
} finally {
RpcUtils.terminateRpcServices(testingTimeout, rpcService1, rpcService2);
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskExecutionDecorationTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskExecutionDecorationTest.java
index 51e4803..1d7f33e 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskExecutionDecorationTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskExecutionDecorationTest.java
@@ -17,6 +17,8 @@
package org.apache.flink.streaming.runtime.tasks;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.checkpoint.CheckpointFailureReason;
import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.checkpoint.CheckpointType;
@@ -48,7 +50,10 @@ public class StreamTaskExecutionDecorationTest {
@Test
public void testAbortCheckpointOnBarrierIsDecorated() throws Exception {
- task.abortCheckpointOnBarrier(1, null);
+ task.abortCheckpointOnBarrier(
+ 1,
+ new CheckpointException(
+ CheckpointFailureReason.CHECKPOINT_DECLINED_ON_CANCELLATION_BARRIER));
Assert.assertTrue("execution decorator was not called", decorator.wasCalled());
}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/CheckpointFailureManagerITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/CheckpointFailureManagerITCase.java
new file mode 100644
index 0000000..2b06a97
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/CheckpointFailureManagerITCase.java
@@ -0,0 +1,214 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.test.checkpointing;
+
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.core.fs.CloseableRegistry;
+import org.apache.flink.runtime.checkpoint.CheckpointFailureManager;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.client.JobExecutionException;
+import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.state.AbstractSnapshotStrategy;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.runtime.state.DefaultOperatorStateBackend;
+import org.apache.flink.runtime.state.DefaultOperatorStateBackendBuilder;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.runtime.state.OperatorStateBackend;
+import org.apache.flink.runtime.state.OperatorStateHandle;
+import org.apache.flink.runtime.state.SnapshotResult;
+import org.apache.flink.runtime.state.memory.MemoryStateBackend;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
+import org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.test.util.TestUtils;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import javax.annotation.Nonnull;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Optional;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.RunnableFuture;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.mockito.Mockito.mock;
+
+/** Tests to verify end-to-end logic of checkpoint failure manager. */
+public class CheckpointFailureManagerITCase extends TestLogger {
+ private static MiniClusterWithClientResource cluster;
+
+ @Before
+ public void setup() throws Exception {
+ Configuration configuration = new Configuration();
+
+ cluster =
+ new MiniClusterWithClientResource(
+ new MiniClusterResourceConfiguration.Builder()
+ .setConfiguration(configuration)
+ .build());
+ cluster.before();
+ }
+
+ @AfterClass
+ public static void shutDownExistingCluster() {
+ if (cluster != null) {
+ cluster.after();
+ cluster = null;
+ }
+ }
+
+ @Test(timeout = 10000)
+ public void testAsyncCheckpointFailureTriggerJobFailed() throws Exception {
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.enableCheckpointing(100);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+ env.setStateBackend(new AsyncFailureStateBackend());
+ env.addSource(new StringGeneratingSourceFunction()).addSink(new DiscardingSink<>());
+ JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(env.getStreamGraph());
+ try {
+ // assert that the job only execute checkpoint once and only failed once.
+ TestUtils.submitJobAndWaitForResult(
+ cluster.getClusterClient(), jobGraph, getClass().getClassLoader());
+ } catch (JobExecutionException jobException) {
+ Optional<FlinkRuntimeException> throwable =
+ ExceptionUtils.findThrowable(jobException, FlinkRuntimeException.class);
+ Assert.assertTrue(throwable.isPresent());
+ Assert.assertEquals(
+ CheckpointFailureManager.EXCEEDED_CHECKPOINT_TOLERABLE_FAILURE_MESSAGE,
+ throwable.get().getMessage());
+ }
+ // assert that the job only failed once.
+ Assert.assertEquals(1, StringGeneratingSourceFunction.INITIALIZE_TIMES.get());
+ }
+
+ private static class StringGeneratingSourceFunction extends RichParallelSourceFunction<String>
+ implements CheckpointedFunction {
+ private static final long serialVersionUID = 1L;
+ private static final ListStateDescriptor<Long> stateDescriptor =
+ new ListStateDescriptor<>("emitted", Long.class);
+
+ private final byte[] randomBytes = new byte[10];
+
+ private ListState<Long> listState;
+ private long emitted = 0L;
+
+ private volatile boolean isRunning = true;
+
+ public static final AtomicInteger INITIALIZE_TIMES = new AtomicInteger(0);
+
+ @Override
+ public void snapshotState(FunctionSnapshotContext context) throws Exception {
+ listState.clear();
+ listState.add(emitted);
+ }
+
+ @Override
+ public void initializeState(FunctionInitializationContext context) throws Exception {
+ listState = context.getOperatorStateStore().getListState(stateDescriptor);
+ INITIALIZE_TIMES.addAndGet(1);
+ }
+
+ @Override
+ public void run(SourceContext<String> ctx) throws Exception {
+ while (isRunning) {
+ ThreadLocalRandom.current().nextBytes(randomBytes);
+ synchronized (ctx.getCheckpointLock()) {
+ ctx.collect(new String(randomBytes));
+ emitted += 1;
+ }
+ Thread.sleep(10);
+ }
+ }
+
+ @Override
+ public void cancel() {
+ isRunning = false;
+ }
+ }
+
+ private static class AsyncFailureStateBackend extends MemoryStateBackend {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public OperatorStateBackend createOperatorStateBackend(
+ Environment env,
+ String operatorIdentifier,
+ @Nonnull Collection<OperatorStateHandle> stateHandles,
+ CloseableRegistry cancelStreamRegistry) {
+ return new DefaultOperatorStateBackendBuilder(
+ env.getUserCodeClassLoader().asClassLoader(),
+ env.getExecutionConfig(),
+ true,
+ stateHandles,
+ cancelStreamRegistry) {
+ @Override
+ @SuppressWarnings("unchecked")
+ public DefaultOperatorStateBackend build() {
+ return new DefaultOperatorStateBackend(
+ executionConfig,
+ cancelStreamRegistry,
+ new HashMap<>(),
+ new HashMap<>(),
+ new HashMap<>(),
+ new HashMap<>(),
+ mock(AbstractSnapshotStrategy.class)) {
+ @Nonnull
+ @Override
+ public RunnableFuture<SnapshotResult<OperatorStateHandle>> snapshot(
+ long checkpointId,
+ long timestamp,
+ @Nonnull CheckpointStreamFactory streamFactory,
+ @Nonnull CheckpointOptions checkpointOptions) {
+
+ return new FutureTask<>(
+ () -> {
+ throw new Exception("Expected async snapshot exception.");
+ });
+ }
+ };
+ }
+ }.build();
+ }
+
+ @Override
+ public AsyncFailureStateBackend configure(ReadableConfig config, ClassLoader classLoader) {
+ return this;
+ }
+ }
+}
[flink] 02/03: [FLINK-20675][checkpointing] Only decline async
checkpoint failure when task is still running
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tangyun pushed a commit to branch release-1.12
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d1f8f41b773ec160d4ccee8f6b82e63afd78c901
Author: Yun Tang <my...@live.com>
AuthorDate: Mon Jan 25 00:57:44 2021 +0800
[FLINK-20675][checkpointing] Only decline async checkpoint failure when task is still running
---
.../runtime/tasks/AsyncCheckpointRunnable.java | 40 +++++++++++++++-------
.../flink/streaming/runtime/tasks/StreamTask.java | 2 +-
.../tasks/SubtaskCheckpointCoordinator.java | 2 +-
.../tasks/SubtaskCheckpointCoordinatorImpl.java | 23 +++++++------
.../runtime/tasks/AsyncCheckpointRunnableTest.java | 27 +++++++++++----
.../runtime/tasks/LocalStateForwardingTest.java | 3 +-
.../tasks/SubtaskCheckpointCoordinatorTest.java | 12 +++----
.../tasks/TestSubtaskCheckpointCoordinator.java | 2 +-
8 files changed, 71 insertions(+), 40 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/AsyncCheckpointRunnable.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/AsyncCheckpointRunnable.java
index 0f2d3e4..a4ee45d 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/AsyncCheckpointRunnable.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/AsyncCheckpointRunnable.java
@@ -36,6 +36,7 @@ import java.io.Closeable;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
+import java.util.function.Supplier;
import static org.apache.flink.util.Preconditions.checkNotNull;
import static org.apache.flink.util.Preconditions.checkState;
@@ -49,6 +50,7 @@ final class AsyncCheckpointRunnable implements Runnable, Closeable {
private final String taskName;
private final Consumer<AsyncCheckpointRunnable> registerConsumer;
private final Consumer<AsyncCheckpointRunnable> unregisterConsumer;
+ private final Supplier<Boolean> isTaskRunning;
private final Environment taskEnvironment;
public boolean isRunning() {
@@ -78,7 +80,8 @@ final class AsyncCheckpointRunnable implements Runnable, Closeable {
Consumer<AsyncCheckpointRunnable> register,
Consumer<AsyncCheckpointRunnable> unregister,
Environment taskEnvironment,
- AsyncExceptionHandler asyncExceptionHandler) {
+ AsyncExceptionHandler asyncExceptionHandler,
+ Supplier<Boolean> isTaskRunning) {
this.operatorSnapshotsInProgress = checkNotNull(operatorSnapshotsInProgress);
this.checkpointMetaData = checkNotNull(checkpointMetaData);
@@ -89,6 +92,7 @@ final class AsyncCheckpointRunnable implements Runnable, Closeable {
this.unregisterConsumer = unregister;
this.taskEnvironment = checkNotNull(taskEnvironment);
this.asyncExceptionHandler = checkNotNull(asyncExceptionHandler);
+ this.isTaskRunning = isTaskRunning;
}
@Override
@@ -237,19 +241,29 @@ final class AsyncCheckpointRunnable implements Runnable, Closeable {
+ '.',
e);
- // We only report the exception for the original cause of fail and cleanup.
- // Otherwise this followup exception could race the original exception in failing
- // the task.
- try {
- taskEnvironment.declineCheckpoint(
+ if (isTaskRunning.get()) {
+ // We only report the exception for the original cause of fail and cleanup.
+ // Otherwise this followup exception could race the original exception in
+ // failing the task.
+ try {
+ taskEnvironment.declineCheckpoint(
+ checkpointMetaData.getCheckpointId(),
+ new CheckpointException(
+ CheckpointFailureReason.CHECKPOINT_ASYNC_EXCEPTION,
+ checkpointException));
+ } catch (Exception unhandled) {
+ AsynchronousException asyncException = new AsynchronousException(unhandled);
+ asyncExceptionHandler.handleAsyncException(
+ "Failure in asynchronous checkpoint materialization",
+ asyncException);
+ }
+ } else {
+ // We never decline checkpoint after task is not running to avoid unexpected job
+ // failover, which caused by exceeding checkpoint tolerable failure threshold.
+ LOG.warn(
+ "As task is already not running, no longer decline checkpoint {}.",
checkpointMetaData.getCheckpointId(),
- new CheckpointException(
- CheckpointFailureReason.CHECKPOINT_ASYNC_EXCEPTION,
- checkpointException));
- } catch (Exception unhandled) {
- AsynchronousException asyncException = new AsynchronousException(unhandled);
- asyncExceptionHandler.handleAsyncException(
- "Failure in asynchronous checkpoint materialization", asyncException);
+ checkpointException);
}
currentState = AsyncCheckpointState.DISCARDED;
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 03fe18c..4ec9515 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
@@ -1005,7 +1005,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> extends Ab
checkpointOptions,
checkpointMetrics,
operatorChain,
- this::isCanceled);
+ this::isRunning);
});
return true;
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinator.java
index 376420e..7da9e3a 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinator.java
@@ -59,7 +59,7 @@ public interface SubtaskCheckpointCoordinator extends Closeable {
CheckpointOptions checkpointOptions,
CheckpointMetricsBuilder checkpointMetrics,
OperatorChain<?, ?> operatorChain,
- Supplier<Boolean> isCanceled)
+ Supplier<Boolean> isRunning)
throws Exception;
/**
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
index 951c30c..5f3ffbf 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
@@ -238,7 +238,7 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
CheckpointOptions options,
CheckpointMetricsBuilder metrics,
OperatorChain<?, ?> operatorChain,
- Supplier<Boolean> isCanceled)
+ Supplier<Boolean> isRunning)
throws Exception {
checkNotNull(options);
@@ -296,8 +296,8 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
new HashMap<>(operatorChain.getNumberOfOperators());
try {
if (takeSnapshotSync(
- snapshotFutures, metadata, metrics, options, operatorChain, isCanceled)) {
- finishAndReportAsync(snapshotFutures, metadata, metrics, options);
+ snapshotFutures, metadata, metrics, options, operatorChain, isRunning)) {
+ finishAndReportAsync(snapshotFutures, metadata, metrics, isRunning);
} else {
cleanup(snapshotFutures, metadata, metrics, new Exception("Checkpoint declined"));
}
@@ -497,7 +497,7 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
Map<OperatorID, OperatorSnapshotFutures> snapshotFutures,
CheckpointMetaData metadata,
CheckpointMetricsBuilder metrics,
- CheckpointOptions options) {
+ Supplier<Boolean> isRunning) {
// we are transferring ownership over snapshotInProgressList for cleanup to the thread,
// active on submit
asyncOperationsThreadPool.execute(
@@ -510,7 +510,8 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
registerConsumer(),
unregisterConsumer(),
env,
- asyncExceptionHandler));
+ asyncExceptionHandler,
+ isRunning));
}
private Consumer<AsyncCheckpointRunnable> registerConsumer() {
@@ -535,7 +536,7 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
CheckpointMetricsBuilder checkpointMetrics,
CheckpointOptions checkpointOptions,
OperatorChain<?, ?> operatorChain,
- Supplier<Boolean> isCanceled)
+ Supplier<Boolean> isRunning)
throws Exception {
for (final StreamOperatorWrapper<?, ?> operatorWrapper :
@@ -573,7 +574,7 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
checkpointOptions,
operatorChain,
operatorWrapper.getStreamOperator(),
- isCanceled,
+ isRunning,
channelStateWriteResult,
storage));
}
@@ -598,13 +599,13 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
CheckpointOptions checkpointOptions,
OperatorChain<?, ?> operatorChain,
StreamOperator<?> op,
- Supplier<Boolean> isCanceled,
+ Supplier<Boolean> isRunning,
ChannelStateWriteResult channelStateWriteResult,
CheckpointStreamFactory storage)
throws Exception {
OperatorSnapshotFutures snapshotInProgress =
checkpointStreamOperator(
- op, checkpointMetaData, checkpointOptions, storage, isCanceled);
+ op, checkpointMetaData, checkpointOptions, storage, isRunning);
if (op == operatorChain.getMainOperator()) {
snapshotInProgress.setInputChannelStateFuture(
channelStateWriteResult
@@ -679,7 +680,7 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
CheckpointMetaData checkpointMetaData,
CheckpointOptions checkpointOptions,
CheckpointStreamFactory storageLocation,
- Supplier<Boolean> isCanceled)
+ Supplier<Boolean> isRunning)
throws Exception {
try {
return op.snapshotState(
@@ -688,7 +689,7 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
checkpointOptions,
storageLocation);
} catch (Exception ex) {
- if (!isCanceled.get()) {
+ if (isRunning.get()) {
LOG.info(ex.getMessage(), ex);
}
throw ex;
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/AsyncCheckpointRunnableTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/AsyncCheckpointRunnableTest.java
index 59a20a9..8c7bbc7 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/AsyncCheckpointRunnableTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/AsyncCheckpointRunnableTest.java
@@ -37,12 +37,22 @@ import org.junit.Test;
import java.util.HashMap;
import java.util.Map;
+import java.util.function.Supplier;
/** Tests for {@link AsyncCheckpointRunnable}. */
public class AsyncCheckpointRunnableTest {
@Test
- public void testAsyncCheckpointException() {
+ public void testDeclineWithAsyncCheckpointExceptionWhenRunning() {
+ testAsyncCheckpointException(() -> true);
+ }
+
+ @Test
+ public void testDeclineWithAsyncCheckpointExceptionWhenNotRunning() {
+ testAsyncCheckpointException(() -> false);
+ }
+
+ private void testAsyncCheckpointException(Supplier<Boolean> isTaskRunning) {
final Map<OperatorID, OperatorSnapshotFutures> snapshotsInProgress = new HashMap<>();
snapshotsInProgress.put(
new OperatorID(),
@@ -66,13 +76,18 @@ public class AsyncCheckpointRunnableTest {
r -> {},
r -> {},
environment,
- (msg, ex) -> {});
+ (msg, ex) -> {},
+ isTaskRunning);
runnable.run();
- Assert.assertTrue(environment.getCause() instanceof CheckpointException);
- Assert.assertSame(
- ((CheckpointException) environment.getCause()).getCheckpointFailureReason(),
- CheckpointFailureReason.CHECKPOINT_ASYNC_EXCEPTION);
+ if (isTaskRunning.get()) {
+ Assert.assertTrue(environment.getCause() instanceof CheckpointException);
+ Assert.assertSame(
+ ((CheckpointException) environment.getCause()).getCheckpointFailureReason(),
+ CheckpointFailureReason.CHECKPOINT_ASYNC_EXCEPTION);
+ } else {
+ Assert.assertNull(environment.getCause());
+ }
}
private static class TestEnvironment extends StreamMockEnvironment {
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/LocalStateForwardingTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/LocalStateForwardingTest.java
index 3fcf08c..f925c34 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/LocalStateForwardingTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/LocalStateForwardingTest.java
@@ -125,7 +125,8 @@ public class LocalStateForwardingTest extends TestLogger {
asyncCheckpointRunnable -> {},
asyncCheckpointRunnable -> {},
testStreamTask.getEnvironment(),
- testStreamTask);
+ testStreamTask,
+ () -> true);
checkpointMetrics.setAlignmentDurationNanos(0L);
checkpointMetrics.setBytesProcessedDuringAlignment(0L);
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorTest.java
index 893ec1f..17adac8 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorTest.java
@@ -168,7 +168,7 @@ public class SubtaskCheckpointCoordinatorTest {
new CheckpointOptions(SAVEPOINT, CheckpointStorageLocationReference.getDefault()),
new CheckpointMetricsBuilder(),
operatorChain,
- () -> false);
+ () -> true);
assertEquals(false, broadcastedPriorityEvent.get());
}
@@ -191,7 +191,7 @@ public class SubtaskCheckpointCoordinatorTest {
new CheckpointMetricsBuilder(),
new OperatorChain<>(
new NoOpStreamTask<>(new DummyEnvironment()), new NonRecordWriter<>()),
- () -> false);
+ () -> true);
}
@Test
@@ -248,7 +248,7 @@ public class SubtaskCheckpointCoordinatorTest {
CheckpointOptions.forCheckpointWithDefaultLocation(),
new CheckpointMetricsBuilder(),
operatorChain,
- () -> true);
+ () -> false);
assertFalse(checkpointOperator.isCheckpointed());
assertEquals(-1, stateManager.getReportedCheckpointId());
assertEquals(0, subtaskCheckpointCoordinator.getAbortedCheckpointSize());
@@ -299,7 +299,7 @@ public class SubtaskCheckpointCoordinatorTest {
CheckpointOptions.forCheckpointWithDefaultLocation(),
new CheckpointMetricsBuilder(),
operatorChain,
- () -> true);
+ () -> false);
assertEquals(1, recordOrEvents.size());
Object recordOrEvent = recordOrEvents.get(0);
@@ -354,7 +354,7 @@ public class SubtaskCheckpointCoordinatorTest {
CheckpointOptions.forCheckpointWithDefaultLocation(),
new CheckpointMetricsBuilder(),
operatorChain,
- () -> true);
+ () -> false);
rawKeyedStateHandleFuture.awaitRun();
assertEquals(1, subtaskCheckpointCoordinator.getAsyncCheckpointRunnableSize());
assertFalse(rawKeyedStateHandleFuture.isCancelled());
@@ -384,7 +384,7 @@ public class SubtaskCheckpointCoordinatorTest {
CheckpointOptions.forCheckpointWithDefaultLocation(),
new CheckpointMetricsBuilder(),
operatorChain,
- () -> true);
+ () -> false);
subtaskCheckpointCoordinator.notifyCheckpointAborted(
checkpointId, operatorChain, () -> true);
assertEquals(0, subtaskCheckpointCoordinator.getAbortedCheckpointSize());
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TestSubtaskCheckpointCoordinator.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TestSubtaskCheckpointCoordinator.java
index eeed342..59de2d9 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TestSubtaskCheckpointCoordinator.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TestSubtaskCheckpointCoordinator.java
@@ -78,7 +78,7 @@ public class TestSubtaskCheckpointCoordinator implements SubtaskCheckpointCoordi
CheckpointOptions checkpointOptions,
CheckpointMetricsBuilder checkpointMetrics,
OperatorChain<?, ?> operatorChain,
- Supplier<Boolean> isCanceled) {}
+ Supplier<Boolean> isRunning) {}
@Override
public void notifyCheckpointComplete(