This is an automated email from the ASF dual-hosted git repository.
merlimat pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 1f45e889087 [improve][broker] PIP-486 (PR2): controller entry-bucket
budget per segment (#26118)
1f45e889087 is described below
commit 1f45e889087f33215ab15aeac9e284e931bd44a7
Author: Matteo Merli <[email protected]>
AuthorDate: Tue Jun 30 10:56:23 2026 -0700
[improve][broker] PIP-486 (PR2): controller entry-bucket budget per segment
(#26118)
---
.../apache/pulsar/broker/ServiceConfiguration.java | 12 ++
.../pulsar/broker/admin/v2/ScalableTopics.java | 7 +-
.../broker/service/scalable/EntryBucketSplits.java | 64 +++++++++++
.../service/scalable/ScalableTopicController.java | 19 ++-
.../service/scalable/ScalableTopicService.java | 4 +-
.../broker/service/scalable/SegmentLayout.java | 15 ++-
.../scalable/AutoScalePolicyEvaluatorTest.java | 2 +-
.../service/scalable/DagWatchSessionTest.java | 2 +-
.../service/scalable/EntryBucketSplitsTest.java | 128 +++++++++++++++++++++
.../ScalableTopicControllerAutoScaleTest.java | 2 +-
.../scalable/ScalableTopicControllerTest.java | 10 +-
.../broker/service/scalable/SegmentLayoutTest.java | 38 +++---
.../scalable/SubscriptionCoordinatorTest.java | 2 +-
13 files changed, 268 insertions(+), 37 deletions(-)
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 987a7ab8385..67112f19a77 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
@@ -1428,6 +1428,18 @@ public class ServiceConfiguration implements
PulsarConfiguration {
)
private int scalableTopicMinSegments = 1;
+ @FieldContext(
+ dynamic = true,
+ category = CATEGORY_POLICIES,
+ doc = "Total entry-bucket budget per scalable topic. Entry-buckets
are the unit of key-shared "
+ + "consumption parallelism within a segment, so this
budget is how many consumers can "
+ + "share a single segment's keys. It is distributed across
the topic's segments (each "
+ + "gets floor(budget / segmentCount), at least 1): a
single-segment topic starts with "
+ + "the whole budget, and as the topic splits into more
segments each segment settles "
+ + "toward 1 bucket (full batching)."
+ )
+ private int scalableTopicEntryBucketBudget = 4;
+
@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 5a365ffa523..ff9b16ccf20 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
@@ -188,7 +188,9 @@ public class ScalableTopics extends AdminResource {
}
Map<String, String> props = properties != null ?
properties : Map.of();
ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(
- numInitialSegments, props);
+ numInitialSegments,
+
pulsar().getConfiguration().getScalableTopicEntryBucketBudget(),
+ props);
return resources().createScalableTopicAsync(tn, metadata)
.thenCompose(ignored ->
createInitialSegmentTopicsAsync(tn, metadata));
})
@@ -331,7 +333,8 @@ public class ScalableTopics extends AdminResource {
return precheck.thenApply(__ -> partitions);
}).thenCompose(partitions -> {
ScalableTopicMetadata metadata =
-
ScalableTopicController.createMigratedMetadata(persistentBase, partitions);
+
ScalableTopicController.createMigratedMetadata(persistentBase, partitions,
+
pulsar().getConfiguration().getScalableTopicEntryBucketBudget());
return createMigratedChildTopicsAsync(scalableName, metadata)
.thenCompose(__ ->
resources().createScalableTopicAsync(scalableName, metadata))
.thenCompose(__ ->
terminateLegacyTopicsAsync(persistentBase, partitions));
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
new file mode 100644
index 00000000000..1cfd013982e
--- /dev/null
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/EntryBucketSplits.java
@@ -0,0 +1,64 @@
+/*
+ * 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.broker.service.scalable;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.pulsar.common.scalable.HashRange;
+
+/**
+ * PIP-486 controller boundary-selection policy: turns a segment's
entry-bucket count {@code N} into the
+ * equal-width split points stored on the segment — the ascending, inclusive
start hashes of buckets
+ * {@code 1..N-1} (bucket 0 implicitly starts at {@code 0x0000}). {@code N =
1} is a single bucket
+ * spanning the whole ring, i.e. no splits.
+ *
+ * <p>The entry-bucket ring is always the full {@code [0x0000, 0xFFFF]} (the
low 16 bits of the key's
+ * {@code Murmur3_32} hash), independent of the segment's segment-routing hash
range. The split values
+ * must match the producer's bucketing (a key falls in the bucket equal to the
number of splits {@code <=}
+ * its entry-bucket hash).
+ *
+ * <p>Equal-width is the initial policy; the wire format is range-based, so a
future controller could
+ * place arbitrary (traffic-balanced) boundaries with no wire or dispatch
change.
+ */
+final class EntryBucketSplits {
+
+ private EntryBucketSplits() {
+ }
+
+ /**
+ * The number of entry-buckets a segment gets from a topic budget shared
across its segments:
+ * {@code floor(budget / segmentCount)}, but at least 1.
+ */
+ static int bucketsForBudget(int budget, int segmentCount) {
+ return Math.max(1, budget / segmentCount);
+ }
+
+ /** Equal-width split points for {@code bucketCount} buckets; empty when
{@code bucketCount <= 1}. */
+ static List<Integer> equalWidth(int bucketCount) {
+ if (bucketCount <= 1) {
+ return List.of();
+ }
+ int ringSize = HashRange.MAX_HASH + 1;
+ List<Integer> splits = new ArrayList<>(bucketCount - 1);
+ for (int i = 1; i < bucketCount; i++) {
+ splits.add((int) ((long) i * ringSize / bucketCount));
+ }
+ return splits;
+ }
+}
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 a85187df28c..aebeca97c6a 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
@@ -1436,6 +1436,7 @@ public class ScalableTopicController {
* Create initial segment layout for a new scalable topic.
*/
public static ScalableTopicMetadata createInitialMetadata(int
numInitialSegments,
+ int entryBucketBudget,
Map<String, String>
properties) {
if (numInitialSegments < 1) {
throw new IllegalArgumentException("Must have at least 1 segment");
@@ -1444,12 +1445,17 @@ public class ScalableTopicController {
int rangeSize = (HashRange.MAX_HASH + 1) / numInitialSegments;
Map<Long, SegmentInfo> segments = new LinkedHashMap<>();
+ // PIP-486: share the topic's entry-bucket budget equally across the
initial segments.
+ List<Integer> entryBucketSplits = EntryBucketSplits.equalWidth(
+ EntryBucketSplits.bucketsForBudget(entryBucketBudget,
numInitialSegments));
+
long nowMs = System.currentTimeMillis();
for (int i = 0; i < numInitialSegments; i++) {
int start = i * rangeSize;
int end = (i == numInitialSegments - 1) ? HashRange.MAX_HASH :
(start + rangeSize - 1);
HashRange range = HashRange.of(start, end);
- SegmentInfo segment = SegmentInfo.active(i, range, 0, nowMs);
+ SegmentInfo segment = SegmentInfo.active(i, range, 0, nowMs)
+ .withEntryBucketSplits(entryBucketSplits);
segments.put((long) i, segment);
}
@@ -1484,11 +1490,17 @@ public class ScalableTopicController {
* @param partitions the source partition count; {@code <= 0} means
non-partitioned
*/
public static ScalableTopicMetadata createMigratedMetadata(TopicName
persistentBase,
- int partitions)
{
+ int partitions,
+ int
entryBucketBudget) {
int n = Math.max(partitions, 1);
long nowMs = System.currentTimeMillis();
Map<Long, SegmentInfo> segments = new LinkedHashMap<>();
+ // PIP-486: the active children share the topic's entry-bucket budget.
The sealed legacy parents
+ // take no new writes, so they keep a single bucket (no splits).
+ List<Integer> childEntryBucketSplits = EntryBucketSplits.equalWidth(
+ EntryBucketSplits.bucketsForBudget(entryBucketBudget, n));
+
// Child IDs are N..2N-1; every child lists every parent (full fan-in).
List<Long> childIds = new ArrayList<>(n);
for (int j = 0; j < n; j++) {
@@ -1516,7 +1528,8 @@ public class ScalableTopicController {
long segId = n + j;
int start = j * rangeSize;
int end = (j == n - 1) ? HashRange.MAX_HASH : (start + rangeSize -
1);
- SegmentInfo child = SegmentInfo.active(segId, HashRange.of(start,
end), parentIds, 0, nowMs);
+ SegmentInfo child = SegmentInfo.active(segId, HashRange.of(start,
end), parentIds, 0, nowMs)
+ .withEntryBucketSplits(childEntryBucketSplits);
segments.put(segId, child);
}
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 2489840a2f0..3e5081e3985 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
@@ -144,7 +144,9 @@ public class ScalableTopicService {
}
ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(
- numInitialSegments, properties);
+ numInitialSegments,
+
brokerService.getPulsar().getConfiguration().getScalableTopicEntryBucketBudget(),
+ properties);
// Write the scalable metadata FIRST, then materialize the underlying
segment topics.
// The metadata is the source of truth: its presence is what defines
whether the topic
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 c823fb75e91..9618ac5df4e 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
@@ -176,11 +176,15 @@ public class SegmentLayout {
long childId1 = nextSegmentId;
long childId2 = nextSegmentId + 1;
+ // PIP-486: a split divides the parent's entry-buckets between its
children — N/2 each (at least
+ // 1) — so the topic's total stays ≈ the budget as it fans out into
more, narrower segments.
+ List<Integer> childEntryBucketSplits =
+ EntryBucketSplits.equalWidth(Math.max(1, segment.bucketCount()
/ 2));
SegmentInfo sealedParent = segment.sealed(newEpoch, nowMs,
List.of(childId1, childId2));
SegmentInfo child1 = SegmentInfo.active(childId1, splitRanges[0],
- List.of(segmentId), newEpoch, nowMs);
+ List.of(segmentId), newEpoch,
nowMs).withEntryBucketSplits(childEntryBucketSplits);
SegmentInfo child2 = SegmentInfo.active(childId2, splitRanges[1],
- List.of(segmentId), newEpoch, nowMs);
+ List.of(segmentId), newEpoch,
nowMs).withEntryBucketSplits(childEntryBucketSplits);
Map<Long, SegmentInfo> newSegments = new LinkedHashMap<>(allSegments);
newSegments.put(segmentId, sealedParent);
@@ -217,10 +221,15 @@ public class SegmentLayout {
long mergedId = nextSegmentId;
HashRange mergedRange = seg1.hashRange().merge(seg2.hashRange());
+ // 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<Integer> mergedEntryBucketSplits =
+ EntryBucketSplits.equalWidth(seg1.bucketCount() +
seg2.bucketCount());
SegmentInfo sealed1 = seg1.sealed(newEpoch, nowMs, List.of(mergedId));
SegmentInfo sealed2 = seg2.sealed(newEpoch, nowMs, List.of(mergedId));
SegmentInfo merged = SegmentInfo.active(mergedId, mergedRange,
- List.of(segmentId1, segmentId2), newEpoch, nowMs);
+ List.of(segmentId1, segmentId2), newEpoch, nowMs)
+ .withEntryBucketSplits(mergedEntryBucketSplits);
Map<Long, SegmentInfo> newSegments = new LinkedHashMap<>(allSegments);
newSegments.put(segmentId1, sealed1);
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 b85845de647..e4420411607 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
@@ -68,7 +68,7 @@ public class AutoScalePolicyEvaluatorTest {
private static SegmentLayout initialLayout(int segments) {
return SegmentLayout.fromMetadata(
- ScalableTopicController.createInitialMetadata(segments,
Map.of()));
+ ScalableTopicController.createInitialMetadata(segments, 4,
Map.of()));
}
/** A load sample with the given rates, last modified {@code ageMs} ago. */
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/DagWatchSessionTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/DagWatchSessionTest.java
index f1967c7e959..f513f7ae45d 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/DagWatchSessionTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/DagWatchSessionTest.java
@@ -441,7 +441,7 @@ public class DagWatchSessionTest {
public void testOnMetadataChangedAfterCloseIsNoop() {
session.close();
// Build a minimal metadata object; close should short-circuit before
any work runs.
- ScalableTopicMetadata md =
ScalableTopicController.createInitialMetadata(1, Map.of());
+ ScalableTopicMetadata md =
ScalableTopicController.createInitialMetadata(1, 4, Map.of());
session.onMetadataChanged(md);
verify(ctx, never()).writeAndFlush(any());
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
new file mode 100644
index 00000000000..040d4495f99
--- /dev/null
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/EntryBucketSplitsTest.java
@@ -0,0 +1,128 @@
+/*
+ * 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.broker.service.scalable;
+
+import static org.testng.Assert.assertEquals;
+import java.util.List;
+import java.util.Map;
+import org.apache.pulsar.broker.resources.ScalableTopicMetadata;
+import org.apache.pulsar.common.naming.TopicName;
+import org.testng.annotations.Test;
+
+/**
+ * PIP-486: the controller's entry-bucket budget — equal-width split
computation and how it is shared
+ * across a topic's segments on create / migrate / split / merge.
+ */
+public class EntryBucketSplitsTest {
+
+ // --- equalWidth ---
+
+ @Test
+ public void testEqualWidthSingleBucketHasNoSplits() {
+ assertEquals(EntryBucketSplits.equalWidth(1), List.of());
+ assertEquals(EntryBucketSplits.equalWidth(0), List.of());
+ }
+
+ @Test
+ public void testEqualWidthPowersOfTwoAreExact() {
+ assertEquals(EntryBucketSplits.equalWidth(2), List.of(0x8000));
+ assertEquals(EntryBucketSplits.equalWidth(4), List.of(0x4000, 0x8000,
0xC000));
+ }
+
+ @Test
+ public void testEqualWidthNonPowerOfTwoRoundsDown() {
+ // 65536/3 = 21845.33, 2*65536/3 = 43690.67
+ assertEquals(EntryBucketSplits.equalWidth(3), List.of(21845, 43690));
+ }
+
+ // --- bucketsForBudget ---
+
+ @Test
+ public void testBucketsForBudgetFloorsAtLeastOne() {
+ assertEquals(EntryBucketSplits.bucketsForBudget(4, 1), 4);
+ assertEquals(EntryBucketSplits.bucketsForBudget(4, 2), 2);
+ assertEquals(EntryBucketSplits.bucketsForBudget(4, 4), 1);
+ assertEquals(EntryBucketSplits.bucketsForBudget(4, 8), 1); // floor
never drops below 1
+ assertEquals(EntryBucketSplits.bucketsForBudget(4, 3), 1); // 4/3 -> 1
+ }
+
+ // --- createInitialMetadata shares the budget across segments ---
+
+ @Test
+ public void testInitialSingleSegmentTakesWholeBudget() {
+ ScalableTopicMetadata md =
ScalableTopicController.createInitialMetadata(1, 4, Map.of());
+ assertEquals(md.getSegments().get(0L).bucketCount(), 4);
+ }
+
+ @Test
+ public void testInitialBudgetSplitAcrossSegments() {
+ ScalableTopicMetadata md =
ScalableTopicController.createInitialMetadata(2, 4, Map.of());
+ assertEquals(md.getSegments().get(0L).bucketCount(), 2);
+ assertEquals(md.getSegments().get(1L).bucketCount(), 2);
+ }
+
+ @Test
+ public void testInitialSegmentsAtOrAboveBudgetSettleAtOneBucket() {
+ ScalableTopicMetadata md =
ScalableTopicController.createInitialMetadata(4, 4, Map.of());
+ for (long id = 0; id < 4; id++) {
+ assertEquals(md.getSegments().get(id).bucketCount(), 1);
+ }
+ }
+
+ // --- createMigratedMetadata: children share the budget, sealed legacy
parents keep one bucket ---
+
+ @Test
+ public void testMigratedChildrenShareBudgetParentsHaveOneBucket() {
+ TopicName base = TopicName.get("persistent://tenant/ns/topic");
+ ScalableTopicMetadata md =
ScalableTopicController.createMigratedMetadata(base, 2, 4);
+ // parents 0..1 (sealed legacy), children 2..3 (active).
+ assertEquals(md.getSegments().get(0L).bucketCount(), 1);
+ assertEquals(md.getSegments().get(1L).bucketCount(), 1);
+ assertEquals(md.getSegments().get(2L).bucketCount(), 2);
+ assertEquals(md.getSegments().get(3L).bucketCount(), 2);
+ }
+
+ // --- split halves buckets, merge sums them ---
+
+ @Test
+ public void testSplitHalvesParentBuckets() {
+ ScalableTopicMetadata md =
ScalableTopicController.createInitialMetadata(1, 4, Map.of());
+ SegmentLayout afterSplit =
SegmentLayout.fromMetadata(md).splitSegment(0, 0L);
+ // parent 0 had N=4; the two children (ids 1, 2) get N=2 each.
+ assertEquals(afterSplit.getAllSegments().get(1L).bucketCount(), 2);
+ assertEquals(afterSplit.getAllSegments().get(2L).bucketCount(), 2);
+ }
+
+ @Test
+ public void testSplitOfSingleBucketStaysSingleBucket() {
+ ScalableTopicMetadata md =
ScalableTopicController.createInitialMetadata(4, 4, Map.of());
+ SegmentLayout afterSplit =
SegmentLayout.fromMetadata(md).splitSegment(0, 0L);
+ long firstChild =
afterSplit.getAllSegments().get(0L).childIds().get(0);
+
assertEquals(afterSplit.getAllSegments().get(firstChild).bucketCount(), 1);
+ }
+
+ @Test
+ public void testMergeSumsChildBuckets() {
+ // 4 segments, each N=1; merging two adjacent ones recovers N = 1 + 1
= 2.
+ ScalableTopicMetadata md =
ScalableTopicController.createInitialMetadata(4, 4, Map.of());
+ SegmentLayout merged = SegmentLayout.fromMetadata(md).mergeSegments(0,
1, 0L);
+ long mergedId = merged.getAllSegments().get(0L).childIds().get(0);
+ assertEquals(merged.getAllSegments().get(mergedId).bucketCount(), 2);
+ }
+}
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 6b1fc907696..c7a5ac04762 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
@@ -153,7 +153,7 @@ public class ScalableTopicControllerAutoScaleTest {
private void startController(int initialSegments) throws Exception {
resources.createScalableTopicAsync(topicName,
- ScalableTopicController.createInitialMetadata(initialSegments,
Map.of())).get();
+ ScalableTopicController.createInitialMetadata(initialSegments,
4, Map.of())).get();
controller = new ScalableTopicController(topicName, resources,
brokerService,
coordinationService);
controller.initialize().get();
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerTest.java
index 959a6a11064..52d7a25adb9 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerTest.java
@@ -99,7 +99,7 @@ public class ScalableTopicControllerTest {
// Seed the topic's initial metadata so initialize() has something to
load.
ScalableTopicMetadata metadata =
-
ScalableTopicController.createInitialMetadata(INITIAL_SEGMENTS, Map.of());
+
ScalableTopicController.createInitialMetadata(INITIAL_SEGMENTS, 4, Map.of());
resources.createScalableTopicAsync(topicName, metadata).get();
// --- Mock the BrokerService / PulsarService / PulsarAdmin chain ---
@@ -478,7 +478,7 @@ public class ScalableTopicControllerTest {
@Test
public void testCreateInitialMetadataDefaults() {
- ScalableTopicMetadata md =
ScalableTopicController.createInitialMetadata(4, Map.of());
+ ScalableTopicMetadata md =
ScalableTopicController.createInitialMetadata(4, 4, Map.of());
assertEquals(md.getEpoch(), 0);
assertEquals(md.getNextSegmentId(), 4);
assertEquals(md.getSegments().size(), 4);
@@ -487,7 +487,7 @@ public class ScalableTopicControllerTest {
@Test
public void testCreateInitialMetadataRejectsZeroSegments() {
assertThrows(IllegalArgumentException.class,
- () -> ScalableTopicController.createInitialMetadata(0,
Map.of()));
+ () -> ScalableTopicController.createInitialMetadata(0, 4,
Map.of()));
}
// --- close / lifecycle ---
@@ -827,7 +827,7 @@ public class ScalableTopicControllerTest {
// 3-partition source → 3 sealed legacy parents (ids 0..2, full range,
wrapping
// each -partition-K) + 3 active range-based children (ids 3..5, full
fan-in).
TopicName base = TopicName.get("persistent://tenant/ns/my-topic");
- ScalableTopicMetadata md =
ScalableTopicController.createMigratedMetadata(base, 3);
+ ScalableTopicMetadata md =
ScalableTopicController.createMigratedMetadata(base, 3, 4);
assertEquals(md.getEpoch(), 0L);
assertEquals(md.getNextSegmentId(), 6L);
@@ -869,7 +869,7 @@ public class ScalableTopicControllerTest {
// Non-partitioned source (partitions <= 0) → 1 sealed legacy parent
wrapping the
// base persistent:// topic + 1 active child covering the full range.
TopicName base = TopicName.get("persistent://tenant/ns/np-topic");
- ScalableTopicMetadata md =
ScalableTopicController.createMigratedMetadata(base, 0);
+ ScalableTopicMetadata md =
ScalableTopicController.createMigratedMetadata(base, 0, 4);
assertEquals(md.getNextSegmentId(), 2L);
assertEquals(md.getSegments().size(), 2);
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 afcab4c8b98..fae5afaa100 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
@@ -34,7 +34,7 @@ public class SegmentLayoutTest {
@Test
public void testInitialLayout() {
- ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(2, Map.of());
+ ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(2, 4, Map.of());
SegmentLayout layout = SegmentLayout.fromMetadata(metadata);
assertEquals(layout.getEpoch(), 0);
@@ -54,7 +54,7 @@ public class SegmentLayoutTest {
@Test
public void testSingleSegmentInitialLayout() {
- ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(1, Map.of());
+ ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(1, 4, Map.of());
SegmentLayout layout = SegmentLayout.fromMetadata(metadata);
assertEquals(layout.getActiveSegments().size(), 1);
@@ -64,7 +64,7 @@ public class SegmentLayoutTest {
@Test
public void testFourSegmentInitialLayout() {
- ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(4, Map.of());
+ ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(4, 4, Map.of());
SegmentLayout layout = SegmentLayout.fromMetadata(metadata);
assertEquals(layout.getActiveSegments().size(), 4);
@@ -81,7 +81,7 @@ public class SegmentLayoutTest {
@Test
public void testFindActiveSegment() {
- ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(2, Map.of());
+ ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(2, 4, Map.of());
SegmentLayout layout = SegmentLayout.fromMetadata(metadata);
SegmentInfo found = layout.findActiveSegment(0x1000);
@@ -93,7 +93,7 @@ public class SegmentLayoutTest {
@Test
public void testSplitSegment() {
- ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(2, Map.of());
+ ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(2, 4, Map.of());
SegmentLayout layout = SegmentLayout.fromMetadata(metadata);
SegmentLayout afterSplit = layout.splitSegment(0, 0L);
@@ -122,7 +122,7 @@ public class SegmentLayoutTest {
@Test
public void testSplitNonActiveSegment() {
- ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(2, Map.of());
+ ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(2, 4, Map.of());
SegmentLayout layout = SegmentLayout.fromMetadata(metadata);
SegmentLayout afterSplit = layout.splitSegment(0, 0L);
@@ -133,7 +133,7 @@ public class SegmentLayoutTest {
@Test
public void testMergeSegments() {
// Start with 2 segments, split seg-0, then merge the children back
- ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(2, Map.of());
+ ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(2, 4, Map.of());
SegmentLayout layout = SegmentLayout.fromMetadata(metadata);
SegmentLayout afterSplit = layout.splitSegment(0, 0L); // seg-2
[0000-3fff], seg-3 [4000-7fff]
@@ -155,7 +155,7 @@ public class SegmentLayoutTest {
@Test
public void testSplitRecordsWallClockTimestamps() {
- ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(1, Map.of());
+ ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(1, 4, Map.of());
SegmentLayout layout = SegmentLayout.fromMetadata(metadata);
long splitAt = 1_700_000_000_000L;
@@ -179,7 +179,7 @@ public class SegmentLayoutTest {
@Test
public void testMergeRecordsWallClockTimestamps() {
- ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(2, Map.of());
+ ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(2, 4, Map.of());
SegmentLayout layout = SegmentLayout.fromMetadata(metadata);
long splitAt = 1_700_000_000_000L;
@@ -199,7 +199,7 @@ public class SegmentLayoutTest {
@Test
public void testMergeNonAdjacentSegments() {
- ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(4, Map.of());
+ ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(4, 4, Map.of());
SegmentLayout layout = SegmentLayout.fromMetadata(metadata);
// Segments 0 and 2 are not adjacent
@@ -208,7 +208,7 @@ public class SegmentLayoutTest {
@Test
public void testPruneSegment() {
- ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(2, Map.of());
+ ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(2, 4, Map.of());
SegmentLayout layout = SegmentLayout.fromMetadata(metadata);
SegmentLayout afterSplit = layout.splitSegment(0, 0L);
@@ -223,7 +223,7 @@ public class SegmentLayoutTest {
@Test
public void testCannotPruneActiveSegment() {
- ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(2, Map.of());
+ ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(2, 4, Map.of());
SegmentLayout layout = SegmentLayout.fromMetadata(metadata);
assertThrows(IllegalArgumentException.class, () ->
layout.pruneSegment(0));
@@ -231,7 +231,7 @@ public class SegmentLayoutTest {
@Test
public void testGetChildren() {
- ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(1, Map.of());
+ ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(1, 4, Map.of());
SegmentLayout layout = SegmentLayout.fromMetadata(metadata);
SegmentLayout afterSplit = layout.splitSegment(0, 0L);
@@ -243,7 +243,7 @@ public class SegmentLayoutTest {
@Test
public void testGetParents() {
- ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(1, Map.of());
+ ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(1, 4, Map.of());
SegmentLayout layout = SegmentLayout.fromMetadata(metadata);
SegmentLayout afterSplit = layout.splitSegment(0, 0L);
@@ -257,7 +257,7 @@ public class SegmentLayoutTest {
@Test
public void testGetLineage() {
- ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(1, Map.of());
+ ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(1, 4, Map.of());
SegmentLayout layout = SegmentLayout.fromMetadata(metadata);
SegmentLayout afterSplit = layout.splitSegment(0, 0L);
@@ -268,7 +268,7 @@ public class SegmentLayoutTest {
@Test
public void testToMetadata() {
- ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(2, Map.of("key", "value"));
+ ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(2, 4, Map.of("key", "value"));
// A layout mutation must round-trip every non-layout field, not just
properties.
metadata.setAutoScalePolicy(org.apache.pulsar.common.policies.data.AutoScalePolicyOverride
.builder().enabled(false).maxSegments(8).build());
@@ -286,7 +286,7 @@ public class SegmentLayoutTest {
@Test
public void testNextSegmentIdAdvances() {
- ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(2, Map.of());
+ ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(2, 4, Map.of());
SegmentLayout layout = SegmentLayout.fromMetadata(metadata);
assertEquals(layout.getNextSegmentId(), 2);
@@ -300,7 +300,7 @@ public class SegmentLayoutTest {
@Test
public void testMergeDepthZeroForNeverMergedSegments() {
SegmentLayout layout = SegmentLayout.fromMetadata(
- ScalableTopicController.createInitialMetadata(2, Map.of()));
+ ScalableTopicController.createInitialMetadata(2, 4, Map.of()));
assertEquals(layout.mergeDepth(0), 0);
assertEquals(layout.mergeDepth(1), 0);
@@ -314,7 +314,7 @@ public class SegmentLayoutTest {
public void testMergeDepthCountsMergesInLineage() {
// split(0) → {1,2}; merge(1,2) → {3}; split(3) → {4,5}.
SegmentLayout layout = SegmentLayout
- .fromMetadata(ScalableTopicController.createInitialMetadata(1,
Map.of()))
+ .fromMetadata(ScalableTopicController.createInitialMetadata(1,
4, Map.of()))
.splitSegment(0, 0L)
.mergeSegments(1, 2, 0L)
.splitSegment(3, 0L);
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 3fa16caec49..57c797639c3 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
@@ -56,7 +56,7 @@ public class SubscriptionCoordinatorTest {
@BeforeMethod
public void setup() {
topicName = TopicName.get("topic://tenant/ns/my-topic");
- ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(4, Map.of());
+ ScalableTopicMetadata metadata =
ScalableTopicController.createInitialMetadata(4, 4, Map.of());
initialLayout = SegmentLayout.fromMetadata(metadata);
resources = mock(ScalableTopicResources.class);
// All persistence ops succeed
