[fix] [client] call redeliver 1 msg but did 2 msgs

[poorbarcode]

Motivation

Issue: call redeliver 1 message, but it will redeliver 2 messages, you can reproduce the issue by the new test testBatchMessageNAck

Modifications

fix the bug

Documentation

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

Matching PR in forked repository

PR in forked repository: x

[codecov-commenter]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 74.25%. Comparing base (bbc6224) to head (bfa194c).
Report is 887 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #23943      +/-   ##
============================================
+ Coverage     73.57%   74.25%   +0.68%     
+ Complexity    32624     2375   -30249     
============================================
  Files          1877     1853      -24     
  Lines        139502   143735    +4233     
  Branches      15299    16334    +1035     
============================================
+ Hits         102638   106730    +4092     
+ Misses        28908    28596     -312     
- Partials       7956     8409     +453     

Flag	Coverage Δ
inttests	26.71% <9.09%> (+2.13%)	⬆️
systests	23.21% <9.09%> (-1.12%)	⬇️
unittests	73.76% <100.00%> (+0.91%)	⬆️

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	79.44% <100.00%> (+1.87%)	⬆️
...n/util/collections/GrowableArrayBlockingQueue.java	97.00% <100.00%> (+1.56%)	⬆️

... and 1036 files with indirect coverage changes

[nodece]

@poorbarcode Could you fix the build error on the branch-3.0?

import static com.google.common.base.Preconditions.checkState;

This import is not used.

[poorbarcode]

@nodece

Could you fix the build error on the branch-3.0?

import static com.google.common.base.Preconditions.checkState;

This import is not used.

@lhotari has fixed, Thanks

[BewareMyPower]

When it's false, even without this change, testBatchMessageNAck could still pass

[poorbarcode]

Yes, it only affects bacthed messages

[BewareMyPower]

It's meaningless here. pause() should only affect the message listener

[lhotari]

In the Java client, pause() will pause sending new permits, so it's not specific to message listener.

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

Lines 1867 to 1880 in 1220951

	protected void increaseAvailablePermits(ClientCnx currentCnx, int delta) {
	int available = AVAILABLE_PERMITS_UPDATER.addAndGet(this, delta);
	while (available >= getCurrentReceiverQueueSize() / 2 && !paused) {
	if (AVAILABLE_PERMITS_UPDATER.compareAndSet(this, available, 0)) {
	if (log.isDebugEnabled()) {
	log.debug("[{}] Sending permit-cmd to broker with available permits = {}", topic, available);
	}
	sendFlowPermitsToBroker(currentCnx, available);
	break;
	} else {
	available = AVAILABLE_PERMITS_UPDATER.get(this);
	}
	}
	}

[dao-jun]

pause is for stop receiving msgs from broker