[improve][broker] Optimize consumer selection for shared subscriptions - #26593

Motivation

Shared subscriptions perform priority checks during consumer selection even when every consumer has the same priority. With mixed priorities, round-robin wraparound can also rescan higher-priority consumers to locate the current priority group.

Reducing this work lowers selection overhead while preserving priority ordering and round-robin behavior.

Modifications

• Extract selection logic into ConsumerPrioritySelector, allowing isolated unit tests and JMH benchmarks.
• Track consumer counts by priority using Int2ObjectMap<MutableInt>. Remove entries when their counts reach zero and skip priority checks when only one priority remains.
• Reuse priority-group boundaries and eliminate redundant list reads during mixed-priority selection.
• Synchronize selection and membership updates on the dispatcher monitor. Preserve cleanup of inconsistent consumer list/set state by rebuilding counts during bulk-removal recovery.
• Keep the refactoring, benchmark, and optimization in separate commits for reproducible comparisons.

Verifying this change

• Make sure that the change passes the CI checks.

Added tests cover:

• Selection and availability-check order across 262,144 priority-group, availability-mask, and cursor combinations.
• Three complete round-robin cycles with 50 consumers, followed by exhaustion and recovery, at zero and nonzero priority.
• Membership changes, priority-count transitions, equal consumer identities, and bulk-removal recovery.

Local validation passed:

• ConsumerPrioritySelectorTest
• The four existing persistent dispatcher priority tests, with quarantine exclusions cleared and retries disabled
• Selected dispatcher add/remove and cleanup tests
• ./gradlew quickCheck
• ./gradlew :microbench:shadowJar

JMH comparison

Compared baseline 2d540fa0853 with optimization 9c88617c944 using identical benchmark sources and settings across 26 parameter combinations.

Environment: Apple M3 Max, macOS arm64, Corretto 25.0.4, JMH 1.37; two forks, three 500-ms warmup iterations, five 500-ms measurement iterations, one thread, and the GC profiler. Builds and tests did not overlap measurements.

Values below are mean nanoseconds per selection. Ranges cover priority offsets zero and three; speedups compare matching cases.

No GC occurred, and measured allocation was approximately zero. The 256-consumer mixed-priority FLOW case at offset zero had a wide confidence interval and remains inconclusive.

These measurements isolate selection cost and exclude dispatcher locking, membership updates, and message processing. Linux x86_64 confirmation is needed before extrapolating to broker throughput.

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

• 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

Selection and membership helpers explicitly synchronize on the existing dispatcher monitor so the consumer list and priority counts remain consistent. Existing dispatch call sites already hold that monitor.