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:12:40 UTC
[flink] branch release-1.12 updated (ec06587 -> f393fdf)
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a change to branch release-1.12
in repository https://gitbox.apache.org/repos/asf/flink.git.
from ec06587 [FLINK-21104][network] Priority notification after cancellation.
new 5e05cd3 [FLINK-21215][task] Do not overwrite the original CheckpointFailureReason in AsyncCheckpointRunnable
new f393fdf [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:
.../runtime/io/SingleCheckpointBarrierHandler.java | 2 +-
.../runtime/tasks/AsyncCheckpointRunnable.java | 14 ++++-
.../runtime/io/UnalignedControllerTest.java | 32 +++++++++++
.../runtime/tasks/AsyncCheckpointRunnableTest.java | 62 ++++++++++++++++------
4 files changed, 91 insertions(+), 19 deletions(-)
[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 release-1.12
in repository https://gitbox.apache.org/repos/asf/flink.git
commit f393fdfd8e7780dc4a5ba0e2a82792bbef7b0576
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.
---
.../runtime/io/SingleCheckpointBarrierHandler.java | 2 +-
.../runtime/io/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/SingleCheckpointBarrierHandler.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/SingleCheckpointBarrierHandler.java
index 4099a73..ca79dba 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/SingleCheckpointBarrierHandler.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/SingleCheckpointBarrierHandler.java
@@ -187,8 +187,8 @@ public class SingleCheckpointBarrierHandler extends CheckpointBarrierHandler {
currentCheckpointId = Math.max(cancelledId, currentCheckpointId);
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/UnalignedControllerTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/UnalignedControllerTest.java
index 336f6f4..9789603 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/UnalignedControllerTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/UnalignedControllerTest.java
@@ -48,6 +48,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;
@@ -57,6 +59,7 @@ import java.util.Random;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
+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;
@@ -554,6 +557,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);
@@ -974,6 +987,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, Throwable 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.
[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 release-1.12
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 5e05cd3ece7ca32e3e7a36aece0687dcdbb2a8f8
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 | 62 ++++++++++++++++------
2 files changed, 58 insertions(+), 18 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 a4ee45d..ac08bf5 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
@@ -34,6 +34,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;
@@ -246,11 +247,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 8c7bbc7..d6f509b 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,20 +66,10 @@ 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.assertTrue(environment.getCause() instanceof CheckpointException);
Assert.assertSame(
((CheckpointException) environment.getCause()).getCheckpointFailureReason(),
@@ -90,6 +79,47 @@ public class AsyncCheckpointRunnableTest {
}
}
+ @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(((CheckpointException) 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 {
Throwable cause = null;