[improve][broker] PIP-486 (PR4): bucket-shared consumption for scale-up fan-out

[merlimat]

Motivation

Continues PIP-486 after the foundation PRs (#26114, #26115, #26118, #26129). Those left the machinery dormant: producers stamp every batch's entry-bucket hash range, the controller computes per-segment buckets, and the proto/consumer plumbing exists — but every segment is still consumed by a single Exclusive consumer.

This PR activates bucket-shared consumption for the scale-up case: when stream consumers outnumber segments (and PIP-483 doesn't split — e.g. at max-segments, or auto split/merge disabled), the controller fans a segment out across the surplus consumers by entry-bucket, and the broker dispatches each whole entry to the consumer owning its bucket — batching stays intact, no per-key hashing, no entry filtering.

Scope note: this covers stable membership (owners settle, then consume). Per-key ordering across an ownership change of a live bucket (the blocked-bucket handoff) is the next PR in the series.

Modifications

Wire format + gate

• KeySharedMeta.entryBucketDispatch (field 5, default false): the explicit signal that a Key_Shared subscription dispatches whole entries by their stamped entry-bucket range instead of hashing each message's key. Plumbed internally (ConsumerConfigurationData → Commands.newSubscribe); only the scalable stream consumer sets it, so plain Key_Shared subscriptions are untouched even for stamped entries.

Broker — dispatcher

• PersistentStickyKeyDispatcherMultipleConsumers.getStickyKeyHash: when the flag is set and the entry is stamped, the entry routes by entry_hash_min (with the reserved-0 nudge). The bucket hash is written through getOrUpdateCachedStickyKeyHash so pending acks, redelivery, and draining all see the same hash dispatch used (Consumer.sendMessages stores the entry's cached hash into pendingAcks; an uncached value blows up MessageRedeliveryController.add when a consumer closes with pending acks). Unstamped entries fall back to the message's sticky-key hash — the same low-16 Murmur value the producer would have stamped, so batched and non-batched messages of one key land in the same bucket.

Broker — controller

• SubscriptionCoordinator.computeAssignment: "segments first, entry-buckets absorb the surplus". While consumers don't outnumber segments each whole segment keeps one owner (empty bucketRanges → Exclusive, unchanged). Surplus consumers are handed to segments with spare bucket capacity (at most bucketCount() owners per segment); each owner takes a contiguous slice of the buckets. Consumers beyond total bucket capacity stay idle. Deterministic for leader failover.

Client — v5 stream consumer

• Shared segments subscribe Key_Shared STICKY with the owned bucket ranges and set the dispatch flag; whole segments stay Exclusive.
• Rebalances converge through retries: a subscribe rejected while another consumer still holds a segment or overlapping ranges (ConsumerBusy / ConsumerAssignError — both terminal at the v4 layer) is retried with backoff, both on assignment updates (reconcile loop) and on the initial subscribe (bounded by the client operation timeout). A segment whose owned ranges changed awaits its own consumer's close before re-subscribing. ScalableConsumerClient replays the current assignment when a listener registers, closing the lost-update window between registerConsumer and setListener.
• Cumulative acks are translated for bucket-shared segments: Key_Shared forbids cumulative acks (client-rejected, broker-ignored), so the consumer tracks its delivered-but-unacked ids per shared segment and turns "ack up to the vector position" into individual acks of exactly the ids it received — its buckets' share. Whole segments keep cumulative acks.

Verifications

• PersistentStickyKeyDispatcherMultipleConsumersTest: the hook routes stamped entries by bucket only with the flag; nudges entry_hash_min == 0 to 1; falls back to the key hash unstamped; and without the flag a stamped entry still routes by key.
• SubscriptionCoordinatorTest: a lone consumer keeps a bucketed segment whole; two consumers split an N=4 segment into disjoint contiguous halves tiling the ring; owners cap at bucketCount with the surplus idle.
• V5EntryBucketDispatchTest: end-to-end — a lone consumer on a bucketed segment (Exclusive path), and two stream consumers sharing an N=4 segment: every key lands wholly on one consumer in send order, both owners receive, and after the translated acks a fresh consumer finds nothing to redeliver. (The test pins the layout via setAutoScalePolicy(enabled=false) — with more consumers than segments PIP-483 would otherwise split, which is the preferred first lever.)
• Regression sweep: the scalable broker suite and all v5 stream-consumer suites pass.