[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.

Author: 1996fanrui

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.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 @@ private Optional<BufferOrEvent> handleEvent(BufferOrEvent bufferOrEvent) throws
                     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);
     }