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 2021/02/01 17:06:44 UTC
[flink] branch master updated (52c540a -> a6eeacd)
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 52c540a [FLINK-21150][table-planner-blink] Introduce ExecEdge to connect two ExecNodes
new 37a6e51 [FLINK-21215][task] Do not overwrite the original CheckpointFailureReason in AsyncCheckpointRunnable
new a6eeacd [FLINK-21215][task] Optimise order of cancellation AsyncCheckpointRunnable and notifing JobManager
The 2 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:
.../SingleCheckpointBarrierHandler.java | 2 +-
.../runtime/tasks/AsyncCheckpointRunnable.java | 14 ++++-
.../io/checkpointing/UnalignedControllerTest.java | 32 +++++++++++
.../runtime/tasks/AsyncCheckpointRunnableTest.java | 64 ++++++++++++++++------
4 files changed, 92 insertions(+), 20 deletions(-)
[flink] 01/02: [FLINK-21215][task] Do not overwrite the original
CheckpointFailureReason in AsyncCheckpointRunnable
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 37a6e514d3c7653a8dd799a1f633f29501931169
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Mon Feb 1 09:52:30 2021 +0100
[FLINK-21215][task] Do not overwrite the original CheckpointFailureReason in AsyncCheckpointRunnable
Before this change, original failure reason would be hidden and replaced with CHECKPOINT_ASYNC_EXCEPTION
---
.../runtime/tasks/AsyncCheckpointRunnable.java | 14 ++++-
.../runtime/tasks/AsyncCheckpointRunnableTest.java | 64 ++++++++++++++++------
2 files changed, 59 insertions(+), 19 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 59809b1..4c3fd7b 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
@@ -35,6 +35,7 @@ import org.slf4j.LoggerFactory;
import java.io.Closeable;
import java.util.Map;
+import java.util.Optional;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Supplier;
@@ -263,11 +264,20 @@ final class AsyncCheckpointRunnable implements Runnable, Closeable {
// Otherwise this followup exception could race the original exception in
// failing the task.
try {
+ Optional<CheckpointException> underlyingCheckpointException =
+ ExceptionUtils.findThrowable(
+ checkpointException, CheckpointException.class);
+
+ // If this failure is already a CheckpointException, do not overwrite the
+ // original CheckpointFailureReason
+ CheckpointFailureReason reportedFailureReason =
+ underlyingCheckpointException
+ .map(exception -> exception.getCheckpointFailureReason())
+ .orElse(CheckpointFailureReason.CHECKPOINT_ASYNC_EXCEPTION);
taskEnvironment.declineCheckpoint(
checkpointMetaData.getCheckpointId(),
new CheckpointException(
- CheckpointFailureReason.CHECKPOINT_ASYNC_EXCEPTION,
- checkpointException));
+ reportedFailureReason, checkpointException));
} catch (Exception unhandled) {
AsynchronousException asyncException = new AsynchronousException(unhandled);
asyncExceptionHandler.handleAsyncException(
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 50551f1..0e4642e 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,22 +37,21 @@ 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 testDeclineWithAsyncCheckpointExceptionWhenRunning() {
- testAsyncCheckpointException(() -> true);
+ testAsyncCheckpointException(true);
}
@Test
public void testDeclineWithAsyncCheckpointExceptionWhenNotRunning() {
- testAsyncCheckpointException(() -> false);
+ testAsyncCheckpointException(false);
}
- private void testAsyncCheckpointException(Supplier<Boolean> isTaskRunning) {
+ private void testAsyncCheckpointException(boolean isTaskRunning) {
final Map<OperatorID, OperatorSnapshotFutures> snapshotsInProgress = new HashMap<>();
snapshotsInProgress.put(
new OperatorID(),
@@ -67,28 +66,59 @@ public class AsyncCheckpointRunnableTest {
final TestEnvironment environment = new TestEnvironment();
final AsyncCheckpointRunnable runnable =
- new AsyncCheckpointRunnable(
- snapshotsInProgress,
- new CheckpointMetaData(1, 1L),
- new CheckpointMetricsBuilder(),
- 1L,
- "Task Name",
- r -> {},
- r -> {},
- environment,
- (msg, ex) -> {},
- isTaskRunning);
+ createAsyncRunnable(snapshotsInProgress, environment, isTaskRunning);
runnable.run();
- if (isTaskRunning.get()) {
+ if (isTaskRunning) {
Assert.assertSame(
- (environment.getCause()).getCheckpointFailureReason(),
+ environment.getCause().getCheckpointFailureReason(),
CheckpointFailureReason.CHECKPOINT_ASYNC_EXCEPTION);
} else {
Assert.assertNull(environment.getCause());
}
}
+ @Test
+ public void testDeclineAsyncCheckpoint() {
+ CheckpointFailureReason originalReason =
+ CheckpointFailureReason.CHECKPOINT_DECLINED_INPUT_END_OF_STREAM;
+
+ final Map<OperatorID, OperatorSnapshotFutures> snapshotsInProgress = new HashMap<>();
+ snapshotsInProgress.put(
+ new OperatorID(),
+ new OperatorSnapshotFutures(
+ DoneFuture.of(SnapshotResult.empty()),
+ DoneFuture.of(SnapshotResult.empty()),
+ DoneFuture.of(SnapshotResult.empty()),
+ DoneFuture.of(SnapshotResult.empty()),
+ ExceptionallyDoneFuture.of(new CheckpointException(originalReason)),
+ DoneFuture.of(SnapshotResult.empty())));
+
+ final TestEnvironment environment = new TestEnvironment();
+ final AsyncCheckpointRunnable runnable =
+ createAsyncRunnable(snapshotsInProgress, environment, true);
+ runnable.run();
+
+ Assert.assertSame(environment.getCause().getCheckpointFailureReason(), originalReason);
+ }
+
+ private AsyncCheckpointRunnable createAsyncRunnable(
+ Map<OperatorID, OperatorSnapshotFutures> snapshotsInProgress,
+ TestEnvironment environment,
+ boolean isTaskRunning) {
+ return new AsyncCheckpointRunnable(
+ snapshotsInProgress,
+ new CheckpointMetaData(1, 1L),
+ new CheckpointMetricsBuilder(),
+ 1L,
+ "Task Name",
+ r -> {},
+ r -> {},
+ environment,
+ (msg, ex) -> {},
+ () -> isTaskRunning);
+ }
+
private static class TestEnvironment extends StreamMockEnvironment {
CheckpointException cause = null;
[flink] 02/02: [FLINK-21215][task] Optimise order of cancellation
AsyncCheckpointRunnable and notifing JobManager
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit a6eeacdb237d32c29f95dbdb6f68ac5e7c81fb2f
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Mon Feb 1 14:34:24 2021 +0100
[FLINK-21215][task] Optimise order of cancellation AsyncCheckpointRunnable and notifing JobManager
The order of cancelling AsyncCheckpointRunnable and notifing the JM was sub-optimal.
Instead of notifing the JM first, we were first cancelling the future, which could
create a race condition, were the secondary failure was reported first, causing
slightly less readable logs/error messages.
---
.../SingleCheckpointBarrierHandler.java | 2 +-
.../io/checkpointing/UnalignedControllerTest.java | 32 ++++++++++++++++++++++
2 files changed, 33 insertions(+), 1 deletion(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/SingleCheckpointBarrierHandler.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/SingleCheckpointBarrierHandler.java
index 6a0c148..f2e2381 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/SingleCheckpointBarrierHandler.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/SingleCheckpointBarrierHandler.java
@@ -257,8 +257,8 @@ public class SingleCheckpointBarrierHandler extends CheckpointBarrierHandler {
Math.max(lastCancelledOrCompletedCheckpointId, cancelledId);
numBarriersReceived = 0;
controller.abortPendingCheckpoint(cancelledId, exception);
- allBarriersReceivedFuture.completeExceptionally(exception);
notifyAbort(cancelledId, exception);
+ allBarriersReceivedFuture.completeExceptionally(exception);
}
@Override
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/UnalignedControllerTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/UnalignedControllerTest.java
index 9aace51..05de6aa 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/UnalignedControllerTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/UnalignedControllerTest.java
@@ -49,6 +49,8 @@ import org.junit.After;
import org.junit.Before;
import org.junit.Test;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
@@ -59,6 +61,7 @@ import java.util.stream.Collectors;
import java.util.stream.IntStream;
import static org.apache.flink.runtime.state.CheckpointStorageLocationReference.getDefault;
+import static org.apache.flink.util.Preconditions.checkState;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertThat;
@@ -556,6 +559,16 @@ public class UnalignedControllerTest {
}
@Test
+ public void testNotifyAbortCheckpointBeforeCanellingAsyncCheckpoint() throws Exception {
+ ValidateAsyncFutureNotCompleted handler = new ValidateAsyncFutureNotCompleted(1);
+ inputGate = createInputGate(2, handler);
+ handler.setInputGate(inputGate);
+ addSequence(inputGate, createBarrier(1, 0), createCancellationBarrier(1, 1));
+
+ addSequence(inputGate, createEndOfPartition(0), createEndOfPartition(1));
+ }
+
+ @Test
public void testSingleChannelAbortCheckpoint() throws Exception {
ValidatingCheckpointHandler handler = new ValidatingCheckpointHandler(1);
inputGate = createInputGate(1, handler);
@@ -975,6 +988,25 @@ public class UnalignedControllerTest {
}
}
+ static class ValidateAsyncFutureNotCompleted extends ValidatingCheckpointHandler {
+ private @Nullable CheckpointedInputGate inputGate;
+
+ public ValidateAsyncFutureNotCompleted(long nextExpectedCheckpointId) {
+ super(nextExpectedCheckpointId);
+ }
+
+ @Override
+ public void abortCheckpointOnBarrier(long checkpointId, CheckpointException cause) {
+ super.abortCheckpointOnBarrier(checkpointId, cause);
+ checkState(inputGate != null);
+ assertFalse(inputGate.getAllBarriersReceivedFuture(checkpointId).isDone());
+ }
+
+ public void setInputGate(CheckpointedInputGate inputGate) {
+ this.inputGate = inputGate;
+ }
+ }
+
/**
* Specific {@link AbstractInvokable} implementation to record and validate which checkpoint id
* is executed and how many checkpoints are executed.