[improve][broker] PIP-486: rebucket rollover and the segments-vs-buckets auto-scale policy

[merlimat]

Motivation

Continues PIP-486 after #26208. The consumption path is complete, but the controller had only one scale-up axis: it always split on consumer surplus (until maxSegments), and a segment's entry-bucket count was fixed at creation. Two use cases from the PIP were unserved:

• UC#2 — a low-throughput topic with many consumers should not materialize physical segments (each a managed ledger with cursors) just for consumer count; it should grow entry-buckets instead.
• UC#3 — at maxSegments, consumer surplus beyond the existing bucket capacity had nowhere to go: the extra consumers idled forever.

Both need the missing primitive first: an operation that changes a segment's bucket count at all — the rebucket rollover the PIP prescribes.

Modifications

Rebucket rollover (mechanism)

• SegmentLayout.rebucketSegment(segmentId, newSplits): seal the segment and create a single same-range successor carrying the new entry-bucket boundary list. A segment's bucketing stays immutable for its life; changing N rides the ordinary seal → successor flow, so producers redirect through the standard segment-gone retry and ordered consumption preserves per-key order with no new machinery.
• ScalableTopicController.rebucketSegment(segmentId, bucketCount): the split scaffold reused — successor topic with subscriptions provisioned before the parent is terminated, metadata CAS last, then notify. Stamps a rebucket cooldown.
• scalableTopicEntryBucketMaxPerSegment (default 1024): hard ceiling on a single segment's bucket count.
• Manual admin lever mirroring split/merge: POST .../rebucket/{segmentId}?bucketCount=N and admin.scalableTopics().rebucketSegment(...). The admin metadata DTO now exposes each segment's entryBucketSplits (the broker always serialized them; the client DTO dropped the field), so operators can see a segment's bucketing.

Segments-vs-buckets policy (evaluator)

• New AutoScaleDecision.Rebucket(segmentId, newBucketCount, reason), and the consumer-surplus trigger becomes a two-lane decision (tryConsumerScale):
  • Split when traffic justifies a physical segment: the busiest segment's inbound rate is at or above the new splitVsRebucketMinMsgRateInThreshold floor (default 1k msg/s) and the topic is under maxSegments — today's behavior, unchanged above the floor.
  • Buckets otherwise (below the floor, or at the segment cap): if the existing bucket capacity absorbs the surplus, do nothing — the broker-side fan-out already serves it; if not, rebucket the smallest-bucketed segment to the smallest power of two that gives every consumer a bucket, capped by the per-segment ceiling.
• Raising is fast and coalesced: the rollover is sized from the live consumer count (one operation absorbs a whole burst, unlike the split lane's one-doubling-per-cooldown), joins during an evaluation collapse into one follow-up pass, and a post-rollover follow-up evaluation runs after the scalableTopicRebucketCooldownSeconds (default 60) cooldown for anything that accumulated meanwhile.
• No automatic scale-down: spiky consumer counts must not flap the bucketing. A lazy scale-down can follow separately if operational experience calls for it.
• Both new tunables are overridable per namespace/topic via AutoScalePolicyOverride; the bucket ceiling stays broker-level.

Behavior change to note: a cold topic no longer splits on consumer count — surplus consumers ride entry-buckets until traffic crosses the floor. Two pre-existing controller tests that exercised the split convergence chain on cold topics now pin the floor to zero (the chain is their subject; the bucket lane has its own coverage).

Verifications

• SegmentLayoutTest: same-range successor, lineage/DAG edges, timestamps, invalid targets (unknown, sealed, unchanged splits).
• V5SegmentRebucketTest: a mid-flow 4→8 rollover preserves per-key order end to end — the sealed parent must be fully consumed and acked before the successor is served — plus admin-level rejection cases (out-of-range count, unchanged bucketing, unknown or sealed segment).
• AutoScalePolicyEvaluatorTest: the full decision matrix — split above the floor, absorb within capacity, UC#2 rebucket below the floor, UC#3 rebucket at maxSegments, both cooldowns, ceiling clamp and idle-when-maxed, power-of-two sizing.
• V5AutoRebucketTest (UC#2 e2e): five consumers on a cold one-segment topic trigger an automatic 4→8 rollover with no split, and keyed traffic then fans out across the consumers with per-key order intact and every key wholly on one consumer.
• Full regression sweep: every v5 client suite, all scalable broker suites, and checkstyle across the touched modules.

[void-ptr974]

Thanks for adding this coverage. While walking through the 4→8 rollover, I found a case that this assertion may currently allow.

After the parent drains, SubscriptionCoordinator.isAssignable() still returns true for the sealed parent via !segment.isActive(), while the successor also becomes eligible. With five consumers, computeAssignment() includes both segments and distributes their owner counts as [3, 2]. Three consumers are then assigned only to the already-drained/terminated parent, while only two receive successor traffic. The evaluator sees the active successor capacity as eight and will not correct this later, so receiving >= 2 passes this exact outcome even though the comment above says all five consumers should be served.

Could we keep a sealed parent assignable only until its ID is present in drainedSegmentIds, while preserving the current undrained and no-drain-checker behavior? A deterministic assignment test could verify: before drain, parent owners = 4, successor owners = 0, idle = 1; after drain, parent owners = 0, successor owners = 5, idle = 0.

[merlimat]

Confirmed — isAssignable returned true for any sealed segment unconditionally, and with surplus consumers the post-drain rebalance really did pin part of the group to the dead parent (nothing ever corrected it, since the evaluator counts active capacity only). Fixed in a5218c6: a sealed segment is now assignable only until it lands in drainedSegmentIds, keeping the undrained and no-drain-checker behavior; markSegmentsDrained already rebalances, so retirement takes effect the moment the drain is detected.

Added your deterministic assignment test (undrained: parent owners 4 / successor 0 / idle 1 → drained: 0 / 5 / 0) and replaced this e2e's weak receiving >= 2 with a broker-side assertion that all five consumers attach to the successor.

[void-ptr974]

Thanks for working on this. I found a multi-segment case that I would like to confirm.

With four active N=1 segments at maxSegments and ten consumers, the target bucket count is calculated as N=4, but each decision rebuckets only one segment. The capacity therefore changes as follows:

[1,1,1,1] -> [4,1,1,1] -> [4,4,1,1]

After the first rollover, the total active capacity is seven, so three consumers remain unassigned until another evaluation after the topic-wide rebucket cooldown. Following the same behavior, the PIP example with 64 N=1 segments and 200 consumers appears to require about 46 cooldown-separated rollovers.

I would like to confirm whether this staged convergence is the intended behavior for a multi-segment topic, and whether the “one rollover” expectation applies only when there is a single active segment.

[void-ptr974]

One related point I wanted to clarify, following the convergence case above. Setting aside the drained-parent assignment issue from the other thread, the earlier example looks at how long it takes to reach enough active capacity; this example looks at the steady state left once that condition is met.

With four active N=1 segments, maxSegments=4, maxEntryBucketsPerSegment >= 8, and one STREAM subscription jumping directly to 17 consumers, the evaluator produces:

[1,1,1,1] -> [8,1,1,1] -> [8,8,1,1]

Aggregate capacity is then 18, so no further rebucket is triggered. Looking only at the active layout, the 17 owners can be distributed as [8,7,1,1]. With gradual consumer growth, the same final group size can instead leave [8,4,4,4], distributing the owners as [5,4,4,4].

I can see the trade-off here: the burst path reaches sufficient capacity in only two rollovers, reducing topic, cursor, metadata, and handoff work, while the resulting steady-state fan-out is more uneven and depends on the group’s arrival history.

I would like to confirm whether this is the intended policy boundary: prioritizing aggregate capacity and rollover convergence, while accepting uneven per-segment fan-out as the steady-state trade-off. Clarifying that boundary would also help separate the behavior intended in this change from possible follow-up policy work.

[merlimat]

Not intended to stay that way — fixed in 332a920: one decision now carries every segment below the common per-segment target, and dispatch rolls the batch sequentially, so a multi-segment topic converges in a single evaluation with one cooldown for the whole batch. Your 4×N=1 / 10-consumer case goes [1,1,1,1] → [4,4,4,4] in one shot (added as an evaluator test), and the PIP's 64-segment / 200-consumer example likewise converges in one evaluation. A mid-batch failure aborts the remainder; the post-rollover follow-up evaluation retries it after the cooldown.

[merlimat]

The one-shot change in 332a920 settles this too: every below-target segment reaches the same target in the same decision, so the steady state is uniform regardless of arrival history — the 17-consumer burst ends at [8,8,8,8] rather than [8,8,1,1] (the capacity overshoot from power-of-two rounding is the accepted cost). Added a test for a partially-rebucketed layout converging to the uniform target.