You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/10/12 19:18:33 UTC
(pulsar) 08/27: [fix][broker] Hold a scalable-topic segment until every ancestor is drained (#26850)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit e5b0a8dbd6e4922128225f8ab6565e5be8600ea1
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Sat Oct 10 19:52:43 2026 -0700
[fix][broker] Hold a scalable-topic segment until every ancestor is drained (#26850)
(cherry picked from commit be41491e7d9e5ead75ce53736a9b9bfaedfe7c93)
---
.../service/scalable/SubscriptionCoordinator.java | 75 ++++----
.../scalable/SubscriptionCoordinatorTest.java | 79 +++++++++
.../api/v5/V5StreamConsumerOrderingTest.java | 192 +++++++++++++++++++++
3 files changed, 313 insertions(+), 33 deletions(-)
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 437203e1f626..73950aba78ec 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
@@ -20,10 +20,12 @@ package org.apache.pulsar.broker.service.scalable;
import io.github.merlimat.slog.Logger;
import java.time.Duration;
+import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Comparator;
+import java.util.Deque;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
@@ -103,9 +105,9 @@ public class SubscriptionCoordinator {
/**
* Sealed segments confirmed drained for this subscription (cursor at end). In-memory
* only — on controller-leader failover the new leader rediscovers drain status by
- * polling. {@link #computeAssignment} consults this set to decide which active
- * children of a split / merge are eligible to be assigned (children stay blocked
- * until <em>every</em> sealed parent is drained, so message order isn't broken).
+ * polling. {@link #computeAssignment} consults this set to decide which segments
+ * of a split / merge are eligible to be assigned (a segment stays blocked until
+ * <em>every</em> ancestor is drained, so message order isn't broken).
*/
private final Set<Long> drainedSegmentIds = ConcurrentHashMap.newKeySet();
@@ -385,15 +387,17 @@ public class SubscriptionCoordinator {
// --- Drain tracking ---
/**
- * A segment is assignable to consumers when:
- * <ul>
- * <li>it's sealed — there's no harm in always handing it out (the v4 layer drains or
- * sees {@code TopicTerminated} immediately if already drained); or</li>
- * <li>it's active <b>and</b> every parent in the current layout has been drained for
- * this subscription. Without the parent-drain check we'd hand a consumer the
- * child of a just-split segment immediately, breaking per-key ordering against
- * any unread messages still sitting in the parent.</li>
- * </ul>
+ * A segment, sealed or active, is assignable to consumers once every ancestor in the
+ * current layout has been drained for this subscription. The ancestors hold the older
+ * messages of the segment's keys: handing the segment out earlier would break per-key
+ * ordering against them. Every ancestor, not just the parents: a parent can be drained
+ * while its own parent is not (an empty segment has no backlog), and the segment's keys
+ * can still have unread messages there. An ancestor missing from the layout was pruned,
+ * so its data is gone, and so are its own ancestors (the GC prunes ancestors first).
+ *
+ * <p>A sealed segment stops being assignable once it is drained itself: it carries no
+ * traffic, and keeping it would pin consumers to a dead segment forever (the drain
+ * rebalance would spread the group across it and its successor).
*
* <p>If no {@link SegmentDrainChecker} was configured (e.g., the simple test
* constructor), the parent-drain ordering is disabled and every segment is treated
@@ -403,20 +407,23 @@ public class SubscriptionCoordinator {
if (drainChecker == null) {
return true;
}
- if (!segment.isActive()) {
- // A sealed segment is assignable only while it still has backlog to drain. Once
- // fully drained it carries no traffic, and keeping it assignable would pin
- // consumers to a dead segment forever (the drain rebalance would spread the
- // group across it and its successor).
- return !drainedSegmentIds.contains(segment.segmentId());
+ if (!segment.isActive() && drainedSegmentIds.contains(segment.segmentId())) {
+ return false;
}
- for (long parentId : segment.parentIds()) {
- // A parent that's no longer in the DAG has been pruned (its data is gone), so
- // treat it as drained — there's nothing to wait on.
- if (layout.getAllSegments().containsKey(parentId)
- && !drainedSegmentIds.contains(parentId)) {
+ Map<Long, SegmentInfo> segments = layout.getAllSegments();
+ Set<Long> visited = new HashSet<>();
+ Deque<Long> ancestors = new ArrayDeque<>(segment.parentIds());
+ while (!ancestors.isEmpty()) {
+ long ancestorId = ancestors.pop();
+ SegmentInfo ancestor = segments.get(ancestorId);
+ if (ancestor == null || !visited.add(ancestorId)) {
+ // Pruned, or already reached through another parent (after a merge).
+ continue;
+ }
+ if (!drainedSegmentIds.contains(ancestorId)) {
return false;
}
+ ancestors.addAll(ancestor.parentIds());
}
return true;
}
@@ -602,17 +609,19 @@ public class SubscriptionCoordinator {
* 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,
- * and those may live on segments that have since been sealed by a split / merge.
+ * <p><b>DAG replay.</b> The assignment includes the <em>sealed</em> segments of the
+ * DAG that still have backlog. A fresh EARLIEST subscription needs to read messages
+ * produced before it joined, and those may live on segments that have since been
+ * sealed by a split / merge. They are read one generation at a time, by the
+ * parent-drain ordering below.
*
- * <p><b>Parent-drain ordering.</b> An <em>active</em> child segment is only assigned
- * once <em>every</em> parent in the DAG has been drained for this subscription
- * (tracked in {@link #drainedSegmentIds}). Without this guard a consumer would be
- * handed an active child immediately after a split and start receiving new messages
- * for some key while the same key's pre-split messages still sit unread on the sealed
- * parent — breaking per-key ordering. Initial active segments (those with no parents
- * in the layout) are unaffected and assigned right away.
+ * <p><b>Parent-drain ordering.</b> A segment, sealed or active, is only assigned once
+ * <em>every</em> ancestor in the DAG has been drained for this subscription (tracked
+ * in {@link #drainedSegmentIds}). Without this guard a consumer would be handed a
+ * child immediately after a split and start receiving new messages for some key while
+ * the same key's pre-split messages still sit unread on the sealed parent — breaking
+ * per-key ordering. Initial segments (those with no ancestors in the layout) are
+ * unaffected and assigned right away.
*
* <p>The client side (per-segment v4 consumer) drains a sealed-but-still-present
* segment naturally and closes it on {@code TopicTerminated}; a sealed-and-already-
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 be2443a0ea86..6a7cfd567752 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
@@ -203,6 +203,74 @@ public class SubscriptionCoordinatorTest {
}
}
+ /**
+ * A sealed segment is held back like an active one: after 0 splits into 4 and 5 and 5
+ * splits again into 6 and 7, a subscription still behind on 0 must not read 5, which
+ * holds newer messages of the same keys.
+ */
+ @Test
+ public void testSealedChildBlockedUntilParentDrained() throws Exception {
+ SubscriptionCoordinator orderedCoordinator = coordinatorDrainedOnlyByTest();
+ try {
+ orderedCoordinator.registerConsumer("consumer-1", 1L, mock(TransportCnx.class)).get();
+ orderedCoordinator.onLayoutChange(initialLayout.splitSegment(0, 0L).splitSegment(5, 0L)).get();
+ assertEquals(assignedTo(orderedCoordinator, "consumer-1"), Set.of(0L, 1L, 2L, 3L),
+ "only the root of the split lineage may be read");
+
+ orderedCoordinator.markSegmentsDrained(Set.of(0L));
+ assertEquals(assignedTo(orderedCoordinator, "consumer-1"), Set.of(1L, 2L, 3L, 4L, 5L),
+ "0's children are read once 0 is drained, 5's children still wait");
+
+ orderedCoordinator.markSegmentsDrained(Set.of(5L));
+ assertEquals(assignedTo(orderedCoordinator, "consumer-1"), Set.of(1L, 2L, 3L, 4L, 6L, 7L));
+ } finally {
+ orderedCoordinator.close();
+ }
+ }
+
+ /**
+ * A segment that received nothing before it was split is drained as soon as it is sealed,
+ * while its own parent can still have backlog: its children must keep waiting for that
+ * parent, not just for the empty segment.
+ */
+ @Test
+ public void testGrandchildrenBlockedUntilRootDrained() throws Exception {
+ SubscriptionCoordinator orderedCoordinator = coordinatorDrainedOnlyByTest();
+ try {
+ orderedCoordinator.registerConsumer("consumer-1", 1L, mock(TransportCnx.class)).get();
+ orderedCoordinator.onLayoutChange(initialLayout.splitSegment(0, 0L).splitSegment(5, 0L)).get();
+
+ orderedCoordinator.markSegmentsDrained(Set.of(5L));
+ assertEquals(assignedTo(orderedCoordinator, "consumer-1"), Set.of(0L, 1L, 2L, 3L),
+ "5's children must wait for 0, not just for 5");
+
+ orderedCoordinator.markSegmentsDrained(Set.of(0L));
+ assertEquals(assignedTo(orderedCoordinator, "consumer-1"), Set.of(1L, 2L, 3L, 4L, 6L, 7L));
+ } finally {
+ orderedCoordinator.close();
+ }
+ }
+
+ /** A merged segment waits for the ancestors of both segments it merges. */
+ @Test
+ public void testMergedChildBlockedUntilGrandparentDrained() throws Exception {
+ SubscriptionCoordinator orderedCoordinator = coordinatorDrainedOnlyByTest();
+ try {
+ orderedCoordinator.registerConsumer("consumer-1", 1L, mock(TransportCnx.class)).get();
+ // 0 splits into 4 and 5, which merge back into 6.
+ orderedCoordinator.onLayoutChange(initialLayout.splitSegment(0, 0L).mergeSegments(4, 5, 0L)).get();
+
+ orderedCoordinator.markSegmentsDrained(Set.of(4L, 5L));
+ assertEquals(assignedTo(orderedCoordinator, "consumer-1"), Set.of(0L, 1L, 2L, 3L),
+ "6 must wait for 0, not just for 4 and 5");
+
+ orderedCoordinator.markSegmentsDrained(Set.of(0L));
+ assertEquals(assignedTo(orderedCoordinator, "consumer-1"), Set.of(1L, 2L, 3L, 6L));
+ } finally {
+ orderedCoordinator.close();
+ }
+ }
+
/**
* After the drain poller has backed off, a fresh consumer registration must cancel
* the long-delay scheduled task and re-arm at the initial delay — otherwise the new
@@ -656,6 +724,17 @@ public class SubscriptionCoordinatorTest {
assertEquals(coordinator.getConsumers().size(), 0);
}
+ /** A coordinator with parent-drain ordering, whose segments only drain through markSegmentsDrained. */
+ private SubscriptionCoordinator coordinatorDrainedOnlyByTest() {
+ return new SubscriptionCoordinator("test-sub", topicName, initialLayout, resources, scheduler,
+ Duration.ofMillis(200), (segment, sub) -> CompletableFuture.completedFuture(false),
+ Duration.ofMillis(50), Duration.ofSeconds(5));
+ }
+
+ private static Set<Long> assignedTo(SubscriptionCoordinator c, String consumerName) {
+ return new HashSet<>(segmentIds(findByName(c.currentAssignment(), consumerName)));
+ }
+
private static ConsumerAssignment findByName(Map<ConsumerSession, ConsumerAssignment> m, String name) {
return m.entrySet().stream()
.filter(e -> name.equals(e.getKey().getConsumerName()))
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5StreamConsumerOrderingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5StreamConsumerOrderingTest.java
new file mode 100644
index 000000000000..d23f2d8a4817
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5StreamConsumerOrderingTest.java
@@ -0,0 +1,192 @@
+/*
+ * 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 static org.testng.Assert.assertTrue;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+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.awaitility.Awaitility;
+import org.testng.annotations.Test;
+
+/**
+ * Per-key ordering of a {@link StreamConsumer} across a lineage of splits: a key's messages on a
+ * segment must all be delivered before its messages on the segment's descendants, however many
+ * generations apart.
+ */
+public class V5StreamConsumerOrderingTest extends V5ClientBaseTest {
+
+ private static final int KEYS = 16;
+
+ /**
+ * A subscription created after two splits replays both sealed generations: it must not read
+ * the second one while the first still has backlog, since it holds newer messages of the same
+ * keys.
+ */
+ @Test
+ public void testPerKeyOrderAcrossTwoSplits() throws Exception {
+ String topic = newTopic();
+ @Cleanup
+ Producer<String> producer = newProducer(topic);
+
+ Map<String, List<String>> sent = new HashMap<>();
+ long root = activeSegmentIds(topic).get(0);
+ publish(producer, "gen0", 50, sent);
+ split(topic, root, 2);
+ publish(producer, "gen1", 50, sent);
+ split(topic, activeSegmentIds(topic).get(0), 3);
+ publish(producer, "gen2", 50, sent);
+
+ @Cleanup
+ StreamConsumer<String> consumer = newConsumer(topic);
+ // Nothing is received yet, so the root keeps its backlog.
+ assertNotReadWhileRootHasBacklog(topic, root);
+ assertPerKeyOrder(receive(consumer, total(sent)), sent);
+ }
+
+ /**
+ * A segment split before anything was published to it has no backlog, so it is drained right
+ * away while its parent can still have backlog: its children must wait for the parent too.
+ */
+ @Test
+ public void testPerKeyOrderAcrossAnEmptySegment() throws Exception {
+ String topic = newTopic();
+ @Cleanup
+ Producer<String> producer = newProducer(topic);
+ @Cleanup
+ StreamConsumer<String> consumer = newConsumer(topic);
+
+ Map<String, List<String>> sent = new HashMap<>();
+ long root = activeSegmentIds(topic).get(0);
+ publish(producer, "gen0", 50, sent);
+ split(topic, root, 2);
+ split(topic, activeSegmentIds(topic).get(0), 3);
+ publish(producer, "gen2", 50, sent);
+
+ assertNotReadWhileRootHasBacklog(topic, root);
+ assertPerKeyOrder(receive(consumer, total(sent)), sent);
+ }
+
+ // --- Helpers ---
+
+ private String newTopic() throws Exception {
+ String topic = newScalableTopic(1);
+ // Only the admin splits below may change the layout.
+ admin.scalableTopics().setAutoScalePolicy(topic,
+ AutoScalePolicyOverride.builder().enabled(false).build());
+ return topic;
+ }
+
+ private Producer<String> newProducer(String topic) throws Exception {
+ return v5Client.newProducer(Schema.string())
+ .topic(topic)
+ .create();
+ }
+
+ private StreamConsumer<String> newConsumer(String topic) throws Exception {
+ return v5Client.newStreamConsumer(Schema.string())
+ .topic(topic)
+ .subscriptionName("sub")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST)
+ .subscribe();
+ }
+
+ /** Publish {@code perKey} rounds of one message per key, keys interleaved. */
+ private static void publish(Producer<String> producer, String prefix, int perKey,
+ Map<String, List<String>> sent) {
+ List<CompletableFuture<MessageId>> sends = new ArrayList<>();
+ for (int i = 0; i < perKey; i++) {
+ for (int k = 0; k < KEYS; k++) {
+ String key = "key-" + k;
+ String value = key + "-" + prefix + "-" + i;
+ sends.add(producer.async().newMessage().key(key).value(value).send());
+ sent.computeIfAbsent(key, __ -> new ArrayList<>()).add(value);
+ }
+ }
+ CompletableFuture.allOf(sends.toArray(CompletableFuture[]::new)).join();
+ }
+
+ private static Map<String, List<String>> receive(StreamConsumer<String> consumer, int count)
+ throws Exception {
+ Map<String, List<String>> received = new HashMap<>();
+ for (int i = 0; i < count; i++) {
+ // A segment is read only once the controller's drain poll, which backs off, sees its
+ // ancestors drained.
+ Message<String> msg = consumer.receive(Duration.ofSeconds(30));
+ assertNotNull(msg, "missed message #" + i + " of " + count);
+ received.computeIfAbsent(msg.key().orElseThrow(), __ -> new ArrayList<>()).add(msg.value());
+ // A sealed segment is drained once the subscription has no backlog on it.
+ consumer.acknowledgeCumulative(msg.id());
+ }
+ return received;
+ }
+
+ private static int total(Map<String, List<String>> messages) {
+ return messages.values().stream().mapToInt(List::size).sum();
+ }
+
+ private static void assertPerKeyOrder(Map<String, List<String>> received, Map<String, List<String>> sent) {
+ for (var entry : sent.entrySet()) {
+ assertEquals(received.getOrDefault(entry.getKey(), List.of()), entry.getValue(),
+ "per-key order for " + entry.getKey());
+ }
+ }
+
+ /** For a while, no consumer reads any descendant of the root, which still has backlog. */
+ private void assertNotReadWhileRootHasBacklog(String topic, long root) throws Exception {
+ Set<Long> descendants = new HashSet<>(admin.scalableTopics().getMetadata(topic).getSegments().keySet());
+ descendants.remove(root);
+ // Long enough for the controller's first drain polls.
+ Awaitility.await().during(Duration.ofSeconds(5)).atMost(Duration.ofSeconds(10)).untilAsserted(() -> {
+ for (var subscription : admin.scalableTopics().getStats(topic).getSubscriptions().values()) {
+ for (long segmentId : descendants) {
+ var onSegment = subscription.getSegments().get(segmentId);
+ assertTrue(onSegment == null || onSegment.getConsumerCount() == 0,
+ "segment " + segmentId + " is read while the root still has backlog");
+ }
+ }
+ });
+ }
+
+ private void split(String topic, long segmentId, int expectedActive) throws Exception {
+ admin.scalableTopics().splitSegment(topic, segmentId);
+ Awaitility.await().untilAsserted(() -> assertEquals(activeSegmentIds(topic).size(), expectedActive));
+ }
+
+ private List<Long> activeSegmentIds(String topic) throws Exception {
+ List<Long> ids = new ArrayList<>();
+ for (var seg : admin.scalableTopics().getMetadata(topic).getSegments().values()) {
+ if (seg.isActive()) {
+ ids.add(seg.getSegmentId());
+ }
+ }
+ return ids;
+ }
+}