[fix][client] Fix client to handle permits for discarded and failed decrypt batch-message

[rdhabalia]

Motivation

Currently if Pulsar-client fails to decrypt batch-message and if CrypotFailureAction is configured as Discard then after a while consumer at client side runs out of permit even though internal queue has space so, consumer gets stuck and broker is not dispatching messages to consumer though consumer is suppose to receive more messages. It happens because Client lib doesn't increase permits while discarding batch messages accordingly number of messages in the batch-message but instead it only increrements permit by one which causes consumer to go out of available permit and stops consuming more messages.

[pulsar-client-io-1-1] WARN org.apache.pulsar.client.impl.ConsumerImpl - [persistent://t1/c1/n1/t1][test][746e5][1001074382-11645--1] Discarding message since decryption failed and config is set to discard
:
 "consumers" : [ {
        "msgRateOut" : 0,
        "msgThroughputOut" : 0,
        "availablePermits" : -50,

Modifications

Fix client consumer to handle batch message when client fails to decrypt message and send new available permits according to number of messages in that batch to avoid stuck consumer for the user.

Verifying this change

• [ ] Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

• Added integration tests for end-to-end deployment with large payloads (10MB)
• Extended integration test for recovery after broker failure

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
• [x] doc-not-needed
• [ ] doc-complete

Matching PR in forked repository

PR in forked repository:

[codecov-commenter]

Codecov Report

Attention: Patch coverage is 70.37037% with 8 lines in your changes missing coverage. Please review.

Project coverage is 73.43%. Comparing base (bbc6224) to head (51982ee). Report is 465 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #23068      +/-   ##
============================================
- Coverage     73.57%   73.43%   -0.15%     
- Complexity    32624    33189     +565     
============================================
  Files          1877     1915      +38     
  Lines        139502   144016    +4514     
  Branches      15299    15731     +432     
============================================
+ Hits         102638   105754    +3116     
- Misses        28908    30146    +1238     
- Partials       7956     8116     +160     

Flag	Coverage Δ
inttests	27.84% <0.00%> (+3.26%)	:arrow_up:
systests	24.74% <0.00%> (+0.42%)	:arrow_up:
unittests	72.48% <70.37%> (-0.36%)	:arrow_down:

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

Files	Coverage Δ
...va/org/apache/pulsar/client/impl/ConsumerImpl.java	78.65% <70.37%> (+1.08%)	:arrow_up:

... and 505 files with indirect coverage changes