[GitHub] [pulsar] shibd commented on a diff in pull request #16336: [improve][java-client] Improve performance of multi-topic consumer with more than one IO thread

[GitBox <gi...@apache.org>]

shibd commented on code in PR #16336:
URL: https://github.com/apache/pulsar/pull/16336#discussion_r913056805


##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java:
##########
@@ -244,13 +245,19 @@ private void startReceivingMessages(List<ConsumerImpl<T>> newConsumers) {
     }
 
     private void receiveMessageFromConsumer(ConsumerImpl<T> consumer) {
-        consumer.receiveAsync().thenAcceptAsync(message -> {
+        CompletableFuture<List<Message<T>>> messagesFuture;
+        if (consumer.numMessagesInQueue() >= 10) {

Review Comment:
   Can we add a new `batchReceive`  implementation to directly get as many messages in `incomingMessages`? This way you don't need to judge the size of the `numMessagesInQueue`.
   
   Similar to just running this logic.
   ``` java
                   MessagesImpl<T> messages = getNewMessagesImpl();
                   Message<T> msgPeeked = incomingMessages.peek();
                   while (msgPeeked != null && messages.canAdd(msgPeeked)) {
                       Message<T> msg = incomingMessages.poll();
                       if (msg != null) {
                           messageProcessed(msg);
                           if (!isValidConsumerEpoch(msg)) {
                               msgPeeked = incomingMessages.peek();
                               continue;
                           }
                           Message<T> interceptMsg = beforeConsume(msg);
                           messages.add(interceptMsg);
                       }
                       msgPeeked = incomingMessages.peek();
                   }
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org