You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "1996fanrui (via GitHub)" <gi...@apache.org> on 2026/07/06 15:20:23 UTC
[PR] [FLINK-39522] Rework StreamTask recovery: async future chain, consume only after filtering completes [flink]
1996fanrui opened a new pull request, #28659:
URL: https://github.com/apache/flink/pull/28659
This PR depends on https://github.com/apache/flink/pull/28651 and https://github.com/apache/flink/pull/28652
Please only review the last 3 commits (prefixed [FLINK-39522]).
## What is the purpose of the change
[FLINK-39522] Rework StreamTask recovery: async future chain, consume only after filtering completes.
It's part 3 of FLINK-38544 (checkpointing during recovery of unaligned checkpoint state). StreamTask channel-state recovery is restructured into an asynchronous future chain on the channelIOExecutor, and recovery completion is unified on the gates' state-consumed futures. With checkpointing-during-recovery enabled, the task now consumes recovered state only after rescale filtering has fully completed (previously consumption overlapped with filtering); the filtered buffers are held in the recovered channels' memory queues and handed to the physical channels through the push interface at conversion. The now-unused filtering-progress gate API is removed. Non-CDR observable behavior is preserved; the defer-finish commit fixes an existing race where a finite task could finish before recovery completed. Checkpoints during recovery are declined at this stage.
## Brief change log
- [FLINK-39522] Restructure StreamTask channel-state recovery into an async future chain
- [FLINK-39522][network] Remove recovery flags and the filtering-complete future from the gate API
- [FLINK-39522] Defer task finish until recovery completes
## Verifying this change
- Existing and adapted tests: TaskCheckpointingBehaviourTest, StreamTaskTest, MultipleInputStreamTaskTest, SingleInputGateTest, UnionInputGateTest, RecoveredInputChannelTest, SequentialChannelStateReaderImplTest
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39522] Rework StreamTask recovery: async future chain, consume only after filtering completes [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #28659:
URL: https://github.com/apache/flink/pull/28659#issuecomment-4894612327
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "18c128c597adf07b7b37fde059b6e64082649980",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "18c128c597adf07b7b37fde059b6e64082649980",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 18c128c597adf07b7b37fde059b6e64082649980 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39522] Rework StreamTask recovery: async future chain, consume only after filtering completes [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #28659:
URL: https://github.com/apache/flink/pull/28659#discussion_r3598709891
##########
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:
NIT: drop the comment?
##########
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();
+ }
+ // FLINK-38544 transitional: removed when the spilling backend lands. In-memory backend:
+ // readInputChannelState first filters the recovered state completely (the filter output
+ // accumulates in the recovered channels' in-memory queues), then conversion pushes it into
+ // the physical channels through the push interface (requestPartitions(true)). The final
+ // form instead fetches the filter output to disk and drains it incrementally from here.
+ return CompletableFuture.runAsync(
+ () -> readInputChannelState(reader, inputGates), channelIOExecutor)
+ .thenCompose(ign -> requestPartitions(inputGates, true))
+ .thenCompose(
+ ign ->
+ completeAll(
+ Arrays.stream(inputGates)
+ .map(InputGate::getStateConsumedFuture)
+ .collect(Collectors.toList())));
+ }
+
+ private CompletableFuture<Void> recoverChannelsWithoutCheckpointing(
+ SequentialChannelStateReader reader, IndexedInputGate[] inputGates) {
+ // Feed recovered channel state on the IO thread. This is intentionally NOT part of the
+ // completion gate below: a gate's stateConsumedFuture only completes once the consumer (the
+ // default action, running in the restore mailbox loop) has drained the end-of-state
+ // sentinel that readInputChannelState pushes, so gating on stateConsumedFuture already
+ // implies the read has finished. Including the read future in the gate would defer
+ // suspend() past the restore loop's first default action, letting input be processed before
+ // recovery completes -- causing record loss and "Mailbox loop interrupted before recovery
+ // was finished". Read failures are surfaced via asyncExceptionHandler inside
+ // readInputChannelState.
+ channelIOExecutor.execute(() -> readInputChannelState(reader, inputGates));
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ for (InputGate inputGate : inputGates) {
+ CompletableFuture<Void> stateConsumed = inputGate.getStateConsumedFuture();
+ futures.add(stateConsumed);
+ // Convert and request partitions for each gate as soon as ITS OWN recovered state is
+ // drained, independent of the other gates. Deferring conversion until every gate has
+ // drained deadlocks selective-reading multi-input operators: such an operator only
+ // drains the selected input's end-of-state sentinel, so an unselected gate would never
+ // drain (it is read only after conversion) while conversion would wait for it to drain
+ // first. suspend() is still gated on completeAll(futures) below.
+ stateConsumed.thenRun(
+ () ->
+ mainMailboxExecutor.execute(
+ () -> inputGate.requestPartitions(false),
+ "Input gate request partitions"));
+ }
+ return completeAll(futures);
+ }
- // Must set the flag on input gates BEFORE starting the async read task, because
- // finishReadRecoveredState() checks this flag to complete bufferFilteringCompleteFuture.
- for (IndexedInputGate inputGate : inputGates) {
- inputGate.setCheckpointingDuringRecoveryEnabled(checkpointingDuringRecoveryEnabled);
+ private void readInputChannelState(
+ SequentialChannelStateReader reader, IndexedInputGate[] inputGates) {
+ try {
+ reader.readInputData(inputGates, createRecordFilterContext());
+
+ for (IndexedInputGate gate : inputGates) {
+ gate.finishReadRecoveredState(); // this is called from IO thread - is that fine?
Review Comment:
I think so :)
##########
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());
Review Comment:
I don't remember if this change from `thenRun` to `whenComplete` was strictly necessary.
However, I don't see any downsides of using `whenComplete` here either.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39522] Rework StreamTask recovery: async future chain, consume only after filtering completes [flink]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39522] Rework StreamTask recovery: async future chain, consume only after filtering completes [flink]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #28659:
URL: https://github.com/apache/flink/pull/28659#discussion_r3602777054
##########
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();
+ }
+ // FLINK-38544 transitional: removed when the spilling backend lands. In-memory backend:
+ // readInputChannelState first filters the recovered state completely (the filter output
+ // accumulates in the recovered channels' in-memory queues), then conversion pushes it into
+ // the physical channels through the push interface (requestPartitions(true)). The final
+ // form instead fetches the filter output to disk and drains it incrementally from here.
+ return CompletableFuture.runAsync(
+ () -> readInputChannelState(reader, inputGates), channelIOExecutor)
+ .thenCompose(ign -> requestPartitions(inputGates, true))
+ .thenCompose(
+ ign ->
+ completeAll(
+ Arrays.stream(inputGates)
+ .map(InputGate::getStateConsumedFuture)
+ .collect(Collectors.toList())));
+ }
+
+ private CompletableFuture<Void> recoverChannelsWithoutCheckpointing(
+ SequentialChannelStateReader reader, IndexedInputGate[] inputGates) {
+ // Feed recovered channel state on the IO thread. This is intentionally NOT part of the
+ // completion gate below: a gate's stateConsumedFuture only completes once the consumer (the
+ // default action, running in the restore mailbox loop) has drained the end-of-state
+ // sentinel that readInputChannelState pushes, so gating on stateConsumedFuture already
+ // implies the read has finished. Including the read future in the gate would defer
+ // suspend() past the restore loop's first default action, letting input be processed before
+ // recovery completes -- causing record loss and "Mailbox loop interrupted before recovery
+ // was finished". Read failures are surfaced via asyncExceptionHandler inside
+ // readInputChannelState.
+ channelIOExecutor.execute(() -> readInputChannelState(reader, inputGates));
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ for (InputGate inputGate : inputGates) {
+ CompletableFuture<Void> stateConsumed = inputGate.getStateConsumedFuture();
+ futures.add(stateConsumed);
+ // Convert and request partitions for each gate as soon as ITS OWN recovered state is
+ // drained, independent of the other gates. Deferring conversion until every gate has
+ // drained deadlocks selective-reading multi-input operators: such an operator only
+ // drains the selected input's end-of-state sentinel, so an unselected gate would never
+ // drain (it is read only after conversion) while conversion would wait for it to drain
+ // first. suspend() is still gated on completeAll(futures) below.
+ stateConsumed.thenRun(
+ () ->
+ mainMailboxExecutor.execute(
+ () -> inputGate.requestPartitions(false),
+ "Input gate request partitions"));
+ }
+ return completeAll(futures);
+ }
- // Must set the flag on input gates BEFORE starting the async read task, because
- // finishReadRecoveredState() checks this flag to complete bufferFilteringCompleteFuture.
- for (IndexedInputGate inputGate : inputGates) {
- inputGate.setCheckpointingDuringRecoveryEnabled(checkpointingDuringRecoveryEnabled);
+ private void readInputChannelState(
+ SequentialChannelStateReader reader, IndexedInputGate[] inputGates) {
+ try {
+ reader.readInputData(inputGates, createRecordFilterContext());
+
+ for (IndexedInputGate gate : inputGates) {
+ gate.finishReadRecoveredState(); // this is called from IO thread - is that fine?
Review Comment:
comment is removed.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39522] Rework StreamTask recovery: async future chain, consume only after filtering completes [flink]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #28659:
URL: https://github.com/apache/flink/pull/28659#discussion_r3602817583
##########
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());
Review Comment:
It's necessary rather than just harmless: the recovery chain can complete exceptionally (e.g. `requestPartitions(...)` does `future.completeExceptionally(t)`, and `completeAll(stateConsumedFutures)` propagates gate failures). We need `suspend()` to run in that case too so the failure is surfaced via `get()` in `restoreInternal`. `thenRun` skips the exceptional path, `whenComplete` doesn't.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39522] Rework StreamTask recovery: async future chain, consume only after filtering completes [flink]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui merged PR #28659:
URL: https://github.com/apache/flink/pull/28659
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org