1996fanrui commented on code in PR #28652:
URL: https://github.com/apache/flink/pull/28652#discussion_r3555242894
##########
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:
> Can we use CountDownLatch in both then?
done via 84987dddbcfedc4e350cbbcad3eb0f5e21b08312
--
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]
