[improve][broker] PIP-379: Enable the use of the classic implementation of Key_Shared / Shared with feature flag

[lhotari]

Motivation

PIP-379 contains:

Slightly unrelated to PIP-379 changes, there's a need to ensure that users upgrading from Pulsar 3.x can revert to the "recently joined consumers" logic (before PIP-282) in case of possible regressions caused by PIP-379. Since PIP-282 is also new in Pulsar 4.0.0, there needs to be a feature flag that toggles between the PIP-379 implementation for Key_Shared and the "recently joined consumers" logic before PIP-282. Implemention details for this feature toggle can be handled in the pull request for implementing this.

This PR copies PersistentDispatcherMultipleConsumers and PersistentStickyKeyDispatcherMultipleConsumers classes from branch-3.3, renames them to have a Classic suffix, adapts the code base, adapts most relevant tests and adds the feature flag.

Modifications

• Add configuration options:
  • subscriptionKeySharedUseClassicPersistentImplementation - For persistent Key_Shared subscriptions, enables the use of the classic implementation of the Key_Shared subscription that was used before Pulsar 4.0.0 and PIP-379
  • subscriptionSharedUseClassicPersistentImplementation - For persistent Shared subscriptions, enables the use of the classic implementation of the Shared subscription that was used before Pulsar 4.0.0.
• extract AbstractPersistentDispatcherMultipleConsumers class to be shared by PersistentDispatcherMultipleConsumers and PersistentDispatcherMultipleConsumersClassic
• add interface StickyKeyDispatcher which is implemented by both PersistentStickyKeyDispatcherMultipleConsumers and PersistentStickyKeyDispatcherMultipleConsumersClassic
• Decouple broker code base to depend on AbstractPersistentDispatcherMultipleConsumers and StickyKeyDispatcher
• Restore readPositionWhenJoining from state before PIP-282
• Restore consumersAfterMarkDeletePosition from state before PIP-282
• Add method isClassic() that can be used to detect the dispatcher implementation type
• Rename duplicate KeySharedSubscriptionTest to KeySharedSubscriptionMaxUnackedMessagesTest
• Don't require hash in replay with classic implementations
  • there's a "feature" or "bug" in the classic implementation that the hash is not added to pending acks, and when the messages get added back to replay, they don't contain hashes. This could be addressed separately in branch-3.3 if it's needed.
• Use Integer.MAX_VALUE-1 as the range end in Classic when using consistent hashing based selector since PIP-379 reduced the range size 65535
• Implement test parameterization solution KeySharedImplementationType which works with TestNG tests
• Test both PIP-379 and Classic in KeySharedSubscriptionTest using the test parameterization solution
• Test both PIP-379 and Classic in KeySharedSubscriptionMaxUnackedMessagesTest using the test parameterization solution
• Reduce test flakiness
  • Revisit receiveAndCheckDistribution to use multiple threads for receiving simultaneously
• Copy unit tests for PersistentDispatcherMultipleConsumersClassic and PersistentStickyKeyDispatcherMultipleConsumersClassic from branch-3.3

Documentation

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

[codecov-commenter]

Codecov Report

Attention: Patch coverage is 58.60656% with 404 lines in your changes missing coverage. Please review.

Project coverage is 74.40%. Comparing base (bbc6224) to head (59448a6).
Report is 653 commits behind head on master.

Files with missing lines	Patch %	Lines
.../PersistentDispatcherMultipleConsumersClassic.java	44.81%	287 Missing and 80 partials ⚠️
...ntStickyKeyDispatcherMultipleConsumersClassic.java	87.20%	18 Missing and 14 partials ⚠️
...ker/service/persistent/PersistentSubscription.java	91.66%	0 Missing and 2 partials ⚠️
...ervice/persistent/MessageRedeliveryController.java	80.00%	0 Missing and 1 partial ⚠️
...sistent/PersistentDispatcherMultipleConsumers.java	75.00%	1 Missing ⚠️
...sar/broker/service/persistent/PersistentTopic.java	87.50%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #23424      +/-   ##
============================================
+ Coverage     73.57%   74.40%   +0.83%     
- Complexity    32624    34913    +2289     
============================================
  Files          1877     1949      +72     
  Lines        139502   146867    +7365     
  Branches      15299    16168     +869     
============================================
+ Hits         102638   109283    +6645     
- Misses        28908    29198     +290     
- Partials       7956     8386     +430     

Flag	Coverage Δ
inttests	27.31% <2.97%> (+2.72%)	⬆️
systests	24.38% <2.97%> (+0.06%)	⬆️
unittests	73.77% <58.60%> (+0.93%)	⬆️

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

Files with missing lines	Coverage Δ
...org/apache/pulsar/broker/ServiceConfiguration.java	98.98% <100.00%> (-0.42%)	⬇️
...broker/delayed/AbstractDelayedDeliveryTracker.java	86.27% <ø> (ø)
...r/delayed/BucketDelayedDeliveryTrackerFactory.java	93.47% <ø> (+2.04%)	⬆️
...broker/delayed/InMemoryDelayedDeliveryTracker.java	88.13% <ø> (ø)
...delayed/InMemoryDelayedDeliveryTrackerFactory.java	95.23% <ø> (+3.57%)	⬆️
...r/delayed/bucket/BucketDelayedDeliveryTracker.java	83.12% <ø> (-0.58%)	⬇️
...r/service/AbstractDispatcherMultipleConsumers.java	85.24% <ø> (+2.98%)	⬆️
...rg/apache/pulsar/broker/service/BrokerService.java	84.11% <100.00%> (+3.33%)	⬆️
...ava/org/apache/pulsar/broker/service/Consumer.java	85.50% <100.00%> (-1.16%)	⬇️
...a/org/apache/pulsar/broker/service/Dispatcher.java	65.00% <ø> (+5.00%)	⬆️
... and 11 more

... and 608 files with indirect coverage changes