[improve][broker] PIP-486 (PR5): broker-side entry-bucket handoff for orderly consumer scaling

[merlimat]

Motivation

Continues PIP-486 after #26114, #26115, #26118, #26129, #26173. PR4 activated bucket-shared consumption for stable membership and left the hard part open: preserving per-key order across an ownership change of a live bucket (a consumer joining, leaving, or crashing mid-traffic).

The design follows three constraints: every language SDK has to implement whatever protocol we pick, so the client surface must stay minimal; transitions should be orderly, with no transient errors and retries when a consumer joins; and a consumer must not re-subscribe for buckets it keeps. The answer is to move the handoff into the broker, at entry-bucket granularity — exactly how Key_Shared AUTO_SPLIT hands over hash ranges today, but per whole bucket instead of per key. Consumers keep consuming throughout a membership change; a bucket move is observable only as a short pause on that bucket.

The mechanism that makes this nearly free is canonical-hash normalization: the dispatcher normalizes every entry's hash to its bucket's canonical value (the bucket's start hash, nudged to 1 for bucket 0 since 0 is the reserved "not set" sentinel). One hash value per bucket means the existing per-hash Key_Shared machinery — pending acks, replay-order blocking, DrainingHashesTracker — operates at exactly bucket granularity, with no new tracking code.

Modifications

Broker — dispatcher seams

• PersistentStickyKeyDispatcherMultipleConsumers gains a protected constructor taking a StickyKeyConsumerSelector and drainingHashesRequired, plus handleConsumerAdded/handleConsumerRemoved and shouldBlockDispatch hooks (defaults preserve AUTO_SPLIT behavior exactly).
• PR4's entryBucketDispatch field and getStickyKeyHash hook are reverted out of the parent: plain Key_Shared dispatch is back to its pre-PR4 form, and bucket logic lives only in the subclass. hasSameKeySharedPolicy rejects a flag mismatch.

Broker — entry-bucket dispatcher

• EntryBucketConsumerSelector: the segment's bucket boundaries are fixed at construction; the only moving part is membership. The N buckets are spread deterministically over the connected consumers sorted by (name, id) — consumer j of k owns the contiguous slice [j*N/k, (j+1)*N/k) — so a membership change moves as few whole buckets as possible, and membership diffs come from the existing ConsumerHashAssignmentsSnapshot.resolveImpactedConsumers.
• PersistentEntryBucketDispatcherMultipleConsumers (selected by PersistentSubscription when the consumer's KeySharedMeta carries entryBucketDispatch): the bucket boundaries arrive in the subscribe itself (KeySharedMeta.hashRanges = the segment's full boundary list, immutable for the segment's life; validated as ascending, contiguous, tiling the 16-bit ring). A joiner declaring different boundaries is rejected (ConsumerAssignException) — a mismatch is a stale layout, not a race. getStickyKeyHash routes a stamped entry by canonical(entry_hash_min) written through the cached-hash path, and unstamped entries by the key's hash normalized to the same canonical value. drainingHashesRequired=true gives the per-bucket handoff: a moving bucket is withheld from its new owner until every message the prior owner has in flight for it is acked, then flows in order.

Controller

• SubscriptionCoordinator: every sharer of a bucketed segment now receives the segment's full boundary list in bucket_ranges (identical for all sharers) instead of a per-consumer slice; empty still means sole owner → Exclusive. The controller only decides which consumers share a segment; the bucket→consumer spread is computed broker-side from live membership, so a join/leave needs no controller round-trip and no re-subscribes.

Client — v5 stream consumer

• The only client-side transition left is the mode flip (Exclusive ⇄ Key_Shared) when a segment's bucket_ranges flips between empty and non-empty, or a segment drop. Before closing the per-segment consumer the client now drains: delivery of that segment is paused and the close waits until every already-delivered message is acked (shared segments: the individual-ack translation queue is empty; whole segments: the cumulative high-water mark reached the latest delivered id). Completion is ack-driven, not time-based, so the flip cannot discard or reorder in-flight messages.
• The bucket_ranges proto comment is refreshed to the boundaries-for-all semantics. No wire-format changes: field shapes from PR3 and the PR4 flag are unchanged, and no new commands are introduced.

Verifications

• EntryBucketConsumerSelectorTest: bucket-index/canonical-hash math; deterministic, add-order-independent spread; more consumers than buckets; key-hash normalization; impacted-bucket reporting on join/leave; merged assignment snapshot.
• PersistentEntryBucketDispatcherMultipleConsumersTest: boundary-list validation accept/reject cases.
• PersistentStickyKeyDispatcherMultipleConsumersTest: the stamped-range routing test now exercises the subclass and asserts canonical values; a stamped entry on a plain Key_Shared subscription still routes by key.
• V5EntryBucketDispatchTest: the PR4 e2e tests, plus testConsumerJoiningMidTrafficPreservesPerKeyOrder — a second stream consumer joins while a keyed stream is in flight (per-message acks), and per-key order is asserted across the broker-side handoff with nothing lost or duplicated.
• Regression sweep: all v5 client suites, the scalable broker suites, the sticky-key dispatcher suite, and checkstyle pass.

[void-ptr974]

Thanks for the detailed write-up. The broker-side handoff approach looks really nice to me, especially the canonical bucket hash trick to reuse the existing Key_Shared draining path.

One thing I wanted to sanity-check: drainAndClose() waits for messages already delivered to the app to be acked before closing the old v4 consumer. But in subscribeAssigned(), it looks like the old segment future is removed/replaced in segmentConsumers before that drain completes, while ackSegmentUpTo() uses segmentConsumers.get(segmentId) to decide where to send the ack.

Could an app ack during the drain miss the old v4 consumer, or in the resubscribe case land on the new one instead? I may be missing another guard here, but otherwise the drain seems to depend on an ack path that no longer points at the consumer being drained.