[fix] [client] fix huge permits if acked a half batched message

[poorbarcode]

Motivation

Context:

• The broker decreases the permits after delivering messages to the client.
  • It only decreases the un-acked part for permits when delivering batched messages to the client.
• The client acquires more permits after consuming messages.
  • (Highlight) It acquires the whole batch for permits even if it only received half of the batched message.

Reproduce step:

• The permits were initialized as 1000
• Send a batch message containing 100 single messages.
• Ack a half of batched messages(2 single messages).
• Redeliver the batch message.
• Receive the 98 messages left on the topic.
• (Highlight) permits is 1002.

One our ENV, the permits went to 100,000+(it was initialized as 50), and leading the set pendingAcks went to a huge value. This may causes a client side OOM.

Modifications

• The client acquires more permits after consuming messages.
  • It acquires the half batch for permits if it only received half of the batched message.

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository: x

[Technoboy-]

Some tests are failed due to this fix

[codelipenghui]

Nice catch!

Actually, we have another option to fix this issue.

Now, the broker side uses the ackedCount - totalMessages to decrease the permits.

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java

Line 353 in 1b1cfb5

MESSAGE_PERMITS_UPDATER.addAndGet(this, ackedCount - totalMessages);

If we change to - totalMessages, we don't need to catch up the fix for other clients. But it's not good for long term because it will lead to more flow permits requests.

[codelipenghui]

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

Line 1195 in 1b1cfb5

if (ackBitSet != null && !ackBitSet.get(index)) {

You can also change here to use the newly added method isSingleMessageAcked

[poorbarcode]

You can also change here to use the newly added method isSingleMessageAcked

Sory, I forgot that place, improved