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 b8053046290 [improve][client] Make v5 scalable consumer receiveAsync
truly non-blocking (#26131)
b8053046290 is described below
commit b8053046290fc0740b669b10604b052a4b3e5945
Author: Matteo Merli <[email protected]>
AuthorDate: Thu Jul 2 14:55:02 2026 -0700
[improve][client] Make v5 scalable consumer receiveAsync truly non-blocking
(#26131)
---
.../client/impl/v5/AsyncCheckpointConsumerV5.java | 11 +-
.../client/impl/v5/AsyncStreamConsumerV5.java | 11 +-
.../client/impl/v5/MultiTopicQueueConsumer.java | 30 +--
.../client/impl/v5/MultiTopicStreamConsumer.java | 76 +------
.../client/impl/v5/ScalableCheckpointConsumer.java | 73 ++----
.../client/impl/v5/ScalableQueueConsumer.java | 34 +--
.../client/impl/v5/ScalableStreamConsumer.java | 68 ++----
.../pulsar/client/impl/v5/V5ReceiveQueue.java | 253 +++++++++++++++++++++
.../pulsar/client/impl/v5/V5ReceiveQueueTest.java | 248 ++++++++++++++++++++
9 files changed, 565 insertions(+), 239 deletions(-)
diff --git
a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/AsyncCheckpointConsumerV5.java
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/AsyncCheckpointConsumerV5.java
index ace455e7ba9..7c8944708ea 100644
---
a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/AsyncCheckpointConsumerV5.java
+++
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/AsyncCheckpointConsumerV5.java
@@ -48,16 +48,7 @@ final class AsyncCheckpointConsumerV5<T> implements
AsyncCheckpointConsumer<T> {
@Override
public CompletableFuture<List<Message<T>>> receiveMulti(int maxMessages,
Duration timeout) {
- return CompletableFuture.supplyAsync(() -> {
- try {
- var msgs = consumer.receiveMulti(maxMessages, timeout);
- List<Message<T>> result = new java.util.ArrayList<>();
- msgs.forEach(result::add);
- return result;
- } catch (Exception e) {
- throw new java.util.concurrent.CompletionException(e);
- }
- });
+ return consumer.receiveMultiAsync(maxMessages, timeout);
}
@Override
diff --git
a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/AsyncStreamConsumerV5.java
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/AsyncStreamConsumerV5.java
index 36afa102db0..0955e357e9b 100644
---
a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/AsyncStreamConsumerV5.java
+++
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/AsyncStreamConsumerV5.java
@@ -49,16 +49,7 @@ final class AsyncStreamConsumerV5<T> implements
AsyncStreamConsumer<T> {
@Override
public CompletableFuture<List<Message<T>>> receiveMulti(int
maxNumMessages, Duration timeout) {
- return CompletableFuture.supplyAsync(() -> {
- try {
- var msgs = consumer.receiveMulti(maxNumMessages, timeout);
- List<Message<T>> result = new java.util.ArrayList<>();
- msgs.forEach(result::add);
- return result;
- } catch (Exception e) {
- throw new java.util.concurrent.CompletionException(e);
- }
- });
+ return consumer.receiveMultiAsync(maxNumMessages, timeout);
}
@Override
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 52d956e1b58..e02eb325eb8 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
@@ -31,7 +31,6 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
-import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.pulsar.client.api.v5.Message;
@@ -85,7 +84,7 @@ final class MultiTopicQueueConsumer<T> implements
QueueConsumerImpl<T> {
private final ScalableTopicsWatcher watcher;
private final ConcurrentHashMap<String, PerTopicState<T>> perTopic = new
ConcurrentHashMap<>();
- private final LinkedTransferQueue<MessageV5<T>> mux = new
LinkedTransferQueue<>();
+ private final V5ReceiveQueue<T> mux;
private volatile boolean closed = false;
private final AsyncQueueConsumerV5<T> asyncView;
@@ -103,6 +102,8 @@ final class MultiTopicQueueConsumer<T> implements
QueueConsumerImpl<T> {
this.propertyFilters = propertyFilters;
this.subscriptionName = consumerConf.getSubscriptionName();
this.watcher = watcher;
+ this.mux = new V5ReceiveQueue<>(
+ client.v4Client().externalExecutorProvider().getExecutor(),
client.v4Client().timer());
this.log = LOG.with()
.attr("namespace", namespace)
.attr("subscription", subscriptionName)
@@ -170,7 +171,7 @@ final class MultiTopicQueueConsumer<T> implements
QueueConsumerImpl<T> {
// thread; per-segment v4 receive loops fire this sink directly.
java.util.function.Consumer<MessageV5<T>> sink = msg -> {
if (!closed) {
- mux.add(msg.withTopicOverride(topicName));
+ mux.offer(msg.withTopicOverride(topicName));
}
};
return dagWatch.start()
@@ -299,22 +300,12 @@ final class MultiTopicQueueConsumer<T> implements
QueueConsumerImpl<T> {
@Override
public Message<T> receive() throws PulsarClientException {
- try {
- return mux.take();
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new PulsarClientException("Receive interrupted", e);
- }
+ return mux.take();
}
@Override
public Message<T> receive(Duration timeout) throws PulsarClientException {
- try {
- return mux.poll(timeout.toMillis(), TimeUnit.MILLISECONDS);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new PulsarClientException("Receive interrupted", e);
- }
+ return mux.poll(timeout);
}
@Override
@@ -372,13 +363,7 @@ final class MultiTopicQueueConsumer<T> implements
QueueConsumerImpl<T> {
@Override
public CompletableFuture<Message<T>> receiveAsync() {
- return CompletableFuture.supplyAsync(() -> {
- try {
- return receive();
- } catch (PulsarClientException e) {
- throw new CompletionException(e);
- }
- });
+ return mux.receiveAsync();
}
@Override
@@ -388,6 +373,7 @@ final class MultiTopicQueueConsumer<T> implements
QueueConsumerImpl<T> {
}
closed = true;
watcher.close();
+ mux.close();
// Cancel pending retries for topics that never finished subscribing
(they're not in
// perTopic, so the closeTopic loop below wouldn't reach them).
retryTimeouts.values().forEach(Timeout::cancel);
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 ffacdcad11d..ca189af4ea0 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
@@ -31,7 +31,6 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
-import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.pulsar.client.api.v5.Message;
@@ -81,7 +80,7 @@ final class MultiTopicStreamConsumer<T> implements
StreamConsumer<T> {
private final ScalableTopicsWatcher watcher;
private final ConcurrentHashMap<String, PerTopic<T>> perTopic = new
ConcurrentHashMap<>();
- private final LinkedTransferQueue<MessageV5<T>> mux = new
LinkedTransferQueue<>();
+ private final V5ReceiveQueue<T> mux;
/**
* Tracks the latest delivered message id per (parent topic, segment id)
across
@@ -107,6 +106,8 @@ final class MultiTopicStreamConsumer<T> implements
StreamConsumer<T> {
this.propertyFilters = propertyFilters;
this.subscriptionName = consumerConf.getSubscriptionName();
this.watcher = watcher;
+ this.mux = new V5ReceiveQueue<>(
+ client.v4Client().externalExecutorProvider().getExecutor(),
client.v4Client().timer());
this.log = LOG.with()
.attr("namespace", namespace)
.attr("subscription", subscriptionName)
@@ -292,50 +293,17 @@ final class MultiTopicStreamConsumer<T> implements
StreamConsumer<T> {
@Override
public Message<T> receive() throws PulsarClientException {
- try {
- return mux.take();
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new PulsarClientException("Receive interrupted", e);
- }
+ return mux.take();
}
@Override
public Message<T> receive(Duration timeout) throws PulsarClientException {
- try {
- return mux.poll(timeout.toMillis(), TimeUnit.MILLISECONDS);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new PulsarClientException("Receive interrupted", e);
- }
+ return mux.poll(timeout);
}
@Override
public Messages<T> receiveMulti(int maxNumMessages, Duration timeout)
throws PulsarClientException {
- // Block for up to `timeout` waiting for the first message, then drain
whatever
- // else is immediately available up to maxNumMessages. Same shape as
the single
- // topic StreamConsumer.
- long deadline = System.nanoTime() + timeout.toNanos();
- List<Message<T>> batch = new ArrayList<>();
- try {
- long remaining = deadline - System.nanoTime();
- while (batch.size() < maxNumMessages && remaining > 0) {
- MessageV5<T> msg = mux.poll(remaining, TimeUnit.NANOSECONDS);
- if (msg == null) {
- break;
- }
- batch.add(msg);
- remaining = deadline - System.nanoTime();
- }
- // Opportunistic drain of anything else already queued.
- List<MessageV5<T>> tail = new ArrayList<>();
- mux.drainTo(tail, maxNumMessages - batch.size());
- batch.addAll(tail);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new PulsarClientException("Receive interrupted", e);
- }
- return new MessagesV5<>(batch);
+ return new MessagesV5<>(mux.receiveMulti(maxNumMessages, timeout));
}
@Override
@@ -398,6 +366,7 @@ final class MultiTopicStreamConsumer<T> implements
StreamConsumer<T> {
}
closed = true;
watcher.close();
+ mux.close();
// Cancel pending retries for topics that never finished subscribing
(they're not in
// perTopic, so the closeTopic loop below wouldn't reach them).
retryTimeouts.values().forEach(Timeout::cancel);
@@ -479,7 +448,7 @@ final class MultiTopicStreamConsumer<T> implements
StreamConsumer<T> {
MessageIdV5 newId = new MessageIdV5(
origId.v4MessageId(), origId.segmentId(),
origId.positionVector(), parentTopic, snapshot);
- mux.add(new MessageV5<>(msg.v4Message(), newId, parentTopic));
+ mux.offer(new MessageV5<>(msg.v4Message(), newId, parentTopic));
}
// --- Per-topic state ---
@@ -503,40 +472,17 @@ final class MultiTopicStreamConsumer<T> implements
StreamConsumer<T> {
private final class AsyncStreamConsumerV5Multi implements
AsyncStreamConsumer<T> {
@Override
public CompletableFuture<Message<T>> receive() {
- return CompletableFuture.supplyAsync(() -> {
- try {
- return MultiTopicStreamConsumer.this.receive();
- } catch (PulsarClientException e) {
- throw new CompletionException(e);
- }
- });
+ return mux.receiveAsync();
}
@Override
public CompletableFuture<Message<T>> receive(Duration timeout) {
- return CompletableFuture.supplyAsync(() -> {
- try {
- return MultiTopicStreamConsumer.this.receive(timeout);
- } catch (PulsarClientException e) {
- throw new CompletionException(e);
- }
- });
+ return mux.receiveAsync(timeout);
}
@Override
public CompletableFuture<List<Message<T>>> receiveMulti(int
maxNumMessages, Duration timeout) {
- return CompletableFuture.supplyAsync(() -> {
- try {
- Messages<T> ms =
MultiTopicStreamConsumer.this.receiveMulti(maxNumMessages, timeout);
- List<Message<T>> out = new ArrayList<>();
- for (Message<T> m : ms) {
- out.add(m);
- }
- return out;
- } catch (PulsarClientException e) {
- throw new CompletionException(e);
- }
- });
+ return mux.receiveMultiAsync(maxNumMessages, timeout);
}
@Override
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 8f8540af4ea..8e3521af188 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
@@ -28,8 +28,6 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
-import java.util.concurrent.LinkedTransferQueue;
-import java.util.concurrent.TimeUnit;
import
org.apache.pulsar.client.api.PulsarClientException.AlreadyClosedException;
import org.apache.pulsar.client.api.Reader;
import org.apache.pulsar.client.api.v5.Checkpoint;
@@ -81,7 +79,7 @@ final class ScalableCheckpointConsumer<T> implements
CheckpointConsumer<T> {
private final ConcurrentHashMap<Long, CompletableFuture<Reader<T>>>
segmentReaders = new ConcurrentHashMap<>();
private final ConcurrentHashMap<Long,
org.apache.pulsar.client.api.MessageId> lastReceivedPositions =
new ConcurrentHashMap<>();
- private final LinkedTransferQueue<MessageV5<T>> messageQueue = new
LinkedTransferQueue<>();
+ private final V5ReceiveQueue<T> receiveQueue;
private volatile boolean closed = false;
private final AsyncCheckpointConsumerV5<T> asyncView;
@@ -99,6 +97,8 @@ final class ScalableCheckpointConsumer<T> implements
CheckpointConsumer<T> {
this.topicName = topicName;
this.startPosition = startPosition;
this.consumerName = consumerName;
+ this.receiveQueue = new V5ReceiveQueue<>(
+ client.v4Client().externalExecutorProvider().getExecutor(),
client.v4Client().timer());
this.log = LOG.with().attr("topic", topicName).build();
this.asyncView = new AsyncCheckpointConsumerV5<>(this);
}
@@ -168,51 +168,18 @@ final class ScalableCheckpointConsumer<T> implements
CheckpointConsumer<T> {
@Override
public Message<T> receive() throws PulsarClientException {
- try {
- return advanceCheckpoint(messageQueue.take());
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new PulsarClientException("Receive interrupted", e);
- }
+ return advanceCheckpoint(receiveQueue.take());
}
@Override
public Message<T> receive(Duration timeout) throws PulsarClientException {
- try {
- return advanceCheckpoint(messageQueue.poll(timeout.toMillis(),
TimeUnit.MILLISECONDS));
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new PulsarClientException("Receive interrupted", e);
- }
+ return advanceCheckpoint(receiveQueue.poll(timeout));
}
@Override
public Messages<T> receiveMulti(int maxMessages, Duration timeout) throws
PulsarClientException {
- List<Message<T>> batch = new ArrayList<>();
- long deadlineNanos = System.nanoTime() + timeout.toNanos();
-
- while (batch.size() < maxMessages) {
- long remainingNanos = deadlineNanos - System.nanoTime();
- if (remainingNanos <= 0) {
- break;
- }
- try {
- MessageV5<T> msg = messageQueue.poll(remainingNanos,
TimeUnit.NANOSECONDS);
- if (msg == null) {
- break;
- }
- batch.add(advanceCheckpoint(msg));
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new PulsarClientException("Receive interrupted", e);
- }
- // Drain whatever else is immediately ready up to maxMessages.
- List<Message<T>> drained = new ArrayList<>();
- messageQueue.drainTo(drained, maxMessages - batch.size());
- for (Message<T> drainedMsg : drained) {
- batch.add(advanceCheckpoint(drainedMsg));
- }
- }
+ List<Message<T>> batch = receiveQueue.receiveMulti(maxMessages,
timeout);
+ batch.forEach(this::advanceCheckpoint);
return new MessagesV5<>(batch);
}
@@ -258,22 +225,17 @@ final class ScalableCheckpointConsumer<T> implements
CheckpointConsumer<T> {
// --- Async internals ---
CompletableFuture<Message<T>> receiveAsync() {
- return CompletableFuture.supplyAsync(() -> {
- try {
- return receive();
- } catch (PulsarClientException e) {
- throw new CompletionException(e);
- }
- });
+ return receiveQueue.receiveAsync().thenApply(this::advanceCheckpoint);
}
CompletableFuture<Message<T>> receiveAsync(Duration timeout) {
- return CompletableFuture.supplyAsync(() -> {
- try {
- return receive(timeout);
- } catch (PulsarClientException e) {
- throw new CompletionException(e);
- }
+ return
receiveQueue.receiveAsync(timeout).thenApply(this::advanceCheckpoint);
+ }
+
+ CompletableFuture<List<Message<T>>> receiveMultiAsync(int maxMessages,
Duration timeout) {
+ return receiveQueue.receiveMultiAsync(maxMessages,
timeout).thenApply(batch -> {
+ batch.forEach(this::advanceCheckpoint);
+ return batch;
});
}
@@ -283,6 +245,7 @@ final class ScalableCheckpointConsumer<T> implements
CheckpointConsumer<T> {
CompletableFuture<Void> closeAsync() {
closed = true;
+ receiveQueue.close();
try {
sourceHandle.close();
} catch (Exception e) {
@@ -415,7 +378,7 @@ 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.
- messageQueue.add(new MessageV5<>(v4Msg, segmentId));
+ receiveQueue.offer(new MessageV5<>(v4Msg, segmentId));
if (!closed) {
startReadLoop(reader, segmentId);
}
@@ -431,7 +394,7 @@ final class ScalableCheckpointConsumer<T> implements
CheckpointConsumer<T> {
.TopicTerminatedException) {
// Sealed segment fully drained server-side. Close the reader
and drop
// it from the map so resources are released; the segment's
data has
- // already crossed into messageQueue.
+ // already crossed into receiveQueue.
log.info().attr("segmentId", segmentId)
.log("Sealed segment drained, closing reader");
segmentReaders.remove(segmentId);
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 3ead54e6955..fef86429bf3 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
@@ -29,7 +29,6 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
-import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -79,10 +78,10 @@ final class ScalableQueueConsumer<T> implements
QueueConsumerImpl<T>, DagWatchCl
*/
private final ConcurrentHashMap<Long,
CompletableFuture<org.apache.pulsar.client.api.Consumer<T>>>
segmentConsumers = new ConcurrentHashMap<>();
- private final LinkedTransferQueue<MessageV5<T>> messageQueue = new
LinkedTransferQueue<>();
+ private final V5ReceiveQueue<T> receiveQueue;
/**
* Where each per-segment receive loop deposits a freshly-arrived message.
Defaults
- * to enqueueing on {@link #messageQueue} for the user's {@link
#receive()} to pull;
+ * to enqueueing on {@link #receiveQueue} for the user's {@link
#receive()} to pull;
* the multi-topic wrapper overrides this to forward directly into its
shared
* multiplexed queue, so no per-topic pump thread is needed.
*/
@@ -127,10 +126,12 @@ final class ScalableQueueConsumer<T> implements
QueueConsumerImpl<T>, DagWatchCl
this.subscriptionName = consumerConf.getSubscriptionName();
this.dlqPolicy = dlqPolicy;
this.dlqTopic = dlqPolicy == null ? null : resolveDlqTopic(dlqPolicy);
- // Default sink enqueues on the local messageQueue for
receive()/receive(timeout).
+ // Default sink enqueues on the local receiveQueue for
receive()/receive(timeout).
// Multi-topic mode passes a sink that forwards into the shared mux
instead — no
// per-topic pump thread needed.
- this.messageSink = messageSink != null ? messageSink :
messageQueue::add;
+ this.receiveQueue = new V5ReceiveQueue<>(
+ client.v4Client().externalExecutorProvider().getExecutor(),
client.v4Client().timer());
+ this.messageSink = messageSink != null ? messageSink :
receiveQueue::offer;
this.log = LOG.with().attr("topic", topicName).attr("subscription",
subscriptionName).build();
this.asyncView = new AsyncQueueConsumerV5<>(this);
}
@@ -209,22 +210,12 @@ final class ScalableQueueConsumer<T> implements
QueueConsumerImpl<T>, DagWatchCl
@Override
public Message<T> receive() throws PulsarClientException {
- try {
- return messageQueue.take();
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new PulsarClientException("Receive interrupted", e);
- }
+ return receiveQueue.take();
}
@Override
public Message<T> receive(Duration timeout) throws PulsarClientException {
- try {
- return messageQueue.poll(timeout.toMillis(),
TimeUnit.MILLISECONDS);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new PulsarClientException("Receive interrupted", e);
- }
+ return receiveQueue.poll(timeout);
}
@Override
@@ -281,18 +272,13 @@ final class ScalableQueueConsumer<T> implements
QueueConsumerImpl<T>, DagWatchCl
@Override
public CompletableFuture<Message<T>> receiveAsync() {
- return CompletableFuture.supplyAsync(() -> {
- try {
- return receive();
- } catch (PulsarClientException e) {
- throw new CompletionException(e);
- }
- });
+ return receiveQueue.receiveAsync();
}
@Override
public CompletableFuture<Void> closeAsync() {
closed = true;
+ receiveQueue.close();
dagWatch.close();
List<CompletableFuture<Void>> futures = new ArrayList<>();
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 3e622baf471..df13bd965f7 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
@@ -28,8 +28,6 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
-import java.util.concurrent.LinkedTransferQueue;
-import java.util.concurrent.TimeUnit;
import org.apache.pulsar.client.api.KeySharedPolicy;
import org.apache.pulsar.client.api.Range;
import org.apache.pulsar.client.api.SubscriptionType;
@@ -92,10 +90,10 @@ final class ScalableStreamConsumer<T>
private final ConcurrentHashMap<Long,
org.apache.pulsar.client.api.MessageId> latestDelivered =
new ConcurrentHashMap<>();
- private final LinkedTransferQueue<MessageV5<T>> messageQueue = new
LinkedTransferQueue<>();
+ private final V5ReceiveQueue<T> receiveQueue;
/**
* Where each per-segment receive loop deposits a freshly-arrived message.
Defaults
- * to enqueueing on {@link #messageQueue} for the user's {@link
#receive()} to pull;
+ * to enqueueing on {@link #receiveQueue} for the user's {@link
#receive()} to pull;
* the multi-topic wrapper overrides this to forward into its shared
multiplexed
* queue, applying its own multi-topic position-vector capture in the
process.
*/
@@ -117,7 +115,9 @@ final class ScalableStreamConsumer<T>
this.session = session;
this.topicName = topicName;
this.subscriptionName = consumerConf.getSubscriptionName();
- this.messageSink = messageSink != null ? messageSink :
messageQueue::add;
+ this.receiveQueue = new V5ReceiveQueue<>(
+ client.v4Client().externalExecutorProvider().getExecutor(),
client.v4Client().timer());
+ this.messageSink = messageSink != null ? messageSink :
receiveQueue::offer;
this.log = LOG.with().attr("topic", topicName).attr("subscription",
subscriptionName).build();
this.asyncView = new AsyncStreamConsumerV5<>(this);
}
@@ -199,48 +199,17 @@ final class ScalableStreamConsumer<T>
@Override
public Message<T> receive() throws PulsarClientException {
- try {
- return messageQueue.take();
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new PulsarClientException("Receive interrupted", e);
- }
+ return receiveQueue.take();
}
@Override
public Message<T> receive(Duration timeout) throws PulsarClientException {
- try {
- return messageQueue.poll(timeout.toMillis(),
TimeUnit.MILLISECONDS);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new PulsarClientException("Receive interrupted", e);
- }
+ return receiveQueue.poll(timeout);
}
@Override
public Messages<T> receiveMulti(int maxNumMessages, Duration timeout)
throws PulsarClientException {
- List<Message<T>> batch = new ArrayList<>();
- long deadlineNanos = System.nanoTime() + timeout.toNanos();
-
- while (batch.size() < maxNumMessages) {
- long remainingNanos = deadlineNanos - System.nanoTime();
- if (remainingNanos <= 0) {
- break;
- }
- try {
- MessageV5<T> msg = messageQueue.poll(remainingNanos,
TimeUnit.NANOSECONDS);
- if (msg == null) {
- break;
- }
- batch.add(msg);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new PulsarClientException("Receive interrupted", e);
- }
- // Drain any immediately available messages
- messageQueue.drainTo(batch, maxNumMessages - batch.size());
- }
- return new MessagesV5<>(batch);
+ return new MessagesV5<>(receiveQueue.receiveMulti(maxNumMessages,
timeout));
}
@Override
@@ -292,27 +261,20 @@ final class ScalableStreamConsumer<T>
// --- Async internals ---
CompletableFuture<Message<T>> receiveAsync() {
- return CompletableFuture.supplyAsync(() -> {
- try {
- return receive();
- } catch (PulsarClientException e) {
- throw new CompletionException(e);
- }
- });
+ return receiveQueue.receiveAsync();
}
CompletableFuture<Message<T>> receiveAsync(Duration timeout) {
- return CompletableFuture.supplyAsync(() -> {
- try {
- return receive(timeout);
- } catch (PulsarClientException e) {
- throw new CompletionException(e);
- }
- });
+ return receiveQueue.receiveAsync(timeout);
+ }
+
+ CompletableFuture<List<Message<T>>> receiveMultiAsync(int maxNumMessages,
Duration timeout) {
+ return receiveQueue.receiveMultiAsync(maxNumMessages, timeout);
}
CompletableFuture<Void> closeAsync() {
closed = true;
+ receiveQueue.close();
session.close();
List<CompletableFuture<Void>> futures = new ArrayList<>();
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
new file mode 100644
index 00000000000..92a8e479417
--- /dev/null
+++
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/V5ReceiveQueue.java
@@ -0,0 +1,253 @@
+/*
+ * 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 io.netty.util.Timeout;
+import io.netty.util.Timer;
+import java.time.Duration;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+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;
+
+/**
+ * Async-native, single-consumer 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>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.
+ */
+final class V5ReceiveQueue<T> {
+
+ private final ExecutorService executor;
+ private final Timer timer;
+
+ // Both 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<>();
+ private boolean closed = false;
+
+ V5ReceiveQueue(ExecutorService executor, Timer timer) {
+ this.executor = executor;
+ this.timer = timer;
+ }
+
+ /**
+ * 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.
+ */
+ void offer(Message<T> msg) {
+ executor.execute(() -> {
+ if (closed) {
+ return;
+ }
+ CompletableFuture<Message<T>> waiter = pollWaiter();
+ if (waiter != null) {
+ waiter.complete(msg);
+ } else {
+ buffer.add(msg);
+ }
+ });
+ }
+
+ /** Receive a message, completing as soon as one is available. Never
blocks a thread. */
+ CompletableFuture<Message<T>> receiveAsync() {
+ CompletableFuture<Message<T>> result = new CompletableFuture<>();
+ executor.execute(() -> {
+ if (closed) {
+ result.completeExceptionally(alreadyClosed());
+ return;
+ }
+ Message<T> msg = buffer.poll();
+ if (msg != null) {
+ result.complete(msg);
+ } else {
+ pendingReceives.add(result);
+ }
+ });
+ return result;
+ }
+
+ /**
+ * Receive a message, completing with {@code null} if none arrives within
{@code timeout}.
+ * The timeout is armed on the client timer; no thread is parked while
waiting.
+ */
+ CompletableFuture<Message<T>> receiveAsync(Duration timeout) {
+ CompletableFuture<Message<T>> result = new CompletableFuture<>();
+ executor.execute(() -> {
+ if (closed) {
+ result.completeExceptionally(alreadyClosed());
+ return;
+ }
+ Message<T> msg = buffer.poll();
+ if (msg != null) {
+ result.complete(msg);
+ 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);
+ 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());
+ });
+ return result;
+ }
+
+ /**
+ * 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
+ * already buffered, repeating until the batch is full or the deadline
passes.
+ */
+ CompletableFuture<List<Message<T>>> receiveMultiAsync(int maxMessages,
Duration timeout) {
+ long deadlineNanos = System.nanoTime() + timeout.toNanos();
+ CompletableFuture<List<Message<T>>> result = new CompletableFuture<>();
+ collectMulti(new ArrayList<>(), maxMessages, deadlineNanos, result);
+ return result;
+ }
+
+ private void collectMulti(List<Message<T>> batch, int max, long
deadlineNanos,
+ CompletableFuture<List<Message<T>>> result) {
+ if (batch.size() >= max) {
+ result.complete(batch);
+ return;
+ }
+ long remainingNanos = deadlineNanos - System.nanoTime();
+ if (remainingNanos <= 0) {
+ result.complete(batch);
+ return;
+ }
+ receiveAsync(Duration.ofNanos(remainingNanos)).whenComplete((msg, ex)
-> {
+ if (ex != null) {
+ result.completeExceptionally(ex);
+ } else if (msg == null) {
+ result.complete(batch);
+ } else {
+ batch.add(msg);
+ drainReady(batch, max).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) {
+ CompletableFuture<Void> done = new CompletableFuture<>();
+ executor.execute(() -> {
+ Message<T> m;
+ while (batch.size() < max && (m = buffer.poll()) != null) {
+ batch.add(m);
+ }
+ done.complete(null);
+ });
+ return done;
+ }
+
+ // --- Blocking views, for the synchronous receive() API. Block only the
caller's thread. ---
+
+ Message<T> take() throws PulsarClientException {
+ try {
+ return receiveAsync().get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new PulsarClientException("Receive interrupted", e);
+ } catch (ExecutionException e) {
+ throw unwrap(e);
+ }
+ }
+
+ Message<T> poll(Duration timeout) throws PulsarClientException {
+ try {
+ return receiveAsync(timeout).get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new PulsarClientException("Receive interrupted", e);
+ } catch (ExecutionException e) {
+ throw unwrap(e);
+ }
+ }
+
+ List<Message<T>> receiveMulti(int maxMessages, Duration timeout) throws
PulsarClientException {
+ try {
+ return receiveMultiAsync(maxMessages, timeout).get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new PulsarClientException("Receive interrupted", e);
+ } catch (ExecutionException e) {
+ throw unwrap(e);
+ }
+ }
+
+ /** Fail any outstanding receives so blocked/awaiting callers wake instead
of hanging forever. */
+ void close() {
+ executor.execute(() -> {
+ closed = true;
+ CompletableFuture<Message<T>> waiter;
+ while ((waiter = pendingReceives.poll()) != null) {
+ if (!waiter.isDone()) {
+ waiter.completeExceptionally(alreadyClosed());
+ }
+ }
+ buffer.clear();
+ });
+ }
+
+ private CompletableFuture<Message<T>> pollWaiter() {
+ CompletableFuture<Message<T>> waiter;
+ // Skip futures already completed by cancellation or timeout.
+ while ((waiter = pendingReceives.poll()) != null) {
+ if (!waiter.isDone()) {
+ return waiter;
+ }
+ }
+ return null;
+ }
+
+ private static PulsarClientException alreadyClosed() {
+ return new PulsarClientException.AlreadyClosedException("Consumer
already closed");
+ }
+
+ private static PulsarClientException unwrap(ExecutionException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof PulsarClientException pce) {
+ return pce;
+ }
+ return new PulsarClientException(cause != null ? cause : e);
+ }
+}
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
new file mode 100644
index 00000000000..61d326c18aa
--- /dev/null
+++
b/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/V5ReceiveQueueTest.java
@@ -0,0 +1,248 @@
+/*
+ * 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 static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
+import io.netty.util.HashedWheelTimer;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import org.apache.pulsar.client.api.v5.Message;
+import org.apache.pulsar.client.api.v5.MessageId;
+import org.apache.pulsar.client.api.v5.PulsarClientException;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+public class V5ReceiveQueueTest {
+
+ private ExecutorService executor;
+ private HashedWheelTimer timer;
+ private V5ReceiveQueue<Integer> queue;
+
+ @BeforeMethod
+ public void setup() {
+ executor = Executors.newSingleThreadExecutor();
+ timer = new HashedWheelTimer();
+ queue = new V5ReceiveQueue<>(executor, timer);
+ }
+
+ @AfterMethod(alwaysRun = true)
+ public void teardown() {
+ executor.shutdownNow();
+ timer.stop();
+ }
+
+ @Test
+ public void bufferedMessageIsDeliveredToReceiveAsync() throws Exception {
+ Message<Integer> m = msg(1);
+ queue.offer(m);
+ assertSame(queue.receiveAsync().get(5, TimeUnit.SECONDS), m);
+ }
+
+ @Test
+ public void receiveAsyncCompletesWhenMessageArrivesLater() throws
Exception {
+ CompletableFuture<Message<Integer>> f = queue.receiveAsync();
+ assertTrue(!f.isDone());
+ Message<Integer> m = msg(1);
+ queue.offer(m);
+ assertSame(f.get(5, TimeUnit.SECONDS), m);
+ }
+
+ @Test
+ public void deliversInFifoOrder() throws Exception {
+ Message<Integer> m1 = msg(1);
+ Message<Integer> m2 = msg(2);
+ Message<Integer> m3 = msg(3);
+ queue.offer(m1);
+ queue.offer(m2);
+ queue.offer(m3);
+ assertSame(queue.receiveAsync().get(5, TimeUnit.SECONDS), m1);
+ assertSame(queue.receiveAsync().get(5, TimeUnit.SECONDS), m2);
+ assertSame(queue.receiveAsync().get(5, TimeUnit.SECONDS), m3);
+ }
+
+ @Test
+ public void timedReceiveReturnsNullOnTimeout() throws Exception {
+ assertNull(queue.receiveAsync(Duration.ofMillis(150)).get(5,
TimeUnit.SECONDS));
+ }
+
+ @Test
+ public void messageBeatsTimeout() throws Exception {
+ CompletableFuture<Message<Integer>> f =
queue.receiveAsync(Duration.ofSeconds(30));
+ Message<Integer> m = msg(1);
+ queue.offer(m);
+ assertSame(f.get(5, TimeUnit.SECONDS), m);
+ }
+
+ @Test
+ public void blockingTakeReturnsBufferedMessage() throws Exception {
+ Message<Integer> m = msg(1);
+ queue.offer(m);
+ assertSame(queue.take(), m);
+ }
+
+ @Test
+ public void blockingPollTimesOutToNull() throws Exception {
+ assertNull(queue.poll(Duration.ofMillis(150)));
+ }
+
+ @Test
+ public void closeFailsPendingReceive() {
+ CompletableFuture<Message<Integer>> f = queue.receiveAsync();
+ queue.close();
+ ExecutionException ex = expectThrows(ExecutionException.class, () ->
f.get(5, TimeUnit.SECONDS));
+ assertTrue(ex.getCause() instanceof
PulsarClientException.AlreadyClosedException);
+ }
+
+ @Test
+ public void receiveAfterCloseFailsFast() {
+ queue.close();
+ ExecutionException ex = expectThrows(ExecutionException.class,
+ () -> queue.receiveAsync().get(5, TimeUnit.SECONDS));
+ assertTrue(ex.getCause() instanceof
PulsarClientException.AlreadyClosedException);
+ }
+
+ @Test
+ public void cancelledWaiterDoesNotConsumeMessage() throws Exception {
+ CompletableFuture<Message<Integer>> cancelled = queue.receiveAsync();
+ cancelled.cancel(true);
+ Message<Integer> m = msg(1);
+ queue.offer(m);
+ // The message must reach a live receiver, not be swallowed by the
cancelled one.
+ assertSame(queue.receiveAsync().get(5, TimeUnit.SECONDS), m);
+ }
+
+ @Test
+ public void receiveMultiReturnsUpToMax() throws Exception {
+ for (int i = 1; i <= 5; i++) {
+ queue.offer(msg(i));
+ }
+ List<Message<Integer>> batch = queue.receiveMulti(3,
Duration.ofSeconds(5));
+ assertEquals(batch.size(), 3);
+ assertEquals(batch.get(0).value(), Integer.valueOf(1));
+ assertEquals(batch.get(1).value(), Integer.valueOf(2));
+ assertEquals(batch.get(2).value(), Integer.valueOf(3));
+ }
+
+ @Test
+ public void receiveMultiReturnsPartialOnTimeout() throws Exception {
+ queue.offer(msg(1));
+ queue.offer(msg(2));
+ List<Message<Integer>> batch = queue.receiveMultiAsync(5,
Duration.ofMillis(300))
+ .get(5, TimeUnit.SECONDS);
+ assertEquals(batch.size(), 2);
+ }
+
+ @Test
+ public void receiveMultiEmptyOnTimeoutWithNoMessages() throws Exception {
+ List<Message<Integer>> batch = queue.receiveMultiAsync(5,
Duration.ofMillis(150))
+ .get(5, TimeUnit.SECONDS);
+ assertTrue(batch.isEmpty());
+ }
+
+ private static Message<Integer> msg(int id) {
+ return new IntMessage(id);
+ }
+
+ /** Minimal {@link Message} whose {@link #value()} carries a test id;
other accessors are inert. */
+ private static final class IntMessage implements Message<Integer> {
+ private final int id;
+
+ IntMessage(int id) {
+ this.id = id;
+ }
+
+ @Override
+ public Integer value() {
+ return id;
+ }
+
+ @Override
+ public byte[] data() {
+ return new byte[0];
+ }
+
+ @Override
+ public MessageId id() {
+ return null;
+ }
+
+ @Override
+ public Optional<String> key() {
+ return Optional.empty();
+ }
+
+ @Override
+ public Map<String, String> properties() {
+ return Map.of();
+ }
+
+ @Override
+ public Instant publishTime() {
+ return Instant.EPOCH;
+ }
+
+ @Override
+ public Optional<Instant> eventTime() {
+ return Optional.empty();
+ }
+
+ @Override
+ public long sequenceId() {
+ return id;
+ }
+
+ @Override
+ public Optional<String> producerName() {
+ return Optional.empty();
+ }
+
+ @Override
+ public String topic() {
+ return "test";
+ }
+
+ @Override
+ public int redeliveryCount() {
+ return 0;
+ }
+
+ @Override
+ public int size() {
+ return 0;
+ }
+
+ @Override
+ public Optional<String> replicatedFrom() {
+ return Optional.empty();
+ }
+ }
+}
