[improve][broker] Speed up auto-split Key_Shared consumer selection

[lhotari]

Motivation

The AUTO_SPLIT Key_Shared selector performs a ConcurrentSkipListMap.ceilingEntry lookup for every dispatched message. Consumer membership changes are rare compared with message selection, and those changes are already serialized. A concurrent ordered-map traversal therefore adds pointer chasing and comparison work to the hot path while membership is stable.

Modifications

• Keep the existing maps as the authoritative membership and range-assignment state.
• After each successful synchronized membership change, publish an immutable snapshot containing sorted primitive range ends and their consumers.
• Select consumers by binary-searching one volatile snapshot read.
• Add exhaustive assignment-equivalence coverage after additions and removals, concurrent selection/membership-churn coverage, and an actual-selector microbenchmark.

A selection concurrent with a membership change observes either the complete old snapshot or the complete new snapshot. Once addConsumer or removeConsumer returns, the new snapshot has been published.

Performance

JMH compares the actual selector with the previous ConcurrentSkipListMap.ceilingEntry lookup. Runs used two forks, three 1-second warmups, five 1-second measurements, and GC profiling.

Consumers	Readers	Before	After	Change
2	1	7.098 ns/op	3.745 ns/op	-47.2%
10	1	9.863 ns/op	4.849 ns/op	-50.8%
10	4	14.566 ns/op	5.189 ns/op	-64.4%
50	1	13.842 ns/op	5.681 ns/op	-59.0%
50	4	17.089 ns/op	6.760 ns/op	-60.4%

Both implementations allocate effectively zero on selection. Membership changes allocate the compact snapshot and remain control-plane operations.

A five-million-message scenario with 500 producers and five ten-consumer Key_Shared subscriptions produced two positive matched comparisons:

Pair	Before	After	Change
1	59,312 msg/s	74,128 msg/s	+25.0%
2	59,779 msg/s	72,396 msg/s	+21.1%

Every run delivered all messages to every subscription with zero duplicates, ordering violations, or invalid messages. The whole-host gain is larger than the isolated nanosecond saving directly explains, so the table establishes repeatability and absence of regression rather than attributing the full improvement to this lookup.

Verifying this change

This change added tests and can be verified as follows:

• Exhaustive selection equivalence across the complete test hash range after every add and remove
• Concurrent selection during repeated consumer membership changes
• Existing selector test suite
• ./gradlew :microbench:shadowJar quickCheck
• Two matched end-to-end scenario pairs with ordering and duplication checks

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