1996fanrui commented on code in PR #28652:
URL: https://github.com/apache/flink/pull/28652#discussion_r3609272219
##########
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:
> Could you elaborate more or point me to any docs or previous discussions?
Historically, `LocalInputChannel` and `RemoteInputChannel` only start
working after the subscription to the upstream subpartition has been
established — all of their logic relies on this assumption. CDR breaks this
assumption: the channels are now created directly in recovery mode and
deliver/consume recovered state while the subscription is still being set up.
For simplicity, recovery just waits for the subscription to be established (the
upstreamReady latch) before it is allowed to finish, so all existing
assumptions still hold.
Note that `requestPartition` is asynchronous and may not succeed on the
first attempt: the upstream task may not have been deployed yet, and the
request is retried in the background.
--
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]
