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 8dd3c313a01 [improve][broker] PIP-486 (PR3): entry-bucket foundation
(single-active by default) (#26129)
8dd3c313a01 is described below
commit 8dd3c313a01d869cb8f4d8507e41a7f2e8cba2dc
Author: Matteo Merli <[email protected]>
AuthorDate: Wed Jul 1 11:21:24 2026 -0700
[improve][broker] PIP-486 (PR3): entry-bucket foundation (single-active by
default) (#26129)
---
.../service/scalable/ConsumerAssignment.java | 14 +++-
.../broker/service/scalable/ConsumerSession.java | 6 +-
.../broker/service/scalable/EntryBucketSplits.java | 19 +++++
.../service/scalable/SubscriptionCoordinator.java | 13 +--
.../service/scalable/ConsumerSessionTest.java | 23 ++++-
.../service/scalable/EntryBucketSplitsTest.java | 17 ++++
.../scalable/SubscriptionCoordinatorTest.java | 37 +++++++++
.../client/api/v5/V5EntryBucketDispatchTest.java | 97 ++++++++++++++++++++++
.../pulsar/client/impl/v5/ClientSegmentLayout.java | 4 +-
.../client/impl/v5/ScalableConsumerClient.java | 11 ++-
.../client/impl/v5/ScalableStreamConsumer.java | 36 +++++++-
.../client/impl/v5/ScalableTopicProducer.java | 10 ++-
.../pulsar/client/impl/v5/SegmentRouter.java | 10 ++-
.../client/impl/v5/ScalableTopicProducerTest.java | 18 +++-
.../pulsar/client/impl/v5/SegmentRouterTest.java | 4 +-
pulsar-common/src/main/proto/PulsarApi.proto | 5 ++
16 files changed, 304 insertions(+), 20 deletions(-)
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ConsumerAssignment.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ConsumerAssignment.java
index 7615f64db98..56e29856060 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ConsumerAssignment.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ConsumerAssignment.java
@@ -37,10 +37,20 @@ public record ConsumerAssignment(
/**
* A single segment assignment for a consumer.
+ *
+ * @param bucketRanges PIP-486 entry-bucket hash ranges this consumer owns
within the segment.
+ * Empty means the consumer owns the whole segment
(single bucket) and
+ * subscribes {@code Shared}; non-empty means the
segment is shared by bucket
+ * and the consumer subscribes {@code Key_Shared}
STICKY with exactly these ranges.
*/
public record AssignedSegment(
long segmentId,
HashRange hashRange,
- String underlyingTopicName
- ) {}
+ String underlyingTopicName,
+ List<HashRange> bucketRanges
+ ) {
+ public AssignedSegment {
+ bucketRanges = List.copyOf(bucketRanges);
+ }
+ }
}
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ConsumerSession.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ConsumerSession.java
index b5e1f5c00fc..99061daa73d 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ConsumerSession.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ConsumerSession.java
@@ -26,6 +26,7 @@ import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import lombok.Getter;
import org.apache.pulsar.broker.service.TransportCnx;
+import org.apache.pulsar.common.scalable.HashRange;
/**
* In-memory handle for a consumer registered with the controller leader.
@@ -196,11 +197,14 @@ public class ConsumerSession {
var proto = new
org.apache.pulsar.common.api.proto.ScalableConsumerAssignment()
.setLayoutEpoch(assignment.layoutEpoch());
for (ConsumerAssignment.AssignedSegment seg :
assignment.assignedSegments()) {
- proto.addSegment()
+ var segProto = proto.addSegment()
.setSegmentId(seg.segmentId())
.setHashStart(seg.hashRange().start())
.setHashEnd(seg.hashRange().end())
.setSegmentTopic(seg.underlyingTopicName());
+ for (HashRange bucket : seg.bucketRanges()) {
+
segProto.addBucketRange().setStart(bucket.start()).setEnd(bucket.end());
+ }
}
return proto;
}
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 1cfd013982e..57a6461119e 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
@@ -49,6 +49,25 @@ final class EntryBucketSplits {
return Math.max(1, budget / segmentCount);
}
+ /**
+ * The per-bucket hash ranges (inclusive, over the 16-bit entry-bucket
ring) defined by
+ * {@code splits}. Empty splits yield a single range spanning the whole
ring. The i-th range is
+ * the i-th entry-bucket, so the result has {@code splits.size() + 1}
elements.
+ */
+ static List<HashRange> ranges(List<Integer> splits) {
+ if (splits.isEmpty()) {
+ return List.of(HashRange.of(0, HashRange.MAX_HASH));
+ }
+ List<HashRange> ranges = new ArrayList<>(splits.size() + 1);
+ int start = 0;
+ for (int split : splits) {
+ ranges.add(HashRange.of(start, split - 1));
+ start = split;
+ }
+ ranges.add(HashRange.of(start, HashRange.MAX_HASH));
+ return ranges;
+ }
+
/** Equal-width split points for {@code bucketCount} buckets; empty when
{@code bucketCount <= 1}. */
static List<Integer> equalWidth(int bucketCount) {
if (bucketCount <= 1) {
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 ade25fb1779..a3757c16c6c 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
@@ -579,14 +579,17 @@ public class SubscriptionCoordinator {
int consumerIndex = 0;
for (SegmentInfo segment : sortedSegments) {
- ConsumerSession consumer = sortedConsumers.get(consumerIndex %
sortedConsumers.size());
TopicName segmentTopic = SegmentTopicName.fromParent(topicName,
segment.hashRange(),
segment.segmentId());
+ // PIP-486: assign each whole segment to a single consumer for
efficient single-active
+ // (Exclusive) dispatch — no per-bucket pending tracking. A
segment's entry-buckets let it be
+ // *shared* across multiple consumers, but fanning a segment out
into Key_Shared bucket
+ // ownership is a controller-driven scale-up action handled
separately; by default one
+ // consumer owns the whole segment. Empty bucketRanges signals the
client to subscribe
+ // Exclusive.
+ ConsumerSession consumer = sortedConsumers.get(consumerIndex %
sortedConsumers.size());
assignmentLists.get(consumer).add(new
ConsumerAssignment.AssignedSegment(
- segment.segmentId(),
- segment.hashRange(),
- segmentTopic.toString()
- ));
+ segment.segmentId(), segment.hashRange(),
segmentTopic.toString(), List.of()));
consumerIndex++;
}
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ConsumerSessionTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ConsumerSessionTest.java
index 15e2b10ba30..f2fee9e4bf9 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ConsumerSessionTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ConsumerSessionTest.java
@@ -80,7 +80,7 @@ public class ConsumerSessionTest {
int end = start + 0x3FFF;
segments.add(new ConsumerAssignment.AssignedSegment(
id, HashRange.of(start, end),
- "persistent://tenant/ns/my-scalable-seg-" + id));
+ "persistent://tenant/ns/my-scalable-seg-" + id,
List.of()));
}
return new ConsumerAssignment(epoch, segments);
}
@@ -285,6 +285,27 @@ public class ConsumerSessionTest {
assertEquals(proto.getSegmentAt(1).getHashStart(), 0x4000);
assertEquals(proto.getSegmentAt(2).getSegmentId(), 5L);
assertEquals(proto.getSegmentAt(2).getHashStart(), 0x8000);
+ // Whole-segment (single-active) assignments carry no entry-bucket
ranges.
+ assertEquals(proto.getSegmentAt(0).getBucketRangesCount(), 0);
+ }
+
+ @Test
+ public void testToProtoCarriesEntryBucketRanges() {
+ // PIP-486: a consumer that owns a subset of a segment's entry-buckets
declares those bucket
+ // hash-ranges; toProto must serialize each as an IntRange
(start/end), in order.
+ ConsumerAssignment assignment = new ConsumerAssignment(7L, List.of(
+ new ConsumerAssignment.AssignedSegment(2L,
HashRange.of(0x0000, 0xFFFF),
+ "persistent://tenant/ns/my-scalable-seg-2",
+ List.of(HashRange.of(0x0000, 0x7FFF),
HashRange.of(0x8000, 0xFFFF)))));
+
+ ScalableConsumerAssignment proto = ConsumerSession.toProto(assignment);
+
+ assertEquals(proto.getSegmentsCount(), 1);
+ assertEquals(proto.getSegmentAt(0).getBucketRangesCount(), 2);
+ assertEquals(proto.getSegmentAt(0).getBucketRangeAt(0).getStart(),
0x0000);
+ assertEquals(proto.getSegmentAt(0).getBucketRangeAt(0).getEnd(),
0x7FFF);
+ assertEquals(proto.getSegmentAt(0).getBucketRangeAt(1).getStart(),
0x8000);
+ assertEquals(proto.getSegmentAt(0).getBucketRangeAt(1).getEnd(),
0xFFFF);
}
@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 040d4495f99..e1e1c827aa3 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
@@ -23,6 +23,7 @@ import java.util.List;
import java.util.Map;
import org.apache.pulsar.broker.resources.ScalableTopicMetadata;
import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.scalable.HashRange;
import org.testng.annotations.Test;
/**
@@ -125,4 +126,20 @@ public class EntryBucketSplitsTest {
long mergedId = merged.getAllSegments().get(0L).childIds().get(0);
assertEquals(merged.getAllSegments().get(mergedId).bucketCount(), 2);
}
+
+ // --- ranges: split points -> per-bucket hash ranges ---
+
+ @Test
+ public void testRangesSingleBucketSpansWholeRing() {
+ assertEquals(EntryBucketSplits.ranges(List.of()),
List.of(HashRange.of(0, 0xFFFF)));
+ }
+
+ @Test
+ public void testRangesFromSplits() {
+ assertEquals(EntryBucketSplits.ranges(List.of(0x8000)),
+ List.of(HashRange.of(0, 0x7FFF), HashRange.of(0x8000,
0xFFFF)));
+ assertEquals(EntryBucketSplits.ranges(List.of(0x4000, 0x8000, 0xC000)),
+ List.of(HashRange.of(0, 0x3FFF), HashRange.of(0x4000, 0x7FFF),
+ HashRange.of(0x8000, 0xBFFF), HashRange.of(0xC000,
0xFFFF)));
+ }
}
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 57c797639c3..90b20ac9667 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
@@ -430,6 +430,43 @@ public class SubscriptionCoordinatorTest {
assertTrue(reconnected.isConnected());
}
+ // --- PIP-486 entry-bucket assignment ---
+
+ @Test
+ public void testSingleBucketSegmentsHaveNoBucketRanges() throws Exception {
+ // The default 4-segment layout is N=1 per segment: each whole segment
is assigned to one
+ // consumer with empty bucketRanges (single-active / Exclusive).
+ Map<ConsumerSession, ConsumerAssignment> result =
+ coordinator.registerConsumer("consumer-1", 1L,
mock(TransportCnx.class)).get();
+ for (ConsumerAssignment.AssignedSegment seg : findByName(result,
"consumer-1").assignedSegments()) {
+ assertTrue(seg.bucketRanges().isEmpty());
+ }
+ }
+
+ @Test
+ public void testBucketedSegmentIsAssignedWholeToOneConsumer() throws
Exception {
+ // One segment with N=4 entry-buckets (budget 4 / 1 segment). Even
with several consumers, the
+ // controller assigns the whole segment to a single consumer with
empty bucketRanges (efficient
+ // single-active / Exclusive dispatch); fanning it out into per-bucket
Key_Shared ownership is a
+ // separate controller-driven scale-up action.
+ SubscriptionCoordinator c = new SubscriptionCoordinator("test-sub",
topicName,
+
SegmentLayout.fromMetadata(ScalableTopicController.createInitialMetadata(1, 4,
Map.of())),
+ resources, scheduler, Duration.ofMillis(200));
+ c.registerConsumer("consumer-1", 1L, mock(TransportCnx.class)).get();
+ Map<ConsumerSession, ConsumerAssignment> result =
+ c.registerConsumer("consumer-2", 2L,
mock(TransportCnx.class)).get();
+
+ int owners = 0;
+ for (ConsumerAssignment assignment : result.values()) {
+ for (ConsumerAssignment.AssignedSegment seg :
assignment.assignedSegments()) {
+ assertEquals(seg.segmentId(), 0);
+ assertTrue(seg.bucketRanges().isEmpty());
+ owners++;
+ }
+ }
+ assertEquals(owners, 1);
+ }
+
// --- Helpers ---
private static ConsumerAssignment findByName(Map<ConsumerSession,
ConsumerAssignment> m, String name) {
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5EntryBucketDispatchTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5EntryBucketDispatchTest.java
new file mode 100644
index 00000000000..8de7e821067
--- /dev/null
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5EntryBucketDispatchTest.java
@@ -0,0 +1,97 @@
+/*
+ * 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 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.api.v5.config.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.v5.schema.Schema;
+import org.testng.annotations.Test;
+
+/**
+ * PIP-486 end-to-end: a single stream consumer on an entry-bucketed segment.
+ *
+ * <p>A one-segment scalable topic with the default entry-bucket budget (4)
gives that segment
+ * {@code N = 4} entry-buckets, so the producer batches per-bucket and stamps
each entry's
+ * {@code entry_hash} range. A lone stream consumer owns the whole segment and
subscribes
+ * {@code Exclusive} (single-active dispatch — the controller only fans a
segment out into per-bucket
+ * {@code Key_Shared} ownership on scale-up). This verifies the producer-side
per-bucket batching and
+ * stamping do not disturb ordinary single-active delivery: per-key order is
preserved and no message
+ * is dropped or duplicated.
+ */
+public class V5EntryBucketDispatchTest extends V5ClientBaseTest {
+
+ @Test
+ public void testBucketedSegmentPreservesPerKeyOrderAndDeliversAll() throws
Exception {
+ String topic = newScalableTopic(1);
+
+ @Cleanup
+ Producer<String> producer = v5Client.newProducer(Schema.string())
+ .topic(topic)
+ .create();
+ @Cleanup
+ StreamConsumer<String> consumer =
v5Client.newStreamConsumer(Schema.string())
+ .topic(topic)
+ .subscriptionName("bucket-dispatch")
+
.subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST)
+ .subscribe();
+
+ // 8 keys × 25 messages, interleaved. With per-bucket batching on,
same-key messages must
+ // still arrive in send order (only holds if every entry routes to the
one consumer that owns
+ // the key's bucket), and every message must be delivered exactly once.
+ List<String> keys = List.of("alpha", "bravo", "charlie", "delta",
"echo", "foxtrot", "golf", "hotel");
+ int perKey = 25;
+ Map<String, List<String>> 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);
+ }
+ }
+
+ Map<String, List<String>> received = new HashMap<>();
+ for (String k : keys) {
+ received.put(k, new ArrayList<>());
+ }
+ int total = keys.size() * perKey;
+ MessageId last = null;
+ for (int i = 0; i < total; i++) {
+ Message<String> msg = consumer.receive(Duration.ofSeconds(5));
+ assertNotNull(msg, "missed message #" + i);
+ String key = msg.key().orElseThrow(() -> new
AssertionError("missing key"));
+ received.get(key).add(msg.value());
+ last = msg.id();
+ }
+ consumer.acknowledgeCumulative(last);
+
+ for (String k : keys) {
+ assertEquals(received.get(k), sent.get(k), "per-key order must be
preserved for key=" + k);
+ }
+ }
+}
diff --git
a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ClientSegmentLayout.java
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ClientSegmentLayout.java
index cd0866b80b7..b987150e6a5 100644
---
a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ClientSegmentLayout.java
+++
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ClientSegmentLayout.java
@@ -87,8 +87,10 @@ final class ClientSegmentLayout {
for (int j = 0; j < seg.getEntryBucketSplitsCount(); j++) {
bucketSplits.add(seg.getEntryBucketSplitAt(j));
}
+ // The DAG topology carries the segment's split points (for
producer bucketing); the
+ // consumer's owned bucket ranges come from the assignment, not
the topology.
ActiveSegment ref = new ActiveSegment(seg.getSegmentId(), range,
segTopicName, legacy,
- bucketSplits);
+ bucketSplits, List.of());
if (seg.getState() ==
org.apache.pulsar.common.api.proto.SegmentState.ACTIVE) {
activeSegments.add(ref);
} else if (seg.getState() ==
org.apache.pulsar.common.api.proto.SegmentState.SEALED) {
diff --git
a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java
index edfa1a4a0f8..d09696a190e 100644
---
a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java
+++
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableConsumerClient.java
@@ -314,13 +314,20 @@ final class ScalableConsumerClient implements
ScalableConsumerSession, AutoClose
List<ActiveSegment> segments = new
ArrayList<>(assignment.getSegmentsCount());
for (int i = 0; i < assignment.getSegmentsCount(); i++) {
ScalableAssignedSegment s = assignment.getSegmentAt(i);
+ // PIP-486: the entry-bucket hash ranges this consumer owns within
the segment (empty =
+ // the whole segment). Drives Shared vs Key_Shared STICKY when
subscribing to the segment.
+ List<HashRange> ownedBucketRanges = new
ArrayList<>(s.getBucketRangesCount());
+ for (int j = 0; j < s.getBucketRangesCount(); j++) {
+ var range = s.getBucketRangeAt(j);
+ ownedBucketRanges.add(HashRange.of(range.getStart(),
range.getEnd()));
+ }
segments.add(new ActiveSegment(
s.getSegmentId(),
HashRange.of((int) s.getHashStart(), (int) s.getHashEnd()),
s.getSegmentTopic(),
/*legacyTopicName*/ null,
- /*entryBucketSplits, set by the controller assignment in a
later PIP-486 PR*/
- List.of()));
+ /*entryBucketSplits, producer-only*/ List.of(),
+ ownedBucketRanges));
}
return Collections.unmodifiableList(segments);
}
diff --git
a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableStreamConsumer.java
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableStreamConsumer.java
index 9cc3d7b110e..3e622baf471 100644
---
a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableStreamConsumer.java
+++
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableStreamConsumer.java
@@ -30,6 +30,8 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.TimeUnit;
+import org.apache.pulsar.client.api.KeySharedPolicy;
+import org.apache.pulsar.client.api.Range;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.api.v5.Message;
import org.apache.pulsar.client.api.v5.MessageId;
@@ -42,6 +44,7 @@ import org.apache.pulsar.client.api.v5.schema.Schema;
import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
import org.apache.pulsar.client.impl.v5.SegmentRouter.ActiveSegment;
+import org.apache.pulsar.common.scalable.HashRange;
import org.apache.pulsar.common.scalable.ScalableTopicConstants;
/**
@@ -77,6 +80,9 @@ final class ScalableStreamConsumer<T>
*/
private final ConcurrentHashMap<Long,
CompletableFuture<org.apache.pulsar.client.api.Consumer<T>>>
segmentConsumers = new ConcurrentHashMap<>();
+ // PIP-486: the entry-bucket ranges each segment consumer was last
subscribed with, so a change in
+ // ownership (a consumer joined/left) re-subscribes the segment with the
new ranges.
+ private final ConcurrentHashMap<Long, List<HashRange>> segmentBucketRanges
= new ConcurrentHashMap<>();
/**
* Tracks the latest message ID delivered from each segment. Updated
atomically
@@ -351,6 +357,7 @@ final class ScalableStreamConsumer<T>
.log("Closing consumer for segment removed from
assignment");
entry.getValue().thenAccept(c -> c.closeAsync());
segmentConsumers.remove(entry.getKey());
+ segmentBucketRanges.remove(entry.getKey());
latestDelivered.remove(entry.getKey());
}
}
@@ -358,6 +365,17 @@ final class ScalableStreamConsumer<T>
// Subscribe to newly-assigned segments.
List<CompletableFuture<?>> futures = new ArrayList<>();
for (var seg : assigned) {
+ // PIP-486: if the controller changed which entry-buckets we own
on a segment we are
+ // already subscribed to (a consumer joined or left), re-subscribe
with the new ranges so
+ // the broker's exclusive bucket selector sees a consistent,
non-overlapping assignment.
+ var existing = segmentConsumers.get(seg.segmentId());
+ if (existing != null
+ &&
!seg.ownedBucketRanges().equals(segmentBucketRanges.get(seg.segmentId()))) {
+ log.info().attr("segmentId", seg.segmentId())
+ .log("Re-subscribing segment for changed entry-bucket
ownership");
+ existing.thenAccept(c -> c.closeAsync());
+ segmentConsumers.remove(seg.segmentId());
+ }
futures.add(segmentConsumers.computeIfAbsent(seg.segmentId(),
id -> createSegmentConsumerAsync(seg)));
}
@@ -377,7 +395,23 @@ final class ScalableStreamConsumer<T>
// Legacy segments wrap an externally managed persistent:// topic;
regular ones use the
// computed segment:// URI. attachTopicName() collapses both into the
right URI.
segConf.getTopicNames().add(segment.attachTopicName());
- segConf.setSubscriptionType(SubscriptionType.Exclusive);
+ List<HashRange> ownedBucketRanges = segment.ownedBucketRanges();
+ if (ownedBucketRanges.isEmpty()) {
+ // Single-bucket segment: this consumer owns the whole segment
exclusively (pre-PIP-486).
+ segConf.setSubscriptionType(SubscriptionType.Exclusive);
+ } else {
+ // PIP-486: this consumer owns a subset of the segment's
entry-buckets. Subscribe Key_Shared
+ // STICKY declaring exactly those bucket hash-ranges, so the
broker dispatches each entry to
+ // the consumer owning its bucket; other owners of the same
segment share the subscription
+ // with disjoint ranges.
+ List<Range> ranges = new ArrayList<>(ownedBucketRanges.size());
+ for (HashRange r : ownedBucketRanges) {
+ ranges.add(Range.of(r.start(), r.end()));
+ }
+ segConf.setSubscriptionType(SubscriptionType.Key_Shared);
+
segConf.setKeySharedPolicy(KeySharedPolicy.stickyHashRange().ranges(ranges));
+ }
+ segmentBucketRanges.put(segment.segmentId(), ownedBucketRanges);
// Only legacy segments wrap a persistent:// topic that the
regular-to-scalable
// migration pre-check inspects, so mark just those connections as
V5-managed —
// connections to real segment:// topics are never examined.
diff --git
a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableTopicProducer.java
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableTopicProducer.java
index 027f82a1a62..71b7060de5b 100644
---
a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableTopicProducer.java
+++
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableTopicProducer.java
@@ -592,8 +592,14 @@ final class ScalableTopicProducer<T> implements
Producer<T>, DagWatchClient.Layo
/**
* PIP-486: configure a per-segment producer's batching for
entry-bucketing. End-to-end encryption
* disables batching (an encrypted batch can't be reshaped if re-routed
across a divergent layout);
- * otherwise, when batching is enabled, route the segment's batches by
entry-bucket so the broker can
- * dispatch a whole entry to one consumer. A segment's bucketing is
immutable for its life.
+ * otherwise, when batching is enabled, group the segment's batches by
entry-bucket and stamp each
+ * entry's effective entry-bucket hash range. A segment's bucketing is
immutable for its life.
+ *
+ * <p>The stamp is written for every segment, including single-bucket ones
(N = 1, e.g. the
+ * legacy/synthetic layouts wrapping a regular {@code persistent://}
topic): the effective hash
+ * range is standalone metadata a consumer or a geo-replicator can use to
check whether a batch
+ * still lands cleanly in one bucket of a possibly-different target
layout, independent of how any
+ * single broker dispatches it.
*/
static void applyEntryBucketing(ProducerConfigurationData segConf,
ActiveSegment segment) {
if (segConf.isEncryptionEnabled()) {
diff --git
a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/SegmentRouter.java
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/SegmentRouter.java
index d831fcc60a8..08e872b6780 100644
---
a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/SegmentRouter.java
+++
b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/SegmentRouter.java
@@ -154,11 +154,17 @@ final class SegmentRouter {
* {@code segmentTopicName} otherwise.
*/
record ActiveSegment(long segmentId, HashRange hashRange, String
segmentTopicName,
- String legacyTopicName, List<Integer>
entryBucketSplits) {
+ String legacyTopicName, List<Integer>
entryBucketSplits,
+ List<HashRange> ownedBucketRanges) {
ActiveSegment {
- // PIP-486: entry-bucket split points (empty = single bucket over
the whole ring).
+ // PIP-486: entry-bucket split points (empty = single bucket over
the whole ring). Used by
+ // the producer to bucket its batches.
entryBucketSplits = entryBucketSplits != null ?
List.copyOf(entryBucketSplits) : List.of();
+ // PIP-486: the entry-bucket hash ranges this consumer owns within
the segment (from the
+ // controller assignment). Empty = the whole segment (subscribe
Shared); non-empty = subscribe
+ // Key_Shared STICKY declaring exactly these ranges. Unused on the
producer side.
+ ownedBucketRanges = ownedBucketRanges != null ?
List.copyOf(ownedBucketRanges) : List.of();
}
/** Number of entry-buckets this segment is divided into ({@code
entryBucketSplits.size()+1}). */
diff --git
a/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/ScalableTopicProducerTest.java
b/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/ScalableTopicProducerTest.java
index ac5d30d318e..b193debda9b 100644
---
a/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/ScalableTopicProducerTest.java
+++
b/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/ScalableTopicProducerTest.java
@@ -39,7 +39,13 @@ public class ScalableTopicProducerTest {
private static ActiveSegment segment() {
return new ActiveSegment(0L, HashRange.of(0x0000, 0xFFFF),
"segment://t/n/x/0", null,
- List.of(0x8000));
+ List.of(0x8000), List.of());
+ }
+
+ /** A single-bucket segment (N = 1, e.g. a legacy/synthetic layout for a
regular topic). */
+ private static ActiveSegment singleBucketSegment() {
+ return new ActiveSegment(0L, HashRange.of(0x0000, 0xFFFF),
"segment://t/n/x/0", null,
+ List.of(), List.of());
}
@Test
@@ -50,6 +56,16 @@ public class ScalableTopicProducerTest {
assertTrue(conf.getBatcherBuilder() instanceof
EntryBucketBatcherBuilder);
}
+ @Test
+ public void testSingleBucketSegmentAlsoUsesEntryBucketBatcher() {
+ // N = 1 still uses the entry-bucket batcher so it stamps the
effective hash range — standalone
+ // metadata (e.g. for geo-replication re-routing), even though there
is only one bucket.
+ ProducerConfigurationData conf = new ProducerConfigurationData();
+ ScalableTopicProducer.applyEntryBucketing(conf, singleBucketSegment());
+ assertTrue(conf.isBatchingEnabled());
+ assertTrue(conf.getBatcherBuilder() instanceof
EntryBucketBatcherBuilder);
+ }
+
@Test
public void testEncryptionDisablesBatching() {
ProducerConfigurationData conf = new ProducerConfigurationData();
diff --git
a/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/SegmentRouterTest.java
b/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/SegmentRouterTest.java
index 884658e7db0..97f6242a275 100644
---
a/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/SegmentRouterTest.java
+++
b/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/SegmentRouterTest.java
@@ -34,13 +34,13 @@ public class SegmentRouterTest {
private static ActiveSegment seg(long id, int start, int end) {
return new ActiveSegment(id, HashRange.of(start, end),
"persistent://t/n/seg-" + id, null,
- List.of());
+ List.of(), List.of());
}
/** Build a legacy segment (synthetic-layout entry wrapping an externally
managed persistent:// topic). */
private static ActiveSegment legacySeg(long id, int start, int end, String
underlying) {
return new ActiveSegment(id, HashRange.of(start, end),
"segment://t/n/x/" + id, underlying,
- List.of());
+ List.of(), List.of());
}
// --- route(key, ...) ---
diff --git a/pulsar-common/src/main/proto/PulsarApi.proto
b/pulsar-common/src/main/proto/PulsarApi.proto
index 70e71028ae8..529c806d84d 100644
--- a/pulsar-common/src/main/proto/PulsarApi.proto
+++ b/pulsar-common/src/main/proto/PulsarApi.proto
@@ -945,6 +945,11 @@ message ScalableAssignedSegment {
required uint32 hash_end = 3;
// Fully-qualified segment:// topic name the consumer should attach to.
required string segment_topic = 4;
+ // PIP-486: the entry-bucket hash ranges (16-bit, inclusive) this consumer
owns within the
+ // segment. Empty means the consumer owns the whole segment (single
bucket) and subscribes
+ // Shared; non-empty means the segment is shared by bucket, and the
consumer subscribes
+ // Key_Shared STICKY declaring exactly these ranges.
+ repeated IntRange bucket_ranges = 5;
}
// An assignment of active segments to a single consumer. Carries the layout
epoch
