You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2022/10/07 07:00:52 UTC
[flink] branch master updated (fa8b1232d8e -> a6119121164)
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 fa8b1232d8e [FLINK-29511][rest][docs] Sort schemas in OpenAPI spec
new 127fa954a17 [hotfix] Remove unused parameters in SubtaskCheckpointCoordinatorTest
new c02ceb66c49 [FLINK-28474][checkpoint][hotfix] Remove unused code in ChannelStateWriterImplTest
new 5601da7cc25 [FLINK-28474][checkpoint] Use maxConcurrentCheckpoints as maxRecordAbortedCheckpoints
new a6119121164 [FLINK-28474][checkpoint] Fix the bug ChannelStateWriteResult might not fail after checkpoint abort
The 4 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 | 15 +++++-
.../checkpoint/channel/ChannelStateWriterImpl.java | 7 +++
.../channel/ChannelStateWriterImplTest.java | 3 +-
.../flink/streaming/api/graph/StreamConfig.java | 11 ++++
.../api/graph/StreamingJobGraphGenerator.java | 1 +
.../flink/streaming/runtime/tasks/StreamTask.java | 1 +
.../tasks/SubtaskCheckpointCoordinatorImpl.java | 42 +++++----------
.../tasks/SubtaskCheckpointCoordinatorTest.java | 59 ++++++++++++++++++++--
8 files changed, 103 insertions(+), 36 deletions(-)
[flink] 01/04: [hotfix] Remove unused parameters in SubtaskCheckpointCoordinatorTest
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 127fa954a176a96f1a351b6957108add9516d2aa
Author: fanrui <19...@gmail.com>
AuthorDate: Sat Jul 9 22:36:49 2022 +0800
[hotfix] Remove unused parameters in SubtaskCheckpointCoordinatorTest
---
.../streaming/runtime/tasks/SubtaskCheckpointCoordinatorTest.java | 5 ++---
1 file changed, 2 insertions(+), 3 deletions(-)
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 74a2e36086f..e46ebe03a03 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
@@ -111,7 +111,7 @@ public class SubtaskCheckpointCoordinatorTest {
}
MockWriter writer = new MockWriter();
- SubtaskCheckpointCoordinator coordinator = coordinator(unalignedCheckpointEnabled, writer);
+ SubtaskCheckpointCoordinator coordinator = coordinator(writer);
CheckpointStorageLocationReference locationReference =
CheckpointStorageLocationReference.getDefault();
coordinator.initInputsCheckpoint(
@@ -730,8 +730,7 @@ public class SubtaskCheckpointCoordinatorTest {
public void processWatermarkStatus(WatermarkStatus watermarkStatus) throws Exception {}
}
- private static SubtaskCheckpointCoordinator coordinator(
- boolean unalignedCheckpointEnabled, ChannelStateWriter channelStateWriter)
+ private static SubtaskCheckpointCoordinator coordinator(ChannelStateWriter channelStateWriter)
throws IOException {
return new SubtaskCheckpointCoordinatorImpl(
new TestCheckpointStorageWorkerView(100),
[flink] 03/04: [FLINK-28474][checkpoint] Use maxConcurrentCheckpoints as maxRecordAbortedCheckpoints
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 5601da7cc251eac479fe24167c0f58dbd963072f
Author: fanrui <19...@gmail.com>
AuthorDate: Thu Oct 6 19:23:59 2022 +0800
[FLINK-28474][checkpoint] Use maxConcurrentCheckpoints as maxRecordAbortedCheckpoints
---
.../flink/streaming/api/graph/StreamConfig.java | 11 ++++++++
.../api/graph/StreamingJobGraphGenerator.java | 1 +
.../flink/streaming/runtime/tasks/StreamTask.java | 1 +
.../tasks/SubtaskCheckpointCoordinatorImpl.java | 29 ----------------------
4 files changed, 13 insertions(+), 29 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java
index 479b2004580..476c165c8c0 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java
@@ -519,6 +519,17 @@ public class StreamConfig implements Serializable {
ExecutionCheckpointingOptions.ALIGNED_CHECKPOINT_TIMEOUT, alignedCheckpointTimeout);
}
+ public void setMaxConcurrentCheckpoints(int maxConcurrentCheckpoints) {
+ config.setInteger(
+ ExecutionCheckpointingOptions.MAX_CONCURRENT_CHECKPOINTS, maxConcurrentCheckpoints);
+ }
+
+ public int getMaxConcurrentCheckpoints() {
+ return config.getInteger(
+ ExecutionCheckpointingOptions.MAX_CONCURRENT_CHECKPOINTS,
+ ExecutionCheckpointingOptions.MAX_CONCURRENT_CHECKPOINTS.defaultValue());
+ }
+
/**
* Sets the job vertex level non-chained outputs. The given output list must have the same order
* with {@link JobVertex#getProducedDataSets()}.
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
index a3a173a5d50..200469fb95d 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
@@ -981,6 +981,7 @@ public class StreamingJobGraphGenerator {
config.setCheckpointMode(getCheckpointingMode(checkpointCfg));
config.setUnalignedCheckpointsEnabled(checkpointCfg.isUnalignedCheckpointsEnabled());
config.setAlignedCheckpointTimeout(checkpointCfg.getAlignedCheckpointTimeout());
+ config.setMaxConcurrentCheckpoints(checkpointCfg.getMaxConcurrentCheckpoints());
for (int i = 0; i < vertex.getStatePartitioners().length; i++) {
config.setStatePartitioner(i, vertex.getStatePartitioners()[i]);
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 ece19531c47..dc6a9dff6e4 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
@@ -466,6 +466,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
ExecutionCheckpointingOptions
.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH),
this::prepareInputSnapshot,
+ configuration.getMaxConcurrentCheckpoints(),
BarrierAlignmentUtil.createRegisterTimerCallback(
mainMailboxExecutor, systemTimerService));
resourceCloser.registerCloseable(subtaskCheckpointCoordinator::close);
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 48698a2c13e..faba793d3ca 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
@@ -77,7 +77,6 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
private static final Logger LOG =
LoggerFactory.getLogger(SubtaskCheckpointCoordinatorImpl.class);
- private static final int DEFAULT_MAX_RECORD_ABORTED_CHECKPOINTS = 128;
private static final int CHECKPOINT_EXECUTION_DELAY_LOG_THRESHOLD_MS = 30_000;
@@ -121,34 +120,6 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
*/
private long alignmentCheckpointId;
- SubtaskCheckpointCoordinatorImpl(
- CheckpointStorageWorkerView checkpointStorage,
- String taskName,
- StreamTaskActionExecutor actionExecutor,
- ExecutorService asyncOperationsThreadPool,
- Environment env,
- AsyncExceptionHandler asyncExceptionHandler,
- boolean unalignedCheckpointEnabled,
- boolean enableCheckpointAfterTasksFinished,
- BiFunctionWithException<
- ChannelStateWriter, Long, CompletableFuture<Void>, CheckpointException>
- prepareInputSnapshot,
- DelayableTimer registerTimer)
- throws IOException {
- this(
- checkpointStorage,
- taskName,
- actionExecutor,
- asyncOperationsThreadPool,
- env,
- asyncExceptionHandler,
- unalignedCheckpointEnabled,
- enableCheckpointAfterTasksFinished,
- prepareInputSnapshot,
- DEFAULT_MAX_RECORD_ABORTED_CHECKPOINTS,
- registerTimer);
- }
-
SubtaskCheckpointCoordinatorImpl(
CheckpointStorageWorkerView checkpointStorage,
String taskName,
[flink] 04/04: [FLINK-28474][checkpoint] Fix the bug ChannelStateWriteResult might not fail after checkpoint abort
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 a6119121164634dfdf2d7ba671b8a3a31dc8ab0f
Author: fanrui <19...@gmail.com>
AuthorDate: Tue Oct 4 22:42:16 2022 +0800
[FLINK-28474][checkpoint] Fix the bug ChannelStateWriteResult might not fail after checkpoint abort
---
.../checkpoint/channel/ChannelStateWriter.java | 15 +++++-
.../checkpoint/channel/ChannelStateWriterImpl.java | 7 +++
.../tasks/SubtaskCheckpointCoordinatorImpl.java | 13 +++++-
.../tasks/SubtaskCheckpointCoordinatorTest.java | 54 ++++++++++++++++++++++
4 files changed, 87 insertions(+), 2 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 74eef6c9336..82c13836e2b 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
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.checkpoint.channel;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.state.InputChannelStateHandle;
@@ -72,9 +73,21 @@ public interface ChannelStateWriter extends Closeable {
resultSubpartitionStateHandles.completeExceptionally(e);
}
- boolean isDone() {
+ public boolean isDone() {
return inputChannelStateHandles.isDone() && resultSubpartitionStateHandles.isDone();
}
+
+ @VisibleForTesting
+ public void waitForDone() {
+ try {
+ inputChannelStateHandles.get();
+ } catch (Throwable ignored) {
+ }
+ try {
+ resultSubpartitionStateHandles.get();
+ } catch (Throwable ignored) {
+ }
+ }
}
/**
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 40d09f1a21a..b1091d2b2bf 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
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.checkpoint.channel;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.state.CheckpointStateOutputStream;
@@ -223,6 +224,12 @@ public class ChannelStateWriterImpl implements ChannelStateWriter {
return result;
}
+ // just for test
+ @VisibleForTesting
+ public ChannelStateWriteResult getWriteResult(long checkpointId) {
+ return results.get(checkpointId);
+ }
+
public void open() {
executor.start();
}
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 faba793d3ca..fe36c31efc6 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
@@ -95,6 +95,10 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
/** The IDs of the checkpoint for which we are notified aborted. */
private final Set<Long> abortedCheckpointIds;
+ private final int maxRecordAbortedCheckpoints;
+
+ private long maxAbortedCheckpointId = 0;
+
private long lastCheckpointId;
/** Lock that guards state of AsyncCheckpointRunnable registry. * */
@@ -180,6 +184,7 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
this.prepareInputSnapshot = prepareInputSnapshot;
this.abortedCheckpointIds =
createAbortedCheckpointSetWithLimitSize(maxRecordAbortedCheckpoints);
+ this.maxRecordAbortedCheckpoints = maxRecordAbortedCheckpoints;
this.lastCheckpointId = -1L;
this.closed = false;
this.enableCheckpointAfterTasksFinished = enableCheckpointAfterTasksFinished;
@@ -287,6 +292,10 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
// broadcast cancel checkpoint marker to avoid downstream back-pressure due to
// checkpoint barrier align.
operatorChain.broadcastEvent(new CancelCheckpointMarker(metadata.getCheckpointId()));
+ channelStateWriter.abort(
+ metadata.getCheckpointId(),
+ new CancellationException("checkpoint aborted via notification"),
+ true);
LOG.info(
"Checkpoint {} has been notified as aborted, would not trigger any checkpoint.",
metadata.getCheckpointId());
@@ -432,6 +441,7 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
if (checkpointId > lastCheckpointId) {
// only record checkpoints that have not triggered on task side.
abortedCheckpointIds.add(checkpointId);
+ maxAbortedCheckpointId = Math.max(maxAbortedCheckpointId, checkpointId);
}
}
@@ -544,7 +554,8 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
}
private boolean checkAndClearAbortedStatus(long checkpointId) {
- return abortedCheckpointIds.remove(checkpointId);
+ return abortedCheckpointIds.remove(checkpointId)
+ || checkpointId + maxRecordAbortedCheckpoints < maxAbortedCheckpointId;
}
private void registerAsyncCheckpointRunnable(
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 e46ebe03a03..2ddd9083a7a 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
@@ -82,10 +82,12 @@ import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;
import static org.apache.flink.runtime.checkpoint.CheckpointType.CHECKPOINT;
+import static org.apache.flink.runtime.state.ChannelPersistenceITCase.getStreamFactoryFactory;
import static org.apache.flink.shaded.guava30.com.google.common.util.concurrent.MoreExecutors.newDirectExecutorService;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -569,6 +571,58 @@ public class SubtaskCheckpointCoordinatorTest {
}
}
+ @Test
+ public void testChannelStateWriteResultLeakAndNotFailAfterCheckpointAborted() throws Exception {
+ String taskName = "test";
+ try (MockEnvironment mockEnvironment = MockEnvironment.builder().build();
+ ChannelStateWriterImpl writer =
+ new ChannelStateWriterImpl(taskName, 0, getStreamFactoryFactory());
+ SubtaskCheckpointCoordinator coordinator =
+ new SubtaskCheckpointCoordinatorImpl(
+ new TestCheckpointStorageWorkerView(100),
+ taskName,
+ StreamTaskActionExecutor.IMMEDIATE,
+ newDirectExecutorService(),
+ new DummyEnvironment(),
+ (unused1, unused2) -> {},
+ (unused1, unused2) -> CompletableFuture.completedFuture(null),
+ 1,
+ writer,
+ true,
+ (callable, duration) -> () -> {})) {
+ writer.open();
+ final OperatorChain<?, ?> operatorChain = getOperatorChain(mockEnvironment);
+ int checkpointId = 1;
+ // Abort checkpoint 1
+ coordinator.notifyCheckpointAborted(checkpointId, operatorChain, () -> true);
+
+ coordinator.initInputsCheckpoint(
+ checkpointId,
+ CheckpointOptions.unaligned(
+ CheckpointType.CHECKPOINT,
+ CheckpointStorageLocationReference.getDefault()));
+ ChannelStateWriter.ChannelStateWriteResult writeResult =
+ writer.getWriteResult(checkpointId);
+ assertNotNull(writeResult);
+ assertFalse(writeResult.isDone());
+ assertFalse(writeResult.getInputChannelStateHandles().isCompletedExceptionally());
+ assertFalse(writeResult.getResultSubpartitionStateHandles().isCompletedExceptionally());
+
+ coordinator.checkpointState(
+ new CheckpointMetaData(checkpointId, System.currentTimeMillis()),
+ CheckpointOptions.forCheckpointWithDefaultLocation(),
+ new CheckpointMetricsBuilder(),
+ operatorChain,
+ false,
+ () -> true);
+ assertNull(writer.getWriteResult(checkpointId));
+ writeResult.waitForDone();
+ assertTrue(writeResult.isDone());
+ assertTrue(writeResult.getInputChannelStateHandles().isCompletedExceptionally());
+ assertTrue(writeResult.getResultSubpartitionStateHandles().isCompletedExceptionally());
+ }
+ }
+
private OperatorChain<?, ?> getOperatorChain(MockEnvironment mockEnvironment) throws Exception {
return new RegularOperatorChain<>(
new MockStreamTaskBuilder(mockEnvironment).build(), new NonRecordWriter<>());
[flink] 02/04: [FLINK-28474][checkpoint][hotfix] Remove unused code in ChannelStateWriterImplTest
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 c02ceb66c49975d32e5c2ece3b2b9c9b3d207d57
Author: fanrui <19...@gmail.com>
AuthorDate: Sun Jul 17 18:10:14 2022 +0800
[FLINK-28474][checkpoint][hotfix] Remove unused code in ChannelStateWriterImplTest
---
.../flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java | 3 +--
1 file changed, 1 insertion(+), 2 deletions(-)
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 546c341be32..61c2d80aa33 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
@@ -96,7 +96,6 @@ public class ChannelStateWriterImplTest {
@Test(expected = IllegalArgumentException.class)
public void testAbortClearsResults() throws Exception {
- NetworkBuffer buffer = getBuffer();
runWithSyncWorker(
(writer, worker) -> {
callStart(writer);
@@ -167,7 +166,7 @@ public class ChannelStateWriterImplTest {
public void testAddDataNotStarted() throws Exception {
unwrappingError(
IllegalArgumentException.class,
- () -> runWithSyncWorker(writer -> callAddInputData(writer)));
+ () -> runWithSyncWorker((Consumer<ChannelStateWriter>) this::callAddInputData));
}
@Test(expected = IllegalArgumentException.class)