[Issue 12885]Fix unordered consuming case in Key_Shared subscription.

[Jason918]

Fixes #12885

Motivation

See #12885

Modifications

Restrict that the consumer remove happens after entries readings.

Verifying this change

• Make sure that the change passes the CI checks.

This change is already covered by existing tests, such as KeySharedSubscriptionTest.

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

• Dependencies (does it add or upgrade a dependency): (no)
• The public API: (no)
• The schema: (no)
• The default values of configurations: (no)
• The wire protocol: (no)
• The rest endpoints: (no)
• The admin cli options: (no)
• Anything that affects deployment: (no)

Documentation

Check the box below and label this PR (if you have committer privilege).

Need to update docs?

• no-need-doc
  Bug fix.

[Jason918]

@codelipenghui @lhotari @eolivelli @merlimat Would you please help review this?

[codelipenghui]

Thanks for the case, I think the root cause is there is an ongoing read op to read data from the managed ledger, but after a consumer closed, some messages will be added to the replay queue, after the previous read op completed, the active consumer might get messages out of order by key.

I think we can just add a check after got new messages from the managed ledger? If we have messages in the replay queue, we can just add the new messages to the replay queue and then trigger a new read entries operation, then the new operation will try to replay messages from the replay queue, does this can fix the issue?

[Jason918]

Thanks for the case, I think the root cause is there is an ongoing read op to read data from the managed ledger, but after a consumer closed, some messages will be added to the replay queue, after the previous read op completed, the active consumer might get messages out of order by key.

The root cause of this case is recentlyJoinedConsumers is cleared in Thread3 when c1 is closed. See here.

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java

Line 137 in 5523604

if (consumerList.size() == 1) {

So in the following Thread2, when sendMessagesToConsumers is called for m10-m14. Previously we are using recentlyJoinedConsumers in getRestrictedMaxEntriesForConsumer to prevent sending these messages to c2. But now it's cleared. so "maxMessages" is returned instead of "0". See here:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java

Line 315 in 5523604

PositionImpl maxReadPosition = recentlyJoinedConsumers.get(consumer);

[Jason918]

@codelipenghui Come up with a more simple way for the case. PTAL.
The basic ideal is to check if we got message with smaller position to replay, in this case, unacked messages prefecthed by c1 will be added for replay after consumer closed.

[Jason918]

/pulsarbot run-failure-checks

[codelipenghui]

@merlimat Please help review this PR.

[Jason918]

/pulsarbot run-failure-checks