This is an automated email from the ASF dual-hosted git repository. 1996fanrui pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 98ab58c023ea8abeceab04108bd939e596a0b078 Author: Roman Khachatryan <[email protected]> AuthorDate: Sun Jun 28 22:28:01 2026 +0200 [FLINK-39522] Defer task finish until recovery completes A task that reaches END_OF_INPUT during recovery -- e.g. a bounded/batch operator whose input is already fully available behind a blocking exchange, as seen on TPC-H/TPC-DS HashAggregate -- previously suspended the restore mailbox loop immediately. With checkpointing during recovery the recovery future completes asynchronously, so this could end the restore loop before recovery finished, tripping checkState(allGatesRecoveredFuture.isDone()) in restoreInternal with "Mailbox loop interrupted before recovery was finished". Input is intentionally processed during recovery (so checkpoint barriers can arrive from inputs); the bug is only the premature lifecycle finish. When END_OF_INPUT is reached while recovery is still in progress, suspend only the default action (to avoid busy-spinning) and resume it once recovery completes, via a new recoveryCompletionFuture field set in restoreStateAndGates. processInput then re-fires END_OF_INPUT from the main mailbox loop and finishes normally. Sources remain covered by the no-input-gates short-circuit. Validated with a deterministic repro (bounded BATCH job, recovery window widened): the race reproduces with this deferral disabled and is gone with it enabled. Full regression green: heavy-deployment e2e 6/6; ChainOrderTest, KeyedComplexChainTest, KeyedPartitionWindowedStreamITCase; UnalignedCheckpoint {,Rescale}ITCase (the with-recovered-state path) 11/11 and 50/50. --- .../flink/streaming/runtime/tasks/StreamTask.java | 28 +++++++++++++++++++++- 1 file changed, 27 insertions(+), 1 deletion(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java index 5aec6b8bb44..6d776afb1cd 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java @@ -309,6 +309,14 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> /** TODO it might be replaced by the global IO executor on TaskManager level future. */ private final ExecutorService channelIOExecutor; + /** + * Completes when channel recovery has finished; set by {@link #restoreStateAndGates}. Used to + * defer the lifecycle finish of a task that reaches {@code END_OF_INPUT} during recovery (e.g. + * a bounded operator whose input is already fully available) so that it does not suspend the + * restore mailbox loop before recovery completes. Accessed only on the mailbox/task thread. + */ + @Nullable private CompletableFuture<Void> recoveryCompletionFuture; + // ======================================================== // Final checkpoint / savepoint // ======================================================== @@ -677,6 +685,24 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> notifyEndOfData(); return; case END_OF_INPUT: + if (recoveryCompletionFuture != null && !recoveryCompletionFuture.isDone()) { + // The task consumed all of its input during recovery (e.g. a bounded operator + // whose input is already fully available, as with a blocking batch exchange). + // Do NOT suspend the mailbox processor yet: recovery is still in progress and + // relies on this restore loop to run its remaining mails. Suspending here would + // end the restore loop early and trip "Mailbox loop interrupted before recovery + // was finished" in restoreInternal. Instead suspend only the default action (to + // avoid busy-spinning on repeated END_OF_INPUT) and resume it once recovery + // completes; processInput then re-fires END_OF_INPUT from the main mailbox loop + // and finishes through the normal path below. + MailboxDefaultAction.Suspension suspension = controller.suspendDefaultAction(); + recoveryCompletionFuture.whenComplete( + (ign, t) -> + mainMailboxExecutor.execute( + suspension::resume, + "resume default action after recovery")); + return; + } // Suspend the mailbox processor, it would be resumed in afterInvoke and finished // after all records processed by the downstream tasks. We also suspend the default // actions to avoid repeat executing the empty default operation (namely process @@ -892,7 +918,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> INITIALIZE_STATE_DURATION, initializeStateEndTs - readOutputDataTs); IndexedInputGate[] inputGates = getEnvironment().getAllInputGates(); - CompletableFuture<Void> recoveryCompletionFuture = + recoveryCompletionFuture = CheckpointingOptions.isCheckpointingDuringRecoveryEnabled(getJobConfiguration()) ? recoverChannelsWithCheckpointing(reader, inputGates) : recoverChannelsWithoutCheckpointing(reader, inputGates);
