[improve][broker] PIP-486: rebucket rollover and the segments-vs-buckets auto-scale policy #26434
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -46,6 +46,10 @@ | |
| * @param mergeWindow how long a segment must continuously stay below every merge threshold | ||
| * before it becomes merge-eligible (measured from the load record's | ||
| * metadata-store last-modified time) | ||
| * @param rebucketCooldown minimum interval between automatic entry-bucket rollovers | ||
| * @param splitVsRebucketMinMsgRateIn consumer-driven scale-up splits only at/above this inbound | ||
| * msg/s on the busiest segment; below it, entry-buckets grow instead | ||
| * @param maxEntryBucketsPerSegment hard ceiling on a single segment's entry-bucket count | ||
| * @param splitMsgRateIn inbound msg/s above which a segment is split | ||
| * @param splitBytesRateIn inbound bytes/s above which a segment is split | ||
| * @param splitMsgRateOut outbound (dispatched) msg/s above which a segment is split | ||
|
|
@@ -62,8 +66,11 @@ public record AutoScaleConfig( | |
| int minSegments, | ||
| int maxDagDepth, | ||
| Duration splitCooldown, | ||
| Duration rebucketCooldown, | ||
| Duration mergeCooldown, | ||
| Duration mergeWindow, | ||
| double splitVsRebucketMinMsgRateIn, | ||
| int maxEntryBucketsPerSegment, | ||
| double splitMsgRateIn, | ||
| double splitBytesRateIn, | ||
| double splitMsgRateOut, | ||
|
|
@@ -112,6 +119,10 @@ private static AutoScaleConfig brokerDefaults(ServiceConfiguration conf) { | |
| .minSegments(conf.getScalableTopicMinSegments()) | ||
| .maxDagDepth(conf.getScalableTopicMaxDagDepth()) | ||
| .splitCooldown(Duration.ofSeconds(conf.getScalableTopicSplitCooldownSeconds())) | ||
| .rebucketCooldown(Duration.ofSeconds(conf.getScalableTopicRebucketCooldownSeconds())) | ||
| .splitVsRebucketMinMsgRateIn( | ||
| conf.getScalableTopicSplitVsRebucketMinMsgRateInThreshold()) | ||
| .maxEntryBucketsPerSegment(conf.getScalableTopicEntryBucketMaxPerSegment()) | ||
|
Member
There was a problem hiding this comment. [BUG] the three new policy fields were not added to AutoScaleConfig.validated(), whose sibling invariants they mirror
Concretely:
Two of the three are reachable from
Contributor
Author
There was a problem hiding this comment. Added in 5cb4cf7: non-negative |
||
| .mergeCooldown(Duration.ofSeconds(conf.getScalableTopicMergeCooldownSeconds())) | ||
| .mergeWindow(Duration.ofSeconds(conf.getScalableTopicMergeWindowSeconds())) | ||
| .splitMsgRateIn(conf.getScalableTopicSplitMsgRateInThreshold()) | ||
|
|
@@ -145,6 +156,12 @@ private static AutoScaleConfig applyOverride(AutoScaleConfig base, AutoScalePoli | |
| if (o.getSplitCooldownSeconds() != null) { | ||
| b.splitCooldown(Duration.ofSeconds(o.getSplitCooldownSeconds())); | ||
| } | ||
| if (o.getRebucketCooldownSeconds() != null) { | ||
| b.rebucketCooldown(Duration.ofSeconds(o.getRebucketCooldownSeconds())); | ||
| } | ||
| if (o.getSplitVsRebucketMinMsgRateInThreshold() != null) { | ||
| b.splitVsRebucketMinMsgRateIn(o.getSplitVsRebucketMinMsgRateInThreshold()); | ||
| } | ||
| if (o.getMergeCooldownSeconds() != null) { | ||
| b.mergeCooldown(Duration.ofSeconds(o.getMergeCooldownSeconds())); | ||
| } | ||
|
|
@@ -194,6 +211,7 @@ public AutoScaleConfig validated() { | |
| check(maxSegments >= minSegments, "maxSegments must be >= minSegments"); | ||
| check(maxDagDepth >= 0, "maxDagDepth must be >= 0"); | ||
| check(!splitCooldown.isNegative(), "splitCooldown must not be negative"); | ||
| check(!rebucketCooldown.isNegative(), "rebucketCooldown must not be negative"); | ||
| check(!mergeCooldown.isNegative(), "mergeCooldown must not be negative"); | ||
| check(!mergeWindow.isNegative(), "mergeWindow must not be negative"); | ||
| check(splitMsgRateIn > 0, "splitMsgRateInThreshold must be > 0"); | ||
|
|
@@ -204,6 +222,12 @@ public AutoScaleConfig validated() { | |
| check(mergeBytesRateIn >= 0, "mergeBytesRateInThreshold must be >= 0"); | ||
| check(mergeMsgRateOut >= 0, "mergeMsgRateOutThreshold must be >= 0"); | ||
| check(mergeBytesRateOut >= 0, "mergeBytesRateOutThreshold must be >= 0"); | ||
| // Written as >= so a NaN (reachable via the JSON override) fails the check too. | ||
| check(splitVsRebucketMinMsgRateIn >= 0, | ||
| "splitVsRebucketMinMsgRateInThreshold must be >= 0"); | ||
| check(maxEntryBucketsPerSegment >= 1 && maxEntryBucketsPerSegment | ||
| <= EntryBucketSplits.MAX_BUCKETS, | ||
| "maxEntryBucketsPerSegment must be in [1, " + EntryBucketSplits.MAX_BUCKETS + "]"); | ||
| check(splitMsgRateIn > mergeMsgRateIn, | ||
| "splitMsgRateInThreshold must be > mergeMsgRateInThreshold (hysteresis)"); | ||
| check(splitBytesRateIn > mergeBytesRateIn, | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -68,29 +68,115 @@ public static AutoScaleDecision decide( | |
| AutoScaleConfig config, | ||
| long nowMs, | ||
| long lastSplitAtMs, | ||
| long lastMergeAtMs, | ||
| long lastRebucketAtMs) { | ||
|
|
||
| if (!config.enabled()) { | ||
| return AutoScaleDecision.NONE; | ||
| } | ||
|
|
||
| List<SegmentInfo> active = new ArrayList<>(layout.getActiveSegments().values()); | ||
|
|
||
| AutoScaleDecision consumerScale = tryConsumerScale(active, loadBySegment, | ||
| streamConsumerCount, config, nowMs, lastSplitAtMs, lastRebucketAtMs); | ||
| if (!(consumerScale instanceof AutoScaleDecision.NoAction)) { | ||
| return consumerScale; | ||
| } | ||
|
|
||
| AutoScaleDecision split = trySplit(active, loadBySegment, config, nowMs, lastSplitAtMs); | ||
| if (!(split instanceof AutoScaleDecision.NoAction)) { | ||
| return split; | ||
| } | ||
|
|
||
| return tryMerge(active, layout, loadBySegment, config, nowMs, lastMergeAtMs); | ||
| } | ||
|
|
||
| // --- Consumer-driven scale-up: segments vs entry-buckets (PIP-486) --- | ||
|
|
||
| /** | ||
| * Serve surplus consumers (a subscription with more consumers than active segments) by | ||
| * adding capacity on one of two axes: | ||
| * <ul> | ||
| * <li><b>Split</b> when traffic justifies a physical segment: the busiest segment's | ||
| * inbound rate is at or above {@code splitVsRebucketMinMsgRateIn} and the topic is | ||
| * under {@code maxSegments} — today's "segments first" behavior.</li> | ||
| * <li><b>Rebucket-up</b> otherwise (a low-throughput topic, or the topic is at the | ||
| * segment cap): if the existing entry-bucket capacity cannot absorb the surplus, | ||
| * roll the smallest-bucketed segment over to the smallest power of two that lets | ||
| * every consumer own a bucket, capped at {@code maxEntryBucketsPerSegment}. | ||
| * Raising is fast (one rollover sized to the surplus); lowering is deliberately | ||
| * not automated here — spiky consumer counts must not flap the bucketing.</li> | ||
| * </ul> | ||
| */ | ||
| private static AutoScaleDecision tryConsumerScale( | ||
| List<SegmentInfo> active, | ||
| Map<Long, SegmentLoadSample> loadBySegment, | ||
| Map<String, Integer> streamConsumerCount, | ||
| AutoScaleConfig config, | ||
| long nowMs, | ||
| long lastSplitAtMs, | ||
| long lastRebucketAtMs) { | ||
|
|
||
| int consumers = streamConsumerCount.values().stream() | ||
| .mapToInt(Integer::intValue).max().orElse(0); | ||
| int segments = active.size(); | ||
| if (consumers <= segments) { | ||
| return AutoScaleDecision.NONE; | ||
| } | ||
|
|
||
| SegmentInfo busiest = busiestByMsgRateIn(active, loadBySegment); | ||
| if (busiest == null) { | ||
| return AutoScaleDecision.NONE; | ||
| } | ||
| boolean atSegmentCap = segments >= config.maxSegments(); | ||
| boolean belowSplitFloor = statsOf(busiest.segmentId(), loadBySegment).msgRateIn() | ||
| < config.splitVsRebucketMinMsgRateIn(); | ||
|
|
||
| if (!atSegmentCap && !belowSplitFloor) { | ||
| // Traffic justifies a physical segment. | ||
| if (withinCooldown(nowMs, lastSplitAtMs, config.splitCooldown().toMillis())) { | ||
| return AutoScaleDecision.NONE; | ||
| } | ||
| return new AutoScaleDecision.Split(busiest.segmentId(), "consumer-count"); | ||
| } | ||
|
|
||
| // Bucket lane: absorb the surplus with entry-buckets. | ||
| long capacity = 0; | ||
| for (SegmentInfo segment : active) { | ||
| capacity += segment.bucketCount(); | ||
| } | ||
| if (consumers <= capacity) { | ||
| // The existing buckets already absorb the surplus (broker-side fan-out). | ||
| return AutoScaleDecision.NONE; | ||
| } | ||
| if (withinCooldown(nowMs, lastRebucketAtMs, config.rebucketCooldown().toMillis())) { | ||
| return AutoScaleDecision.NONE; | ||
| } | ||
| // One shot: bring every segment below the common per-segment target up to it in a | ||
| // single decision, so the topic converges to a uniform bucketing in one evaluation — | ||
| // never one segment per cooldown, and no arrival-history-dependent skew. | ||
| int target = Math.min(nextPowerOfTwo(ceilDiv(consumers, segments)), | ||
|
Contributor
There was a problem hiding this comment. Thanks for working on this. I found a multi-segment case that I would like to confirm. With four active 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 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.
Contributor
There was a problem hiding this comment. 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 Aggregate capacity is then 18, so no further rebucket is triggered. Looking only at the active layout, the 17 owners can be distributed as 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.
Contributor
Author
There was a problem hiding this comment. 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
Contributor
Author
There was a problem hiding this comment. 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 |
||
| config.maxEntryBucketsPerSegment()); | ||
| List<Long> below = new ArrayList<>(); | ||
| for (SegmentInfo segment : active) { | ||
| if (segment.bucketCount() < target) { | ||
| below.add(segment.segmentId()); | ||
| } | ||
| } | ||
| if (below.isEmpty()) { | ||
| // Bucket capacity is maxed out; the remaining surplus stays idle. | ||
| return AutoScaleDecision.NONE; | ||
| } | ||
| below.sort(Long::compareTo); | ||
| return new AutoScaleDecision.Rebucket(below, target, | ||
| atSegmentCap ? "at-max-segments" : "below-split-rate-floor"); | ||
| } | ||
|
|
||
| // --- Split pass --- | ||
|
|
||
| private static AutoScaleDecision trySplit( | ||
| List<SegmentInfo> active, | ||
| Map<Long, SegmentLoadSample> loadBySegment, | ||
| AutoScaleConfig config, | ||
| long nowMs, | ||
| long lastSplitAtMs) { | ||
|
|
@@ -102,20 +188,7 @@ private static AutoScaleDecision trySplit( | |
| return AutoScaleDecision.NONE; | ||
| } | ||
|
|
||
| // Load-driven: split the segment with the highest overload score among those over | ||
| // at least one split threshold. | ||
| SegmentInfo hottest = null; | ||
| double hottestScore = 1.0; // strictly over threshold means a per-metric ratio > 1.0 | ||
|
|
@@ -239,6 +312,17 @@ private static double combinedRate(long segmentId, Map<Long, SegmentLoadSample> | |
| return s.msgRateIn() + s.bytesRateIn() + s.msgRateOut() + s.bytesRateOut(); | ||
| } | ||
|
|
||
| /** Ceiling integer division for positive operands. */ | ||
| private static int ceilDiv(int a, int b) { | ||
| return (a + b - 1) / b; | ||
| } | ||
|
|
||
| /** The smallest power of two {@code >= v} (for {@code v >= 1}). */ | ||
| private static int nextPowerOfTwo(int v) { | ||
| int highest = Integer.highestOneBit(v); | ||
| return highest == v ? v : highest << 1; | ||
| } | ||
|
|
||
| private static SegmentInfo busiestByMsgRateIn(List<SegmentInfo> active, | ||
| Map<Long, SegmentLoadSample> load) { | ||
| SegmentInfo best = null; | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
[BUG] the new per-segment bucket ceiling is enforced only on the rebucket path — merge and initial creation both bypass it, and above the 16-bit ring the layout persists before it fails
The doc calls this a "Hard ceiling on a single segment's entry-bucket count", and PIP-486 L136-137 states it as an absolute per-segment bound. It is consulted in exactly one place —
ScalableTopicController.rebucketSegment(:749-755) plus the evaluator'sMath.min(..., config.maxEntryBucketsPerSegment()). The other two producers of a bucket count do not:SegmentLayout.mergeSegmentssets the merged segment toseg1.bucketCount() + seg2.bucketCount()(SegmentLayout.java:227) with no clamp. Merging two segments that were each rebucketed to the default 1024 yields 2048 — above the documented hard ceiling.ScalableTopicController.createInitialMetadataderivesNfromEntryBucketSplits.bucketsForBudget(entryBucketBudget, numInitialSegments)(:1524), also unclamped; the split-child path at:1576does the same.scalableTopicEntryBucketBudgetis itself a dynamic config, so a budget of 2048 on a single-segment topic createsN=2048at creation time.Above the ring size this stops being cosmetic.
EntryBucketSplits.equalWidth(n)computesi * 65536 / n, so forn > 65536consecutive split points collide;EntryBucketSplits.rangesthen buildsHashRange.of(start, split - 1)withsplit == start, andHashRange's compact constructor throwsIllegalArgumentException("end must be >= start"). That happens inSubscriptionCoordinator.computeAssignment— i.e. after the layout has already been committed by the metadata CAS atScalableTopicController.java:770-774. The segment is persisted with boundaries no assignment can materialize, and stays that way until it is pruned.Reaching it requires an operator to raise
scalableTopicEntryBucketMaxPerSegment(or the budget) above 65536, both of which aredynamic = trueand neither of which is validated — see the next comment. A bound atHashRange.MAX_HASH + 1, applied at every point that produces a split list rather than only on the rebucket path, would close all three.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Confirmed on all three fronts, fixed in 5cb4cf7.
EntryBucketSplitsnow carries the absolute ring bound (one bucket per 16-bit hash) applied insideequalWidth/bucketsForBudget, so an over-ring split list can never persist boundaries that later failHashRangeconstruction at assignment time; merges clamp the recovered bucket sum to the configured ceiling, and topic creation clamps the budget-derived count to it (ceiling-aware overloads, with the production call sites passing the config). Both dynamic settings are validated now — see thevalidated()thread. Unit tests cover the ring clamp and the merge clamp.