[fix][client] fix negative message re-delivery twice issue

[aloyszhang]

Motivation

When calling the negativeAcknowledge method, the message is supposed to be re-delivered only one time.
But, actually, it may be re-delivered twice.

There are two problems:

1. For MultiTopicsConsumerImpl, when calling negativeAcknowledge, we didn't remove the message from UnAckedMessageTracker, so this message will be re-delivery by both UnAckedMessageTracker and NegativeAcksTracker
2. For ConsumerImpl, when calling negativeAcknowledge, if the message id is a BatchMessageIdImpl, the messageId will be not removed from the UnAckedMessageTracker, there is because the removing MessageId type(BatchMessageIdImpl) is not the MessageId type of add(MessageIdImpl).
   See:

• add to UnAckedMessageTracker

protected void trackMessage(MessageId messageId, int redeliveryCount) {
        if (conf.getAckTimeoutMillis() > 0 && messageId instanceof MessageIdImpl) {
            MessageId id = MessageIdAdvUtils.discardBatch(messageId);

• remove from UnAckedMessageTracker

public void negativeAcknowledge(Message<?> message) {
        negativeAcksTracker.add(message);

        // Ensure the message is not redelivered for ack-timeout, since we did receive an "ack"
        unAckedMessageTracker.remove(message.getMessageId());
    }

Modifications

1. For MultiTopicsConsumerImpl, remove the message id from UnAckedMessageTracker when calling negativeAcknowledge
2. Keep consistent type for message id adding to and removing from the UnAckedMessageTracker
3. add UnAckedMessageTracker size check for NegativeAcksTest
4. add log for NegativeAcksTracker

Verifying this change

• Make sure that the change passes the CI checks.

Documentation

• doc-not-needed

Matching PR in forked repository

PR in forked repository: aloyszhang#17

[github-actions]

@aloyszhang Please add the following content to your PR description and select a checkbox:

- [ ] `doc` <!-- Your PR contains doc changes -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->

[lordcheng10]

nice catch!

[lordcheng10]

LGTM

[BewareMyPower]

getUnAckedMessageTracker was a public method of ConsumerImpl and MultiTopicsConsumerImpl before this change, if you moved it to ConsumerBase and changed it to protected, there will be a breaking change.

It would be better to add back the getter to ConsumerImpl and MultiTopicsConsumerImpl. i.e. add the following code to them

    @Override
    public UnAckedMessageTracker getUnAckedMessageTracker() {
        return super.getUnAckedMessageTracker();
    }

[codecov-commenter]

Codecov Report

Merging #20750 (305ac2e) into master (09c89cd) will increase coverage by 41.13%.
The diff coverage is 57.14%.

@@              Coverage Diff              @@
##             master   #20750       +/-   ##
=============================================
+ Coverage     31.96%   73.09%   +41.13%     
- Complexity    11812    32141    +20329     
=============================================
  Files          1499     1867      +368     
  Lines        114819   139081    +24262     
  Branches      12454    15306     +2852     
=============================================
+ Hits          36697   101659    +64962     
+ Misses        73260    29353    -43907     
- Partials       4862     8069     +3207     

Flag	Coverage Δ
inttests	24.15% <7.14%> (?)
systests	24.87% <7.14%> (?)
unittests	72.39% <57.14%> (+40.43%)	⬆️

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

Impacted Files	Coverage Δ
...org/apache/pulsar/broker/ServiceConfiguration.java	99.38% <ø> (+2.56%)	⬆️
...pache/pulsar/broker/admin/v2/PersistentTopics.java	80.76% <ø> (+73.50%)	⬆️
...he/pulsar/broker/intercept/BrokerInterceptors.java	46.95% <ø> (+37.63%)	⬆️
...in/java/org/apache/pulsar/admin/cli/CmdTopics.java	78.94% <ø> (ø)
...va/org/apache/pulsar/client/impl/ConsumerBase.java	73.63% <0.00%> (+24.34%)	⬆️
...ache/pulsar/client/impl/UnAckedMessageTracker.java	90.67% <0.00%> (+16.99%)	⬆️
...he/pulsar/client/impl/MultiTopicsConsumerImpl.java	78.45% <50.00%> (+48.90%)	⬆️
...va/org/apache/pulsar/client/impl/ConsumerImpl.java	76.23% <100.00%> (+26.34%)	⬆️
...apache/pulsar/client/impl/NegativeAcksTracker.java	98.03% <100.00%> (+77.20%)	⬆️

... and 1546 files with indirect coverage changes

[liangyepianzhou]

@codelipenghui @Technoboy- This is a bug fix and no public API is added. So it should be cherry-picked to other branches, right?

[lhotari]

Thank you @izumo27 for noticing that this fix is missing from Pulsar 3.0.6 release (comment).

Unfortunately this fix was missing from branch-3.0 and wasn't delivered in Pulsar 3.0.3 as it was planned. I have cherry-picked this to branch-3.0 and this fix will be delivered in Pulsar 3.0.7 release for 3.0.x .