You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by zh...@apache.org on 2020/06/10 16:10:04 UTC
[flink] branch release-1.11 updated (31a17cb -> 1d75ec8)
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a change to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 31a17cb [FLINK-18188][Runtime] Derive JM Off-Heap memory from configured Total Flink Memory minus JVM Heap
new 4030b1b [FLINK-17869][hotfix] Add taskName to ChannelStateWriter log messages
new 36609bb [FLINK-17869][hotfix] Don't pass ChannelStateWrite Future to AsyncCheckpointRunnable
new 6cb8f28 [FLINK-17869][task][checkpointing] Revert "[FLINK-17218][checkpointing] Ensuring that ChannelStateWriter aborts previous checkpoints before a new checkpoint is started."
new d806924 [FLINK-17869][task][checkpointing] Abort channel state write if checkpoint is subsumed
new e14958d [FLINK-17869][task][checkpointing] Increase ChannelStateWriterImpl.DEFAULT_MAX_CHECKPOINTS
new 7bb3ffa [FLINK-17869][task][checkpointing] Ignore out of order checkpoints in SubtaskCheckpointCoordinator
new 0634a2e [FLINK-17869][tests] Unignore UnalignedCheckpointITCase
new 1d75ec8 [FLINK-17869][task][checkpointing] Abort writing of channel state by RPC notification
The 8 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:
.../checkpoint/channel/ChannelStateWriter.java | 29 +++++-------
.../checkpoint/channel/ChannelStateWriterImpl.java | 25 ++++------
.../channel/ChannelStateWriterImplTest.java | 49 +++++++++-----------
.../checkpoint/channel/MockChannelStateWriter.java | 8 +---
.../channel/RecordingChannelStateWriter.java | 5 --
.../runtime/state/ChannelPersistenceITCase.java | 2 +-
.../runtime/io/CheckpointBarrierUnaligner.java | 31 +++++++------
.../runtime/tasks/AsyncCheckpointRunnable.java | 6 ---
.../tasks/SubtaskCheckpointCoordinatorImpl.java | 54 +++++++++++++---------
.../runtime/tasks/LocalStateForwardingTest.java | 2 -
.../tasks/TestSubtaskCheckpointCoordinator.java | 2 +-
.../checkpointing/UnalignedCheckpointITCase.java | 2 -
12 files changed, 95 insertions(+), 120 deletions(-)
[flink] 01/08: [FLINK-17869][hotfix] Add taskName to
ChannelStateWriter log messages
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 4030b1b30a2208e494d639af8767c3a944d87d58
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Wed Jun 3 14:34:30 2020 +0200
[FLINK-17869][hotfix] Add taskName to ChannelStateWriter log messages
---
.../flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
index e6aa9dc..fc8655c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
@@ -102,11 +102,11 @@ public class ChannelStateWriterImpl implements ChannelStateWriter {
LOG.debug("{} starting checkpoint {} ({})", taskName, checkpointId, checkpointOptions);
ChannelStateWriteResult result = new ChannelStateWriteResult();
ChannelStateWriteResult put = results.computeIfAbsent(checkpointId, id -> {
- Preconditions.checkState(results.size() < maxCheckpoints, "results.size() > maxCheckpoints", results.size(), maxCheckpoints);
+ Preconditions.checkState(results.size() < maxCheckpoints, String.format("%s can't start %d, results.size() > maxCheckpoints: %d > %d", taskName, checkpointId, results.size(), maxCheckpoints));
enqueue(new CheckpointStartRequest(checkpointId, result, checkpointOptions.getTargetLocation()), false);
return result;
});
- Preconditions.checkArgument(put == result, "result future already present for checkpoint " + checkpointId);
+ Preconditions.checkArgument(put == result, taskName + " result future already present for checkpoint " + checkpointId);
}
@Override
@@ -156,7 +156,7 @@ public class ChannelStateWriterImpl implements ChannelStateWriter {
public ChannelStateWriteResult getWriteResult(long checkpointId) {
LOG.debug("{} requested write result, checkpoint {}", taskName, checkpointId);
ChannelStateWriteResult result = results.get(checkpointId);
- Preconditions.checkArgument(result != null, "channel state write result not found for checkpoint " + checkpointId);
+ Preconditions.checkArgument(result != null, taskName + " channel state write result not found for checkpoint " + checkpointId);
return result;
}
[flink] 04/08: [FLINK-17869][task][checkpointing] Abort channel
state write if checkpoint is subsumed
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d8069249703bbe7858e0c6a044deb54ce75e3989
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Wed Jun 3 14:25:01 2020 +0200
[FLINK-17869][task][checkpointing] Abort channel state write if checkpoint is subsumed
Motivation: stop writing channel state ASAP if the checkpoint is subsumed
Changes:
1. complete CheckpointBarrierUnaligner.ThreadSafeUnaligner#allBarriersReceivedFuture
2. abort channel state write on its erroneous completion
3. add cleanup parameter to ChannelStateWriter.abort to use cleanup=false
in the call above
---
.../checkpoint/channel/ChannelStateWriter.java | 4 +--
.../checkpoint/channel/ChannelStateWriterImpl.java | 6 +++--
.../channel/ChannelStateWriterImplTest.java | 2 +-
.../checkpoint/channel/MockChannelStateWriter.java | 2 +-
.../runtime/io/CheckpointBarrierUnaligner.java | 31 +++++++++++-----------
.../tasks/SubtaskCheckpointCoordinatorImpl.java | 15 +++++++----
.../tasks/TestSubtaskCheckpointCoordinator.java | 2 +-
7 files changed, 35 insertions(+), 27 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
index 02a3a69..2112444 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
@@ -139,7 +139,7 @@ public interface ChannelStateWriter extends Closeable {
* Aborts the checkpoint and fails pending result for this checkpoint.
* @param cleanup true if {@link #getAndRemoveWriteResult(long)} is not supposed to be called afterwards.
*/
- void abort(long checkpointId, Throwable cause);
+ void abort(long checkpointId, Throwable cause, boolean cleanup);
/**
* Must be called after {@link #start(long, CheckpointOptions)} once.
@@ -174,7 +174,7 @@ public interface ChannelStateWriter extends Closeable {
}
@Override
- public void abort(long checkpointId, Throwable cause) {
+ public void abort(long checkpointId, Throwable cause, boolean cleanup) {
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
index 8996b3b..3e18050 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
@@ -144,11 +144,13 @@ public class ChannelStateWriterImpl implements ChannelStateWriter {
}
@Override
- public void abort(long checkpointId, Throwable cause) {
+ public void abort(long checkpointId, Throwable cause, boolean cleanup) {
LOG.debug("{} aborting, checkpoint {}", taskName, checkpointId);
enqueue(ChannelStateWriteRequest.abort(checkpointId, cause), true); // abort already started
enqueue(ChannelStateWriteRequest.abort(checkpointId, cause), false); // abort enqueued but not started
- results.remove(checkpointId);
+ if (cleanup) {
+ results.remove(checkpointId);
+ }
}
@Override
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
index 0dae88e..d0193dd 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
@@ -290,7 +290,7 @@ public class ChannelStateWriterImplTest {
}
private void callAbort(ChannelStateWriter writer) {
- writer.abort(CHECKPOINT_ID, new TestException());
+ writer.abort(CHECKPOINT_ID, new TestException(), false);
}
private void callFinish(ChannelStateWriter writer) {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/MockChannelStateWriter.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/MockChannelStateWriter.java
index 88bd334..7641d36 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/MockChannelStateWriter.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/MockChannelStateWriter.java
@@ -106,7 +106,7 @@ public class MockChannelStateWriter implements ChannelStateWriter {
}
@Override
- public void abort(long checkpointId, Throwable cause) {
+ public void abort(long checkpointId, Throwable cause, boolean cleanup) {
checkCheckpointId(checkpointId);
channelStateWriteResult.getInputChannelStateHandles().cancel(false);
channelStateWriteResult.getResultSubpartitionStateHandles().cancel(false);
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java
index f978e72..114c03e 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java
@@ -46,6 +46,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.function.Function;
import java.util.stream.IntStream;
+import static org.apache.flink.runtime.checkpoint.CheckpointFailureReason.CHECKPOINT_DECLINED_SUBSUMED;
import static org.apache.flink.util.CloseableIterator.ofElement;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -331,21 +332,21 @@ public class CheckpointBarrierUnaligner extends CheckpointBarrierHandler {
private synchronized void handleNewCheckpoint(CheckpointBarrier barrier) throws IOException {
long barrierId = barrier.getId();
- if (!allBarriersReceivedFuture.isDone() && isCheckpointPending()) {
- // we did not complete the current checkpoint, another started before
- LOG.warn("{}: Received checkpoint barrier for checkpoint {} before completing current checkpoint {}. " +
- "Skipping current checkpoint.",
- handler.taskName,
- barrierId,
- currentReceivedCheckpointId);
-
- // let the task know we are not completing this
- long currentCheckpointId = currentReceivedCheckpointId;
- handler.executeInTaskThread(() ->
- handler.notifyAbort(
- currentCheckpointId,
- new CheckpointException("Barrier id: " + barrierId, CheckpointFailureReason.CHECKPOINT_DECLINED_SUBSUMED)),
- "notifyAbort");
+ if (!allBarriersReceivedFuture.isDone()) {
+ CheckpointException exception = new CheckpointException("Barrier id: " + barrierId, CHECKPOINT_DECLINED_SUBSUMED);
+ if (isCheckpointPending()) {
+ // we did not complete the current checkpoint, another started before
+ LOG.warn("{}: Received checkpoint barrier for checkpoint {} before completing current checkpoint {}. " +
+ "Skipping current checkpoint.",
+ handler.taskName,
+ barrierId,
+ currentReceivedCheckpointId);
+
+ // let the task know we are not completing this
+ final long currentCheckpointId = currentReceivedCheckpointId;
+ handler.executeInTaskThread(() -> handler.notifyAbort(currentCheckpointId, exception), "notifyAbort");
+ }
+ allBarriersReceivedFuture.completeExceptionally(exception);
}
currentReceivedCheckpointId = barrierId;
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 adfaa44..cf8a21e 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
@@ -186,7 +186,7 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
checkpointStorage.clearCacheFor(checkpointId);
- channelStateWriter.abort(checkpointId, cause);
+ channelStateWriter.abort(checkpointId, cause, true);
// notify the coordinator that we decline this checkpoint
env.declineCheckpoint(checkpointId, cause);
@@ -391,7 +391,7 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
CheckpointMetrics metrics,
Exception ex) {
- channelStateWriter.abort(metadata.getCheckpointId(), ex);
+ channelStateWriter.abort(metadata.getCheckpointId(), ex, true);
for (OperatorSnapshotFutures operatorSnapshotResult : operatorSnapshotsInProgress.values()) {
if (operatorSnapshotResult != null) {
try {
@@ -412,9 +412,6 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
}
private void prepareInflightDataSnapshot(long checkpointId) throws IOException {
- prepareInputSnapshot.apply(channelStateWriter, checkpointId)
- .thenAccept(unused -> channelStateWriter.finishInput(checkpointId));
-
ResultPartitionWriter[] writers = env.getAllWriters();
for (ResultPartitionWriter writer : writers) {
for (int i = 0; i < writer.getNumberOfSubpartitions(); i++) {
@@ -427,6 +424,14 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
}
}
channelStateWriter.finishOutput(checkpointId);
+ prepareInputSnapshot.apply(channelStateWriter, checkpointId)
+ .whenComplete((unused, ex) -> {
+ if (ex != null) {
+ channelStateWriter.abort(checkpointId, ex, false /* result is needed and cleaned by getWriteResult */);
+ } else {
+ channelStateWriter.finishInput(checkpointId);
+ }
+ });
}
private void finishAndReportAsync(Map<OperatorID, OperatorSnapshotFutures> snapshotFutures, CheckpointMetaData metadata, CheckpointMetrics metrics, CheckpointOptions options) {
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 3897ab1..9d71a2d 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
@@ -68,7 +68,7 @@ public class TestSubtaskCheckpointCoordinator implements SubtaskCheckpointCoordi
@Override
public void abortCheckpointOnBarrier(long checkpointId, Throwable cause, OperatorChain<?, ?> operatorChain) {
- channelStateWriter.abort(checkpointId, cause);
+ channelStateWriter.abort(checkpointId, cause, true);
}
@Override
[flink] 07/08: [FLINK-17869][tests] Unignore
UnalignedCheckpointITCase
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 0634a2efd2f0ca42d4cb6762015be650c04a3136
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Wed Jun 3 22:12:15 2020 +0200
[FLINK-17869][tests] Unignore UnalignedCheckpointITCase
---
.../org/apache/flink/test/checkpointing/UnalignedCheckpointITCase.java | 2 --
1 file changed, 2 deletions(-)
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointITCase.java
index 47e7007..6211077 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointITCase.java
@@ -44,7 +44,6 @@ import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunctio
import org.apache.flink.util.TestLogger;
import org.apache.commons.lang3.ArrayUtils;
-import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ErrorCollector;
@@ -102,7 +101,6 @@ import static org.hamcrest.Matchers.greaterThan;
* <li>The number of successful checkpoints is indeed {@code >=n}.</li>
* </ul>
*/
-@Ignore
public class UnalignedCheckpointITCase extends TestLogger {
public static final String NUM_INPUTS = "inputs";
public static final String NUM_OUTPUTS = "outputs";
[flink] 03/08: [FLINK-17869][task][checkpointing] Revert
"[FLINK-17218][checkpointing] Ensuring that ChannelStateWriter aborts
previous checkpoints before a new checkpoint is started."
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 6cb8f28cf219f79a36b90a567692fafc5f85a2a2
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Wed Jun 3 14:12:04 2020 +0200
[FLINK-17869][task][checkpointing] Revert "[FLINK-17218][checkpointing] Ensuring that ChannelStateWriter aborts previous checkpoints before a new checkpoint is started."
This reverts commit 24ff415f1b76392f75dea7c3538558d24fcb7058
which introduced a race condition when task thread and netty
thread compete for ChannelStateWriteResult.
Instead, next commits fix it by:
1. Map size validation error will be prevented simply by increasing the limit
2. When a checkpoint is subsumed, it's write result will be removed from on future completion
---
.../checkpoint/channel/ChannelStateWriter.java | 4 +++-
.../checkpoint/channel/ChannelStateWriterImpl.java | 1 -
.../channel/ChannelStateWriterImplTest.java | 19 ++++++++-----------
3 files changed, 11 insertions(+), 13 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
index af2a708..02a3a69 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
@@ -179,7 +179,9 @@ public interface ChannelStateWriter extends Closeable {
@Override
public ChannelStateWriteResult getAndRemoveWriteResult(long checkpointId) {
- return ChannelStateWriteResult.EMPTY;
+ return new ChannelStateWriteResult(
+ CompletableFuture.completedFuture(Collections.emptyList()),
+ CompletableFuture.completedFuture(Collections.emptyList()));
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
index 6158358..8996b3b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
@@ -98,7 +98,6 @@ public class ChannelStateWriterImpl implements ChannelStateWriter {
@Override
public void start(long checkpointId, CheckpointOptions checkpointOptions) {
- results.keySet().forEach(oldCheckpointId -> abort(oldCheckpointId, new Exception("Starting new checkpoint " + checkpointId)));
LOG.debug("{} starting checkpoint {} ({})", taskName, checkpointId, checkpointOptions);
ChannelStateWriteResult result = new ChannelStateWriteResult();
ChannelStateWriteResult put = results.computeIfAbsent(checkpointId, id -> {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
index 9d7a7ea..0dae88e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
@@ -186,19 +186,16 @@ public class ChannelStateWriterImplTest {
unwrappingError(TestException.class, () -> callStart(writer));
}
- @Test
- public void testStartAbortsOldCheckpoints() throws Exception {
- int maxCheckpoints = 10;
- runWithSyncWorker((writer, worker) -> {
- writer.start(0, CheckpointOptions.forCheckpointWithDefaultLocation());
- ChannelStateWriteResult writeResult = writer.getWriteResult(0);
- for (int i = 1; i <= maxCheckpoints; i++) {
+ @Test(expected = IllegalStateException.class)
+ public void testLimit() throws IOException {
+ int maxCheckpoints = 3;
+ try (ChannelStateWriterImpl writer = new ChannelStateWriterImpl(TASK_NAME, getStreamFactoryFactory(), maxCheckpoints)) {
+ writer.open();
+ for (int i = 0; i < maxCheckpoints; i++) {
writer.start(i, CheckpointOptions.forCheckpointWithDefaultLocation());
- worker.processAllRequests();
- assertTrue(writeResult.isDone());
- writeResult = writer.getWriteResult(i);
}
- });
+ writer.start(maxCheckpoints, CheckpointOptions.forCheckpointWithDefaultLocation());
+ }
}
@Test(expected = IllegalStateException.class)
[flink] 02/08: [FLINK-17869][hotfix] Don't pass ChannelStateWrite
Future to AsyncCheckpointRunnable
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 36609bb5dc2aee4061a47f7a767630f1f5912d96
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Wed Jun 3 23:03:52 2020 +0200
[FLINK-17869][hotfix] Don't pass ChannelStateWrite Future to AsyncCheckpointRunnable
OperatorSnapshotFinalizer already waits and holds this future.
ChannelStateWriter.getWriteResult() can then be non-idempotent.
ChannelStateWriter.stop() can then be removed.
---
.../checkpoint/channel/ChannelStateWriter.java | 21 ++++++----------
.../checkpoint/channel/ChannelStateWriterImpl.java | 10 ++------
.../channel/ChannelStateWriterImplTest.java | 28 ++++++++++------------
.../checkpoint/channel/MockChannelStateWriter.java | 6 +----
.../channel/RecordingChannelStateWriter.java | 5 ----
.../runtime/state/ChannelPersistenceITCase.java | 2 +-
.../runtime/tasks/AsyncCheckpointRunnable.java | 6 -----
.../tasks/SubtaskCheckpointCoordinatorImpl.java | 15 +-----------
.../runtime/tasks/LocalStateForwardingTest.java | 2 --
9 files changed, 24 insertions(+), 71 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
index 5dad559..af2a708 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
@@ -123,7 +123,7 @@ public interface ChannelStateWriter extends Closeable {
* Finalize write of channel state data for the given checkpoint id.
* Must be called after {@link #start(long, CheckpointOptions)} and all of the input data of the given checkpoint added.
* When both {@link #finishInput} and {@link #finishOutput} were called the results can be (eventually) obtained
- * using {@link #getWriteResult}
+ * using {@link #getAndRemoveWriteResult}
*/
void finishInput(long checkpointId);
@@ -131,24 +131,21 @@ public interface ChannelStateWriter extends Closeable {
* Finalize write of channel state data for the given checkpoint id.
* Must be called after {@link #start(long, CheckpointOptions)} and all of the output data of the given checkpoint added.
* When both {@link #finishInput} and {@link #finishOutput} were called the results can be (eventually) obtained
- * using {@link #getWriteResult}
+ * using {@link #getAndRemoveWriteResult}
*/
void finishOutput(long checkpointId);
/**
* Aborts the checkpoint and fails pending result for this checkpoint.
+ * @param cleanup true if {@link #getAndRemoveWriteResult(long)} is not supposed to be called afterwards.
*/
void abort(long checkpointId, Throwable cause);
/**
- * Must be called after {@link #start(long, CheckpointOptions)}.
+ * Must be called after {@link #start(long, CheckpointOptions)} once.
+ * @throws IllegalArgumentException if the passed checkpointId is not known.
*/
- ChannelStateWriteResult getWriteResult(long checkpointId);
-
- /**
- * Cleans up the internal state for the given checkpoint.
- */
- void stop(long checkpointId);
+ ChannelStateWriteResult getAndRemoveWriteResult(long checkpointId) throws IllegalArgumentException;
ChannelStateWriter NO_OP = new NoOpChannelStateWriter();
@@ -181,16 +178,12 @@ public interface ChannelStateWriter extends Closeable {
}
@Override
- public ChannelStateWriteResult getWriteResult(long checkpointId) {
+ public ChannelStateWriteResult getAndRemoveWriteResult(long checkpointId) {
return ChannelStateWriteResult.EMPTY;
}
@Override
public void close() {
}
-
- @Override
- public void stop(long checkpointId) {
- }
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
index fc8655c..6158358 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
@@ -153,19 +153,13 @@ public class ChannelStateWriterImpl implements ChannelStateWriter {
}
@Override
- public ChannelStateWriteResult getWriteResult(long checkpointId) {
+ public ChannelStateWriteResult getAndRemoveWriteResult(long checkpointId) {
LOG.debug("{} requested write result, checkpoint {}", taskName, checkpointId);
- ChannelStateWriteResult result = results.get(checkpointId);
+ ChannelStateWriteResult result = results.remove(checkpointId);
Preconditions.checkArgument(result != null, taskName + " channel state write result not found for checkpoint " + checkpointId);
return result;
}
- @Override
- public void stop(long checkpointId) {
- LOG.debug("{} stopping checkpoint {}", taskName, checkpointId);
- results.remove(checkpointId);
- }
-
public void open() {
executor.start();
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
index 8c7d7f2..9d7a7ea 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
@@ -38,7 +38,6 @@ import static org.apache.flink.runtime.state.ChannelPersistenceITCase.getStreamF
import static org.apache.flink.util.CloseableIterator.ofElements;
import static org.apache.flink.util.ExceptionUtils.findThrowable;
import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
/**
@@ -69,9 +68,7 @@ public class ChannelStateWriterImplTest {
ChannelStateWriteResult result;
try (ChannelStateWriterImpl writer = openWriter()) {
callStart(writer);
- result = writer.getWriteResult(CHECKPOINT_ID);
- ChannelStateWriteResult result2 = writer.getWriteResult(CHECKPOINT_ID);
- assertSame(result, result2);
+ result = writer.getAndRemoveWriteResult(CHECKPOINT_ID);
assertFalse(result.resultSubpartitionStateHandles.isDone());
assertFalse(result.inputChannelStateHandles.isDone());
}
@@ -79,22 +76,12 @@ public class ChannelStateWriterImplTest {
assertTrue(result.resultSubpartitionStateHandles.isDone());
}
- @Test(expected = IllegalArgumentException.class)
- public void testResultCleanup() throws IOException {
- try (ChannelStateWriterImpl writer = openWriter()) {
- callStart(writer);
- writer.getWriteResult(CHECKPOINT_ID);
- writer.stop(CHECKPOINT_ID);
- writer.getWriteResult(CHECKPOINT_ID);
- }
- }
-
@Test
public void testAbort() throws Exception {
NetworkBuffer buffer = getBuffer();
runWithSyncWorker((writer, worker) -> {
callStart(writer);
- ChannelStateWriteResult result = writer.getWriteResult(CHECKPOINT_ID);
+ ChannelStateWriteResult result = writer.getAndRemoveWriteResult(CHECKPOINT_ID);
callAddInputData(writer, buffer);
callAbort(writer);
worker.processAllRequests();
@@ -108,9 +95,18 @@ public class ChannelStateWriterImplTest {
NetworkBuffer buffer = getBuffer();
runWithSyncWorker((writer, worker) -> {
callStart(writer);
+ writer.abort(CHECKPOINT_ID, new TestException(), true);
+ writer.getAndRemoveWriteResult(CHECKPOINT_ID);
+ });
+ }
+
+ @Test
+ public void testAbortDoesNotClearsResults() throws Exception {
+ runWithSyncWorker((writer, worker) -> {
+ callStart(writer);
callAbort(writer);
worker.processAllRequests();
- writer.getWriteResult(CHECKPOINT_ID);
+ writer.getAndRemoveWriteResult(CHECKPOINT_ID);
});
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/MockChannelStateWriter.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/MockChannelStateWriter.java
index 0a61066d..88bd334 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/MockChannelStateWriter.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/MockChannelStateWriter.java
@@ -101,7 +101,7 @@ public class MockChannelStateWriter implements ChannelStateWriter {
}
@Override
- public ChannelStateWriteResult getWriteResult(long checkpointId) {
+ public ChannelStateWriteResult getAndRemoveWriteResult(long checkpointId) {
return channelStateWriteResult;
}
@@ -117,8 +117,4 @@ public class MockChannelStateWriter implements ChannelStateWriter {
channelStateWriteResult.getInputChannelStateHandles().cancel(false);
channelStateWriteResult.getResultSubpartitionStateHandles().cancel(false);
}
-
- @Override
- public void stop(long checkpointId) {
- }
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/RecordingChannelStateWriter.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/RecordingChannelStateWriter.java
index d0cfe3f..151e38e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/RecordingChannelStateWriter.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/RecordingChannelStateWriter.java
@@ -81,11 +81,6 @@ public class RecordingChannelStateWriter extends MockChannelStateWriter {
return lastFinishedCheckpointId;
}
- @Override
- public void stop(long checkpointId) {
- lastFinishedCheckpointId = checkpointId;
- }
-
public ListMultimap<InputChannelInfo, Buffer> getAddedInput() {
return addedInput;
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ChannelPersistenceITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ChannelPersistenceITCase.java
index b696c77..18caac0f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ChannelPersistenceITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ChannelPersistenceITCase.java
@@ -110,7 +110,7 @@ public class ChannelPersistenceITCase {
writer.addOutputData(checkpointId, e.getKey(), SEQUENCE_NUMBER_UNKNOWN, e.getValue());
}
writer.finishOutput(checkpointId);
- ChannelStateWriteResult result = writer.getWriteResult(checkpointId);
+ ChannelStateWriteResult result = writer.getAndRemoveWriteResult(checkpointId);
result.getResultSubpartitionStateHandles().join(); // prevent abnormal complete in close
return result;
}
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 e89e962..af10411 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
@@ -33,7 +33,6 @@ import org.slf4j.LoggerFactory;
import java.io.Closeable;
import java.util.Map;
-import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
@@ -60,7 +59,6 @@ final class AsyncCheckpointRunnable implements Runnable, Closeable {
private final Map<OperatorID, OperatorSnapshotFutures> operatorSnapshotsInProgress;
private final CheckpointMetaData checkpointMetaData;
private final CheckpointMetrics checkpointMetrics;
- private final Future<?> channelWrittenFuture;
private final long asyncStartNanos;
private final AtomicReference<AsyncCheckpointState> asyncCheckpointState = new AtomicReference<>(AsyncCheckpointState.RUNNING);
@@ -68,7 +66,6 @@ final class AsyncCheckpointRunnable implements Runnable, Closeable {
Map<OperatorID, OperatorSnapshotFutures> operatorSnapshotsInProgress,
CheckpointMetaData checkpointMetaData,
CheckpointMetrics checkpointMetrics,
- Future<?> channelWrittenFuture,
long asyncStartNanos,
String taskName,
Consumer<AsyncCheckpointRunnable> register,
@@ -79,7 +76,6 @@ final class AsyncCheckpointRunnable implements Runnable, Closeable {
this.operatorSnapshotsInProgress = checkNotNull(operatorSnapshotsInProgress);
this.checkpointMetaData = checkNotNull(checkpointMetaData);
this.checkpointMetrics = checkNotNull(checkpointMetrics);
- this.channelWrittenFuture = checkNotNull(channelWrittenFuture);
this.asyncStartNanos = asyncStartNanos;
this.taskName = checkNotNull(taskName);
this.registerConsumer = register;
@@ -120,8 +116,6 @@ final class AsyncCheckpointRunnable implements Runnable, Closeable {
checkpointMetrics.setAsyncDurationMillis(asyncDurationMillis);
- channelWrittenFuture.get();
-
if (asyncCheckpointState.compareAndSet(AsyncCheckpointState.RUNNING, AsyncCheckpointState.COMPLETED)) {
reportCompletedSnapshotStates(
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 d14594f..adfaa44 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
@@ -28,7 +28,6 @@ import org.apache.flink.runtime.checkpoint.StateObjectCollection;
import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter;
import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter.ChannelStateWriteResult;
import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriterImpl;
-import org.apache.flink.runtime.concurrent.FutureUtils;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.io.network.api.CancelCheckpointMarker;
import org.apache.flink.runtime.io.network.api.CheckpointBarrier;
@@ -65,7 +64,6 @@ import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
import java.util.function.Consumer;
import java.util.function.Supplier;
@@ -432,22 +430,11 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
}
private void finishAndReportAsync(Map<OperatorID, OperatorSnapshotFutures> snapshotFutures, CheckpointMetaData metadata, CheckpointMetrics metrics, CheckpointOptions options) {
- final Future<?> channelWrittenFuture;
- if (includeChannelState(options)) {
- ChannelStateWriteResult writeResult = channelStateWriter.getWriteResult(metadata.getCheckpointId());
- channelWrittenFuture = CompletableFuture.allOf(
- writeResult.getInputChannelStateHandles(),
- writeResult.getResultSubpartitionStateHandles())
- .whenComplete((dummy, ex) -> channelStateWriter.stop(metadata.getCheckpointId()));
- } else {
- channelWrittenFuture = FutureUtils.completedVoidFuture();
- }
// we are transferring ownership over snapshotInProgressList for cleanup to the thread, active on submit
executorService.execute(new AsyncCheckpointRunnable(
snapshotFutures,
metadata,
metrics,
- channelWrittenFuture,
System.nanoTime(),
taskName,
registerConsumer(),
@@ -490,7 +477,7 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
long started = System.nanoTime();
ChannelStateWriteResult channelStateWriteResult = includeChannelState(checkpointOptions) ?
- channelStateWriter.getWriteResult(checkpointId) :
+ channelStateWriter.getAndRemoveWriteResult(checkpointId) :
ChannelStateWriteResult.EMPTY;
CheckpointStreamFactory storage = checkpointStorage.resolveCheckpointStorageLocation(checkpointId, checkpointOptions.getTargetLocation());
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 3e0703f..2eabca4 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
@@ -59,7 +59,6 @@ import javax.annotation.Nullable;
import java.util.HashMap;
import java.util.Map;
-import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;
import java.util.concurrent.RunnableFuture;
@@ -115,7 +114,6 @@ public class LocalStateForwardingTest extends TestLogger {
snapshots,
checkpointMetaData,
checkpointMetrics,
- CompletableFuture.completedFuture(null),
0L,
testStreamTask.getName(),
asyncCheckpointRunnable -> {},
[flink] 08/08: [FLINK-17869][task][checkpointing] Abort writing of
channel state by RPC notification
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 1d75ec8efe3c31f2a5e1a4572b5cfb8000ddbf67
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Thu Jun 4 10:06:15 2020 +0200
[FLINK-17869][task][checkpointing] Abort writing of channel state by RPC
notification
---
.../flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java | 2 ++
1 file changed, 2 insertions(+)
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 5d7a2c9..d904c87 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
@@ -306,6 +306,8 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
}
}
+ channelStateWriter.abort(checkpointId, new CancellationException("checkpoint aborted via notification"), false);
+
for (StreamOperatorWrapper<?, ?> operatorWrapper : operatorChain.getAllOperators(true)) {
try {
operatorWrapper.getStreamOperator().notifyCheckpointAborted(checkpointId);
[flink] 06/08: [FLINK-17869][task][checkpointing] Ignore out of
order checkpoints in SubtaskCheckpointCoordinator
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 7bb3ffa91a9916348d2f0a6a2e6cba4b109be56e
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Wed Jun 3 21:43:56 2020 +0200
[FLINK-17869][task][checkpointing] Ignore out of order checkpoints in SubtaskCheckpointCoordinator
Check (by task thread) whether the current checkpoint was already aborted in the following scenario:
1. on checkpoint barrier ThreadSafeUnaligner sends a mail to start checkpointing (netty thread)
2. on cancellation marker CheckpointBarrierUnaligner aborts it (task thread)
3. task thread processes a mail to start checkpointing
---
.../tasks/SubtaskCheckpointCoordinatorImpl.java | 22 +++++++++++++++++++---
1 file changed, 19 insertions(+), 3 deletions(-)
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 cf8a21e..5d7a2c9 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
@@ -44,7 +44,6 @@ import org.apache.flink.streaming.api.operators.StreamOperator;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.IOUtils;
-import org.apache.flink.util.Preconditions;
import org.apache.flink.util.function.BiFunctionWithException;
import org.slf4j.Logger;
@@ -57,10 +56,12 @@ import java.io.UncheckedIOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
+import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
@@ -183,6 +184,16 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
@Override
public void abortCheckpointOnBarrier(long checkpointId, Throwable cause, OperatorChain<?, ?> operatorChain) throws IOException {
LOG.debug("Aborting checkpoint via cancel-barrier {} for task {}", checkpointId, taskName);
+ lastCheckpointId = Math.max(lastCheckpointId, checkpointId);
+ Iterator<Long> iterator = abortedCheckpointIds.iterator();
+ while (iterator.hasNext()) {
+ long next = iterator.next();
+ if (next < lastCheckpointId) {
+ iterator.remove();
+ } else {
+ break;
+ }
+ }
checkpointStorage.clearCacheFor(checkpointId);
@@ -221,9 +232,14 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
// We generally try to emit the checkpoint barrier as soon as possible to not affect downstream
// checkpoint alignments
+ if (lastCheckpointId >= metadata.getCheckpointId()) {
+ LOG.info("Out of order checkpoint barrier (aborted previously?): {} >= {}", lastCheckpointId, metadata.getCheckpointId());
+ channelStateWriter.abort(metadata.getCheckpointId(), new CancellationException(), true);
+ checkAndClearAbortedStatus(metadata.getCheckpointId());
+ return;
+ }
+
// Step (0): Record the last triggered checkpointId.
- Preconditions.checkArgument(lastCheckpointId < metadata.getCheckpointId(), String.format(
- "Unexpected current checkpoint-id: %s vs last checkpoint-id: %s", metadata.getCheckpointId(), lastCheckpointId));
lastCheckpointId = metadata.getCheckpointId();
if (checkAndClearAbortedStatus(metadata.getCheckpointId())) {
LOG.info("Checkpoint {} has been notified as aborted, would not trigger any checkpoint.", metadata.getCheckpointId());
[flink] 05/08: [FLINK-17869][task][checkpointing] Increase
ChannelStateWriterImpl.DEFAULT_MAX_CHECKPOINTS
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhijiang pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e14958dcd09f7c32ba2ba25fc6a48696fe7eeca2
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Wed Jun 3 15:56:47 2020 +0200
[FLINK-17869][task][checkpointing] Increase ChannelStateWriterImpl.DEFAULT_MAX_CHECKPOINTS
ChannelStateWriter map is cleaned up by the task thread,
so the check in netty thread should take possible delay
into account.
---
.../apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
index 3e18050..89a5247 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
@@ -56,7 +56,7 @@ import static org.apache.flink.runtime.checkpoint.channel.ChannelStateWriteReque
public class ChannelStateWriterImpl implements ChannelStateWriter {
private static final Logger LOG = LoggerFactory.getLogger(ChannelStateWriterImpl.class);
- private static final int DEFAULT_MAX_CHECKPOINTS = 5; // currently, only single in-flight checkpoint is supported
+ private static final int DEFAULT_MAX_CHECKPOINTS = 1000; // includes max-concurrent-checkpoints + checkpoints to be aborted (scheduled via mailbox)
private final String taskName;
private final ChannelStateWriteRequestExecutor executor;