This is an automated email from the ASF dual-hosted git repository.
merlimat 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 92bd0987526 [improve][client] Apply receiver-queue backpressure in v5
scalable consumers (#26151)
92bd0987526 is described below
commit 92bd0987526ee26c88c52e10210d55babdbc9170
Author: Matteo Merli <[email protected]>
AuthorDate: Tue Jul 7 10:26:51 2026 -0700
[improve][client] Apply receiver-queue backpressure in v5 scalable
consumers (#26151)
---
.../apache/pulsar/client/impl/v5/MessageSink.java | 38 +++++++
.../client/impl/v5/MultiTopicQueueConsumer.java | 10 +-
.../client/impl/v5/MultiTopicStreamConsumer.java | 12 +-
.../client/impl/v5/ScalableCheckpointConsumer.java | 17 ++-
.../client/impl/v5/ScalableQueueConsumer.java | 24 ++--
.../client/impl/v5/ScalableStreamConsumer.java | 21 ++--
.../pulsar/client/impl/v5/V5ReceiveQueue.java | 114 +++++++++++++++++--
.../pulsar/client/impl/v5/V5ReceiveQueueTest.java | 121 ++++++++++++++++++++-
8 files changed, 312 insertions(+), 45 deletions(-)
diff --git
a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MessageSink.java
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MessageSink.java
new file mode 100644
index 00000000000..f1f61ac097a
--- /dev/null
+++
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MessageSink.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl.v5;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Where a per-segment receive loop deposits a freshly-arrived message.
+ *
+ * <p>The returned future is the backpressure signal: it completes when the
sink is ready
+ * to accept the next message. Producers gate re-arming their {@code
receiveAsync()} loop
+ * on it, so when the downstream (multiplexed) queue fills up, the per-segment
loops stop
+ * pulling — which lets the underlying v4 consumer's {@code receiverQueueSize}
fill and
+ * stop issuing flow permits, applying backpressure all the way back to the
broker.
+ *
+ * <p>The default sink is {@link V5ReceiveQueue#offer}; the multi-topic
wrappers inject a
+ * sink that forwards into their shared mux so its fullness pauses every
per-topic loop.
+ */
+@FunctionalInterface
+interface MessageSink<T> {
+ CompletableFuture<Void> accept(MessageV5<T> message);
+}
diff --git
a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicQueueConsumer.java
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicQueueConsumer.java
index e02eb325eb8..a83c753437d 100644
---
a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicQueueConsumer.java
+++
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicQueueConsumer.java
@@ -103,7 +103,8 @@ final class MultiTopicQueueConsumer<T> implements
QueueConsumerImpl<T> {
this.subscriptionName = consumerConf.getSubscriptionName();
this.watcher = watcher;
this.mux = new V5ReceiveQueue<>(
- client.v4Client().externalExecutorProvider().getExecutor(),
client.v4Client().timer());
+ client.v4Client().externalExecutorProvider().getExecutor(),
client.v4Client().timer(),
+ consumerConf.getReceiverQueueSize());
this.log = LOG.with()
.attr("namespace", namespace)
.attr("subscription", subscriptionName)
@@ -169,10 +170,11 @@ final class MultiTopicQueueConsumer<T> implements
QueueConsumerImpl<T> {
// Per-topic message sink: tag each delivered message with the parent
scalable
// topic for ack routing + display, and forward into the shared mux.
No pump
// thread; per-segment v4 receive loops fire this sink directly.
- java.util.function.Consumer<MessageV5<T>> sink = msg -> {
- if (!closed) {
- mux.offer(msg.withTopicOverride(topicName));
+ MessageSink<T> sink = msg -> {
+ if (closed) {
+ return CompletableFuture.completedFuture(null);
}
+ return mux.offer(msg.withTopicOverride(topicName));
};
return dagWatch.start()
.thenCompose(layout -> ScalableQueueConsumer.createAsyncImpl(
diff --git
a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicStreamConsumer.java
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicStreamConsumer.java
index ca189af4ea0..7e092569214 100644
---
a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicStreamConsumer.java
+++
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicStreamConsumer.java
@@ -107,7 +107,8 @@ final class MultiTopicStreamConsumer<T> implements
StreamConsumer<T> {
this.subscriptionName = consumerConf.getSubscriptionName();
this.watcher = watcher;
this.mux = new V5ReceiveQueue<>(
- client.v4Client().externalExecutorProvider().getExecutor(),
client.v4Client().timer());
+ client.v4Client().externalExecutorProvider().getExecutor(),
client.v4Client().timer(),
+ consumerConf.getReceiverQueueSize());
this.log = LOG.with()
.attr("namespace", namespace)
.attr("subscription", subscriptionName)
@@ -166,8 +167,7 @@ final class MultiTopicStreamConsumer<T> implements
StreamConsumer<T> {
// single-topic positionVector (computed by ScalableStreamConsumer).
Update
// our cross-topic latestDelivered map, snapshot the full cross-topic
vector,
// and forward to the shared mux. No pump thread.
- java.util.function.Consumer<MessageV5<T>> sink = msg ->
- onPerTopicMessage(topicName, msg);
+ MessageSink<T> sink = msg -> onPerTopicMessage(topicName, msg);
return session.start()
.thenCompose(initialAssignment ->
ScalableStreamConsumer.createAsyncImpl(
@@ -421,9 +421,9 @@ final class MultiTopicStreamConsumer<T> implements
StreamConsumer<T> {
* the only contention is the synchronized snapshot block which guards
* against torn cross-topic views during concurrent deliveries.
*/
- private void onPerTopicMessage(String parentTopic, MessageV5<T> msg) {
+ private CompletableFuture<Void> onPerTopicMessage(String parentTopic,
MessageV5<T> msg) {
if (closed) {
- return;
+ return CompletableFuture.completedFuture(null);
}
MessageIdV5 origId = (MessageIdV5) msg.id();
@@ -448,7 +448,7 @@ final class MultiTopicStreamConsumer<T> implements
StreamConsumer<T> {
MessageIdV5 newId = new MessageIdV5(
origId.v4MessageId(), origId.segmentId(),
origId.positionVector(), parentTopic, snapshot);
- mux.offer(new MessageV5<>(msg.v4Message(), newId, parentTopic));
+ return mux.offer(new MessageV5<>(msg.v4Message(), newId, parentTopic));
}
// --- Per-topic state ---
diff --git
a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableCheckpointConsumer.java
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableCheckpointConsumer.java
index 8e3521af188..a1fa0c20325 100644
---
a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableCheckpointConsumer.java
+++
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableCheckpointConsumer.java
@@ -62,6 +62,9 @@ import
org.apache.pulsar.client.impl.v5.SegmentRouter.ActiveSegment;
final class ScalableCheckpointConsumer<T> implements CheckpointConsumer<T> {
private static final Logger LOG =
Logger.get(ScalableCheckpointConsumer.class);
+ /** Buffer bound for backpressure. The checkpoint consumer has no
ConsumerConfigurationData;
+ * its per-segment readers use the default reader queue size, so mirror
that here. */
+ private static final int DEFAULT_RECEIVER_QUEUE_SIZE = 1000;
private final Logger log;
private final PulsarClientV5 client;
@@ -98,7 +101,8 @@ final class ScalableCheckpointConsumer<T> implements
CheckpointConsumer<T> {
this.startPosition = startPosition;
this.consumerName = consumerName;
this.receiveQueue = new V5ReceiveQueue<>(
- client.v4Client().externalExecutorProvider().getExecutor(),
client.v4Client().timer());
+ client.v4Client().externalExecutorProvider().getExecutor(),
client.v4Client().timer(),
+ DEFAULT_RECEIVER_QUEUE_SIZE);
this.log = LOG.with().attr("topic", topicName).build();
this.asyncView = new AsyncCheckpointConsumerV5<>(this);
}
@@ -378,10 +382,13 @@ final class ScalableCheckpointConsumer<T> implements
CheckpointConsumer<T> {
// taken right after the app received message N could already
point past
// N+1 if the read loop got ahead). The advance happens in
receive() /
// receiveMulti() instead, where the message crosses into
application code.
- receiveQueue.offer(new MessageV5<>(v4Msg, segmentId));
- if (!closed) {
- startReadLoop(reader, segmentId);
- }
+ // Re-arm only once the buffer has room, so a slow consumer pauses
this reader
+ // instead of pre-fetching unboundedly.
+ receiveQueue.offer(new MessageV5<>(v4Msg, segmentId)).thenRun(()
-> {
+ if (!closed) {
+ startReadLoop(reader, segmentId);
+ }
+ });
}).exceptionally(ex -> {
Throwable cause = ex instanceof CompletionException ce &&
ce.getCause() != null
? ce.getCause() : ex;
diff --git
a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableQueueConsumer.java
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableQueueConsumer.java
index fef86429bf3..e1de657659d 100644
---
a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableQueueConsumer.java
+++
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableQueueConsumer.java
@@ -85,7 +85,7 @@ final class ScalableQueueConsumer<T> implements
QueueConsumerImpl<T>, DagWatchCl
* the multi-topic wrapper overrides this to forward directly into its
shared
* multiplexed queue, so no per-topic pump thread is needed.
*/
- private final java.util.function.Consumer<MessageV5<T>> messageSink;
+ private final MessageSink<T> messageSink;
/**
* V5-layer DLQ. Owned at the V5 consumer (not per-segment) so a single
producer
@@ -115,7 +115,7 @@ final class ScalableQueueConsumer<T> implements
QueueConsumerImpl<T>, DagWatchCl
Schema<T> v5Schema,
ConsumerConfigurationData<T> consumerConf,
DagWatchClient dagWatch,
- java.util.function.Consumer<MessageV5<T>>
messageSink,
+ MessageSink<T> messageSink,
DeadLetterPolicy dlqPolicy) {
this.client = client;
this.v5Schema = v5Schema;
@@ -130,7 +130,8 @@ final class ScalableQueueConsumer<T> implements
QueueConsumerImpl<T>, DagWatchCl
// Multi-topic mode passes a sink that forwards into the shared mux
instead — no
// per-topic pump thread needed.
this.receiveQueue = new V5ReceiveQueue<>(
- client.v4Client().externalExecutorProvider().getExecutor(),
client.v4Client().timer());
+ client.v4Client().externalExecutorProvider().getExecutor(),
client.v4Client().timer(),
+ consumerConf.getReceiverQueueSize());
this.messageSink = messageSink != null ? messageSink :
receiveQueue::offer;
this.log = LOG.with().attr("topic", topicName).attr("subscription",
subscriptionName).build();
this.asyncView = new AsyncQueueConsumerV5<>(this);
@@ -179,7 +180,7 @@ final class ScalableQueueConsumer<T> implements
QueueConsumerImpl<T>, DagWatchCl
ConsumerConfigurationData<T> consumerConf,
DagWatchClient dagWatch,
ClientSegmentLayout initialLayout,
- java.util.function.Consumer<MessageV5<T>> messageSink,
+ MessageSink<T> messageSink,
DeadLetterPolicy dlqPolicy) {
ScalableQueueConsumer<T> consumer = new ScalableQueueConsumer<>(
client, v5Schema, consumerConf, dagWatch, messageSink,
dlqPolicy);
@@ -433,14 +434,21 @@ final class ScalableQueueConsumer<T> implements
QueueConsumerImpl<T>, DagWatchCl
private void startReceiveLoop(org.apache.pulsar.client.api.Consumer<T>
v4Consumer, long segmentId) {
v4Consumer.receiveAsync().thenAccept(v4Msg -> {
+ CompletableFuture<Void> ready;
if (shouldGoToDlq(v4Msg)) {
forwardToDlq(v4Msg, v4Consumer);
+ // DLQ-forwarded messages never enter the receive buffer —
nothing to wait on.
+ ready = CompletableFuture.completedFuture(null);
} else {
- messageSink.accept(new MessageV5<>(v4Msg, segmentId));
- }
- if (!closed) {
- startReceiveLoop(v4Consumer, segmentId);
+ ready = messageSink.accept(new MessageV5<>(v4Msg, segmentId));
}
+ // Re-arm only once the sink has room, so a slow consumer pauses
this segment's
+ // receive loop (and the v4 flow-control permits) instead of
buffering unboundedly.
+ ready.thenRun(() -> {
+ if (!closed) {
+ startReceiveLoop(v4Consumer, segmentId);
+ }
+ });
}).exceptionally(ex -> {
Throwable cause = ex instanceof CompletionException ce &&
ce.getCause() != null ? ce.getCause() : ex;
if (closed
diff --git
a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableStreamConsumer.java
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableStreamConsumer.java
index df13bd965f7..b22a62ddd7f 100644
---
a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableStreamConsumer.java
+++
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableStreamConsumer.java
@@ -97,7 +97,7 @@ final class ScalableStreamConsumer<T>
* the multi-topic wrapper overrides this to forward into its shared
multiplexed
* queue, applying its own multi-topic position-vector capture in the
process.
*/
- private final java.util.function.Consumer<MessageV5<T>> messageSink;
+ private final MessageSink<T> messageSink;
private volatile boolean closed = false;
private final AsyncStreamConsumerV5<T> asyncView;
@@ -107,7 +107,7 @@ final class ScalableStreamConsumer<T>
ConsumerConfigurationData<T> consumerConf,
ScalableConsumerClient session,
String topicName,
- java.util.function.Consumer<MessageV5<T>>
messageSink) {
+ MessageSink<T> messageSink) {
this.client = client;
this.v5Schema = v5Schema;
this.v4Schema = SchemaAdapter.toV4(v5Schema);
@@ -116,7 +116,8 @@ final class ScalableStreamConsumer<T>
this.topicName = topicName;
this.subscriptionName = consumerConf.getSubscriptionName();
this.receiveQueue = new V5ReceiveQueue<>(
- client.v4Client().externalExecutorProvider().getExecutor(),
client.v4Client().timer());
+ client.v4Client().externalExecutorProvider().getExecutor(),
client.v4Client().timer(),
+ consumerConf.getReceiverQueueSize());
this.messageSink = messageSink != null ? messageSink :
receiveQueue::offer;
this.log = LOG.with().attr("topic", topicName).attr("subscription",
subscriptionName).build();
this.asyncView = new AsyncStreamConsumerV5<>(this);
@@ -151,7 +152,7 @@ final class ScalableStreamConsumer<T>
ScalableConsumerClient session,
String topicName,
List<ActiveSegment> initialAssignment,
- java.util.function.Consumer<MessageV5<T>> messageSink) {
+ MessageSink<T> messageSink) {
ScalableStreamConsumer<T> consumer = new ScalableStreamConsumer<>(
client, v5Schema, consumerConf, session, topicName,
messageSink);
return consumer.subscribeAssigned(initialAssignment)
@@ -410,11 +411,13 @@ final class ScalableStreamConsumer<T>
// Create the V5 message with the position vector embedded in the
ID
var msgId = new MessageIdV5(v4Msg.getMessageId(), segmentId,
positionVector);
- messageSink.accept(new MessageV5<>(v4Msg, msgId));
-
- if (!closed) {
- startReceiveLoop(v4Consumer, segmentId);
- }
+ // Re-arm only once the sink has room, so a slow consumer pauses
this segment's
+ // receive loop (and the v4 flow-control permits) instead of
buffering unboundedly.
+ messageSink.accept(new MessageV5<>(v4Msg, msgId)).thenRun(() -> {
+ if (!closed) {
+ startReceiveLoop(v4Consumer, segmentId);
+ }
+ });
}).exceptionally(ex -> {
Throwable cause = ex instanceof
java.util.concurrent.CompletionException ce
&& ce.getCause() != null ? ce.getCause() : ex;
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 92a8e479417..e56d7f8ac03 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
@@ -44,39 +44,116 @@ import
org.apache.pulsar.client.api.v5.PulsarClientException;
* <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.
*/
final class V5ReceiveQueue<T> {
+ /** Shared pre-completed capacity grant for the fast path — no allocation,
no executor hop. */
+ private static final CompletableFuture<Void> READY =
CompletableFuture.completedFuture(null);
+
private final ExecutorService executor;
private final Timer timer;
+ /** Producers pause once the buffer reaches this size, and resume once it
drains to half. */
+ private final int highWatermark;
+ private final int lowWatermark;
- // Both touched only on `executor`, so plain (non-concurrent) collections
are safe.
+ // All three touched only on `executor`, so plain (non-concurrent)
collections are safe.
private final ArrayDeque<Message<T>> buffer = new ArrayDeque<>();
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;
- V5ReceiveQueue(ExecutorService executor, Timer timer) {
+ // 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;
+ private volatile boolean producersPaused = false;
+
+ V5ReceiveQueue(ExecutorService executor, Timer timer, int
receiverQueueSize) {
this.executor = executor;
this.timer = timer;
+ this.highWatermark = Math.max(1, receiverQueueSize);
+ this.lowWatermark = highWatermark / 2;
}
/**
* 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.
+ *
+ * @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).
*/
- void offer(Message<T> msg) {
- executor.execute(() -> {
- if (closed) {
- return;
+ 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));
+ return READY;
+ }
+ CompletableFuture<Void> capacity = new CompletableFuture<>();
+ executor.execute(() -> doOffer(msg, 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);
}
- CompletableFuture<Message<T>> waiter = pollWaiter();
- if (waiter != null) {
- waiter.complete(msg);
- } else {
- buffer.add(msg);
+ 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;
+ }
+ // 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)) {
+ capacityWaiters.add(capacity);
+ producersPaused = true;
+ } else {
+ capacity.complete(null);
+ }
+ }
+
+ /** Resume paused producers once the buffer has drained to the low
watermark. */
+ private void maybeResumeProducers() {
+ if (buffer.size() <= lowWatermark && !capacityWaiters.isEmpty()) {
+ CompletableFuture<Void> capacity;
+ while ((capacity = capacityWaiters.poll()) != null) {
+ // Post each completion as its own task (as v4 does) instead
of completing
+ // inline: each grant synchronously runs a segment loop's
re-arm, and a wide
+ // release would otherwise stall queued user receive
completions behind it.
+ CompletableFuture<Void> c = capacity;
+ executor.execute(() -> c.complete(null));
+ }
+ producersPaused = false;
+ }
}
/** Receive a message, completing as soon as one is available. Never
blocks a thread. */
@@ -89,7 +166,9 @@ final class V5ReceiveQueue<T> {
}
Message<T> msg = buffer.poll();
if (msg != null) {
+ approxBufferSize = buffer.size();
result.complete(msg);
+ maybeResumeProducers();
} else {
pendingReceives.add(result);
}
@@ -110,7 +189,9 @@ final class V5ReceiveQueue<T> {
}
Message<T> msg = buffer.poll();
if (msg != null) {
+ approxBufferSize = buffer.size();
result.complete(msg);
+ maybeResumeProducers();
return;
}
long millis = timeout.toMillis();
@@ -174,6 +255,8 @@ final class V5ReceiveQueue<T> {
while (batch.size() < max && (m = buffer.poll()) != null) {
batch.add(m);
}
+ approxBufferSize = buffer.size();
+ maybeResumeProducers();
done.complete(null);
});
return done;
@@ -224,7 +307,14 @@ final class V5ReceiveQueue<T> {
waiter.completeExceptionally(alreadyClosed());
}
}
+ // Release any paused producers so their receive loops re-arm and
observe the close.
+ CompletableFuture<Void> capacity;
+ while ((capacity = capacityWaiters.poll()) != null) {
+ capacity.complete(null);
+ }
+ 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 61d326c18aa..e5644a3c401 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
@@ -51,7 +51,14 @@ public class V5ReceiveQueueTest {
public void setup() {
executor = Executors.newSingleThreadExecutor();
timer = new HashedWheelTimer();
- queue = new V5ReceiveQueue<>(executor, timer);
+ queue = new V5ReceiveQueue<>(executor, timer, 1000);
+ }
+
+ /** Wait for the single-threaded executor to drain queued tasks
(offer/receive run there). */
+ private void flush() throws Exception {
+ CompletableFuture<Void> f = new CompletableFuture<>();
+ executor.execute(() -> f.complete(null));
+ f.get(5, TimeUnit.SECONDS);
}
@AfterMethod(alwaysRun = true)
@@ -168,6 +175,118 @@ public class V5ReceiveQueueTest {
assertTrue(batch.isEmpty());
}
+ @Test
+ public void fastPathGrantsCapacityImmediatelyWhenBufferHasRoom() {
+ // Empty buffer, nobody paused: offer() must return an
already-completed (shared)
+ // future decided on the caller thread — no executor hop before the
loop re-arms.
+ assertTrue(queue.offer(msg(1)).isDone());
+ }
+
+ @Test
+ public void offerBelowWatermarkCompletesImmediately() throws Exception {
+ // receiverQueueSize=4 -> pause at 4, resume at 2.
+ V5ReceiveQueue<Integer> q = new V5ReceiveQueue<>(executor, timer, 4);
+ for (int i = 1; i <= 3; i++) {
+ // Completes right away while there is room in the buffer.
+ q.offer(msg(i)).get(5, TimeUnit.SECONDS);
+ flush();
+ }
+ }
+
+ @Test
+ public void offerAtHighWatermarkPausesUntilDrain() throws Exception {
+ V5ReceiveQueue<Integer> q = new V5ReceiveQueue<>(executor, timer, 4);
+ for (int i = 1; i <= 3; i++) {
+ q.offer(msg(i)).get(5, TimeUnit.SECONDS);
+ // Let the enqueue task run so the fast-path size snapshot is
exact and the
+ // 4th offer deterministically takes the slow path.
+ flush();
+ }
+ // The 4th message hits the high watermark and parks the producer.
+ CompletableFuture<Void> paused = q.offer(msg(4));
+ flush();
+ assertTrue(!paused.isDone());
+
+ // Draining below the low watermark (2) resumes it.
+ q.receiveAsync().get(5, TimeUnit.SECONDS); // buffer 4 -> 3, still
above low
+ assertTrue(!paused.isDone());
+ q.receiveAsync().get(5, TimeUnit.SECONDS); // buffer 3 -> 2, resume
+ paused.get(5, TimeUnit.SECONDS);
+ }
+
+ @Test
+ public void fairnessPausesSubsequentOffersOncePeerIsPaused() throws
Exception {
+ // receiverQueueSize=8 -> pause at 8 (or at >4 once a peer is paused),
resume at 4.
+ V5ReceiveQueue<Integer> q = new V5ReceiveQueue<>(executor, timer, 8);
+ for (int i = 1; i <= 7; i++) {
+ q.offer(msg(i)).get(5, TimeUnit.SECONDS);
+ flush();
+ }
+ CompletableFuture<Void> pausedA = q.offer(msg(8)); // buffer 8 >= high
-> parked
+ flush();
+ assertTrue(!pausedA.isDone());
+
+ for (int i = 0; i < 3; i++) {
+ q.receiveAsync().get(5, TimeUnit.SECONDS); // buffer 8 -> 5,
above low
+ }
+ assertTrue(!pausedA.isDone());
+
+ // v4 fairness clause: with a peer already paused, a new offer above
the low
+ // watermark must park too instead of overtaking the paused one
indefinitely.
+ CompletableFuture<Void> pausedB = q.offer(msg(9)); // buffer 6 >
low(4) with peer paused
+ flush();
+ assertTrue(!pausedB.isDone());
+
+ q.receiveAsync().get(5, TimeUnit.SECONDS); // buffer 6 -> 5
+ q.receiveAsync().get(5, TimeUnit.SECONDS); // buffer 5 -> 4 ==
low, resume all
+ pausedA.get(5, TimeUnit.SECONDS);
+ pausedB.get(5, TimeUnit.SECONDS);
+ }
+
+ @Test
+ public void resumeViaReceiveMultiDrain() throws Exception {
+ V5ReceiveQueue<Integer> q = new V5ReceiveQueue<>(executor, timer, 4);
+ for (int i = 1; i <= 3; i++) {
+ q.offer(msg(i)).get(5, TimeUnit.SECONDS);
+ flush();
+ }
+ CompletableFuture<Void> paused = q.offer(msg(4));
+ flush();
+ assertTrue(!paused.isDone());
+
+ // Batch drain crosses the low watermark (4 -> 1) and must resume the
producer.
+ List<Message<Integer>> batch = q.receiveMulti(3,
Duration.ofSeconds(5));
+ assertEquals(batch.size(), 3);
+ paused.get(5, TimeUnit.SECONDS);
+ }
+
+ @Test
+ public void resumeViaTimedReceiveDrain() throws Exception {
+ V5ReceiveQueue<Integer> q = new V5ReceiveQueue<>(executor, timer, 2);
+ q.offer(msg(1)).get(5, TimeUnit.SECONDS);
+ flush();
+ CompletableFuture<Void> paused = q.offer(msg(2)); // buffer 2 >= high
-> parked
+ flush();
+ assertTrue(!paused.isDone());
+
+ // Timed receive's immediate-poll path (2 -> 1 == low) must resume the
producer.
+ assertEquals(q.poll(Duration.ofSeconds(5)).value(),
Integer.valueOf(1));
+ paused.get(5, TimeUnit.SECONDS);
+ }
+
+ @Test
+ public void closeReleasesPausedProducers() throws Exception {
+ V5ReceiveQueue<Integer> q = new V5ReceiveQueue<>(executor, timer, 2);
+ q.offer(msg(1)).get(5, TimeUnit.SECONDS); // buffer 1, below
watermark
+ flush();
+ CompletableFuture<Void> paused = q.offer(msg(2)); // buffer 2, parked
+ flush();
+ assertTrue(!paused.isDone());
+ // close() must release parked producers so their receive loops
observe the close.
+ q.close();
+ paused.get(5, TimeUnit.SECONDS);
+ }
+
private static Message<Integer> msg(int id) {
return new IntMessage(id);
}
