[fix][broker] Introduce the last sent position to fix message ordering issues in Key_Shared (PIP-282)

[equanz]

PIP: #20776

Motivation

Key_Shared has a mechanism called the "recently joined consumers" to keep message ordering.
However, currently, it doesn't care about some corner cases.
More specifically, we found two out-of-order issues cased by:

1. [issue-1] The race condition in the "recently joined consumers", where consumers can be added before finishing reading and dispatching messages from ledgers.
2. [issue-2] Messages could be added to messagesToRedeliver without consumer-side operations such as unacknowledgement.

We should care about these cases in Key_Shared subscription.
(For more details, please see the PIP document.)

Modifications

1. Introduce the new position (the lastSentPosition and individuallySentPositions) and use them in PersistentStickyKeyDispatcherMultipleConsumers.
2. Rename the consumer stats readPositionWhenJoining to lastSentPositionWhenJoining.
3. Modify the subscription stats (consumersAfterMarkDeletePosition) of the definition.

(For more details, please see the PIP document.)

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• Unit tests for the lastSentPosition and individuallySentPositions
• Test for guaranteed message ordering in corner cases

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

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

    Lines 1281 to 1288 in 43a2dc3

    	// The dispatcher allows same name consumers
    	final StringBuilder stringBuilder = new StringBuilder();
    	stringBuilder.append("consumerName=").append(k.consumerName())
    	.append(", consumerId=").append(k.consumerId());
    	if (k.cnx() != null) {
    	stringBuilder.append(", address=").append(k.cnx().clientAddress());
    	}
    	subStats.consumersAfterMarkDeletePosition.put(stringBuilder.toString(), v.toString());

  • pulsar/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ConsumerStats.java

    Lines 72 to 73 in 43a2dc3

    	/** The last sent position of the cursor when the consumer joining. */
    	String getLastSentPositionWhenJoining();

• Anything that affects deployment

Documentation

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

Matching PR in forked repository

PR in forked repository: equanz#6

[nkurihar]

LGTM

(In our-company we cherry-picked this change and so far no issues have been identified. However, it is better if you could give it another review for further validation.)

[hanmz]

LGTM

[poorbarcode]

/pulsarbot rerun-failure-checks

[codecov-commenter]

Codecov Report

Attention: Patch coverage is 87.93103% with 14 lines in your changes are missing coverage. Please review.

Project coverage is 73.90%. Comparing base (bbc6224) to head (c3b318a).
Report is 179 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #21953      +/-   ##
============================================
+ Coverage     73.57%   73.90%   +0.33%     
- Complexity    32624    33094     +470     
============================================
  Files          1877     1885       +8     
  Lines        139502   140522    +1020     
  Branches      15299    15466     +167     
============================================
+ Hits         102638   103857    +1219     
+ Misses        28908    28616     -292     
- Partials       7956     8049      +93     

Flag	Coverage Δ
inttests	26.69% <44.82%> (+2.11%)	⬆️
systests	24.45% <45.68%> (+0.13%)	⬆️
unittests	73.18% <87.93%> (+0.34%)	⬆️

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

Files	Coverage Δ
...che/bookkeeper/mledger/impl/ManagedCursorImpl.java	79.83% <100.00%> (+0.53%)	⬆️
...che/bookkeeper/mledger/impl/ManagedLedgerImpl.java	81.01% <ø> (+0.35%)	⬆️
...ava/org/apache/pulsar/broker/service/Consumer.java	86.66% <100.00%> (ø)
.../common/policies/data/stats/ConsumerStatsImpl.java	97.61% <100.00%> (ø)
...mon/policies/data/stats/SubscriptionStatsImpl.java	94.21% <100.00%> (+0.09%)	⬆️
...ker/service/persistent/PersistentSubscription.java	77.18% <92.85%> (+0.49%)	⬆️
...ersistentStickyKeyDispatcherMultipleConsumers.java	87.02% <85.55%> (+1.38%)	⬆️

... and 243 files with indirect coverage changes

[poorbarcode]

@equanz Sorry for the late review, I will finish the review tomorrow.

[equanz]

Hi @poorbarcode.
Thank you for your review. What is the current status?

[poorbarcode]

Hi @poorbarcode.
Thank you for your review. What is the current status?

Sorry, I am busy for other things before, I will done this review soon

[equanz]

Rebased to fix conflicts.
BTW, can someone review this? I've been trying to fix the issue for a year now... (cf. #20179)

[poorbarcode]

@codelipenghui Could you take a look?

[equanz]

@codelipenghui @poorbarcode

Looking forward to your reviews.

[codelipenghui]

@equanz @poorbarcode Sorry for the late response, I will review this week.

[equanz]

Hi @codelipenghui . Do you have any comments?

[codelipenghui]

Hi @codelipenghui . Do you have any comments?

Sorry, I'm struggling in other tasks. Please go ahead to merge the PR. @poorbarcode

[poorbarcode]

@equanz Could you please handle the conflicts?

[equanz]

/pulsarbot rerun-failure-checks

[equanz]

@poorbarcode Addressed. PTAL

[equanz]

@poorbarcode Could you take a look?

[lhotari]

@equanz do you have a chance to update the documentation in https://pulsar.apache.org/docs/next/concepts-messaging/#preserving-order-of-processing to reflect the PIP-282 changes?
it seems that "Once the connection has been established, the broker will record the current read position and associate it with the new consumer. The read position is a marker indicating that messages have been dispatched to the consumers up to this point, and after it, no messages have been dispatched yet. The broker will start delivering messages to the new consumer only when all messages up to the read position have been acknowledged." would have to be updated in https://github.com/apache/pulsar-site/blob/main/docs/concepts-messaging.md?plain=1#L797-L801.
/cc @poorbarcode @codelipenghui

[equanz]

I forgot it. I'll fix it.

[equanz]

Here is a fix of the docs.
apache/pulsar-site#953