[fix][client] Fixed stuck issues caused by an exception encountered in the receiveIndividualMessagesFromBatch()

[gosonzhang]

Motivation

Fixed three issues with receiveIndividualMessagesFromBatch() parsing failures:

1. On the consumer side, corrupted batch messages that fail to parse based on formats formed by [a] or other conditions should be discarded to avoid consumption stucks.

2. When parsing a batch of messages and an exception is thrown in the middle, the permits (minus 1) of the remaining unprocessed messages are all leaked, causing consumption stucks.

3. When batch index ack is enabled, messages being processed will be confirmed in advance in the event of a parsing exception.

[a]. #24061

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Documentation

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

Matching PR in forked repository

none

[lhotari]

/pulsarbot rerun-failure-checks

[codecov-commenter]

Codecov Report

❌ Patch coverage is 90.90909% with 2 lines in your changes missing coverage. Please review.
✅ Project coverage is 72.77%. Comparing base (765c46e) to head (36a0257).
⚠️ Report is 115 commits behind head on master.

Files with missing lines	Patch %	Lines
...va/org/apache/pulsar/client/impl/ConsumerImpl.java	90.90%	0 Missing and 2 partials ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #25386      +/-   ##
============================================
+ Coverage     72.72%   72.77%   +0.04%     
- Complexity    34277    34296      +19     
============================================
  Files          1954     1954              
  Lines        154857   154897      +40     
  Branches      17739    17745       +6     
============================================
+ Hits         112627   112724      +97     
+ Misses        33197    33132      -65     
- Partials       9033     9041       +8     

Flag	Coverage Δ
inttests	25.71% <9.09%> (-0.21%)	⬇️
systests	22.48% <9.09%> (-0.09%)	⬇️
unittests	73.75% <90.90%> (+0.05%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
...va/org/apache/pulsar/client/impl/ConsumerImpl.java	81.14% <90.90%> (+0.77%)	⬆️

... and 95 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[gosonzhang]

@lhotari Could you please take another look? I've made the modifications according to your suggestions.

[Denovo1998]

It is not advisable to directly use batchSize - (skipped + processed) to replenish permits. The receiveIndividualMessagesFromBatch() method will filter out already-acknowledged batch indices at the beginning based on the inbound ackSet; these indices have not consumed broker permits. If this is a partial batch redelivery, the current algorithm would include the undelivered indices, potentially leading to permit over-credit.

It might be better to pass in the current inbound ackBitSet, count only the bits after the failing index that are still deliverable, and construct a new ack set based on that.

[Denovo1998]

Commands.deSerializeSingleMessageInBatch() may also throw IndexOutOfBoundsException for truncated payload (triggered by readUnsignedInt() or retainedSlice()), whereas newSingleMessage() currently does not wrap it into IllegalStateException. As a result, encountering a corrupted batch with truncation or length mismatch can still bypass discardCorruptedBatchMessage, and permit leak or stuck issues would persist.