Myasuka commented on a change in pull request #16582:
URL: https://github.com/apache/flink/pull/16582#discussion_r742651490
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
##########
@@ -1352,84 +1352,73 @@ public void triggerCheckpointBarrier(
}
public void notifyCheckpointComplete(final long checkpointID) {
- final AbstractInvokable invokable = this.invokable;
-
- if (executionState == ExecutionState.RUNNING && invokable != null) {
- try {
- invokable.notifyCheckpointCompleteAsync(checkpointID);
- } catch (RejectedExecutionException ex) {
- // This may happen if the mailbox is closed. It means that the
task is shutting
- // down, so we just ignore it.
- LOG.debug(
- "Notify checkpoint complete {} for {} ({}) was
rejected by the mailbox",
- checkpointID,
- taskNameWithSubtask,
- executionId);
- } catch (Throwable t) {
- if (getExecutionState() == ExecutionState.RUNNING) {
- // fail task if checkpoint confirmation failed.
- failExternally(new RuntimeException("Error while
confirming checkpoint", t));
- }
- }
- } else {
- LOG.debug(
- "Ignoring checkpoint commit notification for non-running
task {}.",
- taskNameWithSubtask);
- }
+ notifyCheckpoint(
+ checkpointID,
+ (invokable) ->
invokable.notifyCheckpointCompleteAsync(checkpointID),
+ "Notify checkpoint complete {} for {} ({}) was rejected by the
mailbox",
+ "Error while confirming checkpoint {}.",
+ (t) -> failExternally(new RuntimeException("Error while
confirming checkpoint", t)),
+ "Ignoring checkpoint commit notification for non-running task
{}.");
}
public void notifyCheckpointAborted(
final long checkpointID, final long latestCompletedCheckpointId) {
- final AbstractInvokable invokable = this.invokable;
+ notifyCheckpoint(
+ checkpointID,
+ (invokable) ->
+ invokable.notifyCheckpointAbortAsync(
+ checkpointID, latestCompletedCheckpointId),
+ "Notify checkpoint abort {} for {} ({}) was rejected by the
mailbox",
+ "Error while aborting checkpoint {}.",
+ (t) -> failExternally(new RuntimeException("Error while
aborting checkpoint", t)),
+ "Ignoring checkpoint aborted notification for non-running task
{}.");
+ }
- if (executionState == ExecutionState.RUNNING && invokable != null) {
- try {
- invokable.notifyCheckpointAbortAsync(checkpointID,
latestCompletedCheckpointId);
- } catch (RejectedExecutionException ex) {
- // This may happen if the mailbox is closed. It means that the
task is shutting
- // down, so we just ignore it.
- LOG.debug(
- "Notify checkpoint abort {} for {} ({}) was rejected
by the mailbox",
- checkpointID,
- taskNameWithSubtask,
- executionId);
- } catch (Throwable t) {
- if (getExecutionState() == ExecutionState.RUNNING) {
- // fail task if checkpoint aborted notification failed.
- failExternally(new RuntimeException("Error while aborting
checkpoint", t));
- }
- }
- } else {
- LOG.info(
- "Ignoring checkpoint aborted notification for non-running
task {}.",
- taskNameWithSubtask);
- }
+ public void notifyCheckpointSubsumed(long checkpointID) {
+ notifyCheckpoint(
+ checkpointID,
+ (invokable) ->
invokable.notifyCheckpointSubsumedAsync(checkpointID),
+ "Notify checkpoint subsume {} for {} ({}) was rejected by the
mailbox",
+ "Error while subsuming checkpoint {}.",
+ null,
+ "Ignoring checkpoint subsume notification for non-running task
{}.");
}
- public void notifyCheckpointSubsumed(long checkpointId) {
+ private void notifyCheckpoint(
+ long checkpointId,
+ Consumer<AbstractInvokable> invokableNotifier,
+ String logFormatIfRejected,
+ String logFormatIfFailedToNotify,
+ @Nullable Consumer<Throwable> handleThrowableIfTaskRunning,
+ String logFormatIfNotRunning) {
Review comment:
Thanks for the suggestion of introducing `enum` of
`NotifyCheckpointOperation`, I think this could help reduce code duplication
and also make the code more readable.
--
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]
