Thanks for your review @lhotari . Here's my answer to your questions:

1.Please elaborate more on why you think that this contradicts expected behavior of zero-queue consumers?

The current behavior with MessagePayloadProcessor creates an unpredictable interaction pattern. Here's a detailed breakdown of what happens:

Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName)
        .batchingMaxMessages(5)
        .enableBatching(true)
        .create();

ConsumerImpl<byte[]> consumer = (ConsumerImpl<byte[]>) pulsarClient.newConsumer()
        .topic(topicName)
        .messagePayloadProcessor(new CustomProcessor())
        .subscriptionName(subscriptionName)
        .receiverQueueSize(0)
        .subscribe();

for (int i = 0; i < totalMessages; i++) {
    String message = messagePredicate + i;
    producer.sendAsync(message.getBytes());
}
producer.flush();
for (int i = 0; i < 10; i++) {
    Message<byte[]> receive = consumer.receive();
}

In this case, here's the step of what will happen:

so, as you can see, this zeroQueueConsumer will be blocked indefinitely in this case after consume 1 message. and maybe after consume 2 messages or more, and even won't get blocked, this totally depends on the process speed of user's customize MessagePayloadProcessor.

Key Problems:

• Permit Mismatch: The broker sends more messages (batches) than requested

• Hidden Buffering: Messages accumulate in incomingMessages despite queueSize=0

• Unpredictable Blocking: Behavior depends on processor speed, violating zero-queue guarantees

2.What do you consider as "buffering"?

In my understanding, the fundamental design principle of ZeroQueueConsumer (receiverQueueSize=0) is to be a strictly non-buffering consumer that:

• Operates in pure request-response mode:

  • Each receive() call should trigger exactly one network request to the broker

  • No messages should be retained client-side between receive() calls

3.What is your use case?

We need ZeroQueueConsumer to:

1.Maintain Strict Pull Semantics: Each receive() should trigger exactly one broker interaction

2.Handle Batches Predictably: Either:Reject batches entirely, or Process them atomically without intermediate buffering

In my understanding, the fundamental design principle of ZeroQueueConsumer (receiverQueueSize=0) is to be a strictly non-buffering consumer that:

• Operates in pure request-response mode:

  • Each receive() call should trigger exactly one network request to the broker
  • No messages should be retained client-side between receive() calls

I don't think that this is a correct assumption. In Pulsar, the receive method returns individual messages in a batch.
When the broker sends a batch message to the client, it should be buffered on the client side, also with ZeroQueueConsumer.

It's very much possible that you have faced bugs in the ZeroQueueConsumerImpl implementation. When checking at the implementation, it doesn't seem to make a distinction between batch messages. In Pulsar code, it's bit of a mess that in some cases the permits are about batch messages and some times there are about individual messages. I have created an issue about that in #23263 . It's possible that the information in the issue isn't accurate.

In my understanding, the fundamental design principle of ZeroQueueConsumer (receiverQueueSize=0) is to be a strictly non-buffering consumer that:

• Operates in pure request-response mode:

  • Each receive() call should trigger exactly one network request to the broker
  • No messages should be retained client-side between receive() calls

I don't think that this is a correct assumption. In Pulsar, the receive method returns individual messages in a batch. When the broker sends a batch message to the client, it should be buffered on the client side, also with ZeroQueueConsumer.

It's very much possible that you have faced bugs in the ZeroQueueConsumerImpl implementation. When checking at the implementation, it doesn't seem to make a distinction between batch messages. In Pulsar code, it's bit of a mess that in some cases the permits are about batch messages and some times there are about individual messages. I have created an issue about that in #23263 . It's possible that the information in the issue isn't accurate.

you can reproduce this issue with the example I mentioned above. so, I'm wondering is it ok that consumer gets blocked indefinitely?

I searched for ZeroQueueConsumerImpl related issues.

Now I finally understand the motivation. It appears that ZeroQueueConsumerImpl isn't intended to be used with batch messages at all, which is a bit confusing. The issue #11850 hints that. It refers this method:

I wasn't aware of this. I'll review the changes again.

please add javadoc to ConsumerBuilder.messagePayloadProcessor about the behavior with receiverQueueSize=0 (don't mention ZeroQueueConsumerImpl in ConsumerBuilder javadoc since it is an implementation detail):

This is the receiverQueueSize javadoc:

@lhotari I've added javadoc, please take a look again.

Btw. Not directly related to this PR, but could you check if there's any use for the PulsarClientException.InvalidMessageException that is thrown? Does it get used at all? I think that we have made changes to closing of the consumer, that it will fail all pending operations with PulsarClientException.AlreadyClosedException.

Codecov Report

❌ Patch coverage is 62.50000% with 3 lines in your changes missing coverage. Please review.
✅ Project coverage is 74.37%. Comparing base (dedba1a) to head (bec30ca).
⚠️ Report is 5 commits behind head on master.

@@             Coverage Diff              @@
##             master   #24610      +/-   ##
============================================
- Coverage     74.41%   74.37%   -0.05%     
- Complexity    32808    33176     +368     
============================================
  Files          1881     1881              
  Lines        146776   146823      +47     
  Branches      16857    16862       +5     
============================================
- Hits         109226   109194      -32     
- Misses        28901    28953      +52     
- Partials       8649     8676      +27     

Btw. Not directly related to this PR, but could you check if there's any use for the PulsarClientException.InvalidMessageException that is thrown? Does it get used at all? I think that we have made changes to closing of the consumer, that it will fail all pending operations with PulsarClientException.AlreadyClosedException.

@lhotari OK, I'm glad to do that

Btw. Not directly related to this PR, but could you check if there's any use for the PulsarClientException.InvalidMessageException that is thrown? Does it get used at all? I think that we have made changes to closing of the consumer, that it will fail all pending operations with PulsarClientException.AlreadyClosedException.

Hi, lari, @lhotari
I've checked all the InvalidMessageException usages:

• mostly on the producer side (which isn't related to our current changes)

• few on the consumer side (mainly for message validation purposes)

The current usage looks correct and doesn't need modification. However, I did notice and fix some inaccurate javadocs - please take a look when you have time.

@Technoboy- @lhotari
Hi Jiwei, Lari. I think this PR might needs to be cherry-picked to branch3.x and 4.x because:
1.The issue fixed by this PR also exists in branch3.x and 4.x
2.The javadoc behavior described in this PR is also present in branch3.x and 4.x"