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 2020/06/12 14:50:59 UTC
[flink] branch release-1.11 updated: [FLINK-18252][checkpointing]
Fix savepoint overtaking output data.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.11 by this push:
new f981316 [FLINK-18252][checkpointing] Fix savepoint overtaking output data.
f981316 is described below
commit f981316f3ca0e63769fa625062148ebe75a3ebb3
Author: Arvid Heise <ar...@ververica.com>
AuthorDate: Thu Jun 11 21:18:11 2020 +0200
[FLINK-18252][checkpointing] Fix savepoint overtaking output data.
Currently, a checkpoint/savepoint barrier is always send as a priority events to the output partitions, where it overtakes data. After the fix a barrier is only a priority event iff it's unaligned.
Also CheckpointCoordinator only set unaligned flag if the barrier belongs to a checkpoint.
Ultimately, the unaligned checkpoint config option is not used by SubtaskCheckpointCoordinatorImpl except for initializing the channel state writer. The source of truth is now the CheckpointOptions.
---
.../runtime/checkpoint/CheckpointCoordinator.java | 2 +-
.../tasks/SubtaskCheckpointCoordinatorImpl.java | 16 +++------
.../tasks/SubtaskCheckpointCoordinatorTest.java | 40 +++++++++++++++++++---
3 files changed, 41 insertions(+), 17 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 307ad90..2b6bcfa 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
@@ -747,7 +747,7 @@ public class CheckpointCoordinator {
props.getCheckpointType(),
checkpointStorageLocation.getLocationReference(),
isExactlyOnceMode,
- unalignedCheckpointsEnabled);
+ props.getCheckpointType() == CheckpointType.CHECKPOINT && unalignedCheckpointsEnabled);
// send the messages to the tasks that trigger their checkpoint
for (Execution execution: executions) {
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 d904c87..b2fe147 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
@@ -82,7 +82,6 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
private final AsyncExceptionHandler asyncExceptionHandler;
private final ChannelStateWriter channelStateWriter;
private final StreamTaskActionExecutor actionExecutor;
- private final boolean unalignedCheckpointEnabled;
private final BiFunctionWithException<ChannelStateWriter, Long, CompletableFuture<Void>, IOException> prepareInputSnapshot;
/** The IDs of the checkpoint for which we are notified aborted. */
private final Set<Long> abortedCheckpointIds;
@@ -139,7 +138,6 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
executorService,
env,
asyncExceptionHandler,
- unalignedCheckpointEnabled,
prepareInputSnapshot,
maxRecordAbortedCheckpoints,
unalignedCheckpointEnabled ? openChannelStateWriter(taskName, checkpointStorage) : ChannelStateWriter.NO_OP);
@@ -154,7 +152,6 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
ExecutorService executorService,
Environment env,
AsyncExceptionHandler asyncExceptionHandler,
- boolean unalignedCheckpointEnabled,
BiFunctionWithException<ChannelStateWriter, Long, CompletableFuture<Void>, IOException> prepareInputSnapshot,
int maxRecordAbortedCheckpoints,
ChannelStateWriter channelStateWriter) throws IOException {
@@ -167,7 +164,6 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
this.asyncExceptionHandler = checkNotNull(asyncExceptionHandler);
this.actionExecutor = checkNotNull(actionExecutor);
this.channelStateWriter = checkNotNull(channelStateWriter);
- this.unalignedCheckpointEnabled = unalignedCheckpointEnabled;
this.prepareInputSnapshot = prepareInputSnapshot;
this.abortedCheckpointIds = createAbortedCheckpointSetWithLimitSize(maxRecordAbortedCheckpoints);
this.lastCheckpointId = -1L;
@@ -253,10 +249,10 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
// Step (2): Send the checkpoint barrier downstream
operatorChain.broadcastEvent(
new CheckpointBarrier(metadata.getCheckpointId(), metadata.getTimestamp(), options),
- unalignedCheckpointEnabled);
+ options.isUnalignedCheckpoint());
// Step (3): Prepare to spill the in-flight buffers for input and output
- if (includeChannelState(options)) {
+ if (options.isUnalignedCheckpoint()) {
prepareInflightDataSnapshot(metadata.getCheckpointId());
}
@@ -326,15 +322,11 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
@Override
public void initCheckpoint(long id, CheckpointOptions checkpointOptions) {
- if (includeChannelState(checkpointOptions)) {
+ if (checkpointOptions.isUnalignedCheckpoint()) {
channelStateWriter.start(id, checkpointOptions);
}
}
- private boolean includeChannelState(CheckpointOptions checkpointOptions) {
- return unalignedCheckpointEnabled && !checkpointOptions.getCheckpointType().isSavepoint();
- }
-
@Override
public void close() throws IOException {
List<AsyncCheckpointRunnable> asyncCheckpointRunnables = null;
@@ -499,7 +491,7 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
long checkpointId = checkpointMetaData.getCheckpointId();
long started = System.nanoTime();
- ChannelStateWriteResult channelStateWriteResult = includeChannelState(checkpointOptions) ?
+ ChannelStateWriteResult channelStateWriteResult = checkpointOptions.isUnalignedCheckpoint() ?
channelStateWriter.getAndRemoveWriteResult(checkpointId) :
ChannelStateWriteResult.EMPTY;
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 580ab54..f7655ea 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
@@ -27,6 +27,7 @@ import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.checkpoint.CheckpointType;
import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter;
import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriterImpl;
+import org.apache.flink.runtime.event.AbstractEvent;
import org.apache.flink.runtime.io.network.api.writer.NonRecordWriter;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.operators.testutils.DummyEnvironment;
@@ -58,6 +59,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.RunnableFuture;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import static org.apache.flink.runtime.checkpoint.CheckpointType.CHECKPOINT;
import static org.apache.flink.runtime.checkpoint.CheckpointType.SAVEPOINT;
@@ -75,7 +77,6 @@ public class SubtaskCheckpointCoordinatorTest {
@Test
public void testInitCheckpoint() throws IOException {
assertTrue(initCheckpoint(true, CHECKPOINT));
- assertFalse(initCheckpoint(true, SAVEPOINT));
assertFalse(initCheckpoint(false, CHECKPOINT));
assertFalse(initCheckpoint(false, SAVEPOINT));
}
@@ -91,7 +92,9 @@ public class SubtaskCheckpointCoordinatorTest {
MockWriter writer = new MockWriter();
SubtaskCheckpointCoordinator coordinator = coordinator(unalignedCheckpointEnabled, writer);
- coordinator.initCheckpoint(1L, new CheckpointOptions(checkpointType, CheckpointStorageLocationReference.getDefault()));
+ CheckpointStorageLocationReference locationReference = CheckpointStorageLocationReference.getDefault();
+ CheckpointOptions options = new CheckpointOptions(checkpointType, locationReference, true, unalignedCheckpointEnabled);
+ coordinator.initCheckpoint(1L, options);
return writer.started;
}
@@ -120,9 +123,39 @@ public class SubtaskCheckpointCoordinatorTest {
}
@Test
+ public void testSavepointNotResultingInPriorityEvents() throws Exception {
+ MockEnvironment mockEnvironment = MockEnvironment.builder().build();
+
+ SubtaskCheckpointCoordinator coordinator = new MockSubtaskCheckpointCoordinatorBuilder()
+ .setUnalignedCheckpointEnabled(true)
+ .setEnvironment(mockEnvironment)
+ .build();
+
+ AtomicReference<Boolean> broadcastedPriorityEvent = new AtomicReference<>(null);
+ final OperatorChain<?, ?> operatorChain = new OperatorChain(
+ new MockStreamTaskBuilder(mockEnvironment).build(),
+ new NonRecordWriter<>()) {
+ @Override
+ public void broadcastEvent(AbstractEvent event, boolean isPriorityEvent) throws IOException {
+ super.broadcastEvent(event, isPriorityEvent);
+ broadcastedPriorityEvent.set(isPriorityEvent);
+ }
+ };
+
+ coordinator.checkpointState(
+ new CheckpointMetaData(0, 0),
+ new CheckpointOptions(SAVEPOINT, CheckpointStorageLocationReference.getDefault()),
+ new CheckpointMetrics(),
+ operatorChain,
+ () -> false);
+
+ assertEquals(false, broadcastedPriorityEvent.get());
+ }
+
+ @Test
public void testSkipChannelStateForSavepoints() throws Exception {
SubtaskCheckpointCoordinator coordinator = new MockSubtaskCheckpointCoordinatorBuilder()
- .setUnalignedCheckpointEnabled(false)
+ .setUnalignedCheckpointEnabled(true)
.setPrepareInputSnapshot((u1, u2) -> {
fail("should not prepare input snapshot for savepoint");
return null;
@@ -414,7 +447,6 @@ public class SubtaskCheckpointCoordinatorTest {
newDirectExecutorService(),
new DummyEnvironment(),
(message, unused) -> fail(message),
- unalignedCheckpointEnabled,
(unused1, unused2) -> CompletableFuture.completedFuture(null),
0,
channelStateWriter