[fix][client] Fix negative acknowledgement by messageId

[izumo27]

Motivation

related: #20750

When adding to unAckedMessageTracker, the messageId type is converted to MessageIdImpl.

• https://github.com/apache/pulsar/blob/v3.3.0/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java#L1162-L1167
• https://github.com/apache/pulsar/blob/v3.3.0/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L1820-L1827

If calling ConsumerImpl#negativeAcknowledge(MessageId) and the messageId type is BatchMessageIdImpl, the message will not be removed from the unAckedMessageTracker.
https://github.com/apache/pulsar/blob/v3.3.0/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L814

This will cause both negative ack redeliver and ack timeout redeliver.

Modifications

Convert the messageId type to remove the message from unAckedMessageTracker.

Verifying this change

• Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change added tests and can be verified as follows:

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

PR in forked repository: izumo27#5

[hanmz]

Good catch

[codecov-commenter]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 73.43%. Comparing base (bbc6224) to head (8ad4d2e).
Report is 464 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #23060      +/-   ##
============================================
- Coverage     73.57%   73.43%   -0.15%     
- Complexity    32624    33480     +856     
============================================
  Files          1877     1915      +38     
  Lines        139502   143943    +4441     
  Branches      15299    15725     +426     
============================================
+ Hits         102638   105703    +3065     
- Misses        28908    30126    +1218     
- Partials       7956     8114     +158     

Flag	Coverage Δ
inttests	27.48% <0.00%> (+2.90%)	⬆️
systests	24.72% <0.00%> (+0.39%)	⬆️
unittests	72.51% <100.00%> (-0.34%)	⬇️

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.62% <100.00%> (+1.05%)	⬆️

... and 502 files with indirect coverage changes

[lhotari]

Great catch @izumo27 !

This will cause both negative ack redeliver and ack timeout redeliver.

Would it be possible to add a test case for this scenario?

[izumo27]

@lhotari

Would it be possible to add a test case for this scenario?

This is testing whether the messages are deleted from the unAckedMessageTracker.
https://github.com/apache/pulsar/pull/23060/files#diff-bdc977c55b8ac0f29e331906c5b166ebb868aa8e580eb48b162ff0e5439e3a6dR155
Is it necessary to check whether ack timeout redeliver occurs?

[izumo27]

@lhotari It seems that #20750 has not been cherry-picked into branch 3.0.
https://github.com/apache/pulsar/blob/v3.0.6/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L753-L766