diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java index ecf3ac9409371..c6e098d103563 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java @@ -1468,6 +1468,15 @@ The max allowed delay for delayed delivery (in milliseconds). If the broker rece ) private int scalableTopicEntryBucketBudget = 4; + @FieldContext( + dynamic = true, + category = CATEGORY_POLICIES, + doc = "Hard ceiling on a single segment's entry-bucket count (PIP-486). Bounds both the " + + "manual rebucket operation and the controller's auto rebucket-up; a segment's " + + "bucket count caps how many consumers can share it." + ) + private int scalableTopicEntryBucketMaxPerSegment = 1024; + @FieldContext( dynamic = true, category = CATEGORY_POLICIES, @@ -1486,6 +1495,24 @@ The max allowed delay for delayed delivery (in milliseconds). If the broker rece ) private int scalableTopicSplitCooldownSeconds = 60; + @FieldContext( + dynamic = true, + category = CATEGORY_POLICIES, + doc = "PIP-486 segments-vs-buckets lever: on consumer-driven scale-up, split only if the " + + "busiest segment's inbound msg/s is at or above this floor; below it the " + + "controller grows the segment's entry-buckets instead (a low-throughput topic " + + "should not materialize physical segments just for consumer count)." + ) + private double scalableTopicSplitVsRebucketMinMsgRateInThreshold = 1_000; + + @FieldContext( + dynamic = true, + category = CATEGORY_POLICIES, + doc = "Minimum time (seconds) between automatic entry-bucket rollovers (rebuckets) on a " + + "topic. Coalesces consumer-join bursts, like the split cooldown." + ) + private int scalableTopicRebucketCooldownSeconds = 60; + @FieldContext( dynamic = true, category = CATEGORY_POLICIES, diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/ScalableTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/ScalableTopics.java index ce253f6fdfb94..a95c1d5cc614a 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/ScalableTopics.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/ScalableTopics.java @@ -935,6 +935,50 @@ public void splitSegment( }); } + @POST + @Path("/{tenant}/{namespace}/{topic}/rebucket/{segmentId}") + @Operation(summary = "Rebucket a segment: roll it over to a same-range successor with a new " + + "entry-bucket count.") + @ApiResponses(value = { + @ApiResponse(responseCode = "204", description = "Segment rebucketed successfully"), + @ApiResponse(responseCode = "404", description = "Scalable topic or segment doesn't exist"), + @ApiResponse(responseCode = "412", + description = "Segment is not active or the bucket count is invalid"), + @ApiResponse(responseCode = "500", description = "Internal server error")}) + public void rebucketSegment( + @Suspended final AsyncResponse asyncResponse, + @Parameter(description = "Specify the tenant", required = true) + @PathParam("tenant") String tenant, + @Parameter(description = "Specify the namespace", required = true) + @PathParam("namespace") String namespace, + @Parameter(description = "Specify topic name", required = true) + @PathParam("topic") @Encoded String encodedTopic, + @Parameter(description = "Segment ID to rebucket", required = true) + @PathParam("segmentId") long segmentId, + @Parameter(description = "Entry-bucket count for the successor segment", required = true) + @QueryParam("bucketCount") int bucketCount) { + validateNamespaceName(tenant, namespace); + TopicName tn = TopicName.get(TopicDomain.topic.value(), namespaceName, encodedTopic); + + validateSuperUserAccessAsync() + .thenCompose(__ -> onControllerLeader(tn, + svc -> svc.rebucketSegment(tn, segmentId, bucketCount))) + .thenAccept(__ -> { + log.info().attr("clientAppId", clientAppId()) + .attr("segmentId", segmentId).attr("bucketCount", bucketCount) + .attr("topic", tn) + .log("Rebucketed segment of scalable topic"); + asyncResponse.resume(Response.noContent().build()); + }) + .exceptionally(ex -> { + log.error().attr("clientAppId", clientAppId()) + .attr("segmentId", segmentId).attr("topic", tn) + .exception(ex).log("Failed to rebucket segment"); + resumeAsyncResponseExceptionally(asyncResponse, ex); + return null; + }); + } + @POST @Path("/{tenant}/{namespace}/{topic}/merge/{segmentId1}/{segmentId2}") @Operation(summary = "Merge two adjacent segments into one.") diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/AutoScaleConfig.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/AutoScaleConfig.java index 848b59cfa146b..04a6e1bfdb5f9 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/AutoScaleConfig.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/AutoScaleConfig.java @@ -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()) .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())); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/AutoScaleDecision.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/AutoScaleDecision.java index e8a1713e1f060..9dd74afb16ffc 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/AutoScaleDecision.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/AutoScaleDecision.java @@ -24,7 +24,8 @@ * carries a short {@code reason} string used for logging and metrics. */ public sealed interface AutoScaleDecision - permits AutoScaleDecision.Split, AutoScaleDecision.Merge, AutoScaleDecision.NoAction { + permits AutoScaleDecision.Split, AutoScaleDecision.Merge, AutoScaleDecision.Rebucket, + AutoScaleDecision.NoAction { /** Split {@code segmentId} at its midpoint. */ record Split(long segmentId, String reason) implements AutoScaleDecision { @@ -34,6 +35,16 @@ record Split(long segmentId, String reason) implements AutoScaleDecision { record Merge(long segmentId1, long segmentId2, String reason) implements AutoScaleDecision { } + /** + * Roll every segment in {@code segmentIds} over to a same-range successor with + * {@code newBucketCount} entry-buckets (PIP-486): consumer scale-up served by buckets + * instead of a split. One decision carries the whole batch so a multi-segment topic + * converges to a uniform bucketing in a single evaluation, not one segment per cooldown. + */ + record Rebucket(java.util.List segmentIds, int newBucketCount, String reason) + implements AutoScaleDecision { + } + /** No action this evaluation. */ record NoAction() implements AutoScaleDecision { } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/AutoScalePolicyEvaluator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/AutoScalePolicyEvaluator.java index bcb48ffd6a938..53eb85c47cdfd 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/AutoScalePolicyEvaluator.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/AutoScalePolicyEvaluator.java @@ -68,7 +68,8 @@ public static AutoScaleDecision decide( AutoScaleConfig config, long nowMs, long lastSplitAtMs, - long lastMergeAtMs) { + long lastMergeAtMs, + long lastRebucketAtMs) { if (!config.enabled()) { return AutoScaleDecision.NONE; @@ -76,8 +77,13 @@ public static AutoScaleDecision decide( List active = new ArrayList<>(layout.getActiveSegments().values()); - AutoScaleDecision split = trySplit(active, loadBySegment, streamConsumerCount, - config, nowMs, lastSplitAtMs); + 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; } @@ -85,12 +91,92 @@ public static AutoScaleDecision decide( 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: + * + */ + private static AutoScaleDecision tryConsumerScale( + List active, + Map loadBySegment, + Map 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)), + config.maxEntryBucketsPerSegment()); + List 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 active, Map loadBySegment, - Map streamConsumerCount, AutoScaleConfig config, long nowMs, long lastSplitAtMs) { @@ -102,20 +188,7 @@ private static AutoScaleDecision trySplit( return AutoScaleDecision.NONE; } - // (a) Consumer-driven: per-subscription max. If any managed subscription has more - // consumers than there are active segments, add a segment so the 1:1 assignment can - // give the extra consumer its own segment. Split the busiest segment by msgRateIn so - // the new pair lands where it relieves the most ingest. - int requiredConsumers = streamConsumerCount.values().stream() - .mapToInt(Integer::intValue).max().orElse(0); - if (requiredConsumers > active.size()) { - SegmentInfo target = busiestByMsgRateIn(active, loadBySegment); - if (target != null) { - return new AutoScaleDecision.Split(target.segmentId(), "consumer-count"); - } - } - - // (b) Load-driven: split the segment with the highest overload score among those over + // 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 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 active, Map load) { SegmentInfo best = null; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicController.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicController.java index aebeca97c6ab8..162319a28832c 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicController.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicController.java @@ -108,6 +108,7 @@ public class ScalableTopicController { private volatile long lastSplitAtMs = Long.MIN_VALUE; /** Epoch millis of the last merge on this topic (manual or auto); MIN_VALUE if none. */ private volatile long lastMergeAtMs = Long.MIN_VALUE; + private volatile long lastRebucketAtMs = Long.MIN_VALUE; @Getter private volatile LeaderElectionState leaderState = LeaderElectionState.NoLeader; @@ -378,7 +379,7 @@ private CompletableFuture evaluateAndAct(String trigger) { .thenCombine(collectLoadSamples(), (consumers, load) -> AutoScalePolicyEvaluator.decide(currentLayout, load, consumers, config, clock.millis(), - lastSplitAtMs, lastMergeAtMs)) + lastSplitAtMs, lastMergeAtMs, lastRebucketAtMs)) .thenCompose(decision -> dispatch(decision, config, trigger)); }) .whenComplete((__, ex) -> { @@ -447,6 +448,27 @@ private CompletableFuture dispatch(AutoScaleDecision decision, AutoScaleCo return null; }); } + if (decision instanceof AutoScaleDecision.Rebucket rebucket) { + log.info().attr("segmentIds", rebucket.segmentIds()) + .attr("bucketCount", rebucket.newBucketCount()).attr("reason", rebucket.reason()) + .attr("trigger", trigger).log("Auto rebucket"); + // Roll the whole batch sequentially (each rollover is its own seal → successor → + // CAS → notify). A mid-batch failure aborts the rest; the follow-up evaluation + // retries the remainder after the cooldown. + CompletableFuture chain = CompletableFuture.completedFuture(null); + for (long segmentId : rebucket.segmentIds()) { + chain = chain.thenCompose(__ -> + rebucketSegment(segmentId, rebucket.newBucketCount())) + .thenApply(__ -> null); + } + return chain.thenApply(__ -> { + // Like post-split: a consumer burst may need another rollover once the + // cooldown expires (e.g. when the target was clamped by the per-segment + // bucket ceiling); the chain stops at the first NoAction. + scheduleFollowUpEvaluation(config); + return null; + }); + } if (decision instanceof AutoScaleDecision.Merge merge) { log.info().attr("segmentId1", merge.segmentId1()).attr("segmentId2", merge.segmentId2()) .attr("reason", merge.reason()).attr("trigger", trigger).log("Auto merge"); @@ -714,6 +736,58 @@ public CompletableFuture splitSegment(long segmentId) { }).thenApply(__ -> currentLayout); } + /** + * Rebucket an active segment (PIP-486 rollover): seal it and roll over to a single + * same-range successor with {@code newBucketCount} entry-buckets. A segment's + * bucketing is immutable for its life, so changing it rides the ordinary seal → + * successor flow — producers redirect through the standard segment-gone retry and + * per-key order is preserved by the existing machinery. + * + *

Same ordering invariant as split: the successor topic and subscription cursors are + * created before the parent is terminated, and the metadata CAS lands last. + */ + public CompletableFuture rebucketSegment(long segmentId, int newBucketCount) { + checkLeader(); + int maxBuckets = maxEntryBucketsPerSegment(); + if (newBucketCount < 1 || newBucketCount > maxBuckets) { + return CompletableFuture.failedFuture(new IllegalArgumentException( + "bucketCount must be in [1, " + maxBuckets + "]: " + newBucketCount)); + } + final long nowMs = clock.millis(); + final List newSplits = EntryBucketSplits.equalWidth(newBucketCount); + + // Compute the new layout locally to derive the successor's info (this also validates: + // segment exists, is active, and the splits actually change). + SegmentLayout newLayout = currentLayout.rebucketSegment(segmentId, newSplits, nowMs); + SegmentInfo successor = newLayout.getAllSegments().get(newLayout.getNextSegmentId() - 1); + SegmentInfo parent = currentLayout.getAllSegments().get(segmentId); + String parentTopicName = toSegmentPersistentName(parent); + + return resources.listSubscriptionsAsync(topicName) + .thenCompose(parentSubs -> + createSegmentTopic(successor, new java.util.ArrayList<>(parentSubs))) + .thenCompose(__ -> terminateSegmentTopic(parentTopicName)) + .thenCompose(__ -> resources.updateScalableTopicAsync(topicName, md -> { + SegmentLayout latest = SegmentLayout.fromMetadata(md); + SegmentLayout updated = latest.rebucketSegment(segmentId, newSplits, nowMs); + return updated.toMetadata(md); + })) + .thenCompose(__ -> resources.getScalableTopicMetadataAsync(topicName, true)) + .thenCompose(optMd -> { + currentLayout = SegmentLayout.fromMetadata(optMd.orElseThrow()); + // Start the rebucket cooldown only once the rollover actually happened + // (a failed attempt doesn't burn the cooldown). + lastRebucketAtMs = nowMs; + return notifySubscriptions(currentLayout); + }).thenApply(__ -> currentLayout); + } + + private int maxEntryBucketsPerSegment() { + // Defensive: PulsarService.getConfig() is null in some unit-test mocks. + var config = brokerService.getPulsar().getConfig(); + return config != null ? config.getScalableTopicEntryBucketMaxPerSegment() : 1024; + } + /** * Merge two adjacent active segments. * diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java index 3e5081e3985d2..3cb3a56f1a4ec 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java @@ -180,6 +180,16 @@ public CompletableFuture splitSegment(TopicName topic, long segmentId) { .thenApply(__ -> null); } + /** + * Rebucket a segment (delegates to controller). Same leader contract as + * {@link #splitSegment(TopicName, long)}. + */ + public CompletableFuture rebucketSegment(TopicName topic, long segmentId, int bucketCount) { + return getOrCreateController(topic) + .thenCompose(controller -> controller.rebucketSegment(segmentId, bucketCount)) + .thenApply(__ -> null); + } + /** * Merge two adjacent segments (delegates to controller). Same leader contract as * {@link #splitSegment(TopicName, long)}. diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/SegmentLayout.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/SegmentLayout.java index 9618ac5df4e9a..2cf06fccd1475 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/SegmentLayout.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/SegmentLayout.java @@ -239,6 +239,47 @@ public SegmentLayout mergeSegments(long segmentId1, long segmentId2, long nowMs) return new SegmentLayout(newEpoch, nextSegmentId + 1, newSegments); } + /** + * Produce a new layout by rebucketing a segment: seal it and create a single successor + * with the same hash range but a new entry-bucket boundary list (PIP-486 "rebucket + * rollover"). A segment's bucketing is immutable for its life, so changing N is a layout + * operation: the sealed predecessor drains under its old buckets while the successor takes + * new writes under the new ones — the ordinary seal → successor flow, so per-key order + * across the change is preserved by the existing machinery. + * + * @param segmentId the active segment to rebucket + * @param newSplits the successor's entry-bucket split points (ascending start hashes of + * buckets {@code 1..N-1}; empty = a single bucket spanning the ring) + * @param nowMs wall-clock millis used as the parent's seal time and the successor's + * create time + * @return a new SegmentLayout with the rollover applied + */ + public SegmentLayout rebucketSegment(long segmentId, List newSplits, long nowMs) { + SegmentInfo segment = allSegments.get(segmentId); + if (segment == null) { + throw new IllegalArgumentException("Segment not found: " + segmentId); + } + if (!segment.isActive()) { + throw new IllegalArgumentException("Cannot rebucket non-active segment: " + segmentId); + } + if (newSplits.equals(segment.entryBucketSplits())) { + throw new IllegalArgumentException( + "Segment " + segmentId + " already has the requested entry-bucket splits"); + } + + long newEpoch = epoch + 1; + long successorId = nextSegmentId; + SegmentInfo sealedParent = segment.sealed(newEpoch, nowMs, List.of(successorId)); + SegmentInfo successor = SegmentInfo.active(successorId, segment.hashRange(), + List.of(segmentId), newEpoch, nowMs).withEntryBucketSplits(newSplits); + + Map newSegments = new LinkedHashMap<>(allSegments); + newSegments.put(segmentId, sealedParent); + newSegments.put(successorId, successor); + + return new SegmentLayout(newEpoch, nextSegmentId + 1, newSegments); + } + /** * Prune an expired segment from the DAG. The segment must be sealed and have no * children that are still in the DAG (i.e., children have already been pruned or diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/SubscriptionCoordinator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/SubscriptionCoordinator.java index 22a8ccea0f511..672f6c94e3fcd 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/SubscriptionCoordinator.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/SubscriptionCoordinator.java @@ -344,9 +344,16 @@ public synchronized void close() { * as assignable. */ private boolean isAssignable(SegmentInfo segment, SegmentLayout layout) { - if (drainChecker == null || !segment.isActive()) { + if (drainChecker == null) { return true; } + if (!segment.isActive()) { + // A sealed segment is assignable only while it still has backlog to drain. Once + // fully drained it carries no traffic, and keeping it assignable would pin + // consumers to a dead segment forever (the drain rebalance would spread the + // group across it and its successor). + return !drainedSegmentIds.contains(segment.segmentId()); + } for (long parentId : segment.parentIds()) { // A parent that's no longer in the DAG has been pruned (its data is gone), so // treat it as drained — there's nothing to wait on. diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/AutoScalePolicyEvaluatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/AutoScalePolicyEvaluatorTest.java index e44204116070a..5bc1772802a6e 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/AutoScalePolicyEvaluatorTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/AutoScalePolicyEvaluatorTest.java @@ -54,8 +54,11 @@ private static AutoScaleConfig.AutoScaleConfigBuilder baseConfig() { .minSegments(1) .maxDagDepth(10) .splitCooldown(Duration.ofMinutes(1)) + .rebucketCooldown(Duration.ofMinutes(1)) .mergeCooldown(Duration.ofMinutes(5)) .mergeWindow(Duration.ofMinutes(5)) + .splitVsRebucketMinMsgRateIn(1_000) + .maxEntryBucketsPerSegment(1024) .splitMsgRateIn(SPLIT_MSG_IN) .splitBytesRateIn(SPLIT_BYTES_IN) .splitMsgRateOut(SPLIT_MSG_OUT) @@ -91,7 +94,7 @@ private static AutoScaleDecision decide(SegmentLayout layout, Map consumers, AutoScaleConfig config) { return AutoScalePolicyEvaluator.decide(layout, load, consumers, config, NOW, - NO_PRIOR, NO_PRIOR); + NO_PRIOR, NO_PRIOR, NO_PRIOR); } // --- enable switch --- @@ -110,9 +113,10 @@ public void testDisabledReturnsNoAction() { @Test public void testConsumerDrivenSplitTargetsBusiestSegment() { SegmentLayout layout = initialLayout(2); + // Above the split-vs-rebucket floor (1k msg/s): traffic justifies a physical segment. Map load = Map.of( - 0L, sample(100, 0, 0, 0, old()), - 1L, sample(200, 0, 0, 0, old())); + 0L, sample(1_500, 0, 0, 0, old()), + 1L, sample(2_000, 0, 0, 0, old())); // One subscription with 3 consumers but only 2 segments → need a 3rd segment. AutoScaleDecision d = decide(layout, load, Map.of("sub", 3), baseConfig().build()); assertTrue(d instanceof AutoScaleDecision.Split, d.toString()); @@ -134,25 +138,126 @@ public void testConsumerCountUsesPerSubscriptionMaxNotSum() { } @Test - public void testConsumerDrivenSplitRespectsMaxSegments() { + public void testSurplusAtMaxSegmentsRebuckets() { + // UC#3: at the segment cap the surplus is served by entry-buckets, never a split. + // 2 segments × 2 buckets (budget 4) = capacity 4 < 5 consumers → grow the smallest + // segment to ceil(5/2)=3 rounded up to the power of two, 4. SegmentLayout layout = initialLayout(2); - // Fresh samples so the merge pass can't fire — isolates the split suppression. Map load = Map.of(0L, cold(0), 1L, cold(0)); AutoScaleDecision d = decide(layout, load, Map.of("sub", 5), baseConfig().maxSegments(2).build()); - assertTrue(d instanceof AutoScaleDecision.NoAction, "at maxSegments, no split"); + assertTrue(d instanceof AutoScaleDecision.Rebucket, d.toString()); + AutoScaleDecision.Rebucket r = (AutoScaleDecision.Rebucket) d; + assertEquals(r.segmentIds(), List.of(0L, 1L), + "both below-target segments roll in the one decision"); + assertEquals(r.newBucketCount(), 4); + assertEquals(r.reason(), "at-max-segments"); } @Test public void testConsumerDrivenSplitRespectsSplitCooldown() { SegmentLayout layout = initialLayout(2); - Map load = Map.of(0L, cold(0), 1L, cold(0)); + // Above the floor so the decision stays in the split lane, then blocked by cooldown. + Map load = Map.of( + 0L, sample(1_500, 0, 0, 0, 0), 1L, sample(2_000, 0, 0, 0, 0)); long recentSplit = NOW - Duration.ofSeconds(30).toMillis(); // < 1m cooldown AutoScaleDecision d = AutoScalePolicyEvaluator.decide(layout, load, Map.of("sub", 3), - baseConfig().build(), NOW, recentSplit, NO_PRIOR); + baseConfig().build(), NOW, recentSplit, NO_PRIOR, NO_PRIOR); assertTrue(d instanceof AutoScaleDecision.NoAction, "within split cooldown, no split"); } + // --- consumer-driven rebucket (PIP-486 segments-vs-buckets) --- + + @Test + public void testSurplusBelowFloorWithinCapacityDoesNothing() { + // UC#2 half 1: a cold topic's surplus that fits the existing bucket capacity is + // absorbed by broker-side fan-out — no split, no rollover. + SegmentLayout layout = initialLayout(2); // 2 segments × 2 buckets = capacity 4 + Map load = Map.of(0L, cold(0), 1L, cold(0)); + AutoScaleDecision d = decide(layout, load, Map.of("sub", 4), baseConfig().build()); + assertTrue(d instanceof AutoScaleDecision.NoAction, d.toString()); + } + + @Test + public void testSurplusBelowFloorBeyondCapacityRebuckets() { + // UC#2 half 2: below the floor and beyond capacity → rebucket-up, sized to give every + // consumer a bucket at the next power of two: ceil(5/1)=5 → 8. + SegmentLayout layout = initialLayout(1); // 1 segment × 4 buckets + Map load = Map.of(0L, cold(0)); + AutoScaleDecision d = decide(layout, load, Map.of("sub", 5), baseConfig().build()); + assertTrue(d instanceof AutoScaleDecision.Rebucket, d.toString()); + AutoScaleDecision.Rebucket r = (AutoScaleDecision.Rebucket) d; + assertEquals(r.segmentIds(), List.of(0L)); + assertEquals(r.newBucketCount(), 8); + assertEquals(r.reason(), "below-split-rate-floor"); + } + + @Test + public void testRebucketRespectsRebucketCooldown() { + SegmentLayout layout = initialLayout(1); + Map load = Map.of(0L, cold(0)); + long recentRebucket = NOW - Duration.ofSeconds(30).toMillis(); // < 1m cooldown + AutoScaleDecision d = AutoScalePolicyEvaluator.decide(layout, load, Map.of("sub", 5), + baseConfig().build(), NOW, NO_PRIOR, NO_PRIOR, recentRebucket); + assertTrue(d instanceof AutoScaleDecision.NoAction, "within rebucket cooldown"); + } + + @Test + public void testRebucketCapsAtMaxBucketsPerSegment() { + // A huge surplus is clamped to the per-segment ceiling… + SegmentLayout layout = initialLayout(1); + Map load = Map.of(0L, cold(0)); + AutoScaleDecision d = decide(layout, load, Map.of("sub", 5_000), + baseConfig().maxEntryBucketsPerSegment(8).build()); + assertTrue(d instanceof AutoScaleDecision.Rebucket, d.toString()); + assertEquals(((AutoScaleDecision.Rebucket) d).newBucketCount(), 8); + assertEquals(((AutoScaleDecision.Rebucket) d).segmentIds(), List.of(0L)); + + // …and once the segment is at the ceiling, the remaining surplus stays idle. + SegmentLayout maxed = SegmentLayout.fromMetadata( + ScalableTopicController.createInitialMetadata(1, 8, Map.of())); + AutoScaleDecision none = decide(maxed, load, Map.of("sub", 5_000), + baseConfig().maxEntryBucketsPerSegment(8).build()); + assertTrue(none instanceof AutoScaleDecision.NoAction, none.toString()); + } + + @Test + public void testMultiSegmentSurplusRebucketsAllSegmentsInOneShot() { + // Review scenario: four N=1 segments at the cap and ten consumers must converge in a + // single decision — every segment to the common target 4 (ceil(10/4)=3 → pow2 4), + // never one segment per cooldown. + SegmentLayout layout = SegmentLayout.fromMetadata( + ScalableTopicController.createInitialMetadata(4, 4, Map.of())); // 4 × N=1 + Map load = Map.of( + 0L, cold(0), 1L, cold(0), 2L, cold(0), 3L, cold(0)); + AutoScaleDecision d = decide(layout, load, Map.of("sub", 10), + baseConfig().maxSegments(4).build()); + assertTrue(d instanceof AutoScaleDecision.Rebucket, d.toString()); + AutoScaleDecision.Rebucket r = (AutoScaleDecision.Rebucket) d; + assertEquals(r.segmentIds(), List.of(0L, 1L, 2L, 3L)); + assertEquals(r.newBucketCount(), 4); + } + + @Test + public void testPartiallyRebucketedLayoutConvergesToUniformTarget() { + // A layout left uneven (one segment already rolled to 4, three still at N=1) must + // bring exactly the below-target segments up to the same target — the steady state + // is uniform, not arrival-history-dependent. + SegmentLayout base = SegmentLayout.fromMetadata( + ScalableTopicController.createInitialMetadata(4, 4, Map.of())); // 4 × N=1 + SegmentLayout uneven = base.rebucketSegment(0, EntryBucketSplits.equalWidth(4), 0L); + // Active: successor(id=4, N=4) + segments 1..3 (N=1). Capacity 7 < 10 consumers. + Map load = Map.of( + 1L, cold(0), 2L, cold(0), 3L, cold(0), 4L, cold(0)); + AutoScaleDecision d = decide(uneven, load, Map.of("sub", 10), + baseConfig().maxSegments(4).build()); + assertTrue(d instanceof AutoScaleDecision.Rebucket, d.toString()); + AutoScaleDecision.Rebucket r = (AutoScaleDecision.Rebucket) d; + assertEquals(r.segmentIds(), List.of(1L, 2L, 3L), + "only the below-target segments roll; the already-at-target one is untouched"); + assertEquals(r.newBucketCount(), 4); + } + // --- load-driven split --- @Test @@ -227,7 +332,7 @@ public void testMergeRespectsMergeCooldown() { Map load = Map.of(0L, cold(old()), 1L, cold(old())); long recentMerge = NOW - Duration.ofMinutes(1).toMillis(); // < 5m cooldown AutoScaleDecision d = AutoScalePolicyEvaluator.decide(layout, load, Map.of(), - baseConfig().build(), NOW, NO_PRIOR, recentMerge); + baseConfig().build(), NOW, NO_PRIOR, recentMerge, NO_PRIOR); assertTrue(d instanceof AutoScaleDecision.NoAction, "within merge cooldown, no merge"); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerAutoScaleTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerAutoScaleTest.java index c7a5ac04762f7..4ec3d35512223 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerAutoScaleTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerAutoScaleTest.java @@ -212,6 +212,8 @@ public void testColdSegmentsMerge() throws Exception { @Test public void testConsumerDrivenSplit() throws Exception { + // Zero the split-vs-rebucket floor so the cold test topic exercises the split lane. + config.setScalableTopicSplitVsRebucketMinMsgRateInThreshold(0); startController(1); assertEquals(activeSegmentCount(), 1); @@ -333,7 +335,10 @@ public void testInvalidOverrideCombinationFallsBackToDisabled() throws Exception public void testConsumerBurstConvergesWithoutTicks() throws Exception { // A group of consumers joining in quick succession must converge to one segment // each purely from the event-driven evaluations + post-split follow-up chain — no - // periodic tick and no manual evaluation calls. + // periodic tick and no manual evaluation calls. Zero the split-vs-rebucket floor so + // the cold test topic stays in the split lane (this test is about the convergence + // chain; the bucket lane has its own coverage). + config.setScalableTopicSplitVsRebucketMinMsgRateInThreshold(0); startController(1); for (int i = 1; i <= 4; i++) { controller.registerConsumer("sub", "c" + i, i, ScalableConsumerType.STREAM, diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/SegmentLayoutTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/SegmentLayoutTest.java index fae5afaa100ae..2da9bad009cc3 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/SegmentLayoutTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/SegmentLayoutTest.java @@ -153,6 +153,49 @@ public void testMergeSegments() { assertEquals(merged.parentIds(), List.of(2L, 3L)); } + @Test + public void testRebucketSegment() { + // 1 segment, budget 4 -> the initial segment has N=4 buckets. + ScalableTopicMetadata metadata = ScalableTopicController.createInitialMetadata(1, 4, Map.of()); + SegmentLayout layout = SegmentLayout.fromMetadata(metadata); + + SegmentLayout after = layout.rebucketSegment(0, EntryBucketSplits.equalWidth(8), 123L); + + assertEquals(after.getEpoch(), 1); + assertEquals(after.getActiveSegments().size(), 1); + assertEquals(after.getAllSegments().size(), 2); + + SegmentInfo sealed = after.getAllSegments().get(0L); + assertTrue(sealed.isSealed()); + assertEquals(sealed.childIds(), List.of(1L)); + assertEquals(sealed.sealedAtMs(), 123L); + + // The successor keeps the range and carries the new bucketing. + SegmentInfo successor = after.getAllSegments().get(1L); + assertTrue(successor.isActive()); + assertEquals(successor.parentIds(), List.of(0L)); + assertEquals(successor.hashRange(), sealed.hashRange()); + assertEquals(successor.bucketCount(), 8); + assertEquals(successor.createdAtMs(), 123L); + } + + @Test + public void testRebucketRejectsInvalidTargets() { + ScalableTopicMetadata metadata = ScalableTopicController.createInitialMetadata(1, 4, Map.of()); + SegmentLayout layout = SegmentLayout.fromMetadata(metadata); + + // Unknown segment. + assertThrows(IllegalArgumentException.class, + () -> layout.rebucketSegment(99, EntryBucketSplits.equalWidth(8), 0L)); + // Unchanged bucketing (initial segment already has N=4). + assertThrows(IllegalArgumentException.class, + () -> layout.rebucketSegment(0, EntryBucketSplits.equalWidth(4), 0L)); + // Sealed segment. + SegmentLayout afterSplit = layout.splitSegment(0, 0L); + assertThrows(IllegalArgumentException.class, + () -> afterSplit.rebucketSegment(0, EntryBucketSplits.equalWidth(8), 0L)); + } + @Test public void testSplitRecordsWallClockTimestamps() { ScalableTopicMetadata metadata = ScalableTopicController.createInitialMetadata(1, 4, Map.of()); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/SubscriptionCoordinatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/SubscriptionCoordinatorTest.java index b821820e556cc..3abaaa63177ff 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/SubscriptionCoordinatorTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/SubscriptionCoordinatorTest.java @@ -185,14 +185,18 @@ public void testActiveChildrenBlockedUntilParentDrained() throws Exception { assertFalse(assigned.contains(4L), "child of un-drained parent must be blocked"); assertFalse(assigned.contains(5L), "child of un-drained parent must be blocked"); - // Mark the parent drained — the next poll should pick it up and the children - // must end up assigned. + // Mark the parent drained — the next poll should pick it up: the children + // become assigned and the drained parent retires from the assignment (keeping + // it would pin consumers to a dead segment). drained.add(0L); Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { Set nowAssigned = new HashSet<>(segmentIds( findByName(orderedCoordinator.currentAssignment(), "consumer-1"))); - assertTrue(nowAssigned.containsAll(Set.of(0L, 1L, 2L, 3L, 4L, 5L)), - "after parent drain, all 6 segments must be assigned, got " + nowAssigned); + assertTrue(nowAssigned.containsAll(Set.of(1L, 2L, 3L, 4L, 5L)), + "after parent drain, children + active segments must be assigned, got " + + nowAssigned); + assertFalse(nowAssigned.contains(0L), + "the drained parent must retire from the assignment"); }); } finally { orderedCoordinator.close(); @@ -516,6 +520,71 @@ private SubscriptionCoordinator bucketedCoordinator() { // --- Helpers --- + /** + * PIP-486 review: a sealed parent must retire from the assignable set once drained. + * Rollover topology (parent sealed with 4 buckets, same-range successor with 8), five + * consumers: before the drain the parent's buckets cap the group at 4 owners (successor + * gated, one consumer idle); after the drain every consumer moves to the successor and + * none stays pinned to the dead parent. + */ + @Test + public void testDrainedParentRetiresFromAssignment() throws Exception { + Set drained = ConcurrentHashMap.newKeySet(); + SegmentDrainChecker checker = (segment, sub) -> + CompletableFuture.completedFuture(drained.contains(segment.segmentId())); + SegmentLayout base = SegmentLayout.fromMetadata( + ScalableTopicController.createInitialMetadata(1, 4, Map.of())); + SegmentLayout rolled = base.rebucketSegment(0, EntryBucketSplits.equalWidth(8), 0L); + SubscriptionCoordinator coordinator = new SubscriptionCoordinator("test-sub", + topicName, base, resources, scheduler, Duration.ofMillis(200), + checker, Duration.ofMillis(50), Duration.ofSeconds(5)); + try { + for (int i = 1; i <= 5; i++) { + coordinator.registerConsumer("consumer-" + i, i, mock(TransportCnx.class)).get(); + } + Map before = coordinator.onLayoutChange(rolled).get(); + assertEquals(countAssignedTo(before, 0L), 4, + "undrained parent caps the group at its 4 buckets"); + assertEquals(countAssignedTo(before, 1L), 0, + "successor must be gated while the parent is undrained"); + assertEquals(countIdle(before), 1, "one consumer beyond the parent's capacity idles"); + + drained.add(0L); + coordinator.markSegmentsDrained(Set.of(0L)); + Map after = coordinator.onLayoutChange(rolled).get(); + assertEquals(countAssignedTo(after, 0L), 0, + "a drained parent must not be assigned to anyone"); + assertEquals(countAssignedTo(after, 1L), 5, + "every consumer moves to the successor"); + assertEquals(countIdle(after), 0); + } finally { + coordinator.close(); + } + } + + private static int countAssignedTo(Map m, long segmentId) { + int count = 0; + for (ConsumerAssignment a : m.values()) { + for (var seg : a.assignedSegments()) { + if (seg.segmentId() == segmentId) { + count++; + break; + } + } + } + return count; + } + + private static int countIdle(Map m) { + int count = 0; + for (ConsumerAssignment a : m.values()) { + if (a.assignedSegments().isEmpty()) { + count++; + } + } + return count; + } + private static ConsumerAssignment findByName(Map m, String name) { return m.entrySet().stream() .filter(e -> name.equals(e.getKey().getConsumerName())) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5AutoRebucketTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5AutoRebucketTest.java new file mode 100644 index 0000000000000..7865cad7fc46e --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5AutoRebucketTest.java @@ -0,0 +1,158 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.client.api.v5; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertTrue; +import java.time.Duration; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import lombok.Cleanup; +import org.apache.pulsar.client.api.v5.config.SubscriptionInitialPosition; +import org.apache.pulsar.client.api.v5.schema.Schema; +import org.apache.pulsar.common.policies.data.ScalableTopicMetadata; +import org.awaitility.Awaitility; +import org.testng.annotations.Test; + +/** + * PIP-486 segments-vs-buckets policy, UC#2 end to end: on a low-throughput topic (below the + * split-vs-rebucket rate floor), surplus stream consumers must be served by growing the + * segment's entry-buckets — an automatic rebucket rollover — instead of materializing physical + * segments for consumer count alone. + */ +public class V5AutoRebucketTest extends V5ClientBaseTest { + + @Test + public void testConsumerSurplusOnColdTopicRebucketsInsteadOfSplitting() throws Exception { + String topic = newScalableTopic(1); + String subscription = "auto-rebucket"; + + // Five consumers on a cold one-segment topic (N=4 by default): the per-subscription + // surplus (5 > 1 segment) is below the 1k msg/s split floor and beyond the bucket + // capacity (5 > 4), so the controller must roll the segment over to 8 buckets. + List> consumers = new ArrayList<>(); + for (int i = 0; i < 5; i++) { + consumers.add(track(v5Client.newStreamConsumer(Schema.string()) + .topic(topic) + .subscriptionName(subscription) + .subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST) + .subscribe())); + } + + Awaitility.await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> { + ScalableTopicMetadata md = admin.scalableTopics().getMetadata(topic); + List active = md.getSegments().values().stream() + .filter(ScalableTopicMetadata.SegmentInfo::isActive).toList(); + assertEquals(active.size(), 1, + "a cold topic must not split for consumer count — expected one active segment"); + assertEquals(active.get(0).getEntryBucketSplits().size() + 1, 8, + "the surplus must be absorbed by an automatic rebucket to 8"); + assertEquals(md.getSegments().size(), 2, + "exactly one rollover: sealed parent + successor"); + }); + + // Once the (empty) parent drains, every consumer must move to the successor — none + // may stay pinned to the drained parent, and none may idle. + long successorId = admin.scalableTopics().getMetadata(topic).getSegments().values().stream() + .filter(ScalableTopicMetadata.SegmentInfo::isActive).findFirst().orElseThrow() + .getSegmentId(); + String successorTopic = admin.scalableTopics().getStats(topic).getSegments().values() + .stream().filter(seg -> seg.name().endsWith("-" + successorId)).findFirst() + .orElseThrow().name(); + Awaitility.await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> { + var sub = getTopicReference(successorTopic).orElseThrow().getSubscription(subscription); + assertTrue(sub != null && sub.getConsumers().size() == 5, + "all five consumers must attach to the successor, got " + + (sub == null ? "no subscription" : sub.getConsumers().size())); + }); + + // The rebucketed segment serves all five consumers: keyed traffic must reach them + // collectively, per-key in order, with every key wholly on one consumer. + @Cleanup + Producer producer = v5Client.newProducer(Schema.string()) + .topic(topic) + .create(); + List keys = new ArrayList<>(); + for (int i = 0; i < 16; i++) { + keys.add("key-" + i); + } + int perKey = 5; + Map> sent = new HashMap<>(); + for (String k : keys) { + sent.put(k, new ArrayList<>()); + } + for (int i = 0; i < perKey; i++) { + for (String k : keys) { + String value = k + "-" + i; + producer.newMessage().key(k).value(value).send(); + sent.get(k).add(value); + } + } + + List>> got = new ArrayList<>(); + List drainers = new ArrayList<>(); + for (StreamConsumer consumer : consumers) { + Map> into = new ConcurrentHashMap<>(); + got.add(into); + Thread t = new Thread(() -> { + try { + while (true) { + Message msg = consumer.receive(Duration.ofSeconds(3)); + if (msg == null) { + return; + } + String key = msg.key().orElseThrow(); + into.computeIfAbsent(key, __ -> new ArrayList<>()).add(msg.value()); + consumer.acknowledgeCumulative(msg.id()); + } + } catch (Exception ignored) { + } + }); + t.start(); + drainers.add(t); + } + for (Thread t : drainers) { + t.join(); + } + + int receiving = 0; + for (String k : keys) { + List combined = null; + for (Map> into : got) { + List values = into.get(k); + if (values == null) { + continue; + } + assertTrue(combined == null, "key " + k + " was split across consumers"); + combined = values; + } + assertEquals(combined, sent.get(k), "per-key order/content for key=" + k); + } + for (Map> into : got) { + if (!into.isEmpty()) { + receiving++; + } + } + assertTrue(receiving >= 2, "expected the bucket fan-out to feed several consumers, got " + + receiving); + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5SegmentRebucketTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5SegmentRebucketTest.java new file mode 100644 index 0000000000000..21b21078a3770 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5SegmentRebucketTest.java @@ -0,0 +1,159 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.client.api.v5; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertTrue; +import static org.testng.Assert.expectThrows; +import java.time.Duration; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import lombok.Cleanup; +import org.apache.pulsar.client.admin.PulsarAdminException; +import org.apache.pulsar.client.api.v5.config.SubscriptionInitialPosition; +import org.apache.pulsar.client.api.v5.schema.Schema; +import org.apache.pulsar.common.policies.data.ScalableTopicMetadata; +import org.awaitility.Awaitility; +import org.testng.annotations.Test; + +/** + * PIP-486 rebucket rollover: {@code admin.scalableTopics().rebucketSegment(...)} seals a segment + * and rolls it over to a same-range successor with a new entry-bucket count. The change rides the + * ordinary seal → successor flow, so producers redirect transparently and ordered consumption + * preserves per-key order across the boundary. + */ +public class V5SegmentRebucketTest extends V5ClientBaseTest { + + @Test + public void testRebucketMidFlowPreservesPerKeyOrder() throws Exception { + String topic = newScalableTopic(1); + + @Cleanup + Producer producer = v5Client.newProducer(Schema.string()) + .topic(topic) + .create(); + @Cleanup + StreamConsumer consumer = v5Client.newStreamConsumer(Schema.string()) + .topic(topic) + .subscriptionName("rebucket-sub") + .subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST) + .subscribe(); + + List keys = new ArrayList<>(); + for (int i = 0; i < 16; i++) { + keys.add("key-" + i); + } + int perKey = 10; + Map> sent = new HashMap<>(); + for (String k : keys) { + sent.put(k, new ArrayList<>()); + } + + // Phase 1: lands on the initial segment under its original bucketing (N=4 by default). + for (int i = 0; i < perKey; i++) { + for (String k : keys) { + String value = k + "-" + i; + producer.newMessage().key(k).value(value).send(); + sent.get(k).add(value); + } + } + + long parentId = soleActiveSegmentId(topic); + admin.scalableTopics().rebucketSegment(topic, parentId, 8); + + // The rollover seals the parent and creates one same-range successor with 8 buckets. + Awaitility.await().untilAsserted(() -> { + ScalableTopicMetadata md = admin.scalableTopics().getMetadata(topic); + List active = md.getSegments().values().stream() + .filter(ScalableTopicMetadata.SegmentInfo::isActive).toList(); + assertEquals(active.size(), 1, "rollover must keep exactly one active segment"); + ScalableTopicMetadata.SegmentInfo successor = active.get(0); + assertTrue(successor.getSegmentId() != parentId, "successor must be a new segment"); + assertEquals(successor.getParentIds(), List.of(parentId)); + assertEquals(successor.getEntryBucketSplits().size() + 1, 8, + "successor must carry the new bucket count"); + ScalableTopicMetadata.SegmentInfo parent = md.getSegments().get(parentId); + assertTrue(parent.isSealed(), "parent must be sealed"); + assertEquals(parent.getHashRange().getStart(), successor.getHashRange().getStart()); + assertEquals(parent.getHashRange().getEnd(), successor.getHashRange().getEnd()); + }); + + // Phase 2: the producer re-routes to the successor (transparent segment-gone retry) + // and batches by the new 8-bucket boundaries. + for (int i = perKey; i < perKey * 2; i++) { + for (String k : keys) { + String value = k + "-" + i; + producer.newMessage().key(k).value(value).send(); + sent.get(k).add(value); + } + } + + // Ordered consumption across the rollover: the sealed parent must be fully consumed + // AND acknowledged before the controller serves the successor, so ack as we go (an + // order-sensitive application would) — holding every ack would leave the parent + // undrained and the successor unassigned. + Map> received = new HashMap<>(); + int total = keys.size() * perKey * 2; + for (int i = 0; i < total; i++) { + Message msg = consumer.receive(Duration.ofSeconds(10)); + assertNotNull(msg, "missed message #" + i); + String key = msg.key().orElseThrow(() -> new AssertionError("missing key")); + received.computeIfAbsent(key, __ -> new ArrayList<>()).add(msg.value()); + consumer.acknowledgeCumulative(msg.id()); + } + + for (String k : keys) { + assertEquals(received.get(k), sent.get(k), + "per-key order across the rebucket rollover for key=" + k); + } + } + + @Test + public void testRebucketRejectsInvalidRequests() throws Exception { + String topic = newScalableTopic(1); + long segmentId = soleActiveSegmentId(topic); + + // Out-of-range bucket counts. + expectThrows(PulsarAdminException.class, + () -> admin.scalableTopics().rebucketSegment(topic, segmentId, 0)); + // Unchanged bucketing (the initial segment already has the budget-derived N=4). + expectThrows(PulsarAdminException.class, + () -> admin.scalableTopics().rebucketSegment(topic, segmentId, 4)); + // Unknown segment. + expectThrows(PulsarAdminException.class, + () -> admin.scalableTopics().rebucketSegment(topic, 12345, 8)); + + // A valid rollover still works after the rejections, and the parent (now sealed) + // cannot be rolled over again. + admin.scalableTopics().rebucketSegment(topic, segmentId, 8); + expectThrows(PulsarAdminException.class, + () -> admin.scalableTopics().rebucketSegment(topic, segmentId, 16)); + } + + private long soleActiveSegmentId(String topic) throws Exception { + ScalableTopicMetadata md = admin.scalableTopics().getMetadata(topic); + List active = md.getSegments().values().stream() + .filter(ScalableTopicMetadata.SegmentInfo::isActive).toList(); + assertEquals(active.size(), 1, "expected exactly one active segment"); + return active.get(0).getSegmentId(); + } +} diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/ScalableTopics.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/ScalableTopics.java index b2f7632951e7c..9fed01cf15fd0 100644 --- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/ScalableTopics.java +++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/ScalableTopics.java @@ -324,6 +324,26 @@ CompletableFuture seekSubscriptionAsync(String topic, String subscription, */ CompletableFuture splitSegmentAsync(String topic, long segmentId); + /** + * Rebucket a segment: roll it over to a same-range successor segment with the given + * entry-bucket count (PIP-486). The sealed predecessor drains under its old buckets while + * the successor takes new writes under the new ones. + * + * @param topic Topic name in the format "tenant/namespace/topic" + * @param segmentId ID of the segment to rebucket + * @param bucketCount entry-bucket count for the successor segment + */ + void rebucketSegment(String topic, long segmentId, int bucketCount) throws PulsarAdminException; + + /** + * Rebucket a segment asynchronously. + * + * @param topic Topic name in the format "tenant/namespace/topic" + * @param segmentId ID of the segment to rebucket + * @param bucketCount entry-bucket count for the successor segment + */ + CompletableFuture rebucketSegmentAsync(String topic, long segmentId, int bucketCount); + /** * Merge two adjacent segments into one. * diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/AutoScalePolicyOverride.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/AutoScalePolicyOverride.java index 30ecd78e5a7ab..7444dbbabb280 100644 --- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/AutoScalePolicyOverride.java +++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/AutoScalePolicyOverride.java @@ -61,6 +61,15 @@ public final class AutoScalePolicyOverride { /** Minimum seconds between automatic merges. */ private Long mergeCooldownSeconds; + /** + * PIP-486 segments-vs-buckets floor: on consumer-driven scale-up, split only if the busiest + * segment's inbound msg/s is at or above this; below it, grow entry-buckets instead. + */ + private Double splitVsRebucketMinMsgRateInThreshold; + + /** Minimum seconds between automatic entry-bucket rollovers (rebuckets). */ + private Long rebucketCooldownSeconds; + /** Seconds a segment pair must stay cold before becoming merge-eligible. */ private Long mergeWindowSeconds; diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ScalableTopicMetadata.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ScalableTopicMetadata.java index ce246a1340d66..4e5044f226af3 100644 --- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ScalableTopicMetadata.java +++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ScalableTopicMetadata.java @@ -83,6 +83,14 @@ public static class SegmentInfo { */ private String legacyTopicName; + /** + * PIP-486 entry-bucket split points: the ascending start hashes of buckets + * {@code 1..N-1} within the segment's 16-bit entry-bucket ring (empty = a single + * bucket). The segment has {@code entryBucketSplits.size() + 1} entry-buckets; + * a segment's bucketing is immutable — a rebucket rolls over to a successor. + */ + private List entryBucketSplits; + public boolean isActive() { return "ACTIVE".equals(state); } diff --git a/pulsar-client-admin-api/src/test/java/org/apache/pulsar/common/policies/data/ScalableTopicMetadataTest.java b/pulsar-client-admin-api/src/test/java/org/apache/pulsar/common/policies/data/ScalableTopicMetadataTest.java index 6259d9799a46c..75d874496b654 100644 --- a/pulsar-client-admin-api/src/test/java/org/apache/pulsar/common/policies/data/ScalableTopicMetadataTest.java +++ b/pulsar-client-admin-api/src/test/java/org/apache/pulsar/common/policies/data/ScalableTopicMetadataTest.java @@ -122,7 +122,7 @@ public void testSegmentInfoSealedHelpers() { @Test public void testSegmentInfoHelpersForUnknownStateAreFalse() { ScalableTopicMetadata.SegmentInfo seg = new ScalableTopicMetadata.SegmentInfo( - 0L, hashRange(0, 0xFFFF), "UNKNOWN", List.of(), List.of(), 0L, -1L, null); + 0L, hashRange(0, 0xFFFF), "UNKNOWN", List.of(), List.of(), 0L, -1L, null, null); assertFalse(seg.isActive()); assertFalse(seg.isSealed()); } @@ -131,14 +131,14 @@ public void testSegmentInfoHelpersForUnknownStateAreFalse() { public void testSegmentInfoLegacyFlag() { // A null/empty legacyTopicName is a regular controller-managed segment. ScalableTopicMetadata.SegmentInfo regular = new ScalableTopicMetadata.SegmentInfo( - 1L, hashRange(0, 0xFFFF), "ACTIVE", List.of(), List.of(), 0L, -1L, null); + 1L, hashRange(0, 0xFFFF), "ACTIVE", List.of(), List.of(), 0L, -1L, null, null); assertFalse(regular.isLegacy()); assertNull(regular.getLegacyTopicName()); // A non-empty legacyTopicName marks a legacy segment wrapping a persistent:// topic. ScalableTopicMetadata.SegmentInfo legacy = new ScalableTopicMetadata.SegmentInfo( 0L, hashRange(0, 0xFFFF), "SEALED", List.of(), List.of(2L), 0L, 0L, - "persistent://tenant/ns/x-partition-0"); + "persistent://tenant/ns/x-partition-0", null); assertTrue(legacy.isLegacy()); assertEquals(legacy.getLegacyTopicName(), "persistent://tenant/ns/x-partition-0"); } @@ -186,7 +186,7 @@ private static ScalableTopicMetadata.SegmentInfo activeSegment(long id, int star long createdAtEpoch) { return new ScalableTopicMetadata.SegmentInfo( id, hashRange(start, end), "ACTIVE", - List.of(), List.of(), createdAtEpoch, -1L, null); + List.of(), List.of(), createdAtEpoch, -1L, null, null); } private static ScalableTopicMetadata.SegmentInfo sealedSegment(long id, int start, int end, @@ -196,7 +196,7 @@ private static ScalableTopicMetadata.SegmentInfo sealedSegment(long id, int star long sealedAtEpoch) { return new ScalableTopicMetadata.SegmentInfo( id, hashRange(start, end), "SEALED", - parents, children, createdAtEpoch, sealedAtEpoch, null); + parents, children, createdAtEpoch, sealedAtEpoch, null, null); } private static ScalableTopicMetadata.HashRange hashRange(int start, int end) { diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ScalableTopicsImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ScalableTopicsImpl.java index 068101ad6e9db..9c294a067fd86 100644 --- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ScalableTopicsImpl.java +++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ScalableTopicsImpl.java @@ -286,6 +286,20 @@ public CompletableFuture splitSegmentAsync(String topic, long segmentId) { return asyncPostRequest(path, Entity.entity("", MediaType.APPLICATION_JSON)); } + @Override + public void rebucketSegment(String topic, long segmentId, int bucketCount) + throws PulsarAdminException { + sync(() -> rebucketSegmentAsync(topic, segmentId, bucketCount)); + } + + @Override + public CompletableFuture rebucketSegmentAsync(String topic, long segmentId, int bucketCount) { + TopicName tn = validateTopic(topic); + WebTarget path = topicPath(tn).path("rebucket").path(String.valueOf(segmentId)) + .queryParam("bucketCount", bucketCount); + return asyncPostRequest(path, Entity.entity("", MediaType.APPLICATION_JSON)); + } + // --- Merge --- @Override