[fix][client] Fix reader message filtering issue during blue-green cluster switch

[hrzzzz]

Motivation

When the delayed acknowledgment feature is enabled, the method org.apache.pulsar.client.impl.PersistentAcknowledgmentsGroupingTracker#isDuplicate filters out messages that have already been acknowledged on the client side.

When using the blue-green cluster feature, after switching from the blue cluster to the green cluster, the message IDs generated by the green cluster may be smaller than those from the blue cluster. In this case, if a reader is used for consumption, the method org.apache.pulsar.client.impl.PersistentAcknowledgmentsGroupingTracker#isDuplicate will filter out all messages with IDs less than or equal to lastCumulativeAck, which can result in the loss of messages with IDs less than or equal to lastCumulativeAck after switching to the green cluster.

Modifications

Before switching from the blue cluster to the green cluster, first clear the contents of org.apache.pulsar.client.impl.AcknowledgmentsGroupingTracker

Verifying this change

• Make sure that the change passes the CI checks.

Added test in: org.apache.pulsar.broker.service.ClusterMigrationTest#testMigrationWithReader

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: hrzzzz#11

[thetumbled]

I am curious about how pulsar ensure the two message ids from two clusters corresponding to the same message are equal, or they are just not equal? Not equal may introduce lots of problem.

[lhotari]

I am curious about how pulsar ensure the two message ids from two clusters corresponding to the same message are equal, or they are just not equal? Not equal may introduce lots of problem.

@thetumbled Good question. In the blue-green case the message id comparison most likely doesn't apply.
In general there isn't a way to compare message ids between clusters.

The Shadow Topic feature added a field to CommandSend to send the source cluster message id. It would be useful for translations,

pulsar/pulsar-common/src/main/proto/PulsarApi.proto

Lines 532 to 533 in 9a97c84

	// Message id of this message, currently is used in replicator for shadow topic.
	optional MessageIdData message_id = 9;

. Taking this further and changing geo-replication to include the original message id and original source cluster would be useful for adding individual ack replication support to replicated subscriptions. Replicated subscriptions don't currently support individual acks and being able to compare message ids would be necessary for adding that feature.

[lhotari]

LGTM

[codecov-commenter]

Codecov Report

Attention: Patch coverage is 66.66667% with 1 line in your changes missing coverage. Please review.

Project coverage is 74.30%. Comparing base (bbc6224) to head (6c0c792).
Report is 809 commits behind head on master.

Files with missing lines	Patch %	Lines
...va/org/apache/pulsar/client/impl/ConsumerImpl.java	66.66%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #23693      +/-   ##
============================================
+ Coverage     73.57%   74.30%   +0.73%     
+ Complexity    32624    32126     -498     
============================================
  Files          1877     1838      -39     
  Lines        139502   143056    +3554     
  Branches      15299    16232     +933     
============================================
+ Hits         102638   106303    +3665     
+ Misses        28908    28379     -529     
- Partials       7956     8374     +418     

Flag	Coverage Δ
inttests	26.73% <33.33%> (+2.14%)	⬆️
systests	23.69% <33.33%> (-0.63%)	⬇️
unittests	73.67% <66.66%> (+0.83%)	⬆️

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	80.40% <66.66%> (+2.82%)	⬆️

... and 1005 files with indirect coverage changes