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 c2f0c123409285cb244c23b32cc108dea2a9a990 Author: Rui Fan <[email protected]> AuthorDate: Mon Jul 6 00:31:08 2026 +0200 [FLINK-39521][network] CheckpointedInputGate: consume EndOfFetchedChannelStateEvent When the consume path polls the EndOfFetchedChannelStateEvent sentinel, assert the originating channel is a RecoverableInputChannel and call its onRecoveredStateConsumed(): the channel flips out of recovery, releases any upstream events held back during recovery and reopens the upstream so live data may flow again. The event itself is never delivered to the operator. The sentinel-consumption path is exercised end to end by the Local/RemoteInputChannel recovery tests of this PR (onRecoveredStateConsumed after polling the sentinel) and by the StreamTask recovery tests of the follow-up PRs. --- .../runtime/io/checkpointing/CheckpointedInputGate.java | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/CheckpointedInputGate.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/CheckpointedInputGate.java index ec4cde4de58..26f335386a1 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/CheckpointedInputGate.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/CheckpointedInputGate.java @@ -29,9 +29,11 @@ import org.apache.flink.runtime.io.network.api.EndOfData; import org.apache.flink.runtime.io.network.api.EndOfPartitionEvent; import org.apache.flink.runtime.io.network.api.EventAnnouncement; import org.apache.flink.runtime.io.network.partition.consumer.BufferOrEvent; +import org.apache.flink.runtime.io.network.partition.consumer.EndOfFetchedChannelStateEvent; import org.apache.flink.runtime.io.network.partition.consumer.EndOfOutputChannelStateEvent; import org.apache.flink.runtime.io.network.partition.consumer.InputChannel; import org.apache.flink.runtime.io.network.partition.consumer.InputGate; +import org.apache.flink.runtime.io.network.partition.consumer.RecoverableInputChannel; import org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput; import org.slf4j.Logger; @@ -202,6 +204,15 @@ public class CheckpointedInputGate implements PullingAsyncDataInput<BufferOrEven bufferOrEvent.getChannelInfo()); } else if (bufferOrEvent.getEvent().getClass() == EndOfOutputChannelStateEvent.class) { upstreamRecoveryTracker.handleEndOfRecovery(bufferOrEvent.getChannelInfo()); + } else if (eventClass == EndOfFetchedChannelStateEvent.class) { + // Tail of the recovered buffers: only a RecoverableInputChannel can produce this + // sentinel, so anything else here is a bug rather than something to tolerate. + InputChannel channel = inputGate.getChannel(bufferOrEvent.getChannelInfo()); + checkState( + channel instanceof RecoverableInputChannel, + "EndOfFetchedChannelStateEvent received on a non-recoverable channel %s", + bufferOrEvent.getChannelInfo()); + ((RecoverableInputChannel) channel).onRecoveredStateConsumed(); } return Optional.of(bufferOrEvent); }
