You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2026/10/07 22:33:00 UTC
(pulsar) branch master updated: [fix][broker] Stop checkpoint consumer groups from triggering rebucket rollovers (#26838)
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 47987d29196d [fix][broker] Stop checkpoint consumer groups from triggering rebucket rollovers (#26838)
47987d29196d is described below
commit 47987d29196d7b20167b18c7ca9024da05b5f598
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Wed Oct 7 15:32:52 2026 -0700
[fix][broker] Stop checkpoint consumer groups from triggering rebucket rollovers (#26838)
---
.../apache/pulsar/broker/ServiceConfiguration.java | 4 +-
.../service/scalable/AutoScalePolicyEvaluator.java | 27 ++++--
.../service/scalable/ScalableTopicController.java | 30 ++++++-
.../service/scalable/SubscriptionCoordinator.java | 30 ++++++-
.../scalable/AutoScalePolicyEvaluatorTest.java | 84 +++++++++++++++++--
.../ScalableTopicControllerAutoScaleTest.java | 57 +++++++++++++
.../scalable/SubscriptionCoordinatorTest.java | 50 ++++++++++++
.../pulsar/client/api/v5/V5AutoRebucketTest.java | 35 +++++++-
.../api/v5/V5CheckpointConsumerGroupTest.java | 95 ++++++++++++++++++++++
9 files changed, 393 insertions(+), 19 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 fad7fcd421c6..ea988fbe9069 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
@@ -1512,7 +1512,9 @@ public class ServiceConfiguration implements PulsarConfiguration {
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)."
+ + "should not materialize physical segments just for consumer count). Checkpoint "
+ + "consumer groups drive splits only: they read whole segments, so more buckets "
+ + "would not serve them."
)
private double scalableTopicSplitVsRebucketMinMsgRateInThreshold = 1_000;
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 12bd9676ab21..80c4c3332f88 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
@@ -21,6 +21,7 @@ package org.apache.pulsar.broker.service.scalable;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import org.apache.pulsar.common.scalable.SegmentInfo;
import org.apache.pulsar.common.scalable.SegmentLoadStats;
@@ -57,6 +58,10 @@ public final class AutoScalePolicyEvaluator {
* as zero load with no age (never merge-eligible)
* @param streamConsumerCount per-subscription count of STREAM/CHECKPOINT (controller-managed)
* consumers; QUEUE subscriptions are excluded by the caller
+ * @param bucketSharingSubscriptions the subscriptions in {@code streamConsumerCount} whose
+ * consumers can share a segment by entry-bucket (STREAM). The
+ * others (CHECKPOINT groups, and subscriptions whose consumer type
+ * is not known yet) count toward a consumer-driven split only
* @param config the resolved policy
* @param nowMs current wall-clock time, epoch millis
* @param lastSplitAtMs epoch millis of the last split on this topic (manual or auto),
@@ -69,6 +74,7 @@ public final class AutoScalePolicyEvaluator {
SegmentLayout layout,
Map<Long, SegmentLoadSample> loadBySegment,
Map<String, Integer> streamConsumerCount,
+ Set<String> bucketSharingSubscriptions,
AutoScaleConfig config,
long nowMs,
long lastSplitAtMs,
@@ -82,7 +88,8 @@ public final class AutoScalePolicyEvaluator {
List<SegmentInfo> active = new ArrayList<>(layout.getActiveSegments().values());
AutoScaleDecision consumerScale = tryConsumerScale(active, loadBySegment,
- streamConsumerCount, config, nowMs, lastSplitAtMs, lastRebucketAtMs);
+ streamConsumerCount, bucketSharingSubscriptions, config, nowMs, lastSplitAtMs,
+ lastRebucketAtMs);
if (!(consumerScale instanceof AutoScaleDecision.NoAction)) {
return consumerScale;
}
@@ -110,13 +117,16 @@ public final class AutoScalePolicyEvaluator {
* roll the smallest-bucketed segment over to the smallest power of two that lets
* every consumer own a bucket, capped at {@code maxEntryBucketsPerSegment}.
* Raising is fast (one rollover sized to the surplus); lowering is deliberately
- * not automated here — spiky consumer counts must not flap the bucketing.</li>
+ * not automated here — spiky consumer counts must not flap the bucketing. Only
+ * bucket-sharing subscriptions count here: a checkpoint group's members each read
+ * whole segments, so more buckets add no parallelism for it, only smaller batches.</li>
* </ul>
*/
private static AutoScaleDecision tryConsumerScale(
List<SegmentInfo> active,
Map<Long, SegmentLoadSample> loadBySegment,
Map<String, Integer> streamConsumerCount,
+ Set<String> bucketSharingSubscriptions,
AutoScaleConfig config,
long nowMs,
long lastSplitAtMs,
@@ -145,12 +155,15 @@ public final class AutoScalePolicyEvaluator {
return new AutoScaleDecision.Split(busiest.segmentId(), "consumer-count");
}
- // Bucket lane: absorb the surplus with entry-buckets.
+ // Bucket lane: absorb the bucket-sharing subscriptions' surplus with entry-buckets.
+ int bucketConsumers = streamConsumerCount.entrySet().stream()
+ .filter(e -> bucketSharingSubscriptions.contains(e.getKey()))
+ .mapToInt(Map.Entry::getValue).max().orElse(0);
long capacity = 0;
for (SegmentInfo segment : active) {
capacity += segment.bucketCount();
}
- if (consumers <= capacity) {
+ if (bucketConsumers <= capacity) {
// The existing buckets already absorb the surplus (broker-side fan-out).
return AutoScaleDecision.NONE;
}
@@ -160,7 +173,7 @@ public final class AutoScalePolicyEvaluator {
// 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)),
+ int target = Math.min(nextPowerOfTwo(ceilDiv(bucketConsumers, segments)),
config.maxEntryBucketsPerSegment());
List<Long> below = new ArrayList<>();
for (SegmentInfo segment : active) {
@@ -253,7 +266,9 @@ public final class AutoScalePolicyEvaluator {
// and the rebucket lane cannot grow past the ceiling to recover it. Skip any pair
// whose merge would leave less capacity than the parallelism consumers get today:
// min(consumers, capacity), so an already over-subscribed topic can still take a
- // capacity-preserving merge, and an idle one can always consolidate.
+ // capacity-preserving merge, and an idle one can always consolidate. Every subscription
+ // counts, not only the bucket-sharing ones: the guard can only veto a merge, and one whose
+ // consumer type is not known yet (restored on failover) may turn out to share buckets.
int consumers = streamConsumerCount.values().stream()
.mapToInt(Integer::intValue).max().orElse(0);
long totalCapacity = 0;
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 3787b518c8a6..97019a3d0ad3 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
@@ -23,10 +23,12 @@ import io.github.merlimat.slog.Logger;
import java.time.Clock;
import java.time.Duration;
import java.util.ArrayList;
+import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
@@ -390,10 +392,11 @@ public class ScalableTopicController {
if (!config.enabled()) {
return CompletableFuture.<Void>completedFuture(null);
}
+ Set<String> bucketSharing = bucketSharingSubscriptions();
return collectConsumerCounts()
.thenCombine(collectLoadSamples(), (consumers, load) ->
AutoScalePolicyEvaluator.decide(currentLayout, load,
- consumers, config, clock.millis(),
+ consumers, bucketSharing, config, clock.millis(),
lastSplitAtMs, lastMergeAtMs, lastRebucketAtMs))
.thenCompose(decision -> dispatch(decision, config, trigger));
})
@@ -527,6 +530,22 @@ public class ScalableTopicController {
return CompletableFuture.completedFuture(counts);
}
+ /**
+ * The controller-managed subscriptions whose consumers can share a segment by entry-bucket
+ * (see {@link SubscriptionCoordinator#canShareSegmentsByEntryBucket}). Only these count toward
+ * a rebucket rollover; a checkpoint group, or a subscription restored on failover whose
+ * consumer type is not known yet, counts toward a split only.
+ */
+ private Set<String> bucketSharingSubscriptions() {
+ Set<String> names = new HashSet<>();
+ subscriptions.forEach((name, coordinator) -> {
+ if (coordinator.canShareSegmentsByEntryBucket()) {
+ names.add(name);
+ }
+ });
+ return names;
+ }
+
/** Read the load record (value + Stat modified time) for every active segment. */
private CompletableFuture<Map<Long, SegmentLoadSample>> collectLoadSamples() {
Map<Long, SegmentLoadSample> samples = new ConcurrentHashMap<>();
@@ -877,7 +896,8 @@ public class ScalableTopicController {
*
* <p>The {@code consumerType} is used at coordinator creation time to decide whether
* to enforce parent-drain ordering on assignments — see
- * {@link SubscriptionCoordinator}. The coordinator's setting is fixed at first
+ * {@link SubscriptionCoordinator} — and whether segments may be shared by entry-bucket
+ * (STREAM only). The coordinator's setting is fixed at first
* registration (a subscription's type doesn't change in practice); subsequent
* registers with a different type still work but won't change the ordering policy.
*/
@@ -893,9 +913,13 @@ public class ScalableTopicController {
// The coordinator may have been created on the failover-restore path (consumer
// type unknown then; we defaulted to "no parent-drain enforcement"). Now that we
// know the type, upgrade if it's STREAM. installDrainChecker is a no-op if the
- // coordinator already has a checker, so safe to call unconditionally.
+ // coordinator already has a checker, so safe to call unconditionally. A CHECKPOINT
+ // group must be marked before its first assignment is computed: it never shares a
+ // segment between members.
if (consumerType == ScalableConsumerType.STREAM) {
coordinator.installDrainChecker(this::isSegmentDrained);
+ } else if (consumerType == ScalableConsumerType.CHECKPOINT) {
+ coordinator.markCheckpointGroup();
}
return coordinator.registerConsumer(consumerName, consumerId, cnx)
.thenApply(assignments -> {
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 079296885ff0..437203e1f626 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
@@ -125,6 +125,14 @@ public class SubscriptionCoordinator {
* Once non-null it stays non-null (we don't downgrade to no-ordering mid-flight).
*/
private SegmentDrainChecker drainChecker;
+ /**
+ * Whether this coordinator serves a CHECKPOINT consumer group. Its members read each segment
+ * through their own Reader, with no shared subscription the broker could split entry-buckets
+ * on, so a segment is only ever assigned whole, to a single member. Unknown (false) on the
+ * controller-failover restore path until the first member registers, which happens before any
+ * assignment is sent.
+ */
+ private boolean checkpointGroup;
public SubscriptionCoordinator(String subscriptionName,
TopicName topicName,
@@ -343,6 +351,24 @@ public class SubscriptionCoordinator {
resetAndRearmDrainPoll();
}
+ /**
+ * Mark this coordinator as serving a CHECKPOINT consumer group (see {@link #checkpointGroup}).
+ * Called on every CHECKPOINT register, before the registration computes the assignment.
+ */
+ synchronized void markCheckpointGroup() {
+ checkpointGroup = true;
+ }
+
+ /**
+ * Whether this subscription's consumers can share a segment by entry-bucket: it is a STREAM
+ * subscription (it has a drain checker, installed at creation or on its first STREAM register)
+ * and not a checkpoint group. False on the controller-failover restore path until the first
+ * member registers, since the consumer type is not persisted and it may be a checkpoint group.
+ */
+ synchronized boolean canShareSegmentsByEntryBucket() {
+ return drainChecker != null && !checkpointGroup;
+ }
+
/**
* Stop the periodic drain-status poller. Called by the controller on close. Idempotent.
* Also flips a {@code closed} flag so any {@link #pollDrainStatus()} iteration that's
@@ -623,9 +649,11 @@ public class SubscriptionCoordinator {
// segments out by entry-bucket: each owner of a shared segment takes a contiguous slice of its
// buckets and subscribes Key_Shared STICKY declaring exactly those ranges. A segment absorbs at
// most bucketCount() consumers; consumers beyond the topic's total bucket capacity stay idle.
+ // A CHECKPOINT group never fans out: each member reads a segment through its own Reader, so a
+ // shared segment would be read in full by every sharer. Its surplus members stay idle.
int segmentCount = sortedSegments.size();
int consumerCount = sortedConsumers.size();
- if (consumerCount <= segmentCount) {
+ if (consumerCount <= segmentCount || checkpointGroup) {
int consumerIndex = 0;
for (SegmentInfo segment : sortedSegments) {
TopicName segmentTopic = SegmentTopicName.fromParent(topicName, segment.hashRange(),
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 e9460968f8a5..4ca53f290afd 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
@@ -24,6 +24,7 @@ import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import org.apache.pulsar.common.scalable.SegmentInfo;
import org.apache.pulsar.common.scalable.SegmentLoadStats;
import org.testng.annotations.Test;
@@ -89,12 +90,21 @@ public class AutoScalePolicyEvaluatorTest {
return Duration.ofMinutes(10).toMillis();
}
+ /** Every subscription in {@code consumers} is a stream one, able to share segments by bucket. */
private static AutoScaleDecision decide(SegmentLayout layout,
Map<Long, SegmentLoadSample> load,
Map<String, Integer> consumers,
AutoScaleConfig config) {
- return AutoScalePolicyEvaluator.decide(layout, load, consumers, config, NOW,
- NO_PRIOR, NO_PRIOR, NO_PRIOR);
+ return decide(layout, load, consumers, consumers.keySet(), config);
+ }
+
+ private static AutoScaleDecision decide(SegmentLayout layout,
+ Map<Long, SegmentLoadSample> load,
+ Map<String, Integer> consumers,
+ Set<String> bucketSharingSubscriptions,
+ AutoScaleConfig config) {
+ return AutoScalePolicyEvaluator.decide(layout, load, consumers, bucketSharingSubscriptions,
+ config, NOW, NO_PRIOR, NO_PRIOR, NO_PRIOR);
}
// --- enable switch ---
@@ -162,7 +172,7 @@ public class AutoScalePolicyEvaluatorTest {
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, NO_PRIOR);
+ Set.of("sub"), baseConfig().build(), NOW, recentSplit, NO_PRIOR, NO_PRIOR);
assertTrue(d instanceof AutoScaleDecision.NoAction, "within split cooldown, no split");
}
@@ -198,7 +208,7 @@ public class AutoScalePolicyEvaluatorTest {
Map<Long, SegmentLoadSample> 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);
+ Set.of("sub"), baseConfig().build(), NOW, NO_PRIOR, NO_PRIOR, recentRebucket);
assertTrue(d instanceof AutoScaleDecision.NoAction, "within rebucket cooldown");
}
@@ -258,6 +268,53 @@ public class AutoScalePolicyEvaluatorTest {
assertEquals(r.newBucketCount(), 4);
}
+ // --- consumer-driven scale-up for checkpoint groups (whole segments only) ---
+
+ @Test
+ public void testCheckpointGroupSurplusNeverRebuckets() {
+ // A checkpoint group's members each read whole segments, so more buckets would add no
+ // parallelism for it. Its surplus must not roll a segment over below the split floor…
+ SegmentLayout layout = initialLayout(1); // 1 segment × 4 buckets
+ Map<Long, SegmentLoadSample> load = Map.of(0L, cold(0));
+ AutoScaleDecision d = decide(layout, load, Map.of("group", 5), Set.of(),
+ baseConfig().build());
+ assertTrue(d instanceof AutoScaleDecision.NoAction, d.toString());
+
+ // …nor at the segment cap.
+ SegmentLayout atCap = initialLayout(2); // 2 segments × 2 buckets
+ Map<Long, SegmentLoadSample> atCapLoad = Map.of(0L, cold(0), 1L, cold(0));
+ AutoScaleDecision none = decide(atCap, atCapLoad, Map.of("group", 5), Set.of(),
+ baseConfig().maxSegments(2).build());
+ assertTrue(none instanceof AutoScaleDecision.NoAction, none.toString());
+ }
+
+ @Test
+ public void testCheckpointGroupSurplusStillSplits() {
+ // Above the split floor a new segment is one more member the group can keep busy.
+ SegmentLayout layout = initialLayout(2);
+ Map<Long, SegmentLoadSample> load = Map.of(
+ 0L, sample(1_500, 0, 0, 0, old()),
+ 1L, sample(2_000, 0, 0, 0, old()));
+ AutoScaleDecision d = decide(layout, load, Map.of("group", 3), Set.of(),
+ baseConfig().build());
+ assertTrue(d instanceof AutoScaleDecision.Split, d.toString());
+ AutoScaleDecision.Split s = (AutoScaleDecision.Split) d;
+ assertEquals(s.segmentId(), 1L);
+ assertEquals(s.reason(), "consumer-count");
+ }
+
+ @Test
+ public void testRebucketTargetIgnoresCheckpointGroups() {
+ // A stream subscription of 5 beside a checkpoint group of 9 on one cold N=4 segment: the
+ // rollover is sized to the stream consumers (5 → 8), not to the group (9 → 16).
+ SegmentLayout layout = initialLayout(1);
+ Map<Long, SegmentLoadSample> load = Map.of(0L, cold(0));
+ AutoScaleDecision d = decide(layout, load, Map.of("group", 9, "sub", 5), Set.of("sub"),
+ baseConfig().build());
+ assertTrue(d instanceof AutoScaleDecision.Rebucket, d.toString());
+ assertEquals(((AutoScaleDecision.Rebucket) d).newBucketCount(), 8);
+ }
+
// --- load-driven split ---
@Test
@@ -355,11 +412,24 @@ public class AutoScalePolicyEvaluatorTest {
Map<Long, SegmentLoadSample> load = Map.of(0L, cold(old()), 1L, cold(old()));
long recentRebucket = NOW - Duration.ofSeconds(30).toMillis(); // < 1m rebucketCooldown
AutoScaleDecision d = AutoScalePolicyEvaluator.decide(layout, load, Map.of("sub", 20),
- baseConfig().maxEntryBucketsPerSegment(8).build(), NOW, NO_PRIOR, NO_PRIOR,
- recentRebucket);
+ Set.of("sub"), baseConfig().maxEntryBucketsPerSegment(8).build(), NOW, NO_PRIOR,
+ NO_PRIOR, recentRebucket);
assertTrue(d instanceof AutoScaleDecision.Merge, d.toString());
}
+ @Test
+ public void testMergeGuardAlsoCountsSubscriptionsThatCannotShareBuckets() {
+ // The at-ceiling pair of testMergeSkippedWhenClampWouldCutLiveParallelism, with a
+ // subscription outside the bucket-sharing set: it still vetoes the merge, because a
+ // subscription restored on failover may turn out to be a stream one once its type is known.
+ SegmentLayout layout = SegmentLayout.fromMetadata(
+ ScalableTopicController.createInitialMetadata(2, 16, Map.of()));
+ Map<Long, SegmentLoadSample> load = Map.of(0L, cold(old()), 1L, cold(old()));
+ AutoScaleDecision d = decide(layout, load, Map.of("sub", 12), Set.of(),
+ baseConfig().maxEntryBucketsPerSegment(8).build());
+ assertTrue(d instanceof AutoScaleDecision.NoAction, d.toString());
+ }
+
@Test
public void testMergeRespectsMinSegments() {
SegmentLayout layout = initialLayout(2);
@@ -374,7 +444,7 @@ public class AutoScalePolicyEvaluatorTest {
Map<Long, SegmentLoadSample> 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, NO_PRIOR);
+ Set.of(), baseConfig().build(), NOW, NO_PRIOR, recentMerge, NO_PRIOR);
assertTrue(d instanceof AutoScaleDecision.NoAction, "within merge cooldown, no merge");
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerAutoScaleTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerAutoScaleTest.java
index 05d8a54e588a..935974328e23 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
@@ -508,6 +508,63 @@ public class ScalableTopicControllerAutoScaleTest {
"a hot-segment split must fire — the rollover is not a split");
}
+ /**
+ * A checkpoint group never shares a segment by entry-bucket (each member reads whole
+ * segments through its own Readers), so on a cold topic its surplus members must not trigger
+ * a rebucket rollover: more buckets add no parallelism for the group, they only shrink
+ * producer batches. A stream subscription on the same topic still gets a rollover, sized to
+ * its own consumers rather than to the larger group.
+ */
+ @Test
+ public void testCheckpointGroupSurplusDoesNotRebucket() throws Exception {
+ startController(1); // one segment, N=4
+
+ // Nine members against 4 buckets; every registration runs the event-driven evaluation.
+ for (int i = 1; i <= 9; i++) {
+ controller.registerConsumer("group", "m" + i, i, ScalableConsumerType.CHECKPOINT,
+ mock(TransportCnx.class)).get();
+ }
+ Awaitility.await().during(Duration.ofSeconds(2)).atMost(Duration.ofSeconds(5))
+ .untilAsserted(() -> assertEquals(soleActiveBucketCount(), 4,
+ "a checkpoint group's surplus must not roll the segment over"));
+
+ for (int i = 1; i <= 5; i++) {
+ controller.registerConsumer("sub", "c" + i, 100 + i, ScalableConsumerType.STREAM,
+ mock(TransportCnx.class)).get();
+ }
+ Awaitility.await().atMost(Duration.ofSeconds(10)).untilAsserted(
+ () -> assertEquals(soleActiveBucketCount(), 8,
+ "five stream consumers roll the segment to 8, ignoring the 9-member group"));
+ }
+
+ /**
+ * The consumer type is not persisted, so after a leader failover a restored subscription's
+ * type is unknown until one of its members registers again. It may be a checkpoint group, so
+ * it must not trigger a rebucket rollover until a stream member registers.
+ */
+ @Test
+ public void testRestoredSubscriptionRebucketsOnlyOnceKnownToBeStream() throws Exception {
+ resources.createScalableTopicAsync(topicName,
+ ScalableTopicController.createInitialMetadata(1, 4, Map.of())).get(); // N=4
+ // Registrations persisted under a previous leader.
+ for (int i = 1; i <= 5; i++) {
+ resources.registerConsumerAsync(topicName, "sub", "c" + i).get();
+ }
+ controller = new ScalableTopicController(topicName, resources, brokerService,
+ coordinationService);
+ controller.initialize().get();
+
+ controller.evaluateAutoScaleForTest().get();
+ assertEquals(soleActiveBucketCount(), 4,
+ "a restored subscription of unknown type must not roll the segment over");
+
+ controller.registerConsumer("sub", "c1", 1L, ScalableConsumerType.STREAM,
+ mock(TransportCnx.class)).get();
+ Awaitility.await().atMost(Duration.ofSeconds(10)).untilAsserted(
+ () -> assertEquals(soleActiveBucketCount(), 8,
+ "once a stream member registers, its surplus is served by a rollover"));
+ }
+
@Test
public void testSplitCooldownSurvivesLeaderFailover() throws Exception {
config.setScalableTopicSplitCooldownSeconds(3600);
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 1a2790713912..be2443a0ea86 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
@@ -511,6 +511,56 @@ public class SubscriptionCoordinatorTest {
assertEquals(sharers, 4);
}
+ @Test
+ public void testCheckpointGroupNeverSharesSegment() throws Exception {
+ // One segment with N=4 entry-buckets and three members of a CHECKPOINT group. A member reads
+ // the segment through its own Reader, so instead of fanning out by entry-bucket the segment
+ // goes whole to one member and the surplus members stay idle.
+ SubscriptionCoordinator c = bucketedCoordinator();
+ c.markCheckpointGroup();
+ c.registerConsumer("consumer-1", 1L, mock(TransportCnx.class)).get();
+ c.registerConsumer("consumer-2", 2L, mock(TransportCnx.class)).get();
+ Map<ConsumerSession, ConsumerAssignment> result =
+ c.registerConsumer("consumer-3", 3L, mock(TransportCnx.class)).get();
+
+ assertEquals(result.size(), 3);
+ ConsumerAssignment owner = findByName(result, "consumer-1");
+ assertEquals(owner.assignedSegments().size(), 1);
+ assertTrue(owner.assignedSegments().get(0).bucketRanges().isEmpty());
+ assertTrue(findByName(result, "consumer-2").assignedSegments().isEmpty());
+ assertTrue(findByName(result, "consumer-3").assignedSegments().isEmpty());
+ }
+
+ @Test
+ public void testRestoredCheckpointGroupNeverSharesSegmentOnReconnect() throws Exception {
+ // Controller failover: the restored coordinator doesn't know its consumer type. The
+ // controller marks the CHECKPOINT group before the first member's reconnect computes the
+ // assignment it sends.
+ SubscriptionCoordinator c = bucketedCoordinator();
+ c.restoreConsumers(List.of("consumer-1", "consumer-2", "consumer-3"));
+ c.markCheckpointGroup();
+ Map<ConsumerSession, ConsumerAssignment> result =
+ c.registerConsumer("consumer-2", 2L, mock(TransportCnx.class)).get();
+
+ assertEquals(findByName(result, "consumer-1").assignedSegments().size(), 1);
+ assertTrue(findByName(result, "consumer-1").assignedSegments().get(0).bucketRanges().isEmpty());
+ assertTrue(findByName(result, "consumer-2").assignedSegments().isEmpty());
+ assertTrue(findByName(result, "consumer-3").assignedSegments().isEmpty());
+ }
+
+ @Test
+ public void testOnlyAStreamSubscriptionCanShareSegmentsByEntryBucket() {
+ // No drain checker: the consumer type is unknown (e.g. restored on failover), and the
+ // subscription may be a checkpoint group.
+ assertFalse(coordinator.canShareSegmentsByEntryBucket());
+ // A STREAM register installs the drain checker.
+ coordinator.installDrainChecker((segment, sub) -> CompletableFuture.completedFuture(true));
+ assertTrue(coordinator.canShareSegmentsByEntryBucket());
+ // Once marked as a checkpoint group, the subscription gets whole segments only.
+ coordinator.markCheckpointGroup();
+ assertFalse(coordinator.canShareSegmentsByEntryBucket());
+ }
+
private SubscriptionCoordinator bucketedCoordinator() {
// One segment carrying the whole default budget: N = 4 entry-buckets.
return new SubscriptionCoordinator("test-sub", topicName,
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 f76b825cebc8..2e5b2cb97c84 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
@@ -19,6 +19,7 @@
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 java.nio.charset.StandardCharsets;
import java.time.Duration;
@@ -41,7 +42,8 @@ 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.
+ * segments for consumer count alone. Surplus checkpoint group members must not trigger one: a
+ * checkpoint group never shares a segment by entry-bucket.
*/
public class V5AutoRebucketTest extends V5ClientBaseTest {
@@ -162,6 +164,37 @@ public class V5AutoRebucketTest extends V5ClientBaseTest {
"the key set covers every bucket, so every consumer must receive traffic");
}
+ @Test
+ public void testCheckpointGroupSurplusOnColdTopicDoesNotRebucket() throws Exception {
+ String topic = newScalableTopic(1);
+ String group = "auto-rebucket-checkpoint";
+
+ // The same surplus as above (five members, one cold segment with N=4), but in a checkpoint
+ // group: each member reads whole segments through its own Readers, so more buckets would
+ // add no parallelism, only smaller producer batches. The segment must stay as it is.
+ for (int i = 0; i < 5; i++) {
+ track(v5Client.newCheckpointConsumer(Schema.string())
+ .topic(topic)
+ .consumerGroup(group)
+ .startPosition(Checkpoint.earliest())
+ .create());
+ }
+ Awaitility.await().untilAsserted(() -> {
+ var sub = admin.scalableTopics().getStats(topic).getSubscriptions().get(group);
+ assertNotNull(sub);
+ assertEquals(sub.getConsumers().size(), 5);
+ });
+
+ Awaitility.await().during(Duration.ofSeconds(5)).atMost(Duration.ofSeconds(15))
+ .untilAsserted(() -> {
+ ScalableTopicMetadata md = admin.scalableTopics().getMetadata(topic);
+ assertEquals(md.getSegments().size(), 1,
+ "a checkpoint group's surplus must not trigger a rebucket rollover");
+ assertEquals(md.getSegments().values().iterator().next()
+ .getEntryBucketSplits().size() + 1, 4);
+ });
+ }
+
/**
* 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.
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5CheckpointConsumerGroupTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5CheckpointConsumerGroupTest.java
index b6e8fc10bd82..77bea00cc19e 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5CheckpointConsumerGroupTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5CheckpointConsumerGroupTest.java
@@ -23,11 +23,16 @@ import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import java.time.Duration;
+import java.util.ArrayList;
import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
import java.util.Set;
+import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
import lombok.Cleanup;
import org.apache.pulsar.client.api.v5.schema.Schema;
+import org.apache.pulsar.common.policies.data.AutoScalePolicyOverride;
import org.awaitility.Awaitility;
import org.testng.annotations.Test;
@@ -101,6 +106,77 @@ public class V5CheckpointConsumerGroupTest extends V5ClientBaseTest {
+ " (a=" + aGot.size() + " b=" + bGot.size() + ")");
}
+ @Test
+ public void testMembersOutnumberingSegmentsDeliverEachMessageOnce() throws Exception {
+ // One segment with the default entry-bucket budget (N = 4) and three members: the members
+ // outnumber the segments. A checkpoint consumer reads through per-segment Readers, so the
+ // segment must stay with exactly one member while the surplus members stay idle.
+ String topic = newScalableTopic(1);
+ // Pin the layout: the consumer-count auto-scale rule could otherwise split the segment and
+ // hide the surplus case under test.
+ admin.scalableTopics().setAutoScalePolicy(topic,
+ AutoScalePolicyOverride.builder().enabled(false).build());
+ String group = "group-surplus";
+
+ @Cleanup
+ Producer<String> producer = v5Client.newProducer(Schema.string())
+ .topic(topic)
+ .create();
+
+ // Fixed names make the assignment stable as members join: the controller orders members
+ // by name, so the first member keeps the segment and no rebalance moves it mid-test.
+ int memberCount = 3;
+ List<CheckpointConsumer<String>> members = new ArrayList<>();
+ for (int i = 0; i < memberCount; i++) {
+ members.add(track(v5Client.newCheckpointConsumer(Schema.string())
+ .topic(topic)
+ .consumerGroup(group)
+ .consumerName("member-" + i)
+ .startPosition(Checkpoint.earliest())
+ .create()));
+ }
+ Awaitility.await().untilAsserted(() -> {
+ var sub = admin.scalableTopics().getStats(topic).getSubscriptions().get(group);
+ assertNotNull(sub);
+ assertEquals(sub.getConsumers().size(), memberCount);
+ });
+
+ int n = 100;
+ Set<String> sent = new HashSet<>();
+ for (int i = 0; i < n; i++) {
+ String v = "v-" + i;
+ producer.newMessage().key("k-" + (i % 16)).value(v).send();
+ sent.add(v);
+ }
+
+ Map<String, Integer> deliveries = new ConcurrentHashMap<>();
+ List<Set<String>> perMember = new ArrayList<>();
+ List<Thread> drainers = new ArrayList<>();
+ for (CheckpointConsumer<String> member : members) {
+ Set<String> mine = ConcurrentHashMap.newKeySet();
+ perMember.add(mine);
+ drainers.add(drainTo(member, deliveries, mine));
+ }
+ for (Thread t : drainers) {
+ t.join();
+ }
+
+ assertEquals(deliveries.keySet(), sent, "every message must be delivered across the group");
+ Map<String, Integer> duplicates = new TreeMap<>();
+ deliveries.forEach((v, count) -> {
+ if (count > 1) {
+ duplicates.put(v, count);
+ }
+ });
+ assertTrue(duplicates.isEmpty(), "each message must be delivered exactly once across the group;"
+ + " per-member counts=" + perMember.stream().map(Set::size).toList()
+ + " duplicated=" + duplicates);
+
+ var sub = admin.scalableTopics().getStats(topic).getSubscriptions().get(group);
+ long owners = sub.getConsumers().stream().filter(c -> !c.getSegmentIds().isEmpty()).count();
+ assertEquals(owners, 1, "the single segment must be assigned to exactly one member");
+ }
+
@Test
public void testGroupScalesUp() throws Exception {
String topic = newScalableTopic(2);
@@ -268,6 +344,25 @@ public class V5CheckpointConsumerGroupTest extends V5ClientBaseTest {
return received;
}
+ private Thread drainTo(CheckpointConsumer<String> consumer, Map<String, Integer> deliveries,
+ Set<String> mine) {
+ Thread t = new Thread(() -> {
+ try {
+ while (true) {
+ Message<String> msg = consumer.receive(Duration.ofSeconds(1));
+ if (msg == null) {
+ return;
+ }
+ deliveries.merge(msg.value(), 1, Integer::sum);
+ mine.add(msg.value());
+ }
+ } catch (Exception ignored) {
+ }
+ }, "checkpoint-consumer-drainer");
+ t.start();
+ return t;
+ }
+
private Thread drainTo(CheckpointConsumer<String> consumer, Set<String> all, Set<String> mine) {
Thread t = new Thread(() -> {
try {