1996fanrui commented on code in PR #28652:
URL: https://github.com/apache/flink/pull/28652#discussion_r3609280762
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java:
##########
@@ -123,6 +129,45 @@ public class RemoteInputChannel extends InputChannel {
private final ChannelStatePersister channelStatePersister;
+ /**
+ * Whether the channel is still replaying recovered state. Recovered
buffers delivered by the
+ * spill drain are appended directly to {@link #receivedBuffers}, so the
consume path needs no
+ * recovery-specific branch. Starts {@code false} for channels that do not
need recovery and is
+ * flipped to {@code false} the moment the consume path polls the {@code
+ * EndOfFetchedChannelStateEvent} sentinel that the drain appended after
the last recovered
+ * buffer (see {@link #onRecoveredStateConsumed()}).
+ */
+ @GuardedBy("receivedBuffers")
+ private boolean inRecovery;
+
+ private final CompletableFuture<Void> stateConsumedFuture = new
CompletableFuture<>();
+
+ /**
+ * Sequence number assigned to recovered buffers, starting at {@link
Integer#MIN_VALUE},
+ * consistent with {@link RecoveredInputChannel}.
+ */
+ @GuardedBy("receivedBuffers")
+ private int recoverySequenceNumber = Integer.MIN_VALUE;
+
+ /**
+ * Ordinary (non-priority) upstream events received while recovery is
still in progress. They
+ * cannot enter {@link #receivedBuffers} ahead of the recovered buffers,
so they are stashed
+ * here and appended once recovery delivery finishes. Credit is suppressed
during recovery, so
+ * the upstream can only send events (never data buffers) before {@link
+ * #finishRecoveredBufferDelivery()}.
+ */
+ @GuardedBy("receivedBuffers")
+ private final ArrayDeque<SequenceBuffer> recoveryEventStash = new
ArrayDeque<>();
+
+ /**
+ * One-shot latch that opens once the upstream reader is registered and
the connection is live
+ * (signalled by the first {@link #onBuffer} or by {@link
#releaseAllResources()}).
+ * Recovery-side awaiters block on it before handing off; once open, {@link
+ * CountDownLatch#countDown()} on the hot path is a cheap idempotent
no-op, unlike completing a
+ * {@code CompletableFuture}.
+ */
+ private final CountDownLatch upstreamReady = new CountDownLatch(1);
Review Comment:
### What breaks in LocalInputChannel if the wait is removed?
Consuming the recovery sentinel flips the channel out of recovery, and the
normal consume path takes over. That path asserts that the subpartition view
already exists — polling earlier fails with IllegalStateException("Queried for
a buffer before requesting the subpartition"). Without the wait, a drain that
finishes while the view request is still retrying in the background leads the
task straight into this assertion. The latch opens once the subpartition view
is created.
### What breaks in RemoteInputChannel if the wait is removed?
During recovery the channel announces zero credit; credit notifications are
re-enabled when the sentinel is consumed, at which point the accumulated credit
is sent to the server as an AddCredit message. The server, however, registers
the reader into allReaders only after the subpartition view is created — an
AddCredit arriving earlier finds no reader and fails with
IllegalStateException("No reader for receiverId ... exists"), a
connection-level error that takes down every channel on the same TCP
connection. The latch opens on the first message from the producer, which
proves the reader is registered.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
