(pulsar) branch master updated: [fix][broker] Never share a segment between members of a checkpoint consumer group (#26836)

[mm...@apache.org]

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 813c512ddaf3 [fix][broker] Never share a segment between members of a checkpoint consumer group (#26836)
813c512ddaf3 is described below

commit 813c512ddaf3da61cab622bc27dc049723514013
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Wed Oct 7 17:23:10 2026 -0700

    [fix][broker] Never share a segment between members of a checkpoint consumer group (#26836)
---
 .../scalable/ScalableTopicControllerTest.java      | 26 ++++++++++++++++++++++
 1 file changed, 26 insertions(+)

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerTest.java
index 8fd94517428a..028894e03d27 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerTest.java
@@ -333,6 +333,32 @@ public class ScalableTopicControllerTest {
         assertEquals(resources.listConsumersAsync(topicName, "sub-a").get().size(), 1);
     }
 
+    @Test
+    public void testRestoredCheckpointGroupGetsWholeSegments() throws Exception {
+        // One segment with N = 4 entry-buckets and three persisted members of a checkpoint group.
+        // After a controller failover the members are restored without their consumer type, which
+        // isn't persisted: the first one to reconnect must still get the segment whole.
+        TopicName oneSegment = TopicName.get("topic://tenant/ns/one-segment");
+        resources.createScalableTopicAsync(oneSegment,
+                ScalableTopicController.createInitialMetadata(1, 4, Map.of())).get();
+        resources.createSubscriptionAsync(oneSegment, "group", SubscriptionType.STREAM).get();
+        for (String member : List.of("member-1", "member-2", "member-3")) {
+            resources.registerConsumerAsync(oneSegment, "group", member).get();
+        }
+        ScalableTopicController restored = newController(oneSegment);
+        try {
+            restored.initialize().get();
+            ConsumerAssignment assignment = restored.registerConsumer("group", "member-1", 1L,
+                    ScalableConsumerType.CHECKPOINT, mock(TransportCnx.class)).get();
+
+            assertEquals(assignment.assignedSegments().size(), 1);
+            assertTrue(assignment.assignedSegments().get(0).bucketRanges().isEmpty(),
+                    "a checkpoint member must get the segment whole, not a share of its buckets");
+        } finally {
+            restored.close().join();
+        }
+    }
+
     @Test
     public void testUnregisterConsumerDeletesPersistedEntry() throws Exception {
         controller.initialize().get();