Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r743366210
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SubtaskCheckpointCoordinatorImpl.java
##########
@@ -339,78 +343,107 @@ public void checkpointState(
public void notifyCheckpointComplete(
long checkpointId, OperatorChain<?, ?> operatorChain,
Supplier<Boolean> isRunning)
throws Exception {
- Exception previousException = null;
- if (!isRunning.get()) {
- LOG.debug(
- "Ignoring notification of complete checkpoint {} for
not-running task {}",
- checkpointId,
- taskName);
- } else if (operatorChain.isFinishedOnRestore()) {
- LOG.debug(
- "Ignoring notification of complete checkpoint {} for
finished on restore task {}",
- checkpointId,
- taskName);
- } else {
- LOG.debug(
- "Notification of completed checkpoint {} for task {}",
checkpointId, taskName);
- for (StreamOperatorWrapper<?, ?> operatorWrapper :
- operatorChain.getAllOperators(true)) {
- try {
- operatorWrapper.notifyCheckpointComplete(checkpointId);
- } catch (Exception e) {
- previousException = ExceptionUtils.firstOrSuppressed(e,
previousException);
- }
- }
- }
- env.getTaskStateManager().notifyCheckpointComplete(checkpointId);
- ExceptionUtils.tryRethrowException(previousException);
+ notifyCheckpoint(
+ checkpointId,
+ operatorChain,
+ isRunning,
+ "Ignoring notification of complete checkpoint {} for
not-running task {}",
+ "Ignoring notification of complete checkpoint {} for finished
on restore task {}",
+ "Notification of aborted checkpoint {} for task {}",
+ null,
+ (opw) -> opw.notifyCheckpointComplete(checkpointId),
+ (tsm) -> tsm.notifyCheckpointComplete(checkpointId));
}
@Override
public void notifyCheckpointAborted(
long checkpointId, OperatorChain<?, ?> operatorChain,
Supplier<Boolean> isRunning)
throws Exception {
+ notifyCheckpoint(
+ checkpointId,
+ operatorChain,
+ isRunning,
+ "Ignoring notification of aborted checkpoint {} for
not-running task {}",
+ "Ignoring notification of aborted checkpoint {} for finished
on restore task {}",
+ "Notification of aborted checkpoint {} for task {}",
+ () -> {
+ boolean canceled =
cancelAsyncCheckpointRunnable(checkpointId);
+
+ if (!canceled) {
+ if (checkpointId > lastCheckpointId) {
+ // only record checkpoints that have not triggered
on task side.
+ abortedCheckpointIds.add(checkpointId);
+ }
+ }
+
+ channelStateWriter.abort(
+ checkpointId,
+ new CancellationException("checkpoint aborted via
notification"),
+ false);
+ },
+ (opw) ->
opw.getStreamOperator().notifyCheckpointAborted(checkpointId),
+ (tsm) -> tsm.notifyCheckpointAborted(checkpointId));
+ }
+
+ @Override
+ public void notifyCheckpointSubsumed(
+ long checkpointId, OperatorChain<?, ?> operatorChain,
Supplier<Boolean> isRunning)
+ throws Exception {
+ notifyCheckpoint(
+ checkpointId,
+ operatorChain,
+ isRunning,
+ "Ignoring notification of subsumed checkpoint {} for
not-running task {}",
+ "Ignoring notification of subsumed checkpoint {} for finished
on restore task {}",
+ "Notification of subsumed checkpoint {} for task {}",
+ null,
+ (opw) -> opw.notifyCheckpointSubsumed(checkpointId),
+ null);
+ }
+
+ private void notifyCheckpoint(
+ long checkpointId,
+ OperatorChain<?, ?> operatorChain,
+ Supplier<Boolean> isRunning,
+ String logFormatIfNotRunning,
+ String logFormatIfFinishedOnRestore,
+ String logFormatIfNotificationStart,
+ @Nullable Runnable runnableBeforeNotifyOperator,
+ @Nonnull ThrowingConsumer<StreamOperatorWrapper<?, ?>, Exception>
operatorConsumer,
+ @Nullable ThrowingConsumer<TaskStateManager, Exception>
taskStateManagerConsumer)
+ throws Exception {
Review comment:
The refactor also take the `enum` of `NotifyCheckpointOperation` as
`Task` did.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
