merlimat opened a new pull request, #26173: URL: https://github.com/apache/pulsar/pull/26173
### Motivation Continues PIP-486 after the foundation PRs ([#26114](https://github.com/apache/pulsar/pull/26114), [#26115](https://github.com/apache/pulsar/pull/26115), [#26118](https://github.com/apache/pulsar/pull/26118), [#26129](https://github.com/apache/pulsar/pull/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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
