You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by dw...@apache.org on 2021/08/03 11:22:07 UTC
[flink] 02/04: [FLINK-21080][runtime][checkpoint] The task reports
whether it has called operators' finish method
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 66bf4e6891a0cb271ca940d3814db83ca51164d5
Author: Yun Gao <ga...@gmail.com>
AuthorDate: Wed Jul 28 13:46:47 2021 +0800
[FLINK-21080][runtime][checkpoint] The task reports whether it has called operators' finish method
---
.../checkpoint/StateAssignmentOperation.java | 2 +-
.../runtime/checkpoint/TaskStateSnapshot.java | 28 +++++--
.../runtime/state/TaskLocalStateStoreImpl.java | 2 +-
.../runtime/tasks/AsyncCheckpointRunnable.java | 7 +-
.../flink/streaming/runtime/tasks/StreamTask.java | 5 ++
.../tasks/SubtaskCheckpointCoordinator.java | 1 +
.../tasks/SubtaskCheckpointCoordinatorImpl.java | 4 +
.../runtime/tasks/AsyncCheckpointRunnableTest.java | 2 +
.../runtime/tasks/LocalStateForwardingTest.java | 1 +
.../tasks/StreamTaskFinalCheckpointsTest.java | 94 ++++++++++++++++++++++
.../tasks/SubtaskCheckpointCoordinatorTest.java | 7 ++
.../tasks/TestSubtaskCheckpointCoordinator.java | 1 +
12 files changed, 142 insertions(+), 12 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateAssignmentOperation.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateAssignmentOperation.java
index 4995071..fb23b0a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateAssignmentOperation.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateAssignmentOperation.java
@@ -237,7 +237,7 @@ public class StateAssignmentOperation {
List<OperatorIDPair> operatorIDs,
int subTaskIndex,
Execution currentExecutionAttempt) {
- TaskStateSnapshot taskState = new TaskStateSnapshot(operatorIDs.size());
+ TaskStateSnapshot taskState = new TaskStateSnapshot(operatorIDs.size(), false);
boolean statelessTask = true;
for (OperatorIDPair operatorID : operatorIDs) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/TaskStateSnapshot.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/TaskStateSnapshot.java
index ce3777c..aa65c98 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/TaskStateSnapshot.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/TaskStateSnapshot.java
@@ -58,30 +58,34 @@ public class TaskStateSnapshot implements CompositeStateHandle {
private static final long serialVersionUID = 1L;
public static final TaskStateSnapshot FINISHED_ON_RESTORE =
- new TaskStateSnapshot(new HashMap<>(), true);
+ new TaskStateSnapshot(new HashMap<>(), true, true);
/** Mapping from an operator id to the state of one subtask of this operator. */
private final Map<OperatorID, OperatorSubtaskState> subtaskStatesByOperatorID;
private final boolean isFinishedOnRestore;
+ private final boolean isOperatorsFinished;
+
public TaskStateSnapshot() {
- this(10);
+ this(10, false);
}
- public TaskStateSnapshot(int size) {
- this(new HashMap<>(size));
+ public TaskStateSnapshot(int size, boolean isOperatorsFinished) {
+ this(new HashMap<>(size), false, isOperatorsFinished);
}
public TaskStateSnapshot(Map<OperatorID, OperatorSubtaskState> subtaskStatesByOperatorID) {
- this(subtaskStatesByOperatorID, false);
+ this(subtaskStatesByOperatorID, false, false);
}
private TaskStateSnapshot(
Map<OperatorID, OperatorSubtaskState> subtaskStatesByOperatorID,
- boolean isFinishedOnRestore) {
+ boolean isFinishedOnRestore,
+ boolean isOperatorsFinished) {
this.subtaskStatesByOperatorID = Preconditions.checkNotNull(subtaskStatesByOperatorID);
this.isFinishedOnRestore = isFinishedOnRestore;
+ this.isOperatorsFinished = isOperatorsFinished;
}
/** Returns whether all the operators of the task are already finished on restoring. */
@@ -89,6 +93,11 @@ public class TaskStateSnapshot implements CompositeStateHandle {
return isFinishedOnRestore;
}
+ /** Returns whether all the operators of the task have called finished methods. */
+ public boolean isOperatorsFinished() {
+ return isOperatorsFinished;
+ }
+
/** Returns the subtask state for the given operator id (or null if not contained). */
@Nullable
public OperatorSubtaskState getSubtaskStateByOperatorID(OperatorID operatorID) {
@@ -178,12 +187,13 @@ public class TaskStateSnapshot implements CompositeStateHandle {
TaskStateSnapshot that = (TaskStateSnapshot) o;
return subtaskStatesByOperatorID.equals(that.subtaskStatesByOperatorID)
- && isFinishedOnRestore == that.isFinishedOnRestore;
+ && isFinishedOnRestore == that.isFinishedOnRestore
+ && isOperatorsFinished == that.isOperatorsFinished;
}
@Override
public int hashCode() {
- return Objects.hash(subtaskStatesByOperatorID, isFinishedOnRestore);
+ return Objects.hash(subtaskStatesByOperatorID, isFinishedOnRestore, isOperatorsFinished);
}
@Override
@@ -193,6 +203,8 @@ public class TaskStateSnapshot implements CompositeStateHandle {
+ subtaskStatesByOperatorID
+ ", isFinished="
+ isFinishedOnRestore
+ + ", isOperatorsFinished="
+ + isOperatorsFinished
+ '}';
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/TaskLocalStateStoreImpl.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/TaskLocalStateStoreImpl.java
index a7f8ffb..dd9bcee 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/TaskLocalStateStoreImpl.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/TaskLocalStateStoreImpl.java
@@ -57,7 +57,7 @@ public class TaskLocalStateStoreImpl implements OwnedTaskLocalStateStore {
private static final Logger LOG = LoggerFactory.getLogger(TaskLocalStateStoreImpl.class);
/** Dummy value to use instead of null to satisfy {@link ConcurrentHashMap}. */
- @VisibleForTesting static final TaskStateSnapshot NULL_DUMMY = new TaskStateSnapshot(0);
+ @VisibleForTesting static final TaskStateSnapshot NULL_DUMMY = new TaskStateSnapshot(0, false);
/** JobID from the owning subtask. */
@Nonnull private final JobID jobID;
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 6f72e3f..c077914 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
@@ -53,6 +53,7 @@ final class AsyncCheckpointRunnable implements Runnable, Closeable {
private final String taskName;
private final Consumer<AsyncCheckpointRunnable> unregisterConsumer;
private final boolean isFinishedOnRestore;
+ private final boolean isOperatorsFinished;
private final Supplier<Boolean> isTaskRunning;
private final Environment taskEnvironment;
private final CompletableFuture<Void> finishedFuture = new CompletableFuture<>();
@@ -85,6 +86,7 @@ final class AsyncCheckpointRunnable implements Runnable, Closeable {
Environment taskEnvironment,
AsyncExceptionHandler asyncExceptionHandler,
boolean isFinishedOnRestore,
+ boolean isOperatorsFinished,
Supplier<Boolean> isTaskRunning) {
this.operatorSnapshotsInProgress = checkNotNull(operatorSnapshotsInProgress);
@@ -96,6 +98,7 @@ final class AsyncCheckpointRunnable implements Runnable, Closeable {
this.taskEnvironment = checkNotNull(taskEnvironment);
this.asyncExceptionHandler = checkNotNull(asyncExceptionHandler);
this.isFinishedOnRestore = isFinishedOnRestore;
+ this.isOperatorsFinished = isOperatorsFinished;
this.isTaskRunning = isTaskRunning;
}
@@ -159,9 +162,9 @@ final class AsyncCheckpointRunnable implements Runnable, Closeable {
private SnapshotsFinalizeResult finalizeNonFinishedSnapshots() throws Exception {
TaskStateSnapshot jobManagerTaskOperatorSubtaskStates =
- new TaskStateSnapshot(operatorSnapshotsInProgress.size());
+ new TaskStateSnapshot(operatorSnapshotsInProgress.size(), isOperatorsFinished);
TaskStateSnapshot localTaskOperatorSubtaskStates =
- new TaskStateSnapshot(operatorSnapshotsInProgress.size());
+ new TaskStateSnapshot(operatorSnapshotsInProgress.size(), isOperatorsFinished);
long bytesPersistedDuringAlignment = 0;
for (Map.Entry<OperatorID, OperatorSnapshotFutures> entry :
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 b8afd95..3345913 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
@@ -241,6 +241,9 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> extends Ab
*/
private volatile boolean failing;
+ /** Flags indicating the finished method of all the operators are called. */
+ private boolean finishedOperators;
+
private boolean closedOperators;
/** Thread pool for async snapshot workers. */
@@ -741,6 +744,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> extends Ab
// close all operators in a chain effect way
operatorChain.finishOperators(actionExecutor);
+ finishedOperators = true;
// If checkpoints are enabled, waits for all the records get processed by the downstream
// tasks. During this process, this task could coordinate with its downstream tasks to
@@ -1211,6 +1215,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> extends Ab
checkpointOptions,
checkpointMetrics,
operatorChain,
+ finishedOperators,
this::isRunning);
});
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinator.java
index b9e56a3..c786f0a 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinator.java
@@ -61,6 +61,7 @@ public interface SubtaskCheckpointCoordinator extends Closeable {
CheckpointOptions checkpointOptions,
CheckpointMetricsBuilder checkpointMetrics,
OperatorChain<?, ?> operatorChain,
+ boolean isOperatorsFinished,
Supplier<Boolean> isRunning)
throws Exception;
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 d96d3b0..16c0f04 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
@@ -250,6 +250,7 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
CheckpointOptions options,
CheckpointMetricsBuilder metrics,
OperatorChain<?, ?> operatorChain,
+ boolean isOperatorsFinished,
Supplier<Boolean> isRunning)
throws Exception {
@@ -323,6 +324,7 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
metadata,
metrics,
operatorChain.isFinishedOnRestore(),
+ isOperatorsFinished,
isRunning);
} else {
cleanup(snapshotFutures, metadata, metrics, new Exception("Checkpoint declined"));
@@ -568,6 +570,7 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
CheckpointMetaData metadata,
CheckpointMetricsBuilder metrics,
boolean isFinishedOnRestore,
+ boolean isOperatorsFinished,
Supplier<Boolean> isRunning)
throws IOException {
AsyncCheckpointRunnable asyncCheckpointRunnable =
@@ -581,6 +584,7 @@ class SubtaskCheckpointCoordinatorImpl implements SubtaskCheckpointCoordinator {
env,
asyncExceptionHandler,
isFinishedOnRestore,
+ isOperatorsFinished,
isRunning);
registerAsyncCheckpointRunnable(
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 e8a1090..8725b65 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
@@ -59,6 +59,7 @@ public class AsyncCheckpointRunnableTest {
env,
(msg, ex) -> {},
false,
+ false,
() -> true)
.close();
assertEquals(
@@ -165,6 +166,7 @@ public class AsyncCheckpointRunnableTest {
environment,
(msg, ex) -> {},
isFinishedOnRestore,
+ false,
() -> isTaskRunning);
}
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 f296f4b..1991d5d 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
@@ -128,6 +128,7 @@ public class LocalStateForwardingTest extends TestLogger {
testStreamTask.getEnvironment(),
testStreamTask,
false,
+ false,
() -> true);
checkpointMetrics.setAlignmentDurationNanos(0L);
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskFinalCheckpointsTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskFinalCheckpointsTest.java
index 2a05a0a..8f342ff 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskFinalCheckpointsTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskFinalCheckpointsTest.java
@@ -19,6 +19,8 @@
package org.apache.flink.streaming.runtime.tasks;
import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeutils.base.StringSerializer;
import org.apache.flink.core.testutils.OneShotLatch;
@@ -203,6 +205,81 @@ public class StreamTaskFinalCheckpointsTest {
}
}
+ @Test
+ public void testReportOperatorsFinishedInCheckpoint() throws Exception {
+ ResultPartition[] partitionWriters = new ResultPartition[2];
+ try {
+ for (int i = 0; i < partitionWriters.length; ++i) {
+ partitionWriters[i] =
+ PartitionTestUtils.createPartition(ResultPartitionType.PIPELINED_BOUNDED);
+ partitionWriters[i].setup();
+ }
+
+ try (StreamTaskMailboxTestHarness<String> testHarness =
+ new StreamTaskMailboxTestHarnessBuilder<>(
+ OneInputStreamTask::new, BasicTypeInfo.STRING_TYPE_INFO)
+ .addInput(BasicTypeInfo.STRING_TYPE_INFO, 1)
+ .addAdditionalOutput(partitionWriters)
+ .modifyStreamConfig(
+ config -> {
+ config.setCheckpointingEnabled(true);
+ config.getConfiguration()
+ .set(
+ ExecutionCheckpointingOptions
+ .ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH,
+ true);
+ })
+ .setupOperatorChain(new StatefulOperator())
+ .finishForSingletonOperatorChain(StringSerializer.INSTANCE)
+ .build()) {
+
+ // Trigger the first checkpoint before we call operators' finish method.
+ Future<Boolean> checkpointFuture = triggerCheckpoint(testHarness, 2);
+ processMailTillCheckpointSucceeds(testHarness, checkpointFuture);
+ assertEquals(2, testHarness.getTaskStateManager().getReportedCheckpointId());
+ assertFalse(
+ testHarness
+ .getTaskStateManager()
+ .getJobManagerTaskStateSnapshotsByCheckpointId()
+ .get(2L)
+ .isOperatorsFinished());
+
+ // Trigger the first checkpoint after we call operators' finish method.
+ // The checkpoint is added to the mailbox and will be processed in the
+ // mailbox loop after call operators' finish method in the afterInvoke()
+ // method.
+ testHarness.processEvent(EndOfPartitionEvent.INSTANCE, 0, 0);
+ checkpointFuture = triggerCheckpoint(testHarness, 4);
+ checkState(
+ checkpointFuture instanceof CompletableFuture,
+ "The trigger future should " + " be also CompletableFuture.");
+ ((CompletableFuture<?>) checkpointFuture)
+ .thenAccept(
+ (ignored) -> {
+ for (ResultPartition resultPartition : partitionWriters) {
+ resultPartition.onSubpartitionAllRecordsProcessed(0);
+ }
+ });
+ testHarness.finishProcessing();
+ assertTrue(checkpointFuture.isDone());
+ testHarness.getTaskStateManager().getWaitForReportLatch().await();
+ assertTrue(
+ testHarness
+ .getTaskStateManager()
+ .getJobManagerTaskStateSnapshotsByCheckpointId()
+ .get(4L)
+ .isOperatorsFinished());
+ }
+
+ } finally {
+ for (ResultPartitionWriter writer : partitionWriters) {
+ if (writer != null) {
+ writer.close();
+ }
+ }
+ }
+ }
+
static Future<Boolean> triggerCheckpoint(
StreamTaskMailboxTestHarness<String> testHarness, long checkpointId) {
testHarness.getTaskStateManager().getWaitForReportLatch().reset();
@@ -417,4 +494,21 @@ public class StreamTaskFinalCheckpointsTest {
return lifeCycleMonitor;
}
}
+
+ private static class StatefulOperator extends AbstractStreamOperator<String>
+ implements OneInputStreamOperator<String, String> {
+
+ private ListState<Integer> state;
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ state =
+ context.getOperatorStateStore()
+ .getUnionListState(new ListStateDescriptor<>("test", Integer.class));
+ }
+
+ @Override
+ public void processElement(StreamRecord<String> element) throws Exception {}
+ }
}
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 1614a3f..b443a36 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
@@ -176,6 +176,7 @@ public class SubtaskCheckpointCoordinatorTest {
SAVEPOINT, CheckpointStorageLocationReference.getDefault()),
new CheckpointMetricsBuilder(),
operatorChain,
+ false,
() -> true);
assertEquals(false, broadcastedPriorityEvent.get());
@@ -223,6 +224,7 @@ public class SubtaskCheckpointCoordinatorTest {
forcedAlignedOptions,
new CheckpointMetricsBuilder(),
operatorChain,
+ false,
() -> true);
assertEquals(true, broadcastedPriorityEvent.get());
@@ -249,6 +251,7 @@ public class SubtaskCheckpointCoordinatorTest {
new NoOpStreamTask<>(new DummyEnvironment()),
new NonRecordWriter<>(),
false),
+ false,
() -> true);
}
}
@@ -308,6 +311,7 @@ public class SubtaskCheckpointCoordinatorTest {
CheckpointOptions.forCheckpointWithDefaultLocation(),
new CheckpointMetricsBuilder(),
operatorChain,
+ false,
() -> false);
assertFalse(checkpointOperator.isCheckpointed());
assertEquals(-1, stateManager.getReportedCheckpointId());
@@ -362,6 +366,7 @@ public class SubtaskCheckpointCoordinatorTest {
CheckpointOptions.forCheckpointWithDefaultLocation(),
new CheckpointMetricsBuilder(),
operatorChain,
+ false,
() -> false);
assertEquals(1, recordOrEvents.size());
@@ -418,6 +423,7 @@ public class SubtaskCheckpointCoordinatorTest {
CheckpointOptions.forCheckpointWithDefaultLocation(),
new CheckpointMetricsBuilder(),
operatorChain,
+ false,
() -> false);
rawKeyedStateHandleFuture.awaitRun();
assertEquals(1, subtaskCheckpointCoordinator.getAsyncCheckpointRunnableSize());
@@ -449,6 +455,7 @@ public class SubtaskCheckpointCoordinatorTest {
CheckpointOptions.forCheckpointWithDefaultLocation(),
new CheckpointMetricsBuilder(),
operatorChain,
+ false,
() -> false);
subtaskCheckpointCoordinator.notifyCheckpointAborted(
checkpointId, operatorChain, () -> true);
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 e8ceb37..dc83d84 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
@@ -79,6 +79,7 @@ public class TestSubtaskCheckpointCoordinator implements SubtaskCheckpointCoordi
CheckpointOptions checkpointOptions,
CheckpointMetricsBuilder checkpointMetrics,
OperatorChain<?, ?> operatorChain,
+ boolean isOperatorsFinished,
Supplier<Boolean> isRunning) {}
@Override