rkhachatryan commented on code in PR #28652:
URL: https://github.com/apache/flink/pull/28652#discussion_r3554456411
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java:
##########
@@ -113,48 +159,211 @@ public LocalInputChannel(
this.partitionManager = checkNotNull(partitionManager);
this.taskEventPublisher = checkNotNull(taskEventPublisher);
- this.channelStatePersister = new ChannelStatePersister(stateWriter,
getChannelInfo());
-
- // Migrate recovered buffers from RecoveredInputChannel if provided.
- // These buffers have been filtered but not yet consumed by the Task.
- if (!initialRecoveredBuffers.isEmpty()) {
- final int expectedCount = initialRecoveredBuffers.size();
- // Sequence number starts at Integer.MIN_VALUE, consistent with
RecoveredInputChannel.
- int seqNum = Integer.MIN_VALUE;
- while (!initialRecoveredBuffers.isEmpty()) {
- Buffer buffer = initialRecoveredBuffers.poll();
- // Determine next data type based on the next buffer in the
queue
- Buffer.DataType nextDataType =
- initialRecoveredBuffers.isEmpty()
- ? Buffer.DataType.NONE
- : initialRecoveredBuffers.peek().getDataType();
- // buffersInBacklog is set to 0 as these are recovered buffers
- BufferAndBacklog bufferAndBacklog =
- new BufferAndBacklog(buffer, 0, nextDataType,
seqNum++);
- toBeConsumedBuffers.add(bufferAndBacklog);
+ this.channelStatePersister =
+ new ChannelStatePersister(checkNotNull(stateWriter),
getChannelInfo());
+ this.inRecovery = needsRecovery;
+ this.bufferManager =
+ needsRecovery
+ ? new
BufferManager(inputGate.getMemorySegmentProvider(), this, 0, true)
+ : null;
+ this.networkBuffersPerChannel = networkBuffersPerChannel;
+ this.needsRecovery = needsRecovery;
+ if (!needsRecovery) {
+ stateConsumedFuture.complete(null);
+ }
+ }
+
+ @Override
+ void setup() throws IOException {
+ if (needsRecovery && networkBuffersPerChannel > 0) {
+ bufferManager.requestExclusiveBuffers(networkBuffersPerChannel);
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // RecoverableInputChannel implementation
+ // ------------------------------------------------------------------------
+
+ @Override
+ public void onRecoveredStateBuffer(Buffer buffer) {
+ boolean wasEmpty;
+ synchronized (recoveredBuffers) {
+ if (isReleased) {
+ buffer.recycleBuffer();
+ return;
}
- checkState(
- toBeConsumedBuffers.size() == expectedCount,
- "Buffer migration failed: expected %s buffers but got %s",
- expectedCount,
- toBeConsumedBuffers.size());
+ // Migrate recovered buffers from RecoveredInputChannel. These
buffers have been
+ // filtered but not yet consumed by the Task.
+ wasEmpty = offerRecoveredBuffer(buffer);
+ }
+ if (wasEmpty) {
+ notifyChannelNonEmpty();
+ }
+ }
+
+ @Override
+ public void finishRecoveredBufferDelivery() throws IOException {
+ upstreamReady.join();
+ boolean wasEmpty;
+ synchronized (recoveredBuffers) {
+ checkState(inRecovery, "Recovery delivery already finished.");
+ // Append the sentinel after the last recovered buffer. The
consume path flips out of
+ // recovery only once it polls this sentinel, guaranteeing all
recovered buffers are
+ // consumed first.
+ wasEmpty =
+ offerRecoveredBuffer(
+ EventSerializer.toBuffer(
+ EndOfFetchedChannelStateEvent.INSTANCE,
false));
+ }
+ if (wasEmpty) {
+ notifyChannelNonEmpty();
+ }
+ }
+
+ @Override
+ public Buffer requestRecoveryBufferBlocking() throws InterruptedException,
IOException {
+ checkState(
+ bufferManager != null,
+ "requestRecoveryBufferBlocking called on a Local channel
constructed with"
+ + " needsRecovery=false");
+ upstreamReady.join();
+ return bufferManager.requestBufferBlocking();
+ }
+
+ @Override
+ public void insertRecoveryCheckpointBarrierIfInRecovery(long checkpointId)
throws IOException {
+ boolean wasEmpty = false;
+ synchronized (recoveredBuffers) {
+ if (!isReleased && inRecovery) {
+ wasEmpty =
+ offerRecoveredBuffer(
+ EventSerializer.toBuffer(
+ new
RecoveryCheckpointBarrier(checkpointId), false));
+ }
+ }
+ if (wasEmpty) {
+ notifyChannelNonEmpty();
+ }
+ }
+
+ /**
+ * Flips out of recovery the moment the consume path polls the {@code
+ * EndOfFetchedChannelStateEvent} sentinel, i.e. once all recovered
buffers have been consumed.
+ * Live upstream data may flow again afterwards.
+ */
+ @Override
+ public void onRecoveredStateConsumed() {
+ synchronized (recoveredBuffers) {
+ checkState(inRecovery, "Recovery already finished.");
+ inRecovery = false;
}
+ notifyChannelNonEmpty();
+ stateConsumedFuture.complete(null);
+ }
+
+ @Override
+ public CompletableFuture<Void> getStateConsumedFuture() {
+ return stateConsumedFuture;
+ }
+
+ /**
+ * Appends a recovered buffer (or {@code RecoveryCheckpointBarrier} /
{@code
+ * EndOfFetchedChannelStateEvent} sentinel) to {@link #recoveredBuffers}.
+ *
+ * @return {@code true} iff {@link #recoveredBuffers} transitioned from
empty to non-empty.
+ */
+ private boolean offerRecoveredBuffer(Buffer buffer) {
+ assert Thread.holdsLock(recoveredBuffers);
+ checkState(inRecovery, "Push into recovered buffers after recovery
finished.");
+ boolean wasEmpty = recoveredBuffers.isEmpty();
+ recoveredBuffers.add(buffer);
+ return wasEmpty;
+ }
+
+ private int nextRecoverySequenceNumber() {
+ assert Thread.holdsLock(recoveredBuffers);
+ return recoverySequenceNumber++;
+ }
+
+ /**
+ * Walks {@link #recoveredBuffers} up to the {@link
RecoveryCheckpointBarrier} sentinel matching
+ * {@code checkpointId}, retaining each pre-barrier recovered data buffer
and removing the
+ * sentinel.
+ *
+ * @throws IOException if no sentinel matching {@code checkpointId} is
found (the snapshot
+ * protocol guarantees one must be present while the channel is in
recovery).
+ */
+ private List<Buffer> collectPreRecoveryBarrier(long checkpointId) throws
IOException {
+ assert Thread.holdsLock(recoveredBuffers);
+ List<Buffer> retained = new ArrayList<>();
+ try {
+ Iterator<Buffer> it = recoveredBuffers.iterator();
+ while (it.hasNext()) {
+ Buffer b = it.next();
+ if (isRecoveryCheckpointBarrier(b, checkpointId)) {
+ it.remove();
+ b.recycleBuffer();
+ return retained;
+ }
+ if (b.isBuffer()) {
+ retained.add(b.retainBuffer());
Review Comment:
> If it were to happen, would you be inclined to throw an exception?
Yes, I think so 🤔
- if the encountered barrier is > target then something is definitely wrong
- if it's < target then _probably_ a checkpoint was subsumed; at least I'd
log a warning AND remove + recycle this buffer
- if it's == target then return (OK)
WDYT?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannel.java:
##########
@@ -115,23 +115,49 @@ public void setChannelStateWriter(ChannelStateWriter
channelStateWriter) {
this.channelStateWriter = checkNotNull(channelStateWriter);
}
- public final InputChannel toInputChannel() throws IOException {
- Preconditions.checkState(
- bufferFilteringCompleteFuture.isDone(), "buffer filtering is
not complete");
- if (!inputGate.isCheckpointingDuringRecoveryEnabled()) {
- Preconditions.checkState(
- stateConsumedFuture.isDone(), "recovered state is not
fully consumed");
+ public final InputChannel toInputChannel(boolean needsRecovery) throws
IOException {
+ if (needsRecovery) {
+ return toInputChannelInRecovery();
+ }
+ synchronized (receivedBuffers) {
+ Preconditions.checkState(receivedBuffers.isEmpty(), "Received
buffer should be empty.");
}
- // Extract remaining buffers before conversion.
- // These buffers have been filtered but not yet consumed by the Task.
+ final InputChannel inputChannel =
toInputChannelInternal(needsRecovery);
+ inputChannel.setup();
+ inputChannel.checkpointStopped(lastStoppedCheckpointId);
+ return inputChannel;
+ }
+
+ /**
+ * FLINK-38544 transitional: removed when the spilling backend lands.
Creates the physical
+ * channel in recovery state and synchronously hands every queued
recovered buffer over through
+ * the push interface. The legacy {@link EndOfInputChannelStateEvent} in
the queue is dropped in
+ * translation; the {@link EndOfFetchedChannelStateEvent} sentinel takes
its place. The sentinel
+ * is appended directly instead of via {@link
+ * RecoverableInputChannel#finishRecoveredBufferDelivery()} because that
method waits for
+ * upstream readiness, which cannot happen while the mailbox thread is
still converting channels
+ * (partitions are requested only after conversion).
+ */
+ private InputChannel toInputChannelInRecovery() throws IOException {
final ArrayDeque<Buffer> remainingBuffers;
synchronized (receivedBuffers) {
remainingBuffers = new ArrayDeque<>(receivedBuffers);
receivedBuffers.clear();
}
- final InputChannel inputChannel =
toInputChannelInternal(remainingBuffers);
+ final InputChannel inputChannel = toInputChannelInternal(true);
+ inputChannel.setup();
+ final RecoverableInputChannel recoverableChannel =
(RecoverableInputChannel) inputChannel;
+ for (Buffer buffer : remainingBuffers) {
+ if (isEndOfInputChannelStateEvent(buffer)) {
+ buffer.recycleBuffer();
Review Comment:
I'd still add a defensive check here - it doesn't hurt
(in case this code survives in later PRs somehow)
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java:
##########
@@ -113,48 +159,211 @@ public LocalInputChannel(
this.partitionManager = checkNotNull(partitionManager);
this.taskEventPublisher = checkNotNull(taskEventPublisher);
- this.channelStatePersister = new ChannelStatePersister(stateWriter,
getChannelInfo());
-
- // Migrate recovered buffers from RecoveredInputChannel if provided.
- // These buffers have been filtered but not yet consumed by the Task.
- if (!initialRecoveredBuffers.isEmpty()) {
- final int expectedCount = initialRecoveredBuffers.size();
- // Sequence number starts at Integer.MIN_VALUE, consistent with
RecoveredInputChannel.
- int seqNum = Integer.MIN_VALUE;
- while (!initialRecoveredBuffers.isEmpty()) {
- Buffer buffer = initialRecoveredBuffers.poll();
- // Determine next data type based on the next buffer in the
queue
- Buffer.DataType nextDataType =
- initialRecoveredBuffers.isEmpty()
- ? Buffer.DataType.NONE
- : initialRecoveredBuffers.peek().getDataType();
- // buffersInBacklog is set to 0 as these are recovered buffers
- BufferAndBacklog bufferAndBacklog =
- new BufferAndBacklog(buffer, 0, nextDataType,
seqNum++);
- toBeConsumedBuffers.add(bufferAndBacklog);
+ this.channelStatePersister =
+ new ChannelStatePersister(checkNotNull(stateWriter),
getChannelInfo());
+ this.inRecovery = needsRecovery;
+ this.bufferManager =
+ needsRecovery
+ ? new
BufferManager(inputGate.getMemorySegmentProvider(), this, 0, true)
+ : null;
+ this.networkBuffersPerChannel = networkBuffersPerChannel;
+ this.needsRecovery = needsRecovery;
+ if (!needsRecovery) {
+ stateConsumedFuture.complete(null);
+ }
+ }
+
+ @Override
+ void setup() throws IOException {
+ if (needsRecovery && networkBuffersPerChannel > 0) {
+ bufferManager.requestExclusiveBuffers(networkBuffersPerChannel);
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // RecoverableInputChannel implementation
+ // ------------------------------------------------------------------------
+
+ @Override
+ public void onRecoveredStateBuffer(Buffer buffer) {
+ boolean wasEmpty;
+ synchronized (recoveredBuffers) {
+ if (isReleased) {
+ buffer.recycleBuffer();
+ return;
}
- checkState(
- toBeConsumedBuffers.size() == expectedCount,
- "Buffer migration failed: expected %s buffers but got %s",
- expectedCount,
- toBeConsumedBuffers.size());
+ // Migrate recovered buffers from RecoveredInputChannel. These
buffers have been
+ // filtered but not yet consumed by the Task.
+ wasEmpty = offerRecoveredBuffer(buffer);
+ }
+ if (wasEmpty) {
+ notifyChannelNonEmpty();
+ }
+ }
+
+ @Override
+ public void finishRecoveredBufferDelivery() throws IOException {
+ upstreamReady.join();
+ boolean wasEmpty;
+ synchronized (recoveredBuffers) {
+ checkState(inRecovery, "Recovery delivery already finished.");
+ // Append the sentinel after the last recovered buffer. The
consume path flips out of
+ // recovery only once it polls this sentinel, guaranteeing all
recovered buffers are
+ // consumed first.
+ wasEmpty =
+ offerRecoveredBuffer(
+ EventSerializer.toBuffer(
+ EndOfFetchedChannelStateEvent.INSTANCE,
false));
+ }
+ if (wasEmpty) {
+ notifyChannelNonEmpty();
+ }
+ }
+
+ @Override
+ public Buffer requestRecoveryBufferBlocking() throws InterruptedException,
IOException {
+ checkState(
+ bufferManager != null,
+ "requestRecoveryBufferBlocking called on a Local channel
constructed with"
+ + " needsRecovery=false");
+ upstreamReady.join();
+ return bufferManager.requestBufferBlocking();
+ }
+
+ @Override
+ public void insertRecoveryCheckpointBarrierIfInRecovery(long checkpointId)
throws IOException {
+ boolean wasEmpty = false;
+ synchronized (recoveredBuffers) {
+ if (!isReleased && inRecovery) {
+ wasEmpty =
+ offerRecoveredBuffer(
+ EventSerializer.toBuffer(
+ new
RecoveryCheckpointBarrier(checkpointId), false));
+ }
+ }
+ if (wasEmpty) {
+ notifyChannelNonEmpty();
+ }
+ }
+
+ /**
+ * Flips out of recovery the moment the consume path polls the {@code
+ * EndOfFetchedChannelStateEvent} sentinel, i.e. once all recovered
buffers have been consumed.
+ * Live upstream data may flow again afterwards.
+ */
+ @Override
+ public void onRecoveredStateConsumed() {
+ synchronized (recoveredBuffers) {
+ checkState(inRecovery, "Recovery already finished.");
+ inRecovery = false;
}
+ notifyChannelNonEmpty();
+ stateConsumedFuture.complete(null);
+ }
+
+ @Override
+ public CompletableFuture<Void> getStateConsumedFuture() {
+ return stateConsumedFuture;
+ }
+
+ /**
+ * Appends a recovered buffer (or {@code RecoveryCheckpointBarrier} /
{@code
+ * EndOfFetchedChannelStateEvent} sentinel) to {@link #recoveredBuffers}.
+ *
+ * @return {@code true} iff {@link #recoveredBuffers} transitioned from
empty to non-empty.
+ */
+ private boolean offerRecoveredBuffer(Buffer buffer) {
+ assert Thread.holdsLock(recoveredBuffers);
+ checkState(inRecovery, "Push into recovered buffers after recovery
finished.");
+ boolean wasEmpty = recoveredBuffers.isEmpty();
+ recoveredBuffers.add(buffer);
+ return wasEmpty;
+ }
+
+ private int nextRecoverySequenceNumber() {
+ assert Thread.holdsLock(recoveredBuffers);
+ return recoverySequenceNumber++;
+ }
+
+ /**
+ * Walks {@link #recoveredBuffers} up to the {@link
RecoveryCheckpointBarrier} sentinel matching
+ * {@code checkpointId}, retaining each pre-barrier recovered data buffer
and removing the
+ * sentinel.
+ *
+ * @throws IOException if no sentinel matching {@code checkpointId} is
found (the snapshot
+ * protocol guarantees one must be present while the channel is in
recovery).
+ */
+ private List<Buffer> collectPreRecoveryBarrier(long checkpointId) throws
IOException {
+ assert Thread.holdsLock(recoveredBuffers);
+ List<Buffer> retained = new ArrayList<>();
+ try {
+ Iterator<Buffer> it = recoveredBuffers.iterator();
+ while (it.hasNext()) {
+ Buffer b = it.next();
+ if (isRecoveryCheckpointBarrier(b, checkpointId)) {
+ it.remove();
+ b.recycleBuffer();
+ return retained;
+ }
+ if (b.isBuffer()) {
+ retained.add(b.retainBuffer());
Review Comment:
Same for RemoteInputChannel
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java:
##########
@@ -188,6 +217,11 @@ void setExpectedSequenceNumber(int expectedSequenceNumber)
{
this.expectedSequenceNumber = expectedSequenceNumber;
}
+ @VisibleForTesting
+ void completeUpstreamReadyForTest() {
Review Comment:
Why would it give broader control?
To clarify, my proposal is to pass the future/latch into the channel
constructor.
Prod could use the current constructor that would just pass
```
// prod code:
channel = new Channel();
// test code:
latch = new Latch()
channel = new Channel(latch);
...
latch.coundDown();
```
This is not a blocker though.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java:
##########
@@ -713,30 +889,110 @@ private void checkAnnouncedOnlyOnce(SequenceBuffer
sequenceBuffer) {
}
/**
- * Spills all queued buffers on checkpoint start. If barrier has already
been received (and
- * reordered), spill only the overtaken buffers.
+ * Persists inflight data on checkpoint start. During recovery, persists
recovered buffers
+ * before the matching RecoveryCheckpointBarrier sentinel; after recovery,
uses the normal
+ * remote-channel barrier sequence tracking and persists overtaken live
buffers.
*/
public void checkpointStarted(CheckpointBarrier barrier) throws
CheckpointException {
- synchronized (receivedBuffers) {
- if (barrier.getId() < lastBarrierId) {
- throw new CheckpointException(
- String.format(
- "Sequence number for checkpoint %d is not
known (it was likely been overwritten by a newer checkpoint %d)",
- barrier.getId(), lastBarrierId),
- CheckpointFailureReason
- .CHECKPOINT_SUBSUMED); // currently, at most
one active unaligned
- // checkpoint is possible
- } else if (barrier.getId() > lastBarrierId) {
- // This channel has received some obsolete barrier, older
compared to the
- // checkpointId
- // which we are processing right now, and we should ignore
that obsoleted checkpoint
- // barrier sequence number.
- resetLastBarrier();
+ try {
+ List<Buffer> toPersist;
+ synchronized (receivedBuffers) {
+ if (inRecovery) {
+ toPersist = collectPreRecoveryBarrier(barrier.getId());
+ } else {
+ if (barrier.getId() < lastBarrierId) {
+ // Currently, at most one active unaligned checkpoint
is possible.
+ throw new CheckpointException(
+ String.format(
+ "Sequence number for checkpoint %d is
not known (it was likely been overwritten by a newer checkpoint %d)",
+ barrier.getId(), lastBarrierId),
+ CheckpointFailureReason.CHECKPOINT_SUBSUMED);
+ } else if (barrier.getId() > lastBarrierId) {
+ // This channel has received some obsolete barrier,
older compared to the
+ // checkpointId which we are processing right now, and
we should ignore that
+ // obsoleted checkpoint barrier sequence number.
+ resetLastBarrier();
+ }
+ toPersist = getInflightBuffersUnsafe(barrier.getId());
+ }
+ channelStatePersister.startPersisting(barrier.getId(),
toPersist);
+ }
+ } catch (IOException e) {
+ throw new CheckpointException(
+ "Failed to extract recovered buffers for checkpoint " +
barrier.getId(),
+ CheckpointFailureReason.CHECKPOINT_DECLINED,
+ e);
+ }
+ }
+
+ /**
+ * Walks {@link #receivedBuffers} (skipping priority events) up to the
{@link
+ * RecoveryCheckpointBarrier} sentinel matching {@code checkpointId},
retaining each pre-barrier
+ * recovered data buffer and removing the sentinel. During recovery the
upstream has no credit,
+ * so {@code receivedBuffers} holds only recovered buffers, sentinels, and
priority events — no
+ * live data buffers.
+ *
+ * @throws IOException if no sentinel matching {@code checkpointId} is
found (the snapshot
+ * protocol guarantees one must be present while the channel is in
recovery).
+ */
+ @GuardedBy("receivedBuffers")
+ private List<Buffer> collectPreRecoveryBarrier(long checkpointId) throws
IOException {
+ assert Thread.holdsLock(receivedBuffers);
+ List<Buffer> retained = new ArrayList<>();
+ SequenceBuffer sentinel = null;
+ try {
+ Iterator<SequenceBuffer> it = receivedBuffers.iterator();
+ // Priority events are stored separately at the head and never
carry recovered data.
+ Iterators.advance(it, receivedBuffers.getNumPriorityElements());
+ while (it.hasNext()) {
+ SequenceBuffer sb = it.next();
+ if (isRecoveryCheckpointBarrier(sb.buffer, checkpointId)) {
+ sentinel = sb;
+ break;
+ }
+ // Skip non-data events (e.g. the
EndOfFetchedChannelStateEvent sentinel appended
+ // after the recovered buffers): only recovered data buffers
are snapshotted.
+ if (sb.buffer.isBuffer()) {
Review Comment:
What sequence of events could lead to this?
1. Recovery started
2. drainer reaches EOF and calls `finishRecoveredBufferDelivery()` ->
`EndOfFetchedChannelStateEvent` is added
3. Checkpoint started, barrier arrived
4. it's a high-priority event so it "overtakes"
`EndOfFetchedChannelStateEvent`
5. Task thread starts snapshotting the channel
6. It calls `collectPreRecoveryBarrier()` because this channel is still
`inRecovery` because `EndOfFetchedChannelStateEvent` not consumed
Is that correct? Could you clarify and add a comment in the code?
##########
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:
> We could unify both to CompletableFuture or CountDownLatch to ensure
consistency.
Can we use `CountDownLatch` in both then?
> The purpose is avoid complex race condition or synchronization during
recovery or starting.
Could you elaborate more or point me to any docs or previous discussions?
--
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]
