[FLINK-39018][checkpoint] Support checkpoint for LocalInputChannel by 1996fanrui · Pull Request #27861 · apache/flink

1996fanrui · 2026-03-31T12:20:04Z

This PR depends on #27782 and #27783

What is the purpose of the change

[FLINK-39018][checkpoint] Support checkpoint for LocalInputChannel

Brief change log

[hotfix][network] Fix LocalInputChannel.getBuffersInUseCount to include toBeConsumedBuffers
[FLINK-39018][checkpoint] Support LocalInputChannel checkpoint snapshot for recovered buffers
[FLINK-39018][network] Fix LocalInputChannel priority event and buffer availability for recovered buffers
[FLINK-39018][checkpoint] Notify PriorityEvent to downstream task even if it is blocked to ensure the checkpoint barrier can be handled by downstream task
[FLINK-39018][network] Buffer migration from RecoveredInputChannel to physical channels

Verifying this change

Tons of unit tests

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

flinkbot · 2026-03-31T12:29:25Z

CI report:

f08a818 Azure: FAILURE
bb071b2 Azure: PENDING

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

…de toBeConsumedBuffers

…ot for recovered buffers

.../src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java

...ntime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java

...ava/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionWithReadViewTest.java

1996fanrui · 2026-04-01T22:32:33Z

...ntime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java

+        // Priority events (e.g. unaligned checkpoint barriers) must notify downstream even
+        // when the subpartition is blocked.
+        //
+        // During recovery, once the upstream output channel state is fully restored, a
+        // RECOVERY_COMPLETION event (EndOfOutputChannelStateEvent) is emitted. This event
+        // blocks the subpartition to prevent the upstream from sending new data while the
+        // downstream is still consuming recovered buffers. The subpartition remains blocked
+        // until the downstream finishes consuming all recovered buffers from every channel
+        // and calls resumeConsumption() to unblock.
+        //
+        // If a checkpoint is triggered while the downstream is still consuming recovered
+        // buffers, the upstream receives an unaligned checkpoint barrier and adds it to this
+        // blocked subpartition. The barrier must still be delivered to the downstream
+        // immediately, otherwise the checkpoint will hang until it times out.
+        return buffers.getNumPriorityElements() == 1;


Explaining why Priority events (e.g. unaligned checkpoint barriers) must notify downstream even when the subpartition is blocked.

...test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannelTest.java

pnowojski · 2026-04-02T09:26:02Z

...me/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java

+     * <p><b>Lock ordering note:</b> This method acquires {@code inputChannelsWithData} and then may
+     * indirectly acquire {@code receivedBuffers} (via {@code toInputChannel()} and {@code
+     * releaseAllResources()}). This is the reverse order of {@link
+     * RecoveredInputChannel#onRecoveredStateBuffer}, which acquires {@code receivedBuffers} first
+     * and then {@code inputChannelsWithData} (via {@code notifyChannelNonEmpty()}). This is safe
+     * because {@code convertRecoveredInputChannels()} is only called from {@link
+     * #requestPartitions()}, which happens after all state recovery is complete (buffer filtering
+     * future is done), so {@code onRecoveredStateBuffer()} is no longer being called concurrently.


This sounds fishy and fragile 🤔

Good point. I've narrowed the inputChannelsWithData lock scope in convertRecoveredInputChannels() — moved toInputChannel(), releaseAllResources(), and getBuffersInUseCount() outside the synchronized block to eliminate the reverse lock ordering. The lock now only covers the data structure updates.

.../src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java

…r availability for recovered buffers

…n if it is blocked to ensure the checkpoint barrier can be handled by downstream task Priority events (e.g. unaligned checkpoint barriers) must notify downstream even when the subpartition is blocked. During recovery, once the upstream output channel state is fully restored, a RECOVERY_COMPLETION event (EndOfOutputChannelStateEvent) is emitted. This event blocks the subpartition to prevent the upstream from sending new data while the downstream is still consuming recovered buffers. The subpartition remains blocked until the downstream finishes consuming all recovered buffers from every channel and calls resumeConsumption() to unblock. If a checkpoint is triggered while the downstream is still consuming recovered buffers, the upstream receives an unaligned checkpoint barrier and adds it to this blocked subpartition. The barrier must still be delivered to the downstream immediately, otherwise the checkpoint will hang until it times out.

1996fanrui

Thanks @pnowojski for the review, all comments are addressed, and commits are organized.

… physical channels

.../src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java

pnowojski · 2026-04-03T13:17:52Z

...me/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java

+                    int buffersInUseCount = realInputChannel.getBuffersInUseCount();
+
+                    // Phase 2: Atomically update data structures under the lock.
+                    synchronized (inputChannelsWithData) {


Why the previous code didn't need the lock in the first place? What has changed or what did I miss?

The previous(master branch) code didn't touch inputChannelsWithData at all — it was a simple channel swap without any lock. FLINK-39018 changes introduced buffer migration from RecoveredInputChannel to physical channels, where onRecoveredStateBuffer() now enqueues the channel into inputChannelsWithData via notifyChannelNonEmpty(). So during conversion, we now need to dequeue the old channel and conditionally enqueue the new one, which requires the inputChannelsWithData lock. I've structured it so the lock only covers the queue/map updates (phase 2), while toInputChannel() and releaseAllResources() run outside the lock (phase 1) to avoid reverse lock ordering with onRecoveredStateBuffer().

1996fanrui mentioned this pull request Mar 31, 2026

[FLINK-38543] Change the overall UC restore process, JM and task initialization #27862

Open

1996fanrui force-pushed the 39018/support-checkpoint-for-localinputchannel branch from cf606db to 4fa25ef Compare March 31, 2026 16:36

1996fanrui added 2 commits March 31, 2026 20:35

[hotfix][network] Fix LocalInputChannel.getBuffersInUseCount to inclu…

7ced5fb

…de toBeConsumedBuffers

[FLINK-39018][checkpoint] Support LocalInputChannel checkpoint snapsh…

7c9a38d

…ot for recovered buffers

1996fanrui force-pushed the 39018/support-checkpoint-for-localinputchannel branch from 4fa25ef to b1a7ca7 Compare March 31, 2026 18:37

pnowojski reviewed Apr 1, 2026

View reviewed changes

1996fanrui commented Apr 1, 2026

View reviewed changes

pnowojski reviewed Apr 2, 2026

View reviewed changes

1996fanrui added 2 commits April 2, 2026 17:15

[FLINK-39018][network] Fix LocalInputChannel priority event and buffe…

16fbdbf

…r availability for recovered buffers

1996fanrui force-pushed the 39018/support-checkpoint-for-localinputchannel branch 4 times, most recently from fef5732 to f08a818 Compare April 2, 2026 20:33

1996fanrui commented Apr 2, 2026

View reviewed changes

[FLINK-39018][network] Buffer migration from RecoveredInputChannel to…

bb071b2

… physical channels

1996fanrui force-pushed the 39018/support-checkpoint-for-localinputchannel branch from f08a818 to bb071b2 Compare April 3, 2026 12:45

pnowojski reviewed Apr 3, 2026

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-39018][checkpoint] Support checkpoint for LocalInputChannel#27861

[FLINK-39018][checkpoint] Support checkpoint for LocalInputChannel#27861
1996fanrui wants to merge 5 commits intoapache:masterfrom
1996fanrui:39018/support-checkpoint-for-localinputchannel

1996fanrui commented Mar 31, 2026 •

edited

Loading

Uh oh!

flinkbot commented Mar 31, 2026 •

edited

Loading

Uh oh!

Uh oh!

Uh oh!

Uh oh!

1996fanrui Apr 1, 2026

Uh oh!

Uh oh!

pnowojski Apr 2, 2026

Uh oh!

1996fanrui Apr 2, 2026

Uh oh!

Uh oh!

Uh oh!

1996fanrui left a comment •

edited

Loading

Uh oh!

Uh oh!

pnowojski Apr 3, 2026

Uh oh!

1996fanrui Apr 3, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Conversation

1996fanrui commented Mar 31, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

flinkbot commented Mar 31, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

Uh oh!

Uh oh!

Uh oh!

1996fanrui Apr 1, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

pnowojski Apr 2, 2026

Choose a reason for hiding this comment

Uh oh!

1996fanrui Apr 2, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

1996fanrui left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

pnowojski Apr 3, 2026

Choose a reason for hiding this comment

Uh oh!

1996fanrui Apr 3, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

1996fanrui commented Mar 31, 2026 •

edited

Loading

flinkbot commented Mar 31, 2026 •

edited

Loading

1996fanrui left a comment •

edited

Loading