1996fanrui commented on code in PR #28659:
URL: https://github.com/apache/flink/pull/28659#discussion_r3602775729
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -881,58 +918,120 @@ private CompletableFuture<Void> restoreStateAndGates(
INITIALIZE_STATE_DURATION, initializeStateEndTs -
readOutputDataTs);
IndexedInputGate[] inputGates = getEnvironment().getAllInputGates();
- boolean checkpointingDuringRecoveryEnabled =
-
CheckpointingOptions.isCheckpointingDuringRecoveryEnabled(getJobConfiguration());
+ recoveryCompletionFuture =
+
CheckpointingOptions.isCheckpointingDuringRecoveryEnabled(getJobConfiguration())
+ ? recoverChannelsWithCheckpointing(reader, inputGates)
+ : recoverChannelsWithoutCheckpointing(reader,
inputGates);
+
+ recoveryCompletionFuture.whenComplete((ign, throwable) ->
mailboxProcessor.suspend());
+ return recoveryCompletionFuture;
+ }
+
+ private CompletableFuture<Void> recoverChannelsWithCheckpointing(
+ SequentialChannelStateReader reader, IndexedInputGate[]
inputGates) {
+ if (inputGates.length == 0) {
+ // No input channels to recover (e.g. a source task). Complete
synchronously, exactly
+ // like recoverChannelsWithoutCheckpointing, so that the
recovery-completion suspend()
+ // in restoreStateAndGates is enqueued before the restore mailbox
loop runs the default
+ // action. The asynchronous chain below relies on mailbox mails
that only run while the
+ // restore loop is pumping; with no channel work to do it would
merely defer completion,
+ // letting a fast source finish and suspend the restore loop
before recovery completes
+ // -- tripping "Mailbox loop interrupted before recovery was
finished" in
+ // restoreInternal.
+ return FutureUtils.completedVoidFuture();
Review Comment:
droped
--
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]
