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 39784e085ef4 [improve][broker] PIP-486 (PR4): bucket-shared
consumption for scale-up fan-out (#26173)
39784e085ef4 is described below
commit 39784e085ef491ffe319433dc465ba9575cadc95
Author: Matteo Merli <[email protected]>
AuthorDate: Fri Jul 10 07:00:37 2026 -0700
[improve][broker] PIP-486 (PR4): bucket-shared consumption for scale-up
fan-out (#26173)
---
...istentStickyKeyDispatcherMultipleConsumers.java | 21 ++
.../service/scalable/SubscriptionCoordinator.java | 85 ++++++--
...ntStickyKeyDispatcherMultipleConsumersTest.java | 54 +++++
.../scalable/SubscriptionCoordinatorTest.java | 78 +++++--
.../client/api/v5/V5EntryBucketDispatchTest.java | 119 ++++++++++-
.../client/impl/v5/ScalableConsumerClient.java | 12 ++
.../client/impl/v5/ScalableStreamConsumer.java | 235 +++++++++++++++++++--
.../apache/pulsar/client/impl/ConsumerImpl.java | 2 +-
.../impl/conf/ConsumerConfigurationData.java | 8 +
.../apache/pulsar/common/protocol/Commands.java | 15 ++
pulsar-common/src/main/proto/PulsarApi.proto | 5 +
11 files changed, 575 insertions(+), 59 deletions(-)
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
index 7b7b6596f0ca..d761bc92594b 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
@@ -69,6 +69,9 @@ public class PersistentStickyKeyDispatcherMultipleConsumers
extends PersistentDi
private final boolean allowOutOfOrderDelivery;
private final StickyKeyConsumerSelector selector;
private final boolean drainingHashesRequired;
+ // PIP-486: dispatch whole entries by their producer-stamped entry-bucket
hash range instead of
+ // hashing each message's key. Set only by scalable-topic consumers
sharing a segment by bucket.
+ private final boolean entryBucketDispatch;
private boolean skipNextReplayToTriggerLookAhead = false;
private final KeySharedMode keySharedMode;
@@ -84,6 +87,7 @@ public class PersistentStickyKeyDispatcherMultipleConsumers
extends PersistentDi
this.allowOutOfOrderDelivery = ksm.isAllowOutOfOrderDelivery();
this.keySharedMode = ksm.getKeySharedMode();
+ this.entryBucketDispatch = ksm.isEntryBucketDispatch();
// recent joined consumer tracking is required only for AUTO_SPLIT
mode when out-of-order delivery is disabled
this.drainingHashesRequired =
keySharedMode == KeySharedMode.AUTO_SPLIT &&
!allowOutOfOrderDelivery;
@@ -634,6 +638,23 @@ public class
PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi
@Override
protected int getStickyKeyHash(Entry entry) {
if (entry instanceof EntryAndMetadata entryAndMetadata) {
+ // PIP-486: an entry-bucket subscription routes each whole entry
by its producer-stamped
+ // entry-bucket hash range, so a batch holding one bucket's keys
goes to the bucket's owner
+ // with no per-key hashing. The stamp shares the 16-bit key-hash
space, so it feeds the
+ // selector directly. Cached as THE entry's sticky-key hash so
pending acks, redelivery and
+ // draining all see the value dispatch used. Unstamped entries
(non-batched messages) fall
+ // through to the message's sticky-key hash, which is the same
low-16 Murmur value the
+ // producer would have stamped.
+ if (entryBucketDispatch) {
+ var metadata = entryAndMetadata.getMetadata();
+ if (metadata != null && metadata.hasEntryHashMin()) {
+ return
entryAndMetadata.getOrUpdateCachedStickyKeyHash(stickyKey -> {
+ int bucketHash = metadata.getEntryHashMin();
+ // 0 is reserved as "hash not set"; nudge to 1, which
is still inside bucket 0.
+ return bucketHash == STICKY_KEY_HASH_NOT_SET ? 1 :
bucketHash;
+ });
+ }
+ }
// use the cached sticky key hash if available, otherwise
calculate the sticky key hash and cache it
return
entryAndMetadata.getOrUpdateCachedStickyKeyHash(selector::makeStickyKeyHash);
}
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 a3757c16c6ce..9b33ff7120be 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
@@ -21,6 +21,7 @@ package org.apache.pulsar.broker.service.scalable;
import io.github.merlimat.slog.Logger;
import java.time.Duration;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collection;
import java.util.Comparator;
import java.util.HashSet;
@@ -37,6 +38,7 @@ import lombok.Getter;
import org.apache.pulsar.broker.resources.ScalableTopicResources;
import org.apache.pulsar.broker.service.TransportCnx;
import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.scalable.HashRange;
import org.apache.pulsar.common.scalable.SegmentInfo;
import org.apache.pulsar.common.scalable.SegmentTopicName;
import org.apache.pulsar.common.util.Backoff;
@@ -533,9 +535,9 @@ public class SubscriptionCoordinator {
* Compute a balanced assignment of segments to consumers.
*
* <p>Strategy: sort segments by hash range, then segment id (tiebreak),
sort consumers by
- * name, then round-robin. Deterministic: the same inputs always produce
the same output,
- * so a new leader recomputing assignments after failover gets the same
result as the old
- * leader.
+ * name, then "segments first, entry-buckets absorb the surplus" (see the
inline comment).
+ * Deterministic: the same inputs always produce the same output, so a new
leader recomputing
+ * assignments after failover gets the same result as the old leader.
*
* <p><b>DAG replay.</b> The assignment includes every <em>sealed</em>
segment in the
* DAG. A fresh EARLIEST subscription needs to read messages produced
before it joined,
@@ -577,20 +579,69 @@ public class SubscriptionCoordinator {
assignmentLists.put(consumer, new ArrayList<>());
}
- int consumerIndex = 0;
- for (SegmentInfo segment : sortedSegments) {
- 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(), List.of()));
- consumerIndex++;
+ // PIP-486: "segments first, entry-buckets absorb the surplus". While
consumers don't outnumber
+ // segments, each whole segment goes to a single consumer for
efficient single-active
+ // (Exclusive) dispatch — no per-bucket pending tracking; empty
bucketRanges signals the client
+ // to subscribe Exclusive. Only when consumers outnumber segments does
the controller fan
+ // 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.
+ int segmentCount = sortedSegments.size();
+ int consumerCount = sortedConsumers.size();
+ if (consumerCount <= segmentCount) {
+ int consumerIndex = 0;
+ for (SegmentInfo segment : sortedSegments) {
+ TopicName segmentTopic =
SegmentTopicName.fromParent(topicName, segment.hashRange(),
+ segment.segmentId());
+ ConsumerSession consumer = sortedConsumers.get(consumerIndex %
sortedConsumers.size());
+ assignmentLists.get(consumer).add(new
ConsumerAssignment.AssignedSegment(
+ segment.segmentId(), segment.hashRange(),
segmentTopic.toString(), List.of()));
+ consumerIndex++;
+ }
+ } else {
+ // Per-segment owner counts: one each, then hand the surplus to
segments that still have
+ // bucket capacity, round-robin in segment order.
+ int[] owners = new int[segmentCount];
+ Arrays.fill(owners, 1);
+ int surplus = consumerCount - segmentCount;
+ boolean anyCapacityLeft = true;
+ while (surplus > 0 && anyCapacityLeft) {
+ anyCapacityLeft = false;
+ for (int i = 0; i < segmentCount && surplus > 0; i++) {
+ if (owners[i] < sortedSegments.get(i).bucketCount()) {
+ owners[i]++;
+ surplus--;
+ anyCapacityLeft = true;
+ }
+ }
+ }
+ int consumerIndex = 0;
+ for (int i = 0; i < segmentCount; i++) {
+ SegmentInfo segment = sortedSegments.get(i);
+ TopicName segmentTopic =
SegmentTopicName.fromParent(topicName, segment.hashRange(),
+ segment.segmentId());
+ int k = owners[i];
+ if (k == 1) {
+ ConsumerSession consumer =
sortedConsumers.get(consumerIndex++);
+ assignmentLists.get(consumer).add(new
ConsumerAssignment.AssignedSegment(
+ segment.segmentId(), segment.hashRange(),
segmentTopic.toString(), List.of()));
+ } else {
+ List<HashRange> buckets =
EntryBucketSplits.ranges(segment.entryBucketSplits());
+ int base = buckets.size() / k;
+ int extra = buckets.size() % k;
+ int from = 0;
+ for (int c = 0; c < k; c++) {
+ int size = base + (c < extra ? 1 : 0);
+ List<HashRange> slice =
List.copyOf(buckets.subList(from, from + size));
+ from += size;
+ ConsumerSession consumer =
sortedConsumers.get(consumerIndex++);
+ assignmentLists.get(consumer).add(new
ConsumerAssignment.AssignedSegment(
+ segment.segmentId(), segment.hashRange(),
segmentTopic.toString(), slice));
+ }
+ }
+ }
+ // Consumers past consumerIndex found no segment with spare bucket
capacity: they keep an
+ // empty assignment (idle) until the layout or the group changes.
}
Map<ConsumerSession, ConsumerAssignment> result = new
LinkedHashMap<>();
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersTest.java
index 81bf59fe373a..8e68500f80ac 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersTest.java
@@ -71,6 +71,7 @@ import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.EntryAndMetadata;
import org.apache.pulsar.broker.service.EntryBatchIndexesAcks;
import org.apache.pulsar.broker.service.EntryBatchSizes;
import org.apache.pulsar.broker.service.PendingAcksMap;
@@ -847,6 +848,59 @@ public class
PersistentStickyKeyDispatcherMultipleConsumersTest {
});
}
+ @Test
+ public void testEntryBucketDispatchRoutesByStampedRange() {
+ // PIP-486: with entryBucketDispatch set on the subscription's
KeySharedMeta, a stamped entry
+ // routes as a whole by its entry-bucket hash (entry_hash_min), not by
the message key.
+ PersistentStickyKeyDispatcherMultipleConsumers bucketDispatcher =
+ new PersistentStickyKeyDispatcherMultipleConsumers(topicMock,
cursorMock, subscriptionMock,
+ configMock, new
KeySharedMeta().setKeySharedMode(KeySharedMode.STICKY)
+ .setEntryBucketDispatch(true));
+ EntryImpl entry = createEntry(1, 1, "msg", 1, "some-key");
+ try {
+ MessageMetadata stamped = new MessageMetadata()
+ .setProducerName("p").setSequenceId(1).setPublishTime(1)
+ .setEntryHashMin(0x1234).setEntryHashMax(0x5678);
+ assertEquals(bucketDispatcher.getStickyKeyHash(
+ EntryAndMetadata.create(entry, stamped)), 0x1234);
+
+ // entry_hash_min == 0 collides with the reserved "hash not set"
sentinel, so it is nudged
+ // to 1, which is still inside bucket 0.
+ MessageMetadata zero = new MessageMetadata()
+ .setProducerName("p").setSequenceId(1).setPublishTime(1)
+ .setEntryHashMin(0).setEntryHashMax(0);
+ assertEquals(bucketDispatcher.getStickyKeyHash(
+ EntryAndMetadata.create(entry, zero)), 1);
+
+ // Unstamped entries (non-batched messages) fall back to the
message's sticky-key hash.
+ MessageMetadata unstamped = new MessageMetadata()
+
.setProducerName("p").setSequenceId(1).setPublishTime(1).setPartitionKey("some-key");
+ assertEquals(bucketDispatcher.getStickyKeyHash(
+ EntryAndMetadata.create(entry, unstamped)),
+
bucketDispatcher.getSelector().makeStickyKeyHash("some-key".getBytes(UTF_8)));
+ } finally {
+ entry.release();
+ }
+ }
+
+ @Test
+ public void testStampedEntryStillRoutesByKeyWithoutEntryBucketDispatch() {
+ // Without the entryBucketDispatch flag (any plain Key_Shared
subscription), a stamped entry
+ // keeps dispatching by the message key — the stamp is ignored.
+ EntryImpl entry = createEntry(1, 1, "msg", 1, "some-key");
+ try {
+ MessageMetadata stamped = new MessageMetadata()
+ .setProducerName("p").setSequenceId(1).setPublishTime(1)
+ .setPartitionKey("some-key")
+ .setEntryHashMin(0x1234).setEntryHashMax(0x5678);
+ assertEquals(persistentDispatcher.getStickyKeyHash(
+ EntryAndMetadata.create(entry, stamped)),
+
persistentDispatcher.getSelector().makeStickyKeyHash("some-key".getBytes(UTF_8)));
+ } finally {
+ entry.release();
+ }
+ }
+
private EntryImpl createEntry(long ledgerId, long entryId, String message,
long sequenceId) {
return createEntry(ledgerId, entryId, message, sequenceId, "testKey");
}
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 90b20ac96677..2480d4de0927 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
@@ -27,6 +27,8 @@ import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -40,6 +42,7 @@ import
org.apache.pulsar.broker.resources.ScalableTopicMetadata;
import org.apache.pulsar.broker.resources.ScalableTopicResources;
import org.apache.pulsar.broker.service.TransportCnx;
import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.scalable.HashRange;
import org.awaitility.Awaitility;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
@@ -444,27 +447,74 @@ public class SubscriptionCoordinatorTest {
}
@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));
+ public void testLoneConsumerOwnsBucketedSegmentWhole() throws Exception {
+ // One segment with N=4 entry-buckets, one consumer: no surplus, so
the segment stays whole
+ // (empty bucketRanges -> Exclusive single-active dispatch).
+ SubscriptionCoordinator c = bucketedCoordinator();
+ Map<ConsumerSession, ConsumerAssignment> result =
+ c.registerConsumer("consumer-1", 1L,
mock(TransportCnx.class)).get();
+
+ ConsumerAssignment assignment = findByName(result, "consumer-1");
+ assertEquals(assignment.assignedSegments().size(), 1);
+
assertTrue(assignment.assignedSegments().get(0).bucketRanges().isEmpty());
+ }
+
+ @Test
+ public void testSurplusConsumersFanOutBucketedSegment() throws Exception {
+ // One segment with N=4 entry-buckets, two consumers: the surplus
consumer fans the segment
+ // out — each owner takes a contiguous half of the buckets, disjoint
and tiling the ring.
+ SubscriptionCoordinator c = bucketedCoordinator();
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;
+ assertEquals(result.size(), 2);
+ List<HashRange> owned = new ArrayList<>();
for (ConsumerAssignment assignment : result.values()) {
- for (ConsumerAssignment.AssignedSegment seg :
assignment.assignedSegments()) {
- assertEquals(seg.segmentId(), 0);
- assertTrue(seg.bucketRanges().isEmpty());
- owners++;
+ assertEquals(assignment.assignedSegments().size(), 1);
+ ConsumerAssignment.AssignedSegment seg =
assignment.assignedSegments().get(0);
+ assertEquals(seg.segmentId(), 0);
+ assertEquals(seg.bucketRanges().size(), 2);
+ owned.addAll(seg.bucketRanges());
+ }
+ owned.sort(Comparator.comparingInt(HashRange::start));
+ assertEquals(owned, List.of(
+ HashRange.of(0x0000, 0x3FFF), HashRange.of(0x4000, 0x7FFF),
+ HashRange.of(0x8000, 0xBFFF), HashRange.of(0xC000, 0xFFFF)));
+ }
+
+ @Test
+ public void testFanOutCapsAtBucketCountAndLeavesRestIdle() throws
Exception {
+ // One segment with N=4 entry-buckets, five consumers: four owners
with one bucket each; the
+ // fifth consumer exceeds the topic's bucket capacity and stays idle
(empty assignment).
+ SubscriptionCoordinator c = bucketedCoordinator();
+ for (int i = 1; i <= 4; i++) {
+ c.registerConsumer("consumer-" + i, i,
mock(TransportCnx.class)).get();
+ }
+ Map<ConsumerSession, ConsumerAssignment> result =
+ c.registerConsumer("consumer-5", 5L,
mock(TransportCnx.class)).get();
+
+ assertEquals(result.size(), 5);
+ List<HashRange> owned = new ArrayList<>();
+ int idle = 0;
+ for (ConsumerAssignment assignment : result.values()) {
+ if (assignment.assignedSegments().isEmpty()) {
+ idle++;
+ continue;
}
+ ConsumerAssignment.AssignedSegment seg =
assignment.assignedSegments().get(0);
+ assertEquals(seg.bucketRanges().size(), 1);
+ owned.addAll(seg.bucketRanges());
}
- assertEquals(owners, 1);
+ assertEquals(idle, 1);
+ assertEquals(owned.size(), 4);
+ }
+
+ private SubscriptionCoordinator bucketedCoordinator() {
+ // One segment carrying the whole default budget: N = 4 entry-buckets.
+ return new SubscriptionCoordinator("test-sub", topicName,
+
SegmentLayout.fromMetadata(ScalableTopicController.createInitialMetadata(1, 4,
Map.of())),
+ resources, scheduler, Duration.ofMillis(200));
}
// --- Helpers ---
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
index 8de7e8210679..6539de7b261f 100644
---
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
@@ -19,27 +19,31 @@
package org.apache.pulsar.client.api.v5;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
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.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import lombok.Cleanup;
import org.apache.pulsar.client.api.v5.config.SubscriptionInitialPosition;
import org.apache.pulsar.client.api.v5.schema.Schema;
+import org.apache.pulsar.common.policies.data.AutoScalePolicyOverride;
import org.testng.annotations.Test;
/**
- * PIP-486 end-to-end: a single stream consumer on an entry-bucketed segment.
+ * PIP-486 end-to-end: stream consumers 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.
+ * {@code Exclusive} (single-active dispatch); a second consumer makes the
controller fan the segment
+ * out — each owner takes half the buckets and subscribes {@code Key_Shared}
STICKY with those ranges,
+ * and the broker dispatches each whole entry by its stamped range to the
bucket's owner.
*/
public class V5EntryBucketDispatchTest extends V5ClientBaseTest {
@@ -94,4 +98,109 @@ public class V5EntryBucketDispatchTest extends
V5ClientBaseTest {
assertEquals(received.get(k), sent.get(k), "per-key order must be
preserved for key=" + k);
}
}
+
+ @Test
+ public void testTwoConsumersShareBucketedSegmentByEntryBucket() throws
Exception {
+ String topic = newScalableTopic(1);
+ // Pin the layout: with more consumers than segments, PIP-483 would
otherwise split the
+ // segment ("segments first"). Disabling auto split/merge forces the
controller to serve the
+ // second consumer by fanning the segment out by entry-bucket — the
path under test.
+ admin.scalableTopics().setAutoScalePolicy(topic,
+ AutoScalePolicyOverride.builder().enabled(false).build());
+ String subscription = "bucket-share";
+
+ @Cleanup
+ Producer<String> producer = v5Client.newProducer(Schema.string())
+ .topic(topic)
+ .create();
+ // Two stream consumers on the one-segment (N=4) topic: the controller
fans the segment out,
+ // giving each consumer two of the four buckets (Key_Shared STICKY
under the hood).
+ @Cleanup
+ StreamConsumer<String> a = v5Client.newStreamConsumer(Schema.string())
+ .topic(topic)
+ .subscriptionName(subscription)
+
.subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST)
+ .subscribe();
+ @Cleanup
+ StreamConsumer<String> b = v5Client.newStreamConsumer(Schema.string())
+ .topic(topic)
+ .subscriptionName(subscription)
+
.subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST)
+ .subscribe();
+
+ // 16 keys × 20 messages, interleaved — enough keys to populate all
four buckets with
+ // overwhelming probability, so both owners receive traffic.
+ List<String> keys = new ArrayList<>();
+ for (int i = 0; i < 16; i++) {
+ keys.add("key-" + i);
+ }
+ int perKey = 20;
+ 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>> aGot = new ConcurrentHashMap<>();
+ Map<String, List<String>> bGot = new ConcurrentHashMap<>();
+ Thread ta = drainOrdered(a, aGot);
+ Thread tb = drainOrdered(b, bGot);
+ ta.join();
+ tb.join();
+
+ assertFalse(aGot.isEmpty(), "consumer A received nothing — the segment
did not fan out");
+ assertFalse(bGot.isEmpty(), "consumer B received nothing — the segment
did not fan out");
+
+ // Whole-entry bucket dispatch: every key lands wholly on exactly one
consumer, in send order.
+ for (String k : keys) {
+ List<String> fromA = aGot.get(k);
+ List<String> fromB = bGot.get(k);
+ assertTrue(fromA == null || fromB == null, "key " + k + " was
split across both consumers");
+ List<String> got = fromA != null ? fromA : fromB;
+ assertEquals(got, sent.get(k), "per-key order/content for key=" +
k);
+ }
+
+ // The drainers acked everything they received — for a bucket-shared
segment that goes through
+ // the individual-ack translation (Key_Shared forbids cumulative
acks). A fresh consumer on the
+ // same subscription must therefore find nothing to redeliver.
+ a.close();
+ b.close();
+ @Cleanup
+ StreamConsumer<String> c = v5Client.newStreamConsumer(Schema.string())
+ .topic(topic)
+ .subscriptionName(subscription)
+
.subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST)
+ .subscribe();
+ assertNull(c.receive(Duration.ofSeconds(3)), "acked messages were
redelivered");
+ }
+
+ /** Drains until idle, recording values per key in arrival order, then
acks cumulatively. */
+ private Thread drainOrdered(StreamConsumer<String> consumer, Map<String,
List<String>> into) {
+ Thread t = new Thread(() -> {
+ try {
+ MessageId last = null;
+ while (true) {
+ Message<String> msg =
consumer.receive(Duration.ofSeconds(2));
+ if (msg == null) {
+ if (last != null) {
+ consumer.acknowledgeCumulative(last);
+ }
+ return;
+ }
+ String key = msg.key().orElseThrow(() -> new
AssertionError("missing key"));
+ into.computeIfAbsent(key, __ -> new
ArrayList<>()).add(msg.value());
+ last = msg.id();
+ }
+ } catch (Exception ignored) {
+ }
+ }, "bucket-share-drainer");
+ t.start();
+ return t;
+ }
}
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 d09696a190e4..1597a7b083c6 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
@@ -338,6 +338,18 @@ final class ScalableConsumerClient implements
ScalableConsumerSession, AutoClose
void setListener(AssignmentChangeListener listener) {
this.listener = listener;
+ // Replay the current assignment: an update that raced listener
registration (delivered
+ // after registerConsumer returned but before the caller finished
applying its initial
+ // assignment) would otherwise be lost — there is no periodic refresh
to recover it.
+ // Appliers are idempotent, so a redundant replay of the initial
assignment is a no-op.
+ List<ActiveSegment> current = currentAssignment.get();
+ if (current != null) {
+ try {
+ listener.onAssignmentChange(current, current);
+ } catch (Exception e) {
+ log.error().exception(e).log("Error in assignment change
listener");
+ }
+ }
}
long consumerId() {
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 b22a62ddd7ff..14f26de25d77 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
@@ -27,7 +27,11 @@ import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.pulsar.client.api.KeySharedPolicy;
import org.apache.pulsar.client.api.Range;
import org.apache.pulsar.client.api.SubscriptionType;
@@ -44,19 +48,26 @@ 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;
+import org.apache.pulsar.common.util.Backoff;
+import org.apache.pulsar.common.util.FutureUtil;
/**
* V5 StreamConsumer implementation for scalable topics.
*
- * <p>Maintains per-segment v4 Consumers with Exclusive subscription type.
- * Messages from all segments are multiplexed into a single receive queue.
+ * <p>Maintains a v4 Consumer per assigned segment. A segment this consumer
owns whole is subscribed
+ * {@code Exclusive} (single-active dispatch); a segment shared by
entry-bucket (PIP-486, when the
+ * controller fans it out across several consumers) is subscribed {@code
Key_Shared} STICKY declaring
+ * exactly the owned bucket ranges. Messages from all segments are multiplexed
into a single receive
+ * queue.
*
* <p>Each delivered message carries a <em>position vector</em>: a snapshot of
the
* latest delivered message ID per segment at the moment that message enters
the
* queue. When the application calls {@link #acknowledgeCumulative(MessageId)},
- * every segment is cumulatively acknowledged up to the position recorded in
that
- * vector. This ensures that acknowledging a single message correctly advances
- * all segments, not just the one it came from.
+ * every segment is acknowledged up to the position recorded in that vector —
+ * cumulatively for whole segments, and as individual acks of the delivered
ids for
+ * bucket-shared segments (Key_Shared forbids cumulative acks). This ensures
that
+ * acknowledging a single message correctly advances all segments, not just
the one
+ * it came from.
*/
final class ScalableStreamConsumer<T>
implements StreamConsumer<T>,
ScalableConsumerClient.AssignmentChangeListener {
@@ -90,6 +101,24 @@ final class ScalableStreamConsumer<T>
private final ConcurrentHashMap<Long,
org.apache.pulsar.client.api.MessageId> latestDelivered =
new ConcurrentHashMap<>();
+ /**
+ * PIP-486: for bucket-shared segments — Key_Shared under the hood, where
cumulative acks are not
+ * permitted — the ids delivered to this consumer and not yet acked, in
delivery order. A cumulative
+ * ack up to a segment's vector position is translated into individually
acking every tracked id up
+ * to that position. Whole-segment (Exclusive) consumers have no entry
here and ack cumulatively.
+ */
+ private final ConcurrentHashMap<Long,
ConcurrentLinkedQueue<org.apache.pulsar.client.api.MessageId>>
+ sharedSegmentUnacked = new ConcurrentHashMap<>();
+
+ /** Latest controller assignment; {@link #reconcile()} converges the
segment consumers onto it. */
+ private volatile List<ActiveSegment> latestAssignment;
+ /** Coalesces concurrent reconcile attempts; only one runs at a time. */
+ private final AtomicBoolean reconcileInProgress = new AtomicBoolean(false);
+ private final Backoff reconcileBackoff = Backoff.builder()
+ .initialDelay(Duration.ofMillis(100))
+ .maxBackoff(Duration.ofSeconds(30))
+ .build();
+
private final V5ReceiveQueue<T> receiveQueue;
/**
* Where each per-segment receive loop deposits a freshly-arrived message.
Defaults
@@ -155,7 +184,8 @@ final class ScalableStreamConsumer<T>
MessageSink<T> messageSink) {
ScalableStreamConsumer<T> consumer = new ScalableStreamConsumer<>(
client, v5Schema, consumerConf, session, topicName,
messageSink);
- return consumer.subscribeAssigned(initialAssignment)
+ consumer.latestAssignment = initialAssignment;
+ return consumer.subscribeInitialWithRetry(initialAssignment)
.thenApply(__ -> {
session.setListener(consumer);
return consumer;
@@ -221,10 +251,7 @@ final class ScalableStreamConsumer<T>
// Ack each segment up to the position recorded in the vector
for (var entry : id.positionVector().entrySet()) {
- var future = segmentConsumers.get(entry.getKey());
- if (future != null) {
- future.thenAccept(c ->
c.acknowledgeCumulativeAsync(entry.getValue()));
- }
+ ackSegmentUpTo(entry.getKey(), entry.getValue(), null);
}
}
@@ -235,11 +262,56 @@ final class ScalableStreamConsumer<T>
}
var v4Txn = TransactionV5.unwrap(txn);
for (var entry : id.positionVector().entrySet()) {
- var future = segmentConsumers.get(entry.getKey());
- if (future != null) {
- future.thenAccept(c ->
c.acknowledgeCumulativeAsync(entry.getValue(), v4Txn));
+ ackSegmentUpTo(entry.getKey(), entry.getValue(), v4Txn);
+ }
+ }
+
+ /**
+ * Ack one segment up to the given position. Whole-segment (Exclusive)
consumers ack
+ * cumulatively. PIP-486 bucket-shared segments are Key_Shared underneath,
where cumulative acks
+ * are not permitted — the ack is translated into individually acking
every delivered-but-unacked
+ * id up to the position (exactly the messages this consumer received: its
buckets' share).
+ */
+ private void ackSegmentUpTo(long segmentId,
org.apache.pulsar.client.api.MessageId position,
+
org.apache.pulsar.client.api.transaction.Transaction v4Txn) {
+ var future = segmentConsumers.get(segmentId);
+ if (future == null) {
+ return;
+ }
+ var unacked = sharedSegmentUnacked.get(segmentId);
+ if (unacked == null) {
+ future.thenAccept(c -> {
+ if (v4Txn == null) {
+ c.acknowledgeCumulativeAsync(position);
+ } else {
+ c.acknowledgeCumulativeAsync(position, v4Txn);
+ }
+ });
+ return;
+ }
+ // Redeliveries can enqueue ids out of order, so scan the whole queue
rather than stopping at
+ // the first id past the position. Concurrent acks may race on an id;
individually re-acking
+ // an already-acked id is a harmless no-op.
+ List<org.apache.pulsar.client.api.MessageId> toAck = new ArrayList<>();
+ for (var it = unacked.iterator(); it.hasNext();) {
+ var msgId = it.next();
+ if (msgId.compareTo(position) <= 0) {
+ it.remove();
+ toAck.add(msgId);
}
}
+ if (toAck.isEmpty()) {
+ return;
+ }
+ future.thenAccept(c -> {
+ if (v4Txn == null) {
+ c.acknowledgeAsync(toAck);
+ } else {
+ for (var msgId : toAck) {
+ c.acknowledgeAsync(msgId, v4Txn);
+ }
+ }
+ });
}
@Override
@@ -286,20 +358,112 @@ final class ScalableStreamConsumer<T>
: CompletableFuture.completedFuture(null)));
}
return
CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new))
- .whenComplete((__, ___) -> segmentConsumers.clear());
+ .whenComplete((__, ___) -> {
+ segmentConsumers.clear();
+ sharedSegmentUnacked.clear();
+ });
}
// --- Assignment change handling ---
@Override
public void onAssignmentChange(List<ActiveSegment> newSegments,
List<ActiveSegment> oldSegments) {
- // Fully async: safe to run on the netty IO thread that delivered the
update.
- subscribeAssigned(newSegments).exceptionally(ex -> {
- log.warn().exceptionMessage(ex).log("Failed to apply assignment
update");
- return null;
+ // Store the target and kick off a reconcile; a reconcile already in
flight re-reads
+ // latestAssignment when it finishes. Fully async: safe on the netty
IO thread that
+ // delivered the update.
+ latestAssignment = newSegments;
+ reconcile();
+ }
+
+ /**
+ * Initial subscribe with retries for the transient rebalance rejections.
Joining a group
+ * rebalances it: until a previous owner has released a segment (or shrunk
its declared bucket
+ * ranges), our subscribe is rejected — {@code ConsumerBusy} while it
still holds the segment,
+ * {@code ConsumerAssignError} while its STICKY ranges still overlap ours
— and neither is
+ * retried at the v4 layer. Bounded by the client operation timeout; any
other failure fails
+ * the subscribe immediately, preserving fail-fast for real errors.
+ */
+ private CompletableFuture<Void>
subscribeInitialWithRetry(List<ActiveSegment> assigned) {
+ CompletableFuture<Void> result = new CompletableFuture<>();
+ long deadlineNanos = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(
+ client.v4Client().getConfiguration().getOperationTimeoutMs());
+ attemptInitialSubscribe(assigned, deadlineNanos, result);
+ return result;
+ }
+
+ private void attemptInitialSubscribe(List<ActiveSegment> assigned, long
deadlineNanos,
+ CompletableFuture<Void> result) {
+ subscribeAssigned(assigned).whenComplete((__, ex) -> {
+ if (ex == null) {
+ reconcileBackoff.reset();
+ result.complete(null);
+ return;
+ }
+ Throwable cause = FutureUtil.unwrapCompletionException(ex);
+ boolean transientRebalance = cause instanceof
org.apache.pulsar.client.api
+ .PulsarClientException.ConsumerBusyException
+ || cause instanceof org.apache.pulsar.client.api
+ .PulsarClientException.ConsumerAssignException;
+ if (closed || !transientRebalance || System.nanoTime() >=
deadlineNanos) {
+ result.completeExceptionally(ex);
+ return;
+ }
+ evictFailedSegmentConsumers();
+ Duration delay = reconcileBackoff.next();
+ log.info().attr("delayMs", delay.toMillis()).exceptionMessage(ex)
+ .log("Initial subscribe rejected during rebalance,
retrying after backoff");
+ scheduler().schedule(() -> attemptInitialSubscribe(assigned,
deadlineNanos, result),
+ delay.toMillis(), TimeUnit.MILLISECONDS);
+ });
+ }
+
+ /**
+ * Converge the per-segment consumers onto {@link #latestAssignment},
retrying with backoff.
+ * Retrying here is what makes rebalances converge: during a fan-out or a
bucket move another
+ * consumer may not have released a segment or bucket range yet, so our
subscribe is rejected
+ * ({@code ConsumerBusy} on Exclusive, {@code ConsumerAssignError} on
overlapping STICKY
+ * ranges) — and neither is retried at the v4 layer.
+ */
+ private void reconcile() {
+ if (closed || !reconcileInProgress.compareAndSet(false, true)) {
+ return;
+ }
+ List<ActiveSegment> target = latestAssignment;
+ subscribeAssigned(target).whenComplete((__, ex) -> {
+ reconcileInProgress.set(false);
+ if (closed) {
+ return;
+ }
+ if (ex == null) {
+ reconcileBackoff.reset();
+ // If a newer assignment arrived during this reconcile, run
again to converge.
+ if (latestAssignment != target) {
+ reconcile();
+ }
+ return;
+ }
+ // Evict failed subscribe futures so the next attempt can re-try
them.
+ evictFailedSegmentConsumers();
+ Duration delay = reconcileBackoff.next();
+ log.warn().attr("delayMs", delay.toMillis()).exceptionMessage(ex)
+ .log("Failed to apply assignment update, retrying after
backoff");
+ scheduler().schedule(this::reconcile, delay.toMillis(),
TimeUnit.MILLISECONDS);
});
}
+ private void evictFailedSegmentConsumers() {
+ for (var entry : segmentConsumers.entrySet()) {
+ var future = entry.getValue();
+ if (future.isCompletedExceptionally()) {
+ segmentConsumers.remove(entry.getKey(), future);
+ }
+ }
+ }
+
+ private ScheduledExecutorService scheduler() {
+ return (ScheduledExecutorService)
client.v4Client().getScheduledExecutorProvider().getExecutor();
+ }
+
private CompletableFuture<Void> subscribeAssigned(List<ActiveSegment>
assigned) {
// Controller-driven assignment: the broker's SubscriptionCoordinator
decides
// which segments this consumer owns at any moment. We subscribe to
exactly
@@ -321,6 +485,7 @@ final class ScalableStreamConsumer<T>
entry.getValue().thenAccept(c -> c.closeAsync());
segmentConsumers.remove(entry.getKey());
segmentBucketRanges.remove(entry.getKey());
+ sharedSegmentUnacked.remove(entry.getKey());
latestDelivered.remove(entry.getKey());
}
}
@@ -336,11 +501,19 @@ final class ScalableStreamConsumer<T>
&&
!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());
+ segmentConsumers.remove(seg.segmentId(), existing);
+ futures.add(segmentConsumers.computeIfAbsent(seg.segmentId(),
id ->
+ // Await our own close before re-subscribing: the new
subscribe (a different
+ // type, or ranges overlapping the old declaration) is
rejected while the old
+ // consumer is still attached.
+ existing.handle((c, ex) -> c)
+ .thenCompose(c -> c != null ? c.closeAsync()
+ :
CompletableFuture.completedFuture(null))
+ .thenCompose(__ ->
createSegmentConsumerAsync(seg))));
+ } else {
+ futures.add(segmentConsumers.computeIfAbsent(seg.segmentId(),
+ id -> createSegmentConsumerAsync(seg)));
}
- futures.add(segmentConsumers.computeIfAbsent(seg.segmentId(),
- id -> createSegmentConsumerAsync(seg)));
}
log.info().attr("segments", assignedIds).log("Stream consumer
assignment applied");
@@ -373,8 +546,18 @@ final class ScalableStreamConsumer<T>
}
segConf.setSubscriptionType(SubscriptionType.Key_Shared);
segConf.setKeySharedPolicy(KeySharedPolicy.stickyHashRange().ranges(ranges));
+ // Route whole entries by their producer-stamped entry-bucket
range on this subscription
+ // (the gate that keeps plain Key_Shared subscriptions dispatching
by key).
+ segConf.setEntryBucketDispatch(true);
}
segmentBucketRanges.put(segment.segmentId(), ownedBucketRanges);
+ if (ownedBucketRanges.isEmpty()) {
+ sharedSegmentUnacked.remove(segment.segmentId());
+ } else {
+ // Key_Shared consumers cannot ack cumulatively: track delivered
ids so a cumulative ack
+ // can be translated into individual acks (see
sharedSegmentUnacked).
+ sharedSegmentUnacked.put(segment.segmentId(), new
ConcurrentLinkedQueue<>());
+ }
// 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.
@@ -405,6 +588,13 @@ final class ScalableStreamConsumer<T>
// Update the latest delivered position for this segment
latestDelivered.put(segmentId, v4Msg.getMessageId());
+ // PIP-486 bucket-shared segment: remember the id so a cumulative
ack can be translated
+ // into individual acks (Key_Shared consumers cannot ack
cumulatively).
+ var unacked = sharedSegmentUnacked.get(segmentId);
+ if (unacked != null) {
+ unacked.add(v4Msg.getMessageId());
+ }
+
// Snapshot the position vector (all segments, including this one)
Map<Long, org.apache.pulsar.client.api.MessageId> positionVector =
new HashMap<>(latestDelivered);
@@ -436,6 +626,7 @@ final class ScalableStreamConsumer<T>
log.info().attr("segmentId", segmentId)
.log("Sealed segment drained, closing v4 consumer");
segmentConsumers.remove(segmentId);
+ sharedSegmentUnacked.remove(segmentId);
latestDelivered.remove(segmentId);
v4Consumer.closeAsync();
return null;
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
index cf5f2a44527a..4b6f9660594c 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -943,7 +943,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T>
implements ConnectionHandle
InitialPosition.valueOf(subscriptionInitialPosition.getValue()),
startMessageRollbackDuration, si,
createTopicIfDoesNotExist, conf.getKeySharedPolicy(),
// Use the current epoch to subscribe.
- conf.getSubscriptionProperties(),
CONSUMER_EPOCH.get(this));
+ conf.getSubscriptionProperties(),
CONSUMER_EPOCH.get(this), conf.isEntryBucketDispatch());
cnx.sendRequestWithId(request, requestId).thenRun(() -> {
synchronized (ConsumerImpl.this) {
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java
index f388b726d323..9d3e07dc7491 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java
@@ -406,6 +406,14 @@ public class ConsumerConfigurationData<T> implements
Serializable, Cloneable {
@JsonIgnore
private KeySharedPolicy keySharedPolicy;
+ /**
+ * PIP-486 (internal): the Key_Shared subscription dispatches whole
entries by their
+ * producer-stamped entry-bucket hash range instead of hashing each
message's key. Set only by
+ * the scalable-topic consumer when it shares a segment by entry-bucket.
+ */
+ @JsonIgnore
+ private boolean entryBucketDispatch = false;
+
private boolean batchIndexAckEnabled = true;
private boolean ackReceiptEnabled = false;
diff --git
a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
index a9626a3b4cb6..575664e8b093 100644
---
a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
+++
b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
@@ -650,6 +650,18 @@ public class Commands {
InitialPosition subscriptionInitialPosition, long
startMessageRollbackDurationInSec,
SchemaInfo schemaInfo, boolean createTopicIfDoesNotExist,
KeySharedPolicy keySharedPolicy,
Map<String, String> subscriptionProperties, long consumerEpoch)
{
+ return newSubscribe(topic, subscription, consumerId, requestId,
subType, priorityLevel, consumerName,
+ isDurable, startMessageId, metadata, readCompacted,
isReplicated, subscriptionInitialPosition,
+ startMessageRollbackDurationInSec, schemaInfo,
createTopicIfDoesNotExist, keySharedPolicy,
+ subscriptionProperties, consumerEpoch, false /*
entryBucketDispatch */);
+ }
+
+ public static ByteBuf newSubscribe(String topic, String subscription, long
consumerId, long requestId,
+ SubType subType, int priorityLevel, String consumerName,
boolean isDurable, MessageIdData startMessageId,
+ Map<String, String> metadata, boolean readCompacted, Boolean
isReplicated,
+ InitialPosition subscriptionInitialPosition, long
startMessageRollbackDurationInSec,
+ SchemaInfo schemaInfo, boolean createTopicIfDoesNotExist,
KeySharedPolicy keySharedPolicy,
+ Map<String, String> subscriptionProperties, long consumerEpoch,
boolean entryBucketDispatch) {
BaseCommand cmd = localCmd(Type.SUBSCRIBE);
CommandSubscribe subscribe = cmd.setSubscribe()
.setTopic(topic)
@@ -683,6 +695,9 @@ public class Commands {
KeySharedMeta keySharedMeta = subscribe.setKeySharedMeta();
keySharedMeta.setAllowOutOfOrderDelivery(keySharedPolicy.isAllowOutOfOrderDelivery());
keySharedMeta.setKeySharedMode(convertKeySharedMode(keySharedPolicy.getKeySharedMode()));
+ if (entryBucketDispatch) {
+ keySharedMeta.setEntryBucketDispatch(true);
+ }
if (keySharedPolicy instanceof
KeySharedPolicy.KeySharedPolicySticky) {
List<Range> ranges = ((KeySharedPolicy.KeySharedPolicySticky)
keySharedPolicy)
diff --git a/pulsar-common/src/main/proto/PulsarApi.proto
b/pulsar-common/src/main/proto/PulsarApi.proto
index 529c806d84d4..874ac9713ba5 100644
--- a/pulsar-common/src/main/proto/PulsarApi.proto
+++ b/pulsar-common/src/main/proto/PulsarApi.proto
@@ -361,6 +361,11 @@ message KeySharedMeta {
required KeySharedMode keySharedMode = 1;
repeated IntRange hashRanges = 3;
optional bool allowOutOfOrderDelivery = 4 [default = false];
+ // PIP-486: the subscription dispatches whole entries by their
producer-stamped entry-bucket
+ // hash range (entry_hash_min) instead of hashing each message's key. Set
only by scalable-topic
+ // consumers sharing a segment by entry-bucket; plain Key_Shared
subscriptions leave it unset,
+ // so their (possibly stamped) entries keep dispatching by key.
+ optional bool entryBucketDispatch = 5 [default = false];
}
message CommandSubscribe {
