merlimat opened a new pull request, #26129: URL: https://github.com/apache/pulsar/pull/26129
### Motivation Continues PIP-486 after the foundation ([#26114](https://github.com/apache/pulsar/pull/26114)), the producer entry-bucketing ([#26115](https://github.com/apache/pulsar/pull/26115)), and the controller entry-bucket budget ([#26118](https://github.com/apache/pulsar/pull/26118)). A scalable topic subdivides each segment into `N` *entry-buckets*; the producer batches per-bucket and stamps each entry's `entry_hash_min`/`entry_hash_max` range. This PR lays the broker-side **dispatch-by-bucket foundation** so that, when a segment is shared, the broker can route a whole entry to the consumer that owns its bucket. This PR is deliberately **behavior-neutral**. By design the controller assigns each **whole segment to a single consumer**, so the client subscribes `Exclusive` (single-active dispatch — no per-bucket pending tracking), exactly as before. A segment's `N` sets only the *ceiling* on how wide it can later be shared; fanning a segment out into per-bucket `Key_Shared` ownership (and the multi-consumer bucket handoff) is a controller-driven scale-up action left to a follow-up PR. The machinery below is in place but dormant until then. ### Modifications - **Dispatcher** — `PersistentStickyKeyDispatcherMultipleConsumers.getStickyKeyHash` returns the entry's stamped `entry_hash_min` when present (routing the whole entry by its bucket via the existing selector), falling back to the message's sticky-key hash otherwise. `entry_hash_min == 0` is nudged to `1` (0 is the reserved "hash not set" sentinel, and 1 is still inside bucket 0). - **Controller** — `SubscriptionCoordinator.computeAssignment` assigns each whole segment to one consumer (empty `bucketRanges`); `ConsumerAssignment`/`ConsumerSession` carry a per-segment list of owned entry-bucket ranges, serialized to the new proto field `ScalableAssignedSegment.bucket_ranges` (a list of `IntRange`). `EntryBucketSplits.ranges(splits)` maps split points to the per-bucket hash ranges. - **Client (v5 stream consumer)** — the plumbing reads `bucket_ranges` into `ActiveSegment.ownedBucketRanges`; `ScalableStreamConsumer` subscribes `Exclusive` when it owns a whole segment, and (dormant) `Key_Shared` STICKY declaring its owned ranges when it owns a subset, re-subscribing if its owned ranges change. ### Verifications - `PersistentStickyKeyDispatcherMultipleConsumersTest` — `getStickyKeyHash` routes a stamped entry by `entry_hash_min`, nudges `entry_hash_min == 0` to 1, and falls back to the key hash when unstamped. - `SubscriptionCoordinatorTest` — a bucketed (`N > 1`) segment is still assigned whole to a single consumer with no bucket ranges (single-active). - `ConsumerSessionTest` — `toProto` serializes a segment's entry-bucket ranges as ordered `IntRange`s (and none for a whole-segment assignment). - `EntryBucketSplitsTest` — split-points → per-bucket ranges. - `V5EntryBucketDispatchTest` — a lone stream consumer on an `N = 4` segment (Exclusive) still preserves per-key order and drops/duplicates nothing with per-bucket producer batching and stamping in effect. -- 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]
