You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/09/18 09:28:19 UTC
(pulsar) branch master updated: [improve][client] Pull v5 blocking receives straight from the receive buffer (#26614)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new de49904f1fd6 [improve][client] Pull v5 blocking receives straight from the receive buffer (#26614)
de49904f1fd6 is described below
commit de49904f1fd6219305ade232659c2a183c59b36f
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Fri Sep 18 02:28:09 2026 -0700
[improve][client] Pull v5 blocking receives straight from the receive buffer (#26614)
---
.../pulsar/client/impl/v5/V5ReceiveQueue.java | 251 ++++++++++++++-------
.../pulsar/client/impl/v5/V5ReceiveQueueTest.java | 197 ++++++++++++++++
2 files changed, 368 insertions(+), 80 deletions(-)
diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/V5ReceiveQueue.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/V5ReceiveQueue.java
index e56d7f8ac038..6fde71bb573c 100644
--- a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/V5ReceiveQueue.java
+++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/V5ReceiveQueue.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.client.impl.v5;
+import com.google.common.annotations.VisibleForTesting;
import io.netty.util.Timeout;
import io.netty.util.Timer;
import java.time.Duration;
@@ -30,28 +31,27 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.pulsar.client.api.v5.Message;
import org.apache.pulsar.client.api.v5.PulsarClientException;
+import org.apache.pulsar.common.util.collections.GrowableArrayBlockingQueue;
/**
- * Async-native, single-consumer receive queue shared by the v5 scalable consumers.
+ * Async-native receive queue shared by the v5 scalable consumers.
*
- * <p>Mirrors the v4 {@code ConsumerBase} delivery model: a buffer of ready messages
- * plus a queue of pending receive futures. Both are confined to one pinned executor
- * (obtained from the client's external executor provider, so one thread per consumer),
- * which means a message and a waiter can never cross — no locks, and no lost wakeups.
- * Every receive future is completed on that executor, so user continuations chained on
- * the returned {@link CompletableFuture} never run on a netty IO thread.
+ * <p>Mirrors the v4 {@code ConsumerBase} delivery model. The buffer of ready messages is a
+ * thread-safe blocking queue: producers (the per-segment receive loops, on a v4 client executor)
+ * append to it directly, and the blocking {@link #take()}/{@link #poll(Duration)} pull from it on
+ * the caller's own thread, so in steady state a message never has to cross to another thread just
+ * to be handed over. Only the bookkeeping that needs ordering — the pending {@link #receiveAsync()}
+ * futures and the paused producers — is confined to one pinned executor (obtained from the
+ * client's external executor provider, so one thread per consumer), and every async receive
+ * future is completed there, so user continuations chained on it never run on a netty IO thread.
*
- * <p>This replaces the previous {@code LinkedTransferQueue} + {@code supplyAsync(take())}
- * approach: {@link #receiveAsync()} parks no thread, is cancelable, and honours timeouts
- * via the client timer instead of blocking a {@code ForkJoinPool.commonPool()} worker.
- *
- * <p>Backpressure: {@link #offer} returns a future that completes only when the buffer has
- * room. Producers gate re-arming on it, so the buffer is bounded by {@code receiverQueueSize}
- * plus one in-flight message per producer (each producer offers at most one message before
- * observing the pause) — modelled on v4 {@code MultiTopicsConsumerImpl}'s pause/resume of
- * sub-consumers, including its fairness rule: once any producer is paused, further offers
- * pause at the half-way mark too, so active producers can't hold the buffer above the resume
- * threshold and starve the paused ones.
+ * <p>Backpressure: {@link #offer} returns a future that completes only when the buffer has room.
+ * Producers gate re-arming on it, so the buffer is bounded by {@code receiverQueueSize} plus one
+ * in-flight message per producer (each producer offers at most one message before observing the
+ * pause) — modelled on v4 {@code MultiTopicsConsumerImpl}'s pause/resume of sub-consumers,
+ * including its fairness rule: once any producer is paused, further offers pause at the half-way
+ * mark too, so active producers can't hold the buffer above the resume threshold and starve the
+ * paused ones.
*/
final class V5ReceiveQueue<T> {
@@ -64,18 +64,27 @@ final class V5ReceiveQueue<T> {
private final int highWatermark;
private final int lowWatermark;
- // All three touched only on `executor`, so plain (non-concurrent) collections are safe.
- private final ArrayDeque<Message<T>> buffer = new ArrayDeque<>();
+ /**
+ * Ready messages. Thread-safe: producers append from their own threads, blocking receives
+ * poll from the caller thread, and the executor drains it for pending async receives.
+ */
+ private final GrowableArrayBlockingQueue<Message<T>> buffer = new GrowableArrayBlockingQueue<>();
+
+ // Both touched only on `executor`, so plain (non-concurrent) collections are safe.
private final ArrayDeque<CompletableFuture<Message<T>>> pendingReceives = new ArrayDeque<>();
// Capacity futures handed back to producers that were paused because the buffer was full.
private final ArrayDeque<CompletableFuture<Void>> capacityWaiters = new ArrayDeque<>();
- private boolean closed = false;
- // Snapshots of executor-confined state, readable from producer threads so the offer()
- // fast path can decide without a hop. Written only on `executor`; may lag by the offers
- // still queued, so each producer (one in-flight offer at a time) can overshoot by one.
- private volatile int approxBufferSize = 0;
+ // Snapshots of the executor-confined state, readable from producer and receiver threads so
+ // the hot paths can decide without a hop. Written only on `executor` (except `closed`).
+ /** True while {@link #pendingReceives} may hold a future that the next message belongs to. */
+ private volatile boolean hasPendingReceives = false;
private volatile boolean producersPaused = false;
+ private volatile boolean closed = false;
+
+ /** Test seam: runs on the executor between the pause decision and its publication. */
+ @VisibleForTesting
+ volatile Runnable beforePausePublishedHook;
V5ReceiveQueue(ExecutorService executor, Timer timer, int receiverQueueSize) {
this.executor = executor;
@@ -86,56 +95,55 @@ final class V5ReceiveQueue<T> {
/**
* Deposit a freshly-arrived message. Called from the per-segment receive loops (which
- * run on a v4 client executor). Hands the message straight to a waiting receive future
- * if there is one, otherwise buffers it.
+ * run on a v4 client executor). Appends straight to the buffer — waking a blocked
+ * {@link #take()}/{@link #poll(Duration)} if it was empty — and hands it over on the
+ * executor only if an async receive is waiting.
*
* @return a future that completes when the sink is ready for the next message — right
- * away unless the buffer is filling up, in which case it defers until the consumer
- * drains it below the low watermark (backpressure).
+ * away unless the buffer is full, in which case it defers until the consumer drains it
+ * below the low watermark (backpressure).
*/
CompletableFuture<Void> offer(Message<T> msg) {
- // Fast path, decided on the caller thread: while the buffer is comfortably below the
- // watermarks and nobody is paused, grant capacity with a shared completed future so
- // the (fast-consumer) hot path pays no allocation and no serialized hop through our
- // executor before the segment loop re-arms.
- if (!producersPaused && approxBufferSize < lowWatermark) {
- executor.execute(() -> doOffer(msg, null));
+ if (closed) {
+ // Dropped, like a message arriving on an already-closed v4 consumer.
+ return READY;
+ }
+ buffer.put(msg);
+ if (hasPendingReceives) {
+ // A receiver registering concurrently re-checks the buffer after publishing the
+ // flag, and we read the flag after appending, so one side always sees the other.
+ executor.execute(this::drainToPendingReceives);
+ }
+ // Fast path, decided on the caller thread: while the buffer is below the high watermark
+ // and nobody is paused, grant capacity with a shared completed future so the hot path
+ // pays no allocation and no serialized hop through our executor before the segment
+ // loop re-arms.
+ if (!producersPaused && buffer.size() < highWatermark) {
return READY;
}
CompletableFuture<Void> capacity = new CompletableFuture<>();
- executor.execute(() -> doOffer(msg, capacity));
+ executor.execute(() -> grantCapacity(capacity));
return capacity;
}
- /** Runs on {@code executor}. {@code capacity} is null when the fast path already granted it. */
- private void doOffer(Message<T> msg, CompletableFuture<Void> capacity) {
- if (closed) {
- if (capacity != null) {
- capacity.complete(null);
- }
- return;
- }
- CompletableFuture<Message<T>> waiter = pollWaiter();
- if (waiter != null) {
- // Handed straight to a waiting receiver; the buffer didn't grow.
- waiter.complete(msg);
- if (capacity != null) {
- capacity.complete(null);
- }
- return;
- }
- buffer.add(msg);
- approxBufferSize = buffer.size();
- if (capacity == null) {
- return;
- }
+ /** Runs on {@code executor}: park the producer while the buffer is full, else grant now. */
+ private void grantCapacity(CompletableFuture<Void> capacity) {
// Pause when full — or, once any producer is paused, already at the half-way mark, so
// active producers can't keep the buffer hovering above the resume threshold while the
// paused ones starve (v4 MultiTopicsConsumerImpl's fairness clause).
- if (buffer.size() >= highWatermark
- || (!capacityWaiters.isEmpty() && buffer.size() > lowWatermark)) {
+ if (!closed && (buffer.size() >= highWatermark
+ || (!capacityWaiters.isEmpty() && buffer.size() > lowWatermark))) {
+ Runnable hook = beforePausePublishedHook;
+ if (hook != null) {
+ hook.run();
+ }
capacityWaiters.add(capacity);
producersPaused = true;
+ // A direct receive that drained the buffer between the check above and this
+ // publication saw producersPaused == false and posted no resume. Look again now that
+ // the pause is visible: that receive decrements the size before reading the flag and
+ // we publish the flag before reading the size, so one side always sees the other.
+ maybeResumeProducers();
} else {
capacity.complete(null);
}
@@ -156,6 +164,13 @@ final class V5ReceiveQueue<T> {
}
}
+ /** After a receive pulled straight from the buffer: resume producers once it drained enough. */
+ private void afterDirectPoll() {
+ if (producersPaused && buffer.size() <= lowWatermark) {
+ executor.execute(this::maybeResumeProducers);
+ }
+ }
+
/** Receive a message, completing as soon as one is available. Never blocks a thread. */
CompletableFuture<Message<T>> receiveAsync() {
CompletableFuture<Message<T>> result = new CompletableFuture<>();
@@ -164,13 +179,17 @@ final class V5ReceiveQueue<T> {
result.completeExceptionally(alreadyClosed());
return;
}
- Message<T> msg = buffer.poll();
+ if (result.isDone()) {
+ // Cancelled before we got here: it must not consume a message that has since
+ // been appended straight to the buffer.
+ return;
+ }
+ Message<T> msg = pollBehindPendingReceives(result);
if (msg != null) {
- approxBufferSize = buffer.size();
result.complete(msg);
maybeResumeProducers();
- } else {
- pendingReceives.add(result);
+ } else if (!result.isDone()) {
+ addPendingReceive(result);
}
});
return result;
@@ -187,31 +206,82 @@ final class V5ReceiveQueue<T> {
result.completeExceptionally(alreadyClosed());
return;
}
- Message<T> msg = buffer.poll();
+ if (result.isDone()) {
+ // Cancelled before we got here: it must not consume a message that has since
+ // been appended straight to the buffer.
+ return;
+ }
+ Message<T> msg = pollBehindPendingReceives(result);
if (msg != null) {
- approxBufferSize = buffer.size();
result.complete(msg);
maybeResumeProducers();
return;
}
+ if (result.isDone()) {
+ return;
+ }
long millis = timeout.toMillis();
if (millis <= 0) {
result.complete(null);
return;
}
- pendingReceives.add(result);
Timeout t = timer.newTimeout(ignored -> executor.execute(() -> {
if (!result.isDone()) {
pendingReceives.remove(result);
+ hasPendingReceives = !pendingReceives.isEmpty();
result.complete(null);
}
}), millis, TimeUnit.MILLISECONDS);
// Cancel the timer when the message is handed off (or on close) so it doesn't linger.
result.whenComplete((r, e) -> t.cancel());
+ addPendingReceive(result);
});
return result;
}
+ /**
+ * Runs on {@code executor}: the next buffered message for a new receive — unless older
+ * receives are still waiting. Messages are appended to the buffer without going through the
+ * executor, so a message can sit there while an older receive is parked; that receive was
+ * registered first and must be served first, so hand the older ones what is buffered and
+ * report nothing for the newcomer while any of them remains. Completing an older receive runs
+ * its continuations inline, and one of them may cancel the newcomer: it must then not consume
+ * a message either.
+ */
+ private Message<T> pollBehindPendingReceives(CompletableFuture<?> newcomer) {
+ if (!pendingReceives.isEmpty()) {
+ drainToPendingReceives();
+ if (!pendingReceives.isEmpty() || newcomer.isDone()) {
+ return null;
+ }
+ }
+ return buffer.poll();
+ }
+
+ /** Runs on {@code executor}: park an async receive until the next message arrives. */
+ private void addPendingReceive(CompletableFuture<Message<T>> result) {
+ pendingReceives.add(result);
+ hasPendingReceives = true;
+ // A producer that appended between our poll and this flag store saw the flag clear and
+ // posted no drain, so look again now that the flag is published (both are volatile).
+ drainToPendingReceives();
+ }
+
+ /** Runs on {@code executor}: hand buffered messages to pending async receives, in order. */
+ private void drainToPendingReceives() {
+ CompletableFuture<Message<T>> waiter;
+ while ((waiter = pollWaiter()) != null) {
+ Message<T> msg = buffer.poll();
+ if (msg == null) {
+ pendingReceives.addFirst(waiter);
+ break;
+ }
+ waiter.complete(msg);
+ }
+ hasPendingReceives = !pendingReceives.isEmpty();
+ maybeResumeProducers();
+ }
+
/**
* Receive up to {@code maxMessages}, blocking (asynchronously) up to {@code timeout} for
* the batch. Waits for the first message, then opportunistically drains whatever else is
@@ -242,48 +312,67 @@ final class V5ReceiveQueue<T> {
result.complete(batch);
} else {
batch.add(msg);
- drainReady(batch, max).thenRun(() -> collectMulti(batch, max, deadlineNanos, result));
+ drainReady(batch, max, result).thenRun(() -> collectMulti(batch, max, deadlineNanos, result));
}
});
}
/** Move whatever is already buffered into {@code batch} (up to {@code max} total). */
- private CompletableFuture<Void> drainReady(List<Message<T>> batch, int max) {
+ private CompletableFuture<Void> drainReady(List<Message<T>> batch, int max,
+ CompletableFuture<List<Message<T>>> result) {
CompletableFuture<Void> done = new CompletableFuture<>();
executor.execute(() -> {
Message<T> m;
- while (batch.size() < max && (m = buffer.poll()) != null) {
+ while (batch.size() < max && !result.isDone() && (m = pollBehindPendingReceives(result)) != null) {
batch.add(m);
}
- approxBufferSize = buffer.size();
maybeResumeProducers();
done.complete(null);
});
return done;
}
- // --- Blocking views, for the synchronous receive() API. Block only the caller's thread. ---
+ // --- Blocking views, for the synchronous receive() API. Pull straight from the buffer on the
+ // caller's thread; it parks only while the buffer is empty. ---
Message<T> take() throws PulsarClientException {
+ if (closed) {
+ throw alreadyClosed();
+ }
+ Message<T> msg;
try {
- return receiveAsync().get();
+ msg = buffer.take();
} catch (InterruptedException e) {
+ if (closed) {
+ // close() terminates the buffer, which wakes blocked takers this way.
+ throw alreadyClosed();
+ }
Thread.currentThread().interrupt();
throw new PulsarClientException("Receive interrupted", e);
- } catch (ExecutionException e) {
- throw unwrap(e);
}
+ afterDirectPoll();
+ return msg;
}
Message<T> poll(Duration timeout) throws PulsarClientException {
+ if (closed) {
+ throw alreadyClosed();
+ }
+ Message<T> msg;
try {
- return receiveAsync(timeout).get();
+ msg = buffer.poll(timeout.toNanos(), TimeUnit.NANOSECONDS);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new PulsarClientException("Receive interrupted", e);
- } catch (ExecutionException e) {
- throw unwrap(e);
}
+ if (msg == null) {
+ if (closed) {
+ throw alreadyClosed();
+ }
+ return null;
+ }
+ afterDirectPoll();
+ return msg;
}
List<Message<T>> receiveMulti(int maxMessages, Duration timeout) throws PulsarClientException {
@@ -299,14 +388,17 @@ final class V5ReceiveQueue<T> {
/** Fail any outstanding receives so blocked/awaiting callers wake instead of hanging forever. */
void close() {
+ closed = true;
+ // Wake blocking receivers parked on the buffer; anything offered from now on is dropped.
+ buffer.terminate(null);
executor.execute(() -> {
- closed = true;
CompletableFuture<Message<T>> waiter;
while ((waiter = pendingReceives.poll()) != null) {
if (!waiter.isDone()) {
waiter.completeExceptionally(alreadyClosed());
}
}
+ hasPendingReceives = false;
// Release any paused producers so their receive loops re-arm and observe the close.
CompletableFuture<Void> capacity;
while ((capacity = capacityWaiters.poll()) != null) {
@@ -314,7 +406,6 @@ final class V5ReceiveQueue<T> {
}
producersPaused = false;
buffer.clear();
- approxBufferSize = 0;
});
}
diff --git a/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/V5ReceiveQueueTest.java b/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/V5ReceiveQueueTest.java
index e5644a3c401f..b3580d01dedb 100644
--- a/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/V5ReceiveQueueTest.java
+++ b/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/V5ReceiveQueueTest.java
@@ -30,10 +30,12 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.pulsar.client.api.v5.Message;
import org.apache.pulsar.client.api.v5.MessageId;
import org.apache.pulsar.client.api.v5.PulsarClientException;
@@ -287,6 +289,201 @@ public class V5ReceiveQueueTest {
paused.get(5, TimeUnit.SECONDS);
}
+ @Test
+ public void blockingReceiveDoesNotDependOnTheExecutor() throws Exception {
+ // Stall the executor: nothing posted to it can run until the gate opens.
+ CountDownLatch gate = new CountDownLatch(1);
+ executor.execute(() -> {
+ try {
+ gate.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ try {
+ Message<Integer> m1 = msg(1);
+ Message<Integer> m2 = msg(2);
+ queue.offer(m1);
+ queue.offer(m2);
+ // The blocking receives pull straight from the buffer on the caller thread — no
+ // round trip through the (stalled) executor per message.
+ assertSame(queue.take(), m1);
+ assertSame(queue.poll(Duration.ofSeconds(5)), m2);
+ assertNull(queue.poll(Duration.ofMillis(50)));
+ } finally {
+ gate.countDown();
+ }
+ }
+
+ @Test
+ public void fastPathHoldsUpToTheHighWatermark() throws Exception {
+ // receiverQueueSize=4 -> the fast path (shared completed future, decided on the caller
+ // thread) must apply while the buffer is below 4, not just below the low watermark (2).
+ V5ReceiveQueue<Integer> q = new V5ReceiveQueue<>(executor, timer, 4);
+ CountDownLatch gate = new CountDownLatch(1);
+ executor.execute(() -> {
+ try {
+ gate.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ try {
+ for (int i = 1; i <= 3; i++) {
+ assertTrue(q.offer(msg(i)).isDone(), "offer " + i + " should take the fast path");
+ }
+ // The 4th fills the buffer: the grant is decided on the executor, so it is still
+ // pending while the executor is stalled.
+ assertTrue(!q.offer(msg(4)).isDone());
+ } finally {
+ gate.countDown();
+ }
+ }
+
+ @Test
+ public void directReceiveDuringPausePublicationStillResumesTheProducer() throws Exception {
+ // receiverQueueSize=1: the first message fills the buffer, so its offer goes to the
+ // executor to decide the pause.
+ V5ReceiveQueue<Integer> q = new V5ReceiveQueue<>(executor, timer, 1);
+ AtomicReference<Message<Integer>> taken = new AtomicReference<>();
+ // Between the executor's size check and the pause publication, a blocking receive
+ // drains the buffer. It sees producersPaused == false, so it schedules no resume.
+ q.beforePausePublishedHook = () -> {
+ q.beforePausePublishedHook = null;
+ try {
+ taken.set(q.take());
+ } catch (PulsarClientException e) {
+ throw new RuntimeException(e);
+ }
+ };
+ Message<Integer> m = msg(1);
+ CompletableFuture<Void> capacity = q.offer(m);
+ // The producer must not stay parked against an empty buffer.
+ capacity.get(5, TimeUnit.SECONDS);
+ assertSame(taken.get(), m);
+ }
+
+ @Test
+ public void olderPendingReceiveIsServedBeforeANewerOne() throws Exception {
+ CompletableFuture<Message<Integer>> a = queue.receiveAsync();
+ flush(); // A is registered, waiting on an empty buffer
+ CountDownLatch gate = stallExecutor();
+ try {
+ CompletableFuture<Message<Integer>> b = queue.receiveAsync(); // queued behind the gate
+ Message<Integer> m1 = msg(1);
+ Message<Integer> m2 = msg(2);
+ queue.offer(m1); // appended straight to the buffer; A's drain queues behind B
+ queue.offer(m2);
+ gate.countDown();
+ assertSame(a.get(5, TimeUnit.SECONDS), m1);
+ assertSame(b.get(5, TimeUnit.SECONDS), m2);
+ } finally {
+ gate.countDown();
+ }
+ }
+
+ @Test
+ public void olderPendingReceiveIsServedBeforeANewerTimedOne() throws Exception {
+ CompletableFuture<Message<Integer>> a = queue.receiveAsync(Duration.ofSeconds(30));
+ flush();
+ CountDownLatch gate = stallExecutor();
+ try {
+ CompletableFuture<Message<Integer>> b = queue.receiveAsync(Duration.ofSeconds(30));
+ Message<Integer> m1 = msg(1);
+ Message<Integer> m2 = msg(2);
+ queue.offer(m1);
+ queue.offer(m2);
+ gate.countDown();
+ assertSame(a.get(5, TimeUnit.SECONDS), m1);
+ assertSame(b.get(5, TimeUnit.SECONDS), m2);
+ } finally {
+ gate.countDown();
+ }
+ }
+
+ @Test
+ public void receiveMultiDoesNotOvertakeAnOlderPendingReceive() throws Exception {
+ CompletableFuture<Message<Integer>> a = queue.receiveAsync();
+ flush();
+ CountDownLatch gate = stallExecutor();
+ try {
+ CompletableFuture<List<Message<Integer>>> batch = queue.receiveMultiAsync(2, Duration.ofSeconds(30));
+ Message<Integer> m1 = msg(1);
+ Message<Integer> m2 = msg(2);
+ Message<Integer> m3 = msg(3);
+ queue.offer(m1);
+ queue.offer(m2);
+ queue.offer(m3);
+ gate.countDown();
+ assertSame(a.get(5, TimeUnit.SECONDS), m1);
+ List<Message<Integer>> got = batch.get(5, TimeUnit.SECONDS);
+ assertEquals(got.size(), 2);
+ assertSame(got.get(0), m2);
+ assertSame(got.get(1), m3);
+ } finally {
+ gate.countDown();
+ }
+ }
+
+ @Test
+ public void olderReceiveContinuationCancellingTheNewcomerDoesNotLoseAMessage() throws Exception {
+ CompletableFuture<Message<Integer>> a = queue.receiveAsync();
+ flush();
+ CountDownLatch gate = stallExecutor();
+ try {
+ CompletableFuture<Message<Integer>> b = queue.receiveAsync();
+ // Serving A runs this inline on the executor, right before B would poll for itself.
+ a.thenRun(() -> b.cancel(false));
+ Message<Integer> m1 = msg(1);
+ Message<Integer> m2 = msg(2);
+ queue.offer(m1);
+ queue.offer(m2);
+ gate.countDown();
+ assertSame(a.get(5, TimeUnit.SECONDS), m1);
+ flush();
+ assertTrue(b.isCancelled());
+ // The cancelled B must not have taken m2 with it.
+ assertSame(queue.poll(Duration.ofSeconds(1)), m2);
+ } finally {
+ gate.countDown();
+ }
+ }
+
+ @Test
+ public void olderReceiveContinuationCancellingTheNewcomerTimedDoesNotLoseAMessage() throws Exception {
+ CompletableFuture<Message<Integer>> a = queue.receiveAsync(Duration.ofSeconds(30));
+ flush();
+ CountDownLatch gate = stallExecutor();
+ try {
+ CompletableFuture<Message<Integer>> b = queue.receiveAsync(Duration.ofSeconds(30));
+ a.thenRun(() -> b.cancel(false));
+ Message<Integer> m1 = msg(1);
+ Message<Integer> m2 = msg(2);
+ queue.offer(m1);
+ queue.offer(m2);
+ gate.countDown();
+ assertSame(a.get(5, TimeUnit.SECONDS), m1);
+ flush();
+ assertTrue(b.isCancelled());
+ assertSame(queue.poll(Duration.ofSeconds(1)), m2);
+ } finally {
+ gate.countDown();
+ }
+ }
+
+ /** Block the single executor thread until the returned latch is counted down. */
+ private CountDownLatch stallExecutor() {
+ CountDownLatch gate = new CountDownLatch(1);
+ executor.execute(() -> {
+ try {
+ gate.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ return gate;
+ }
+
private static Message<Integer> msg(int id) {
return new IntMessage(id);
}