From 0313c53600d433b31aa9a617651b385e294218ec Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Fri, 28 Aug 2026 16:20:30 -0700 Subject: [PATCH 1/7] =?UTF-8?q?[improve][broker]=20PIP-486:=20rebucket=20r?= =?UTF-8?q?ollover=20=E2=80=94=20change=20a=20segment's=20entry-bucket=20c?= =?UTF-8?q?ount?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A segment's entry-bucketing is immutable for its life, so changing N is a layout operation: seal the segment and roll over to a single same-range successor carrying the new bucket boundaries. The sealed predecessor drains under its old buckets while the successor takes new writes under the new ones — the ordinary seal → successor flow, so producers redirect through the standard segment-gone retry and ordered consumption preserves per-key order across the boundary with no new machinery. - SegmentLayout.rebucketSegment: seal parent, same-range successor with the new splits, DAG edge; validates active + actually-changed splits. - ScalableTopicController.rebucketSegment(segmentId, bucketCount): reuses the split scaffold (successor topic with subscriptions provisioned before the parent is terminated, metadata CAS last) and stamps a rebucket cooldown for the upcoming auto-scale policy. - scalableTopicEntryBucketMaxPerSegment (default 1024): hard ceiling on a single segment's bucket count, enforced by the operation. - 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. Tests: SegmentLayout unit tests (same-range successor, lineage, stamps, invalid targets) and a rollover e2e — keyed traffic across a mid-flow 4→8 rebucket preserves per-key order through the sealed-parent drain and successor attach, plus admin-level rejection cases. Assisted-by: Claude Code (Fable 5) --- .../pulsar/broker/ServiceConfiguration.java | 9 + .../broker/admin/v2/ScalableTopics.java | 44 +++++ .../scalable/ScalableTopicController.java | 53 ++++++ .../scalable/ScalableTopicService.java | 10 ++ .../service/scalable/SegmentLayout.java | 41 +++++ .../service/scalable/SegmentLayoutTest.java | 43 +++++ .../client/api/v5/V5SegmentRebucketTest.java | 159 ++++++++++++++++++ .../pulsar/client/admin/ScalableTopics.java | 20 +++ .../policies/data/ScalableTopicMetadata.java | 8 + .../admin/internal/ScalableTopicsImpl.java | 14 ++ 10 files changed, 401 insertions(+) create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5SegmentRebucketTest.java 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..66ee9f8c8fb74 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, 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/ScalableTopicController.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicController.java index aebeca97c6ab8..a480e4fbbb60c 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; @@ -714,6 +715,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/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/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/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/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 From b95fbf49a609e13522776a97f3e22c2e2005895a Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Fri, 28 Aug 2026 16:26:31 -0700 Subject: [PATCH 2/7] [improve][broker] PIP-486: segments-vs-buckets auto-scale policy MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extends the auto-scale evaluator with the entry-bucket axis: consumer- driven scale-up now chooses between adding a physical segment and growing a segment's entry-buckets. Decision (pure, in AutoScalePolicyEvaluator.tryConsumerScale): when a subscription has more consumers than active segments — - traffic justifies a segment (busiest segment's inbound rate at or above the new splitVsRebucketMinMsgRateIn floor, default 1k msg/s, and under maxSegments): split, exactly today's behavior; - otherwise (a low-throughput topic, or at the segment cap): serve the surplus with entry-buckets. If the existing bucket capacity absorbs it, do nothing (broker-side fan-out already handles it); if not, rebucket the smallest-bucketed segment to the smallest power of two that gives every consumer a bucket, capped by scalableTopicEntryBucketMaxPerSegment. Raising is fast (one rollover sized to the surplus, its own cooldown, post-rollover follow-up evaluation like post-split); lowering is deliberately not automated — spiky consumer counts must not flap the bucketing. A lazy scale-down can follow separately if operational experience wants it. New tunables: scalableTopicSplitVsRebucketMinMsgRateInThreshold (1k), scalableTopicRebucketCooldownSeconds (60), both overridable per namespace/topic via AutoScalePolicyOverride; the per-segment bucket ceiling stays broker-level. Tests: evaluator decision-matrix units (split above the floor, absorb within capacity, UC#2 rebucket below the floor, UC#3 rebucket at maxSegments, both cooldowns, ceiling clamp + idle-when-maxed) and a 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. Assisted-by: Claude Code (Fable 5) --- .../pulsar/broker/ServiceConfiguration.java | 18 +++ .../service/scalable/AutoScaleConfig.java | 17 +++ .../service/scalable/AutoScaleDecision.java | 10 +- .../scalable/AutoScalePolicyEvaluator.java | 124 ++++++++++++--- .../scalable/ScalableTopicController.java | 15 +- .../AutoScalePolicyEvaluatorTest.java | 84 ++++++++-- .../client/api/v5/V5AutoRebucketTest.java | 143 ++++++++++++++++++ .../data/AutoScalePolicyOverride.java | 9 ++ 8 files changed, 391 insertions(+), 29 deletions(-) create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5AutoRebucketTest.java 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 66ee9f8c8fb74..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 @@ -1495,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/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..de1285409af28 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,13 @@ record Split(long segmentId, String reason) implements AutoScaleDecision { record Merge(long segmentId1, long segmentId2, String reason) implements AutoScaleDecision { } + /** + * Roll {@code segmentId} over to a same-range successor with {@code newBucketCount} + * entry-buckets (PIP-486): consumer scale-up served by buckets instead of a split. + */ + record Rebucket(long segmentId, 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..6a63c5ac1313c 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,83 @@ 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; + } + SegmentInfo smallest = smallestByBucketCount(active); + int target = Math.min(nextPowerOfTwo(ceilDiv(consumers, segments)), + config.maxEntryBucketsPerSegment()); + if (target <= smallest.bucketCount()) { + // Bucket capacity is maxed out; the remaining surplus stays idle. + return AutoScaleDecision.NONE; + } + return new AutoScaleDecision.Rebucket(smallest.segmentId(), 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 +179,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 +303,30 @@ 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; + } + + /** The active segment with the fewest entry-buckets (tie-break on segment id). */ + private static SegmentInfo smallestByBucketCount(List active) { + SegmentInfo best = null; + for (SegmentInfo segment : active) { + if (best == null || segment.bucketCount() < best.bucketCount() + || (segment.bucketCount() == best.bucketCount() + && segment.segmentId() < best.segmentId())) { + best = segment; + } + } + return best; + } + 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 a480e4fbbb60c..c69a326fa7447 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 @@ -379,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) -> { @@ -448,6 +448,19 @@ private CompletableFuture dispatch(AutoScaleDecision decision, AutoScaleCo return null; }); } + if (decision instanceof AutoScaleDecision.Rebucket rebucket) { + log.info().attr("segmentId", rebucket.segmentId()) + .attr("bucketCount", rebucket.newBucketCount()).attr("reason", rebucket.reason()) + .attr("trigger", trigger).log("Auto rebucket"); + return rebucketSegment(rebucket.segmentId(), rebucket.newBucketCount()) + .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"); 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..c3651972bc91b 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,87 @@ 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.segmentId(), 0L, "smallest-bucketed segment, id tie-break"); + 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.segmentId(), 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); + + // …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()); + } + // --- load-driven split --- @Test @@ -227,7 +293,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/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..d9c18ba769ace --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5AutoRebucketTest.java @@ -0,0 +1,143 @@ +/* + * 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"); + }); + + // 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-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; From 32b34d03f35c2580fab9085964edec4ecc3ab2fb Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Fri, 28 Aug 2026 16:31:15 -0700 Subject: [PATCH 3/7] [fix][test] Pin cold-topic consumer-split tests to the split lane MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit testConsumerDrivenSplit and testConsumerBurstConvergesWithoutTicks exercise the event-driven split convergence chain on cold test topics; under the segments-vs-buckets policy a cold topic's consumer surplus now prefers entry-buckets. Zero the split-vs-rebucket floor in these two tests so they keep covering the split lane — the bucket lane has its own evaluator-matrix and e2e coverage. Assisted-by: Claude Code (Fable 5) --- .../scalable/ScalableTopicControllerAutoScaleTest.java | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) 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, From 5b8a0b612195f872db11a0d9b7910f5a98993642 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Fri, 28 Aug 2026 17:49:04 -0700 Subject: [PATCH 4/7] [fix][test] Adapt ScalableTopicMetadataTest to the entryBucketSplits DTO field The new field grew the DTO's all-args constructor; pass null at the existing call sites. Assisted-by: Claude Code (Fable 5) --- .../policies/data/ScalableTopicMetadataTest.java | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) 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) { From a5218c62eeee114b6d72418e2bd93f588c430cff Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Sun, 30 Aug 2026 11:03:45 -0700 Subject: [PATCH 5/7] [fix][broker] Retire drained sealed segments from scalable assignments MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A sealed segment stayed assignable forever: isAssignable returned true for any non-active segment, consulting drainedSegmentIds only to gate children. After a rollover (or split) whose parent fully drained, a rebalance could pin part of the consumer group to the dead parent — permanently, since the evaluator's capacity check counts only active segments and nothing ever re-evaluated the spread (found in review). A sealed segment is now assignable only until it lands in drainedSegmentIds; undrained parents and the no-drain-checker test constructor keep today's behavior. markSegmentsDrained already rebalances, so retirement takes effect the moment the drain is detected. Tests: a deterministic coordinator test on the rollover topology (five consumers: undrained parent caps the group at its 4 buckets with the successor gated and one consumer idle; after the drain all five move to the successor and none stays on the parent); the pre-existing child-gating test now asserts the drained parent retires; the auto-rebucket e2e asserts broker-side that all five consumers attach to the successor instead of the too-weak receiving>=2 check. Assisted-by: Claude Code (Fable 5) --- .../scalable/SubscriptionCoordinator.java | 9 ++- .../scalable/SubscriptionCoordinatorTest.java | 77 ++++++++++++++++++- .../client/api/v5/V5AutoRebucketTest.java | 15 ++++ 3 files changed, 96 insertions(+), 5 deletions(-) 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/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 index d9c18ba769ace..7865cad7fc46e 100644 --- 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 @@ -70,6 +70,21 @@ public void testConsumerSurplusOnColdTopicRebucketsInsteadOfSplitting() throws E "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 From 332a920466d13d6c62c344bc7dac2f9b98e2fa64 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Sun, 30 Aug 2026 11:03:45 -0700 Subject: [PATCH 6/7] [improve][broker] PIP-486: converge multi-segment rebuckets in one shot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The bucket lane rolled one segment per decision, so a multi-segment topic converged one rollover per rebucket cooldown (the PIP's 64-segment / 200-consumer example would have needed ~46 cooldown-separated rollovers), and the aggregate-capacity stopping rule left the final bucketing dependent on the group's arrival history (e.g. [8,8,1,1] from a burst vs [8,4,4,4] from gradual growth). Both raised in review. One decision now carries every segment below the common per-segment target (the power of two seating ceil(consumers/segments), capped by the per-segment ceiling), and dispatch rolls the batch sequentially — the topic converges to a uniform bucketing in a single evaluation, with one cooldown for the whole batch. A mid-batch failure aborts the remainder; the post-rollover follow-up evaluation retries it after the cooldown. Tests: the review's four-segment/ten-consumer scenario converges in one decision to a uniform [4,4,4,4]; a partially-rebucketed layout brings exactly the below-target segments up to the same target; existing single-segment and clamping cases updated to the batch decision shape. Assisted-by: Claude Code (Fable 5) --- .../service/scalable/AutoScaleDecision.java | 9 ++-- .../scalable/AutoScalePolicyEvaluator.java | 28 ++++++------ .../scalable/ScalableTopicController.java | 26 +++++++---- .../AutoScalePolicyEvaluatorTest.java | 43 ++++++++++++++++++- 4 files changed, 76 insertions(+), 30 deletions(-) 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 de1285409af28..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 @@ -36,10 +36,13 @@ record Merge(long segmentId1, long segmentId2, String reason) implements AutoSca } /** - * Roll {@code segmentId} over to a same-range successor with {@code newBucketCount} - * entry-buckets (PIP-486): consumer scale-up served by buckets instead of a split. + * 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(long segmentId, int newBucketCount, String reason) implements AutoScaleDecision { + record Rebucket(java.util.List segmentIds, int newBucketCount, String reason) + implements AutoScaleDecision { } /** No action this evaluation. */ 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 6a63c5ac1313c..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 @@ -152,14 +152,23 @@ private static AutoScaleDecision tryConsumerScale( if (withinCooldown(nowMs, lastRebucketAtMs, config.rebucketCooldown().toMillis())) { return AutoScaleDecision.NONE; } - SegmentInfo smallest = smallestByBucketCount(active); + // 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()); - if (target <= smallest.bucketCount()) { + 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; } - return new AutoScaleDecision.Rebucket(smallest.segmentId(), target, + below.sort(Long::compareTo); + return new AutoScaleDecision.Rebucket(below, target, atSegmentCap ? "at-max-segments" : "below-split-rate-floor"); } @@ -314,19 +323,6 @@ private static int nextPowerOfTwo(int v) { return highest == v ? v : highest << 1; } - /** The active segment with the fewest entry-buckets (tie-break on segment id). */ - private static SegmentInfo smallestByBucketCount(List active) { - SegmentInfo best = null; - for (SegmentInfo segment : active) { - if (best == null || segment.bucketCount() < best.bucketCount() - || (segment.bucketCount() == best.bucketCount() - && segment.segmentId() < best.segmentId())) { - best = segment; - } - } - return best; - } - 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 c69a326fa7447..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 @@ -449,17 +449,25 @@ private CompletableFuture dispatch(AutoScaleDecision decision, AutoScaleCo }); } if (decision instanceof AutoScaleDecision.Rebucket rebucket) { - log.info().attr("segmentId", rebucket.segmentId()) + log.info().attr("segmentIds", rebucket.segmentIds()) .attr("bucketCount", rebucket.newBucketCount()).attr("reason", rebucket.reason()) .attr("trigger", trigger).log("Auto rebucket"); - return rebucketSegment(rebucket.segmentId(), rebucket.newBucketCount()) - .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; - }); + // 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()) 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 c3651972bc91b..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 @@ -148,7 +148,8 @@ public void testSurplusAtMaxSegmentsRebuckets() { baseConfig().maxSegments(2).build()); assertTrue(d instanceof AutoScaleDecision.Rebucket, d.toString()); AutoScaleDecision.Rebucket r = (AutoScaleDecision.Rebucket) d; - assertEquals(r.segmentId(), 0L, "smallest-bucketed segment, id tie-break"); + 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"); } @@ -186,7 +187,7 @@ public void testSurplusBelowFloorBeyondCapacityRebuckets() { 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.segmentId(), 0L); + assertEquals(r.segmentIds(), List.of(0L)); assertEquals(r.newBucketCount(), 8); assertEquals(r.reason(), "below-split-rate-floor"); } @@ -210,6 +211,7 @@ public void testRebucketCapsAtMaxBucketsPerSegment() { 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( @@ -219,6 +221,43 @@ public void testRebucketCapsAtMaxBucketsPerSegment() { 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 From 5cb4cf7e4790a65eb75a6b85681d97ee4c6da930 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Mon, 31 Aug 2026 17:01:50 -0700 Subject: [PATCH 7/7] [fix][broker] Rebucket policy hardening: cooldown seeding, bucket ceiling, follow-up, 412s MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses lhotari's second review round on the PR: 1. Leader failover now seeds lastRebucketAtMs: the cooldown-recovery pass classifies a single-parent child by hash range — same range as its parent means a rebucket successor, a narrower range means a split child. Previously a rollover was charged to the split cooldown (and the rebucket cooldown reset to "never"), so a failover both allowed an immediate second rollover and suppressed genuine splits. 2. The per-segment bucket ceiling is a real hard ceiling: an absolute ring bound (one bucket per 16-bit hash) inside EntryBucketSplits stops over-ring split lists from persisting boundaries that fail HashRange construction at assignment time, after the metadata CAS; merges clamp the recovered bucket sum to the configured ceiling; topic creation clamps the budget-derived count to it (both via ceiling-aware overloads, so the dynamic budget cannot exceed the ceiling). 3. AutoScaleConfig.validated() covers the three new fields: non-negative rebucket cooldown, non-negative (NaN-rejecting) split-vs-rebucket floor, and the ceiling bounded to [1, ring size] — restoring the AutoScalePolicyOverride promise that invalid overrides are rejected. 4. The post-rollover follow-up is scheduled off the REBUCKET cooldown with a "post-rebucket" trigger (previously the split cooldown, which only coincides at the defaults), and from whenComplete, so a mid-batch rollover failure still schedules the retry of the remainder. 5. The rebucket endpoint returns 412 for every segment-level rejection (unknown, sealed, invalid or unchanged bucket count) instead of 500, with the API docs aligned; the controller's layout validation comes back through the returned future instead of throwing synchronously. 6. The UC#2 e2e captures drainer failures instead of swallowing them, generates its keys deterministically to cover all eight buckets, and asserts that every one of the five consumers receives traffic. Tests: failover re-seeding (rollover blocked by the seeded cooldown while a hot-segment split still fires), follow-up on the rebucket cooldown with the split cooldown huge, mid-batch failure retried by the follow-up (fault-injecting controller subclass), merge/ring clamps, the new validated() rejections, and 412 status assertions in the rejection e2e. Assisted-by: Claude Code (Fable 5) --- .../broker/admin/v2/ScalableTopics.java | 18 ++- .../service/scalable/AutoScaleConfig.java | 7 + .../service/scalable/EntryBucketSplits.java | 11 +- .../scalable/ScalableTopicController.java | 81 ++++++++---- .../scalable/ScalableTopicService.java | 1 + .../service/scalable/SegmentLayout.java | 14 +- .../service/scalable/AutoScaleConfigTest.java | 27 ++++ .../scalable/EntryBucketSplitsTest.java | 21 +++ .../ScalableTopicControllerAutoScaleTest.java | 122 ++++++++++++++++++ .../client/api/v5/V5AutoRebucketTest.java | 49 ++++++- .../client/api/v5/V5SegmentRebucketTest.java | 21 +-- 11 files changed, 328 insertions(+), 44 deletions(-) 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 a95c1d5cc614a..bacc7b4abeb98 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 @@ -193,6 +193,7 @@ public void createScalableTopic( ScalableTopicMetadata metadata = ScalableTopicController.createInitialMetadata( numInitialSegments, pulsar().getConfiguration().getScalableTopicEntryBucketBudget(), + pulsar().getConfiguration().getScalableTopicEntryBucketMaxPerSegment(), props); return resources().createScalableTopicAsync(tn, metadata) .thenCompose(ignored -> createInitialSegmentTopicsAsync(tn, metadata)); @@ -941,9 +942,9 @@ public void splitSegment( + "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 = "404", description = "Scalable topic doesn't exist"), + @ApiResponse(responseCode = "412", description = "Segment is unknown, not active, " + + "or the bucket count is invalid or unchanged"), @ApiResponse(responseCode = "500", description = "Internal server error")}) public void rebucketSegment( @Suspended final AsyncResponse asyncResponse, @@ -971,6 +972,17 @@ public void rebucketSegment( asyncResponse.resume(Response.noContent().build()); }) .exceptionally(ex -> { + Throwable cause = FutureUtil.unwrapCompletionException(ex); + if (cause instanceof IllegalArgumentException) { + // Segment-level validation (unknown, sealed, bad or unchanged bucket + // count): a client error, not a server one. + log.info().attr("clientAppId", clientAppId()) + .attr("segmentId", segmentId).attr("topic", tn) + .attr("reason", cause.getMessage()).log("Rebucket rejected"); + asyncResponse.resume(new RestException( + Response.Status.PRECONDITION_FAILED, cause.getMessage())); + return null; + } log.error().attr("clientAppId", clientAppId()) .attr("segmentId", segmentId).attr("topic", tn) .exception(ex).log("Failed to rebucket segment"); 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 04a6e1bfdb5f9..fb29e8cffdb33 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 @@ -211,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"); @@ -221,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, diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/EntryBucketSplits.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/EntryBucketSplits.java index 57a6461119e66..64d6da7dd1b9e 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/EntryBucketSplits.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/EntryBucketSplits.java @@ -38,6 +38,14 @@ */ final class EntryBucketSplits { + /** + * Absolute ceiling on a segment's entry-bucket count: one bucket per hash on the 16-bit + * ring. Beyond it {@link #equalWidth} would produce duplicate split points, whose + * zero-width ranges fail {@code HashRange} construction — at assignment time, after the + * layout was already persisted. + */ + static final int MAX_BUCKETS = HashRange.MAX_HASH + 1; + private EntryBucketSplits() { } @@ -46,7 +54,7 @@ private EntryBucketSplits() { * {@code floor(budget / segmentCount)}, but at least 1. */ static int bucketsForBudget(int budget, int segmentCount) { - return Math.max(1, budget / segmentCount); + return Math.min(Math.max(1, budget / segmentCount), MAX_BUCKETS); } /** @@ -70,6 +78,7 @@ static List ranges(List splits) { /** Equal-width split points for {@code bucketCount} buckets; empty when {@code bucketCount <= 1}. */ static List equalWidth(int bucketCount) { + bucketCount = Math.min(bucketCount, MAX_BUCKETS); if (bucketCount <= 1) { return List.of(); } 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 162319a28832c..d253ee42c6d03 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 @@ -198,26 +198,35 @@ public CompletableFuture initialize() { } /** - * Recover the auto split/merge cooldown clocks after winning leadership. The timestamps - * are in-memory only, but the layout itself records when each segment was created — a - * split's children have exactly one parent, a merge's child has two — so the most recent - * creation time of each class is exactly when the last split / merge happened. Without - * this, every leader failover would reset both cooldowns and e.g. allow an auto merge - * seconds after one just ran on the previous leader. + * Recover the auto split/merge/rebucket cooldown clocks after winning leadership. The + * timestamps are in-memory only, but the layout itself records when each segment was + * created — a merge's child has two parents, and a single-parent child is a split child + * (strictly narrower hash range than its parent) or a rebucket successor (the same hash + * range as its parent) — so the most recent creation time of each class is exactly when + * the last operation of that kind happened. Without this, every leader failover would + * reset the cooldowns and e.g. allow an auto merge or rollover seconds after one just + * ran on the previous leader. */ private void seedAutoScaleCooldownsFromLayout() { long split = Long.MIN_VALUE; long merge = Long.MIN_VALUE; + long rebucket = Long.MIN_VALUE; for (SegmentInfo segment : currentLayout.getAllSegments().values()) { int parents = segment.parentIds().size(); if (parents == 1) { - split = Math.max(split, segment.createdAtMs()); + SegmentInfo parent = currentLayout.getAllSegments().get(segment.parentIds().get(0)); + if (parent != null && parent.hashRange().equals(segment.hashRange())) { + rebucket = Math.max(rebucket, segment.createdAtMs()); + } else { + split = Math.max(split, segment.createdAtMs()); + } } else if (parents >= 2) { merge = Math.max(merge, segment.createdAtMs()); } } lastSplitAtMs = split; lastMergeAtMs = merge; + lastRebucketAtMs = rebucket; } /** @@ -453,21 +462,22 @@ private CompletableFuture dispatch(AutoScaleDecision decision, AutoScaleCo .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. + // CAS → notify). A mid-batch failure aborts the rest; the follow-up evaluation — + // scheduled from whenComplete so it survives that failure — retries the remainder + // once the rebucket cooldown expires. 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; - }); + return chain.whenComplete((__, ___) -> + // 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. Delayed by the + // REBUCKET cooldown — the two cooldowns are configured independently. + scheduleFollowUpEvaluation(config.rebucketCooldown().toMillis() + 1, + "post-rebucket")); } if (decision instanceof AutoScaleDecision.Merge merge) { log.info().attr("segmentId1", merge.segmentId1()).attr("segmentId2", merge.segmentId2()) @@ -485,11 +495,14 @@ private CompletableFuture dispatch(AutoScaleDecision decision, AutoScaleCo * stops naturally at the first evaluation that decides {@code NoAction}. */ private void scheduleFollowUpEvaluation(AutoScaleConfig config) { + scheduleFollowUpEvaluation(config.splitCooldown().toMillis() + 1, "post-split"); + } + + private void scheduleFollowUpEvaluation(long delayMs, String trigger) { if (closed || !isLeader()) { return; } - long delayMs = config.splitCooldown().toMillis() + 1; - scheduler().schedule(() -> runAutoScaleSafely("post-split"), + scheduler().schedule(() -> runAutoScaleSafely(trigger), delayMs, TimeUnit.MILLISECONDS); } @@ -757,8 +770,14 @@ public CompletableFuture rebucketSegment(long segmentId, int newB 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); + // segment exists, is active, and the splits actually change). Argument failures come + // back through the returned future, never as a synchronous throw. + SegmentLayout newLayout; + try { + newLayout = currentLayout.rebucketSegment(segmentId, newSplits, nowMs); + } catch (IllegalArgumentException e) { + return CompletableFuture.failedFuture(e); + } SegmentInfo successor = newLayout.getAllSegments().get(newLayout.getNextSegmentId() - 1); SegmentInfo parent = currentLayout.getAllSegments().get(segmentId); String parentTopicName = toSegmentPersistentName(parent); @@ -802,7 +821,8 @@ public CompletableFuture mergeSegments(long segmentId1, long segm final long nowMs = clock.millis(); // Compute the new layout locally to derive merged segment info - SegmentLayout newLayout = currentLayout.mergeSegments(segmentId1, segmentId2, nowMs); + SegmentLayout newLayout = currentLayout.mergeSegments(segmentId1, segmentId2, nowMs, + maxEntryBucketsPerSegment()); SegmentInfo merged = newLayout.getAllSegments().get(newLayout.getNextSegmentId() - 1); SegmentInfo parent1 = currentLayout.getAllSegments().get(segmentId1); SegmentInfo parent2 = currentLayout.getAllSegments().get(segmentId2); @@ -822,7 +842,8 @@ public CompletableFuture mergeSegments(long segmentId1, long segm // Step 3: Atomic metadata update (only after topic + cursors are ready + parents terminated) .thenCompose(__ -> resources.updateScalableTopicAsync(topicName, md -> { SegmentLayout latest = SegmentLayout.fromMetadata(md); - SegmentLayout updated = latest.mergeSegments(segmentId1, segmentId2, nowMs); + SegmentLayout updated = latest.mergeSegments(segmentId1, segmentId2, nowMs, + maxEntryBucketsPerSegment()); return updated.toMetadata(md); })) .thenCompose(__ -> resources.getScalableTopicMetadataAsync(topicName, true)) @@ -1512,6 +1533,19 @@ private CompletableFuture notifySubscriptions(SegmentLayout layout) { public static ScalableTopicMetadata createInitialMetadata(int numInitialSegments, int entryBucketBudget, Map properties) { + return createInitialMetadata(numInitialSegments, entryBucketBudget, + EntryBucketSplits.MAX_BUCKETS, properties); + } + + /** + * As {@link #createInitialMetadata(int, int, Map)}, clamping each initial segment's + * budget-derived entry-bucket count to {@code maxBucketsPerSegment} (the configured + * per-segment ceiling — the budget is a dynamic setting and must not exceed it). + */ + public static ScalableTopicMetadata createInitialMetadata(int numInitialSegments, + int entryBucketBudget, + int maxBucketsPerSegment, + Map properties) { if (numInitialSegments < 1) { throw new IllegalArgumentException("Must have at least 1 segment"); } @@ -1521,7 +1555,8 @@ public static ScalableTopicMetadata createInitialMetadata(int numInitialSegments // PIP-486: share the topic's entry-bucket budget equally across the initial segments. List entryBucketSplits = EntryBucketSplits.equalWidth( - EntryBucketSplits.bucketsForBudget(entryBucketBudget, numInitialSegments)); + Math.min(EntryBucketSplits.bucketsForBudget(entryBucketBudget, numInitialSegments), + maxBucketsPerSegment)); long nowMs = System.currentTimeMillis(); for (int i = 0; i < numInitialSegments; i++) { 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 3cb3a56f1a4ec..ee8d02d2445d2 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 @@ -146,6 +146,7 @@ public CompletableFuture createScalableTopic(TopicName topic, int numIniti ScalableTopicMetadata metadata = ScalableTopicController.createInitialMetadata( numInitialSegments, brokerService.getPulsar().getConfiguration().getScalableTopicEntryBucketBudget(), + brokerService.getPulsar().getConfiguration().getScalableTopicEntryBucketMaxPerSegment(), properties); // Write the scalable metadata FIRST, then materialize the underlying segment topics. 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 2cf06fccd1475..6f62047314e9d 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 @@ -204,6 +204,16 @@ public SegmentLayout splitSegment(long segmentId, long nowMs) { * @return a new SegmentLayout with the merge applied */ public SegmentLayout mergeSegments(long segmentId1, long segmentId2, long nowMs) { + return mergeSegments(segmentId1, segmentId2, nowMs, EntryBucketSplits.MAX_BUCKETS); + } + + /** + * As {@link #mergeSegments(long, long, long)}, clamping the merged segment's entry-bucket + * count to {@code maxBucketsPerSegment} (the configured per-segment ceiling): the merged + * segment recovers the parents' buckets, but never past the hard ceiling. + */ + public SegmentLayout mergeSegments(long segmentId1, long segmentId2, long nowMs, + int maxBucketsPerSegment) { SegmentInfo seg1 = allSegments.get(segmentId1); SegmentInfo seg2 = allSegments.get(segmentId2); if (seg1 == null || seg2 == null) { @@ -223,8 +233,8 @@ public SegmentLayout mergeSegments(long segmentId1, long segmentId2, long nowMs) // PIP-486: a merge is the inverse of a split — the merged segment recovers both parents' buckets // (N1 + N2), so the topic's total entry-bucket count stays ≈ the budget as segments coalesce. - List mergedEntryBucketSplits = - EntryBucketSplits.equalWidth(seg1.bucketCount() + seg2.bucketCount()); + List mergedEntryBucketSplits = EntryBucketSplits.equalWidth( + Math.min(seg1.bucketCount() + seg2.bucketCount(), maxBucketsPerSegment)); SegmentInfo sealed1 = seg1.sealed(newEpoch, nowMs, List.of(mergedId)); SegmentInfo sealed2 = seg2.sealed(newEpoch, nowMs, List.of(mergedId)); SegmentInfo merged = SegmentInfo.active(mergedId, mergedRange, diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/AutoScaleConfigTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/AutoScaleConfigTest.java index 7a991834461eb..44665d94576a4 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/AutoScaleConfigTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/AutoScaleConfigTest.java @@ -129,6 +129,33 @@ public void testValidationRejectsBadConfig() { negativeCooldown.setScalableTopicSplitCooldownSeconds(-1); assertThrows(IllegalArgumentException.class, () -> AutoScaleConfig.fromBrokerConfig(negativeCooldown)); + + // Negative rebucket cooldown: would silently disable the rollover throttle. + ServiceConfiguration negativeRebucket = new ServiceConfiguration(); + negativeRebucket.setScalableTopicRebucketCooldownSeconds(-1); + assertThrows(IllegalArgumentException.class, + () -> AutoScaleConfig.fromBrokerConfig(negativeRebucket)); + + // Negative or NaN split-vs-rebucket floor: breaks lane selection. + ServiceConfiguration negativeFloor = new ServiceConfiguration(); + negativeFloor.setScalableTopicSplitVsRebucketMinMsgRateInThreshold(-1); + assertThrows(IllegalArgumentException.class, + () -> AutoScaleConfig.fromBrokerConfig(negativeFloor)); + ServiceConfiguration nanFloor = new ServiceConfiguration(); + nanFloor.setScalableTopicSplitVsRebucketMinMsgRateInThreshold(Double.NaN); + assertThrows(IllegalArgumentException.class, + () -> AutoScaleConfig.fromBrokerConfig(nanFloor)); + + // Bucket ceiling outside [1, ring size]: 0 rejects every rebucket, beyond the 16-bit + // ring the split points would collide. + ServiceConfiguration zeroCeiling = new ServiceConfiguration(); + zeroCeiling.setScalableTopicEntryBucketMaxPerSegment(0); + assertThrows(IllegalArgumentException.class, + () -> AutoScaleConfig.fromBrokerConfig(zeroCeiling)); + ServiceConfiguration hugeCeiling = new ServiceConfiguration(); + hugeCeiling.setScalableTopicEntryBucketMaxPerSegment(1 << 20); + assertThrows(IllegalArgumentException.class, + () -> AutoScaleConfig.fromBrokerConfig(hugeCeiling)); } @Test diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/EntryBucketSplitsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/EntryBucketSplitsTest.java index e1e1c827aa3f7..4db2ade5db08a 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/EntryBucketSplitsTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/EntryBucketSplitsTest.java @@ -24,6 +24,7 @@ import org.apache.pulsar.broker.resources.ScalableTopicMetadata; import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.scalable.HashRange; +import org.apache.pulsar.common.scalable.SegmentInfo; import org.testng.annotations.Test; /** @@ -142,4 +143,24 @@ public void testRangesFromSplits() { List.of(HashRange.of(0, 0x3FFF), HashRange.of(0x4000, 0x7FFF), HashRange.of(0x8000, 0xBFFF), HashRange.of(0xC000, 0xFFFF))); } + + @Test + public void testEqualWidthClampsToRingSize() { + // Beyond one bucket per hash the split points would collide and produce zero-width + // ranges; equalWidth clamps so ranges(...) always materializes. + var splits = EntryBucketSplits.equalWidth(EntryBucketSplits.MAX_BUCKETS * 4); + assertEquals(splits.size() + 1, EntryBucketSplits.MAX_BUCKETS); + var ranges = EntryBucketSplits.ranges(splits); + assertEquals(ranges.size(), EntryBucketSplits.MAX_BUCKETS); + } + + @Test + public void testMergeClampsToBucketCeiling() { + // Merging recovers the parents' buckets but never past the configured per-segment + // ceiling. + var md = ScalableTopicController.createInitialMetadata(2, 8, Map.of()); // 2 × N=4 + SegmentLayout merged = SegmentLayout.fromMetadata(md).mergeSegments(0, 1, 0L, 6); + SegmentInfo child = merged.getAllSegments().get(2L); + assertEquals(child.bucketCount(), 6, "sum 8 clamped to the ceiling 6"); + } } 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 4ec3d35512223..2ceb498a471d6 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 @@ -30,6 +30,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.atomic.AtomicBoolean; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.broker.resources.NamespaceResources; @@ -366,6 +367,127 @@ public void testSplitCooldownBlocksSecondSplit() throws Exception { assertEquals(activeSegmentCount(), 3, "second split blocked by cooldown"); } + private int soleActiveBucketCount() throws Exception { + var active = controller.getLayout().get().getActiveSegments().values(); + assertEquals(active.size(), 1, "expected a single active segment"); + return active.iterator().next().bucketCount(); + } + + /** + * PIP-486 review: the post-rollover follow-up must be scheduled off the REBUCKET + * cooldown. With the split cooldown huge and the rebucket cooldown short, consumers + * joining inside the rebucket cooldown are absorbed by the follow-up evaluation — under + * the old post-split scheduling nothing would re-evaluate for an hour. + */ + @Test + public void testPostRebucketFollowUpUsesRebucketCooldown() throws Exception { + config.setScalableTopicSplitCooldownSeconds(3600); + config.setScalableTopicRebucketCooldownSeconds(3); + config.setScalableTopicAutoScaleIntervalSeconds(3600); + startController(1); // one segment, N=4 + + for (int i = 1; i <= 5; i++) { + controller.registerConsumer("sub", "c" + i, i, ScalableConsumerType.STREAM, + mock(TransportCnx.class)).get(); + } + Awaitility.await().atMost(Duration.ofSeconds(10)).untilAsserted( + () -> assertEquals(soleActiveBucketCount(), 8, + "five consumers on a cold topic roll the segment to 8")); + + // More joins land inside the rebucket cooldown: their event-driven evaluations are + // blocked, so only the post-rollover follow-up can serve them. + for (int i = 6; i <= 9; i++) { + controller.registerConsumer("sub", "c" + i, i, ScalableConsumerType.STREAM, + mock(TransportCnx.class)).get(); + } + Awaitility.await().atMost(Duration.ofSeconds(20)).untilAsserted( + () -> assertEquals(soleActiveBucketCount(), 16, + "the follow-up after the rebucket cooldown must absorb the late joins")); + } + + /** + * PIP-486 review: a mid-batch rollover failure must still schedule the follow-up + * evaluation, which retries the remainder once the rebucket cooldown expires. + */ + @Test + public void testMidBatchRebucketFailureStillSchedulesFollowUp() throws Exception { + config.setScalableTopicSplitCooldownSeconds(3600); + config.setScalableTopicRebucketCooldownSeconds(2); + config.setScalableTopicAutoScaleIntervalSeconds(3600); + resources.createScalableTopicAsync(topicName, + ScalableTopicController.createInitialMetadata(2, 4, Map.of())).get(); // 2 × N=2 + AtomicBoolean injectedOnce = new AtomicBoolean(); + controller = new ScalableTopicController(topicName, resources, brokerService, + coordinationService) { + @Override + public CompletableFuture rebucketSegment(long segmentId, + int newBucketCount) { + if (segmentId == 1 && injectedOnce.compareAndSet(false, true)) { + return CompletableFuture.failedFuture(new RuntimeException("injected failure")); + } + return super.rebucketSegment(segmentId, newBucketCount); + } + }; + controller.initialize().get(); + + // Seven consumers, cold topic: batch Rebucket([0,1] → 4). Segment 0 rolls, segment 1 + // fails once; capacity is then 6 < 7, so the follow-up must retry it. + for (int i = 1; i <= 7; i++) { + controller.registerConsumer("sub", "c" + i, i, ScalableConsumerType.STREAM, + mock(TransportCnx.class)).get(); + } + Awaitility.await().atMost(Duration.ofSeconds(20)).untilAsserted(() -> { + var active = controller.getLayout().get().getActiveSegments().values(); + assertEquals(active.size(), 2); + for (var segment : active) { + assertEquals(segment.bucketCount(), 4, + "the follow-up must retry the failed remainder of the batch"); + } + }); + } + + /** + * PIP-486 review: after a leader failover the rollover cooldown must be re-seeded from + * the layout (a same-range single-parent child is a rebucket successor), and the rollover + * must NOT be charged to the split cooldown. + */ + @Test + public void testRebucketCooldownSurvivesLeaderFailover() throws Exception { + config.setScalableTopicSplitCooldownSeconds(3600); + config.setScalableTopicRebucketCooldownSeconds(3600); + config.setScalableTopicAutoScaleIntervalSeconds(3600); + startController(1); // one segment, N=4 + for (int i = 1; i <= 5; i++) { + controller.registerConsumer("sub", "c" + i, i, ScalableConsumerType.STREAM, + mock(TransportCnx.class)).get(); + } + Awaitility.await().atMost(Duration.ofSeconds(10)).untilAsserted( + () -> assertEquals(soleActiveBucketCount(), 8)); + + // Leadership moves. The new leader must know a rollover just ran… + controller.close().join(); + controller = new ScalableTopicController(topicName, resources, brokerService, + coordinationService); + controller.initialize().get(); + + // …so another consumer-driven rollover is blocked by the seeded rebucket cooldown… + controller.registerConsumer("sub", "c9", 9L, ScalableConsumerType.STREAM, + mock(TransportCnx.class)).get(); + controller.evaluateAutoScaleForTest().get(); + assertEquals(soleActiveBucketCount(), 8, + "the seeded rebucket cooldown must block an immediate second rollover"); + + // …while a genuine load-driven split is NOT suppressed: the rollover must not have + // been charged to the split cooldown. + long successorId = controller.getLayout().get().getActiveSegments().keySet() + .iterator().next(); + resources.reportSegmentLoadAsync(topicName, successorId, + new SegmentLoadStats(20_000, 0, 0, 0)).get(); + controller.evaluateAutoScaleForTest().get(); + assertEquals(activeSegmentCount(), 2, + "a hot-segment split must fire — the rollover is not a split"); + } + @Test public void testSplitCooldownSurvivesLeaderFailover() throws Exception { config.setScalableTopicSplitCooldownSeconds(3600); 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 index 7865cad7fc46e..72e121a00d61e 100644 --- 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 @@ -20,16 +20,20 @@ import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertTrue; +import java.nio.charset.StandardCharsets; 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 java.util.concurrent.CopyOnWriteArrayList; 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.apache.pulsar.common.scalable.HashRange; +import org.apache.pulsar.common.scalable.ScalableTopicHashing; import org.awaitility.Awaitility; import org.testng.annotations.Test; @@ -91,10 +95,10 @@ public void testConsumerSurplusOnColdTopicRebucketsInsteadOfSplitting() throws E Producer producer = v5Client.newProducer(Schema.string()) .topic(topic) .create(); - List keys = new ArrayList<>(); - for (int i = 0; i < 16; i++) { - keys.add("key-" + i); - } + // Deterministic key set covering every one of the successor's 8 buckets (at least two + // keys each), so every consumer — each owns at least one bucket — must receive traffic + // and the delivery assertion below can demand all five, not "some". + List keys = bucketCoveringKeys(8, 2); int perKey = 5; Map> sent = new HashMap<>(); for (String k : keys) { @@ -110,6 +114,7 @@ public void testConsumerSurplusOnColdTopicRebucketsInsteadOfSplitting() throws E List>> got = new ArrayList<>(); List drainers = new ArrayList<>(); + List drainerFailures = new CopyOnWriteArrayList<>(); for (StreamConsumer consumer : consumers) { Map> into = new ConcurrentHashMap<>(); got.add(into); @@ -124,7 +129,8 @@ public void testConsumerSurplusOnColdTopicRebucketsInsteadOfSplitting() throws E into.computeIfAbsent(key, __ -> new ArrayList<>()).add(msg.value()); consumer.acknowledgeCumulative(msg.id()); } - } catch (Exception ignored) { + } catch (Throwable t2) { + drainerFailures.add(t2); } }); t.start(); @@ -133,6 +139,7 @@ public void testConsumerSurplusOnColdTopicRebucketsInsteadOfSplitting() throws E for (Thread t : drainers) { t.join(); } + assertTrue(drainerFailures.isEmpty(), "drainer failed: " + drainerFailures); int receiving = 0; for (String k : keys) { @@ -152,7 +159,35 @@ public void testConsumerSurplusOnColdTopicRebucketsInsteadOfSplitting() throws E receiving++; } } - assertTrue(receiving >= 2, "expected the bucket fan-out to feed several consumers, got " - + receiving); + assertEquals(receiving, consumers.size(), + "the key set covers every bucket, so every consumer must receive traffic"); + } + + /** + * Deterministic keys such that every one of {@code buckets} equal-width entry-buckets holds + * at least {@code minPerBucket} keys — computed with the same hashing the producer uses. + */ + private static List bucketCoveringKeys(int buckets, int minPerBucket) { + int bucketWidth = (HashRange.MAX_HASH + 1) / buckets; + int[] counts = new int[buckets]; + List keys = new ArrayList<>(); + for (int i = 0; keys.size() < 64; i++) { + String key = "key-" + i; + int hash = ScalableTopicHashing.entryBucketHash( + ScalableTopicHashing.murmur(key.getBytes(StandardCharsets.UTF_8))); + int bucket = Math.min(hash / bucketWidth, buckets - 1); + if (counts[bucket] < minPerBucket) { + counts[bucket]++; + keys.add(key); + } + boolean covered = true; + for (int c : counts) { + covered &= c >= minPerBucket; + } + if (covered) { + break; + } + } + return keys; } } 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 index 21b21078a3770..b78ff700879ea 100644 --- 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 @@ -132,21 +132,26 @@ public void testRebucketRejectsInvalidRequests() throws Exception { String topic = newScalableTopic(1); long segmentId = soleActiveSegmentId(topic); + // All segment-level rejections are client errors: HTTP 412, never 500. // Out-of-range bucket counts. - expectThrows(PulsarAdminException.class, - () -> admin.scalableTopics().rebucketSegment(topic, segmentId, 0)); + assertEquals(expectThrows(PulsarAdminException.class, + () -> admin.scalableTopics().rebucketSegment(topic, segmentId, 0)) + .getStatusCode(), 412); // Unchanged bucketing (the initial segment already has the budget-derived N=4). - expectThrows(PulsarAdminException.class, - () -> admin.scalableTopics().rebucketSegment(topic, segmentId, 4)); + assertEquals(expectThrows(PulsarAdminException.class, + () -> admin.scalableTopics().rebucketSegment(topic, segmentId, 4)) + .getStatusCode(), 412); // Unknown segment. - expectThrows(PulsarAdminException.class, - () -> admin.scalableTopics().rebucketSegment(topic, 12345, 8)); + assertEquals(expectThrows(PulsarAdminException.class, + () -> admin.scalableTopics().rebucketSegment(topic, 12345, 8)) + .getStatusCode(), 412); // 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)); + assertEquals(expectThrows(PulsarAdminException.class, + () -> admin.scalableTopics().rebucketSegment(topic, segmentId, 16)) + .getStatusCode(), 412); } private long soleActiveSegmentId(String topic) throws Exception {