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/10/17 20:20:11 UTC
(pulsar) 17/22: [improve][client] Don't start a batch receive on an internal consumer without queued messages (#26966)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 8321b9d68e90a6793838c7044145cb4ecd43ad76
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Sat Oct 17 09:31:10 2026 -0700
[improve][client] Don't start a batch receive on an internal consumer without queued messages (#26966)
(cherry picked from commit 9d48d8e6d4e8a4c9bf73ea1be176daab10518a88)
---
.../java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java | 5 ++++-
1 file changed, 4 insertions(+), 1 deletion(-)
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
index 1c1c146f0925..4773f7d66cf1 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
@@ -308,7 +308,10 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
} else {
// Call receiveAsync() if the incoming queue is not full. Because this block is run with
// thenAcceptAsync, there is no chance for recursion that would lead to stack overflow.
- receiveMessageFromConsumer(consumer, messages.size() > 0);
+ // A batch receive takes the consumer's queued messages; when there are none, it would wait for its
+ // timeout with a timer task and a hand-off to this executor for each received message, so a
+ // receiveAsync() waits for the next message instead.
+ receiveMessageFromConsumer(consumer, messages.size() > 0 && consumer.numMessagesInQueue() > 0);
}
}, internalPinnedExecutor).exceptionally(ex -> {
if (ex instanceof PulsarClientException.AlreadyClosedException