You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2026/10/10 22:50:14 UTC
(pulsar) branch master updated: [fix][client] Retry failed initial topic attaches and drop topics removed while attaching in V5 namespace consumers (#26848)
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 3adcccb583ea [fix][client] Retry failed initial topic attaches and drop topics removed while attaching in V5 namespace consumers (#26848)
3adcccb583ea is described below
commit 3adcccb583ea94466c017861512a27d6867e8591
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Sat Oct 10 15:50:07 2026 -0700
[fix][client] Retry failed initial topic attaches and drop topics removed while attaching in V5 namespace consumers (#26848)
---
.../v5/V5NamespaceConsumerTopicAttachTest.java | 441 +++++++++++++++++++++
.../client/impl/v5/MultiTopicQueueConsumer.java | 127 +++---
.../client/impl/v5/MultiTopicStreamConsumer.java | 108 +++--
3 files changed, 579 insertions(+), 97 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5NamespaceConsumerTopicAttachTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5NamespaceConsumerTopicAttachTest.java
new file mode 100644
index 000000000000..c13a0f8a1615
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5NamespaceConsumerTopicAttachTest.java
@@ -0,0 +1,441 @@
+/*
+ * 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.impl.v5;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import io.netty.channel.EventLoopGroup;
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.time.Duration;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicInteger;
+import lombok.Cleanup;
+import lombok.CustomLog;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.v5.Message;
+import org.apache.pulsar.client.api.v5.Producer;
+import org.apache.pulsar.client.api.v5.QueueConsumer;
+import org.apache.pulsar.client.api.v5.StreamConsumer;
+import org.apache.pulsar.client.api.v5.V5ClientBaseTest;
+import org.apache.pulsar.client.api.v5.config.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.v5.schema.Schema;
+import org.apache.pulsar.client.impl.ClientCnx;
+import org.apache.pulsar.client.impl.ConnectionPool;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.ScalableTopicsWatcherSession;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.apache.pulsar.client.impl.metrics.InstrumentProvider;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.scalable.SegmentTopicName;
+import org.apache.pulsar.common.util.netty.EventLoopUtil;
+import org.awaitility.Awaitility;
+import org.testng.annotations.Test;
+
+/**
+ * How a namespace consumer attaches the topics of its matching set when attaching one doesn't simply
+ * succeed:
+ * <ul>
+ * <li>a topic whose first attach fails is retried in the background and attached once it succeeds,
+ * including a topic of the initial snapshot;</li>
+ * <li>a topic that leaves the matching set while it is being attached, because its properties stop
+ * matching the filter or because it is deleted, is not attached once the attach completes, and is
+ * not retried.</li>
+ * </ul>
+ *
+ * <p>The tests inject the faults into the segment subscriptions of the v4 client the consumer runs on.
+ */
+public class V5NamespaceConsumerTopicAttachTest extends V5ClientBaseTest {
+
+ /**
+ * A v4 client that fails the next segment subscription to the topics the test picks, or holds them back
+ * until the test allows them. It records the segment subscriptions and lookups of each scalable topic,
+ * and keeps the namespace watchers registered on its connections.
+ */
+ @CustomLog
+ private static final class FaultyClient extends PulsarClientImpl {
+
+ private final List<ScalableTopicsWatcher> watchers;
+ private final Set<String> failingTopics = ConcurrentHashMap.newKeySet();
+ private final Set<String> heldTopics = ConcurrentHashMap.newKeySet();
+ private final CompletableFuture<Void> attachAllowed = new CompletableFuture<>();
+ private final Map<String, AtomicInteger> attachRequests = new ConcurrentHashMap<>();
+ private final Map<String, AtomicInteger> settledAttaches = new ConcurrentHashMap<>();
+ private final Map<String, List<Consumer<?>>> segmentConsumers = new ConcurrentHashMap<>();
+ private final Map<String, AtomicInteger> lookups = new ConcurrentHashMap<>();
+
+ static FaultyClient create(String serviceUrl) throws Exception {
+ ClientConfigurationData conf = new ClientConfigurationData();
+ conf.setServiceUrl(serviceUrl);
+ conf.setStatsIntervalSeconds(0);
+ EventLoopGroup eventLoopGroup =
+ EventLoopUtil.newEventLoopGroup(1, false, new DefaultThreadFactory("faulty-client-io"));
+ List<ScalableTopicsWatcher> watchers = new CopyOnWriteArrayList<>();
+ ConnectionPool connectionPool = new ConnectionPool(InstrumentProvider.NOOP, conf, eventLoopGroup,
+ () -> new ClientCnx(InstrumentProvider.NOOP, conf, eventLoopGroup) {
+ @Override
+ public void registerScalableTopicsWatcher(long watchId,
+ ScalableTopicsWatcherSession watcher) {
+ watchers.add((ScalableTopicsWatcher) watcher);
+ super.registerScalableTopicsWatcher(watchId, watcher);
+ }
+ }, null);
+ return new FaultyClient(conf, eventLoopGroup, connectionPool, watchers);
+ }
+
+ private FaultyClient(ClientConfigurationData conf, EventLoopGroup eventLoopGroup,
+ ConnectionPool connectionPool, List<ScalableTopicsWatcher> watchers)
+ throws Exception {
+ super(conf, eventLoopGroup, connectionPool);
+ this.watchers = watchers;
+ }
+
+ void failNextAttach(String topic) {
+ failingTopics.add(topic);
+ }
+
+ void hold(String topic) {
+ heldTopics.add(topic);
+ }
+
+ void allowAttach() {
+ attachAllowed.complete(null);
+ }
+
+ /** Fail the segment subscriptions held back so far, and any later one to a held topic. */
+ void failHeldAttaches() {
+ attachAllowed.completeExceptionally(
+ new PulsarClientException.ConnectException("Injected segment subscribe failure"));
+ }
+
+ /** How many segment subscriptions to the topic were requested, whether held, failed or not. */
+ int attachRequests(String topic) {
+ return count(attachRequests, topic);
+ }
+
+ /** How many segment subscriptions to the topic completed, successfully or not. */
+ int settledAttaches(String topic) {
+ return count(settledAttaches, topic);
+ }
+
+ /** How many times the client looked up the scalable topic. */
+ int lookups(String topic) {
+ return count(lookups, topic);
+ }
+
+ /** The scalable topics with a segment subscription still open. */
+ Set<String> attachedTopics() {
+ Set<String> attached = new HashSet<>();
+ segmentConsumers.forEach((topic, consumers) -> {
+ if (consumers.stream().anyMatch(Consumer::isConnected)) {
+ attached.add(topic);
+ }
+ });
+ return attached;
+ }
+
+ /** The watcher of the namespace consumer: the only one this client opens. */
+ ScalableTopicsWatcher watcher() {
+ Awaitility.await().until(() -> !watchers.isEmpty());
+ return watchers.get(0);
+ }
+
+ private static int count(Map<String, AtomicInteger> counters, String topic) {
+ AtomicInteger counter = counters.get(topic);
+ return counter == null ? 0 : counter.get();
+ }
+
+ private static void increment(Map<String, AtomicInteger> counters, String topic) {
+ counters.computeIfAbsent(topic, __ -> new AtomicInteger()).incrementAndGet();
+ }
+
+ // The v4 Schema is qualified: the V5 one is imported.
+ @Override
+ public <T> CompletableFuture<Consumer<T>> subscribeSegmentAsync(
+ ConsumerConfigurationData<T> conf, org.apache.pulsar.client.api.Schema<T> schema) {
+ String topic = SegmentTopicName.getParentTopicName(TopicName.get(conf.getSingleTopic())).toString();
+ increment(attachRequests, topic);
+ CompletableFuture<Consumer<T>> attach;
+ if (failingTopics.remove(topic)) {
+ attach = CompletableFuture.failedFuture(
+ new PulsarClientException.ConnectException("Injected segment subscribe failure"));
+ } else {
+ CompletableFuture<Void> allowed =
+ heldTopics.contains(topic) ? attachAllowed : CompletableFuture.completedFuture(null);
+ attach = allowed.thenCompose(__ -> super.subscribeSegmentAsync(conf, schema))
+ .thenApply(consumer -> {
+ segmentConsumers.computeIfAbsent(topic, __ -> new CopyOnWriteArrayList<>())
+ .add(consumer);
+ return consumer;
+ });
+ }
+ return attach.whenComplete((__, ___) -> increment(settledAttaches, topic));
+ }
+
+ @Override
+ public CompletableFuture<ClientCnx> getConnection(String topic) {
+ increment(lookups, topic);
+ return super.getConnection(topic);
+ }
+
+ // The connection pool and its I/O threads were passed in, so the client doesn't close them.
+ @Override
+ public CompletableFuture<Void> closeAsync() {
+ return super.closeAsync().whenComplete((__, ___) -> {
+ try {
+ getCnxPool().close();
+ } catch (Exception e) {
+ log.warn().exception(e).log("Failed to close the connection pool");
+ }
+ eventLoopGroup.shutdownGracefully();
+ });
+ }
+ }
+
+ @Test
+ public void testQueueConsumerRetriesTopicWhoseFirstAttachFailed() throws Exception {
+ String healthy = newScalableTopic(1);
+ String failing = newScalableTopic(1);
+ PulsarClientV5 client = newFaultyClient();
+ FaultyClient faulty = (FaultyClient) client.v4Client();
+ faulty.failNextAttach(failing);
+
+ @Cleanup
+ QueueConsumer<String> consumer = client.newQueueConsumer(Schema.string())
+ .namespace(getNamespace())
+ .subscriptionName("sub")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST)
+ .subscribe();
+
+ Awaitility.await().untilAsserted(() ->
+ assertThat(((MultiTopicQueueConsumer<String>) consumer).attachedTopicsForTesting())
+ .as("topics the consumer is attached to")
+ .containsExactlyInAnyOrder(healthy, failing));
+ send(failing, "after-failed-attach");
+ Message<String> msg = consumer.receive(Duration.ofSeconds(10));
+ assertThat(msg).as("message sent to the topic whose first attach failed").isNotNull();
+ assertThat(msg.value()).isEqualTo("after-failed-attach");
+ }
+
+ @Test
+ public void testStreamConsumerRetriesTopicWhoseFirstAttachFailed() throws Exception {
+ String healthy = newScalableTopic(1);
+ String failing = newScalableTopic(1);
+ PulsarClientV5 client = newFaultyClient();
+ FaultyClient faulty = (FaultyClient) client.v4Client();
+ faulty.failNextAttach(failing);
+
+ @Cleanup
+ StreamConsumer<String> consumer = client.newStreamConsumer(Schema.string())
+ .namespace(getNamespace())
+ .subscriptionName("sub")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST)
+ .subscribe();
+
+ Awaitility.await().untilAsserted(() ->
+ assertThat(faulty.attachedTopics())
+ .as("topics the consumer is attached to")
+ .containsExactlyInAnyOrder(healthy, failing));
+ send(failing, "after-failed-attach");
+ Message<String> msg = consumer.receive(Duration.ofSeconds(10));
+ assertThat(msg).as("message sent to the topic whose first attach failed").isNotNull();
+ assertThat(msg.value()).isEqualTo("after-failed-attach");
+ }
+
+ @Test
+ public void testQueueConsumerDropsTopicThatLeftTheFilterWhileAttaching() throws Exception {
+ Map<String, String> filter = Map.of("team", "a");
+ String kept = newScalableTopic(filter);
+ PulsarClientV5 client = newFaultyClient();
+ FaultyClient faulty = (FaultyClient) client.v4Client();
+ @Cleanup
+ QueueConsumer<String> consumer = client.newQueueConsumer(Schema.string())
+ .namespace(getNamespace(), filter)
+ .subscriptionName("sub")
+ .subscribe();
+ MultiTopicQueueConsumer<String> namespaceConsumer = (MultiTopicQueueConsumer<String>) consumer;
+
+ String leaving = holdAttachOfNewTopic(faulty, filter);
+ setProperties(leaving, Map.of("team", "b"));
+ awaitRemovedFromMatchingSet(faulty, leaving);
+ faulty.allowAttach();
+ Awaitility.await().until(() -> faulty.settledAttaches(leaving) > 0);
+
+ Awaitility.await().atMost(15, SECONDS).during(3, SECONDS).untilAsserted(() -> {
+ assertThat(namespaceConsumer.attachedTopicsForTesting())
+ .as("topics the consumer is attached to")
+ .containsExactly(kept);
+ assertThat(faulty.attachedTopics())
+ .as("topics with a segment subscription open")
+ .containsExactly(kept);
+ });
+ }
+
+ @Test
+ public void testStreamConsumerDropsTopicThatLeftTheFilterWhileAttaching() throws Exception {
+ Map<String, String> filter = Map.of("team", "a");
+ String kept = newScalableTopic(filter);
+ PulsarClientV5 client = newFaultyClient();
+ FaultyClient faulty = (FaultyClient) client.v4Client();
+ @Cleanup
+ StreamConsumer<String> consumer = client.newStreamConsumer(Schema.string())
+ .namespace(getNamespace(), filter)
+ .subscriptionName("sub")
+ .subscribe();
+
+ String leaving = holdAttachOfNewTopic(faulty, filter);
+ setProperties(leaving, Map.of("team", "b"));
+ awaitRemovedFromMatchingSet(faulty, leaving);
+ faulty.allowAttach();
+ Awaitility.await().until(() -> faulty.settledAttaches(leaving) > 0);
+
+ Awaitility.await().atMost(15, SECONDS).during(3, SECONDS).untilAsserted(() ->
+ assertThat(faulty.attachedTopics())
+ .as("topics with a segment subscription open")
+ .containsExactly(kept));
+ }
+
+ @Test
+ public void testQueueConsumerStopsAttachingTopicDeletedWhileAttaching() throws Exception {
+ String kept = newScalableTopic(1);
+ PulsarClientV5 client = newFaultyClient();
+ FaultyClient faulty = (FaultyClient) client.v4Client();
+ @Cleanup
+ QueueConsumer<String> consumer = client.newQueueConsumer(Schema.string())
+ .namespace(getNamespace())
+ .subscriptionName("sub")
+ .subscribe();
+ MultiTopicQueueConsumer<String> namespaceConsumer = (MultiTopicQueueConsumer<String>) consumer;
+
+ String deleted = holdAttachOfNewTopic(faulty, Map.of());
+ admin.scalableTopics().deleteScalableTopic(deleted, true);
+ awaitRemovedFromMatchingSet(faulty, deleted);
+ int lookupsAtRemoval = faulty.lookups(deleted);
+ faulty.allowAttach();
+ Awaitility.await().until(() -> faulty.settledAttaches(deleted) > 0);
+
+ Awaitility.await().atMost(15, SECONDS).during(3, SECONDS).untilAsserted(() -> {
+ assertThat(faulty.lookups(deleted))
+ .as("lookups of the deleted topic since it was removed")
+ .isEqualTo(lookupsAtRemoval);
+ assertThat(namespaceConsumer.attachedTopicsForTesting())
+ .as("topics the consumer is attached to")
+ .containsExactly(kept);
+ assertThat(faulty.attachedTopics())
+ .as("topics with a segment subscription open")
+ .containsExactly(kept);
+ });
+ assertThatThrownBy(() -> admin.scalableTopics().getMetadata(deleted))
+ .as("the deleted topic stays deleted")
+ .isInstanceOf(PulsarAdminException.NotFoundException.class);
+ }
+
+ @Test
+ public void testStreamConsumerDoesNotRetryTopicThatLeftTheFilterWhileAttaching() throws Exception {
+ Map<String, String> filter = Map.of("team", "a");
+ String kept = newScalableTopic(filter);
+ PulsarClientV5 client = newFaultyClient();
+ FaultyClient faulty = (FaultyClient) client.v4Client();
+ @Cleanup
+ StreamConsumer<String> consumer = client.newStreamConsumer(Schema.string())
+ .namespace(getNamespace(), filter)
+ .subscriptionName("sub")
+ .subscribe();
+
+ String leaving = holdAttachOfNewTopic(faulty, filter);
+ setProperties(leaving, Map.of("team", "b"));
+ awaitRemovedFromMatchingSet(faulty, leaving);
+ int lookupsAtRemoval = faulty.lookups(leaving);
+ faulty.failHeldAttaches();
+ Awaitility.await().until(() -> faulty.settledAttaches(leaving) > 0);
+
+ Awaitility.await().atMost(15, SECONDS).during(3, SECONDS).untilAsserted(() -> {
+ assertThat(faulty.lookups(leaving))
+ .as("lookups of the topic since it left the matching set")
+ .isEqualTo(lookupsAtRemoval);
+ assertThat(faulty.attachedTopics())
+ .as("topics with a segment subscription open")
+ .containsExactly(kept);
+ });
+ }
+
+ // --- Helpers ---
+
+ private PulsarClientV5 newFaultyClient() throws Exception {
+ return track(new PulsarClientV5(FaultyClient.create(getBrokerServiceUrl()), "faulty-client", null));
+ }
+
+ private String newScalableTopicName() {
+ return "topic://" + getNamespace() + "/scalable-" + UUID.randomUUID().toString().substring(0, 8);
+ }
+
+ private String newScalableTopic(Map<String, String> properties) throws Exception {
+ String topic = newScalableTopicName();
+ admin.scalableTopics().createScalableTopic(topic, 1, properties);
+ return topic;
+ }
+
+ /**
+ * Create a topic in the consumer's matching set while holding back its segment subscriptions, and wait
+ * until the consumer is attaching it.
+ */
+ private String holdAttachOfNewTopic(FaultyClient faulty, Map<String, String> properties) throws Exception {
+ String topic = newScalableTopicName();
+ faulty.hold(topic);
+ admin.scalableTopics().createScalableTopic(topic, 1, properties);
+ Awaitility.await().until(() -> faulty.attachRequests(topic) > 0);
+ return topic;
+ }
+
+ /** The admin API can't change the properties of a scalable topic: change its metadata directly. */
+ private void setProperties(String topic, Map<String, String> properties) throws Exception {
+ getPulsar().getPulsarResources().getScalableTopicResources()
+ .updateScalableTopicAsync(TopicName.get(topic), metadata -> {
+ metadata.setProperties(properties);
+ return metadata;
+ })
+ .get(30, SECONDS);
+ }
+
+ /** Wait until the consumer's watcher has received the removal of the topic from the matching set. */
+ private static void awaitRemovedFromMatchingSet(FaultyClient faulty, String topic) {
+ ScalableTopicsWatcher watcher = faulty.watcher();
+ Awaitility.await().until(() -> !watcher.currentSetForTesting().contains(topic));
+ }
+
+ private void send(String topic, String value) throws Exception {
+ @Cleanup
+ Producer<String> producer = v5Client.newProducer(Schema.string())
+ .topic(topic)
+ .create();
+ producer.newMessage()
+ .value(value)
+ .send();
+ }
+}
diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicQueueConsumer.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicQueueConsumer.java
index a83c753437de..b808536d4e02 100644
--- a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicQueueConsumer.java
+++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicQueueConsumer.java
@@ -60,8 +60,8 @@ import org.apache.pulsar.common.naming.TopicName;
* <li>The watcher's {@code Snapshot} replaces the active set; {@code Diff}
* applies removals before additions to handle a rapid remove-then-add of the
* same topic name.</li>
- * <li>Per-topic add failures retry forever with exponential backoff (100 ms
- * initial, 30 min cap).</li>
+ * <li>Per-topic add failures retry with exponential backoff (100 ms initial,
+ * 30 min cap) until the topic attaches or leaves the matching set.</li>
* </ul>
*/
final class MultiTopicQueueConsumer<T> implements QueueConsumerImpl<T> {
@@ -83,6 +83,10 @@ final class MultiTopicQueueConsumer<T> implements QueueConsumerImpl<T> {
private final String subscriptionName;
private final ScalableTopicsWatcher watcher;
+ /**
+ * Every topic of the matching set, from when the consumer starts attaching it until it leaves the set:
+ * while it is being attached, once attached, and while a failed attach waits to be retried.
+ */
private final ConcurrentHashMap<String, PerTopicState<T>> perTopic = new ConcurrentHashMap<>();
private final V5ReceiveQueue<T> mux;
@@ -134,10 +138,10 @@ final class MultiTopicQueueConsumer<T> implements QueueConsumerImpl<T> {
}
/**
- * Open one per-topic consumer per topic in the initial snapshot. Block on every
- * future so {@code subscribeAsync} only resolves once the consumer is fully
- * attached — gives the user the same all-or-nothing semantics as the
- * single-topic builder.
+ * Open one per-topic consumer per topic in the initial snapshot. Wait for the first
+ * attempt on every topic so {@code subscribeAsync} only resolves once the consumer is
+ * fully attached, unless an attempt failed: that topic is then retried in the
+ * background, like a topic that joins the matching set later.
*/
private CompletableFuture<Void> openInitial(List<String> topics) {
if (topics.isEmpty()) {
@@ -145,24 +149,31 @@ final class MultiTopicQueueConsumer<T> implements QueueConsumerImpl<T> {
}
List<CompletableFuture<?>> opens = new ArrayList<>(topics.size());
for (String t : topics) {
- opens.add(openTopic(t, /* retry= */ false));
+ opens.add(openTopic(t));
}
return CompletableFuture.allOf(opens.toArray(CompletableFuture[]::new));
}
/**
- * Subscribe to one topic. When {@code retry} is true, failures schedule a
- * background retry with exponential backoff; the returned future completes as
- * soon as the first attempt finishes (success or failure) so we don't hold up
- * Snapshot / Diff processing.
+ * Subscribe to one topic. Failures schedule a background retry with exponential
+ * backoff until an attempt succeeds or the topic leaves the matching set; the
+ * returned future completes as soon as the first attempt finishes (success or
+ * failure) so we don't hold up Snapshot / Diff processing.
*/
- private CompletableFuture<Void> openTopic(String topicName, boolean retry) {
+ private CompletableFuture<Void> openTopic(String topicName) {
if (closed) {
return CompletableFuture.completedFuture(null);
}
- if (perTopic.containsKey(topicName)) {
+ PerTopicState<T> state = new PerTopicState<>();
+ if (perTopic.putIfAbsent(topicName, state) != null) {
+ // Already attached, being attached, or waiting to retry.
return CompletableFuture.completedFuture(null);
}
+ return attachTopic(topicName, state);
+ }
+
+ /** One attempt at subscribing to a topic; a failure schedules the next one. */
+ private CompletableFuture<Void> attachTopic(String topicName, PerTopicState<T> state) {
TopicName topic = V5Utils.parseScalableTopicInput(topicName);
// A namespace consumer only attaches to topics the watcher reports as existing; it must
// never auto-create one (so a deleted topic can't be resurrected by a reconnecting watch).
@@ -180,14 +191,9 @@ final class MultiTopicQueueConsumer<T> implements QueueConsumerImpl<T> {
.thenCompose(layout -> ScalableQueueConsumer.createAsyncImpl(
client, v5Schema, perTopicConf(topicName), dagWatch, layout, sink, null))
.thenAccept(qc -> {
- if (closed) {
- qc.closeAsync();
- return;
- }
- PerTopicState<T> state = new PerTopicState<>(qc);
- PerTopicState<T> existing = perTopic.putIfAbsent(topicName, state);
- if (existing != null) {
- // Concurrent open; drop the dup.
+ // closeTopic cancels the state when the topic leaves the matching set
+ // while this attempt is in flight: close what it opened.
+ if (closed || !state.consumer.complete(qc)) {
qc.closeAsync();
return;
}
@@ -196,8 +202,9 @@ final class MultiTopicQueueConsumer<T> implements QueueConsumerImpl<T> {
.exceptionally(ex -> {
Throwable cause = ex instanceof CompletionException ce && ce.getCause() != null
? ce.getCause() : ex;
- if (retry && !closed) {
- scheduleRetry(topicName);
+ // No retry once the topic has left the matching set: it may be gone for good.
+ if (!closed && !state.consumer.isDone()) {
+ scheduleRetry(topicName, state);
}
log.warn().attr("topic", topicName).exceptionMessage(cause)
.log("Per-topic subscribe failed");
@@ -205,21 +212,18 @@ final class MultiTopicQueueConsumer<T> implements QueueConsumerImpl<T> {
});
}
- private void scheduleRetry(String topicName) {
+ private void scheduleRetry(String topicName, PerTopicState<T> state) {
long delayMs = nextBackoff(topicName);
log.info().attr("topic", topicName).attr("delayMs", delayMs)
.log("Retrying per-topic subscribe after backoff");
- Timeout timeout = client.v4Client().timer().newTimeout(t -> {
- retryTimeouts.remove(topicName);
- openTopic(topicName, /* retry= */ true);
+ state.retry = client.v4Client().timer().newTimeout(t -> {
+ if (!closed && !state.consumer.isDone()) {
+ attachTopic(topicName, state);
+ }
}, delayMs, TimeUnit.MILLISECONDS);
- retryTimeouts.put(topicName, timeout);
}
private final ConcurrentHashMap<String, AtomicLong> retryDelays = new ConcurrentHashMap<>();
- /** Pending backoff-retry timers, so {@link #closeTopic} can cancel a retry for a topic that
- * dropped out of the match set before the timer fires (otherwise it would resurrect the topic). */
- private final ConcurrentHashMap<String, Timeout> retryTimeouts = new ConcurrentHashMap<>();
/** Returns the next exponential-backoff delay (ms) for a topic and updates the state. */
private long nextBackoff(String topicName) {
@@ -258,17 +262,22 @@ final class MultiTopicQueueConsumer<T> implements QueueConsumerImpl<T> {
*/
private CompletableFuture<Void> closeTopic(String topicName) {
retryDelays.remove(topicName);
- // Cancel any pending backoff retry so a topic that just left the match set can't be
- // re-subscribed when a stale timer fires.
- Timeout retry = retryTimeouts.remove(topicName);
+ PerTopicState<T> state = perTopic.remove(topicName);
+ if (state == null) {
+ return CompletableFuture.completedFuture(null);
+ }
+ // If it isn't attached yet, stop attaching it: an attempt in flight closes the consumer
+ // it opens, and a pending backoff retry must not re-subscribe a topic that just left
+ // the match set.
+ boolean attaching = state.consumer.cancel(false);
+ Timeout retry = state.retry;
if (retry != null) {
retry.cancel();
}
- PerTopicState<T> state = perTopic.remove(topicName);
- if (state == null) {
+ if (attaching) {
return CompletableFuture.completedFuture(null);
}
- return state.consumer.closeAsync()
+ return state.consumer.thenCompose(ScalableQueueConsumer::closeAsync)
.thenRun(() -> log.info().attr("topic", topicName)
.log("Per-topic consumer detached"));
}
@@ -280,7 +289,13 @@ final class MultiTopicQueueConsumer<T> implements QueueConsumerImpl<T> {
*/
@VisibleForTesting
Set<String> attachedTopicsForTesting() {
- return new HashSet<>(perTopic.keySet());
+ Set<String> attached = new HashSet<>();
+ perTopic.forEach((topic, state) -> {
+ if (state.attachedConsumer() != null) {
+ attached.add(topic);
+ }
+ });
+ return attached;
}
// --- QueueConsumer ---
@@ -336,14 +351,15 @@ final class MultiTopicQueueConsumer<T> implements QueueConsumerImpl<T> {
+ " delivered through a multi-topic consumer?");
}
PerTopicState<T> state = perTopic.get(parent);
- if (state == null) {
+ ScalableQueueConsumer<T> consumer = state != null ? state.attachedConsumer() : null;
+ if (consumer == null) {
// Topic was removed between deliver and ack. Fine — broker has dropped the
// session for that topic. Drop the ack silently.
log.debug().attr("topic", parent)
.log("Ack for removed topic; dropping");
return;
}
- action.accept(state.consumer);
+ action.accept(consumer);
}
@Override
@@ -376,10 +392,6 @@ final class MultiTopicQueueConsumer<T> implements QueueConsumerImpl<T> {
closed = true;
watcher.close();
mux.close();
- // Cancel pending retries for topics that never finished subscribing (they're not in
- // perTopic, so the closeTopic loop below wouldn't reach them).
- retryTimeouts.values().forEach(Timeout::cancel);
- retryTimeouts.clear();
List<CompletableFuture<Void>> closes = new ArrayList<>();
for (var topic : new HashSet<>(perTopic.keySet())) {
closes.add(closeTopic(topic));
@@ -405,7 +417,7 @@ final class MultiTopicQueueConsumer<T> implements QueueConsumerImpl<T> {
}
for (String t : target) {
if (!current.contains(t)) {
- openTopic(t, /* retry= */ true);
+ openTopic(t);
resetBackoff(t);
}
}
@@ -418,7 +430,7 @@ final class MultiTopicQueueConsumer<T> implements QueueConsumerImpl<T> {
closeTopic(t);
}
for (String t : added) {
- openTopic(t, /* retry= */ true);
+ openTopic(t);
resetBackoff(t);
}
}
@@ -427,16 +439,23 @@ final class MultiTopicQueueConsumer<T> implements QueueConsumerImpl<T> {
// --- Per-topic state ---
/**
- * Per-topic bookkeeping. Messages flow directly into the shared mux via the
- * sink the wrapper installed on the per-topic consumer at create-time, so
- * there's no pump thread to start/stop here — just hold a reference to the
- * underlying consumer for ack routing and clean shutdown.
+ * Per-topic bookkeeping, from when the topic joins the matching set until it leaves it.
+ * Messages flow directly into the shared mux via the sink the wrapper installed on the
+ * per-topic consumer at create-time, so there's no pump thread to start/stop here — just
+ * hold the underlying consumer, once attached, for ack routing and clean shutdown, and the
+ * pending retry while attaching it fails.
*/
private static final class PerTopicState<T> {
- private final ScalableQueueConsumer<T> consumer;
-
- PerTopicState(ScalableQueueConsumer<T> consumer) {
- this.consumer = consumer;
+ /**
+ * Completes with the per-topic consumer once an attempt attaches it. Cancelled by
+ * {@code closeTopic} if the topic leaves the matching set first, which stops the attempts.
+ */
+ private final CompletableFuture<ScalableQueueConsumer<T>> consumer = new CompletableFuture<>();
+ private volatile Timeout retry;
+
+ /** The per-topic consumer, or null while the topic isn't attached. */
+ ScalableQueueConsumer<T> attachedConsumer() {
+ return consumer.isDone() && !consumer.isCompletedExceptionally() ? consumer.getNow(null) : null;
}
}
}
diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicStreamConsumer.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicStreamConsumer.java
index 951de7980a19..cf637a129227 100644
--- a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicStreamConsumer.java
+++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicStreamConsumer.java
@@ -80,6 +80,10 @@ final class MultiTopicStreamConsumer<T> implements StreamConsumer<T> {
private final String subscriptionName;
private final ScalableTopicsWatcher watcher;
+ /**
+ * Every topic of the matching set, from when the consumer starts attaching it until it leaves the set:
+ * while it is being attached, once attached, and while a failed attach waits to be retried.
+ */
private final ConcurrentHashMap<String, PerTopic<T>> perTopic = new ConcurrentHashMap<>();
private final V5ReceiveQueue<T> mux;
@@ -139,24 +143,40 @@ final class MultiTopicStreamConsumer<T> implements StreamConsumer<T> {
}));
}
+ /**
+ * Open one per-topic consumer per topic in the initial snapshot, waiting for the first attempt on
+ * every topic. A topic whose first attempt failed is retried in the background, like a topic that
+ * joins the matching set later.
+ */
private CompletableFuture<Void> openInitial(List<String> topics) {
if (topics.isEmpty()) {
return CompletableFuture.completedFuture(null);
}
List<CompletableFuture<?>> opens = new ArrayList<>(topics.size());
for (String t : topics) {
- opens.add(openTopic(t, /* retry= */ false));
+ opens.add(openTopic(t));
}
return CompletableFuture.allOf(opens.toArray(CompletableFuture[]::new));
}
- private CompletableFuture<Void> openTopic(String topicName, boolean retry) {
+ /**
+ * Subscribe to one topic, retrying failures in the background until an attempt succeeds or the
+ * topic leaves the matching set. The returned future completes once the first attempt finishes.
+ */
+ private CompletableFuture<Void> openTopic(String topicName) {
if (closed) {
return CompletableFuture.completedFuture(null);
}
- if (perTopic.containsKey(topicName)) {
+ PerTopic<T> state = new PerTopic<>();
+ if (perTopic.putIfAbsent(topicName, state) != null) {
+ // Already attached, being attached, or waiting to retry.
return CompletableFuture.completedFuture(null);
}
+ return attachTopic(topicName, state);
+ }
+
+ /** One attempt at subscribing to a topic; a failure schedules the next one. */
+ private CompletableFuture<Void> attachTopic(String topicName, PerTopic<T> state) {
TopicName topic = V5Utils.parseScalableTopicInput(topicName);
// One ScalableConsumerClient session per topic, same as the single-topic builder.
ScalableConsumerClient session = new ScalableConsumerClient(
@@ -176,13 +196,9 @@ final class MultiTopicStreamConsumer<T> implements StreamConsumer<T> {
client, v5Schema, perTopicConf(topicName), session,
topicName, initialAssignment, sink))
.thenAccept(sc -> {
- if (closed) {
- sc.closeAsync();
- return;
- }
- PerTopic<T> state = new PerTopic<>(sc);
- PerTopic<T> existing = perTopic.putIfAbsent(topicName, state);
- if (existing != null) {
+ // closeTopic cancels the state when the topic leaves the matching set
+ // while this attempt is in flight: close what it opened.
+ if (closed || !state.consumer.complete(sc)) {
sc.closeAsync();
return;
}
@@ -191,8 +207,9 @@ final class MultiTopicStreamConsumer<T> implements StreamConsumer<T> {
.exceptionally(ex -> {
Throwable cause = ex instanceof CompletionException ce && ce.getCause() != null
? ce.getCause() : ex;
- if (retry && !closed) {
- scheduleRetry(topicName);
+ // No retry once the topic has left the matching set: it may be gone for good.
+ if (!closed && !state.consumer.isDone()) {
+ scheduleRetry(topicName, state);
}
log.warn().attr("topic", topicName).exceptionMessage(cause)
.log("Per-topic stream subscribe failed");
@@ -200,21 +217,18 @@ final class MultiTopicStreamConsumer<T> implements StreamConsumer<T> {
});
}
- private void scheduleRetry(String topicName) {
+ private void scheduleRetry(String topicName, PerTopic<T> state) {
long delayMs = nextBackoff(topicName);
log.info().attr("topic", topicName).attr("delayMs", delayMs)
.log("Retrying per-topic stream subscribe");
- Timeout timeout = client.v4Client().timer().newTimeout(t -> {
- retryTimeouts.remove(topicName);
- openTopic(topicName, /* retry= */ true);
+ state.retry = client.v4Client().timer().newTimeout(t -> {
+ if (!closed && !state.consumer.isDone()) {
+ attachTopic(topicName, state);
+ }
}, delayMs, TimeUnit.MILLISECONDS);
- retryTimeouts.put(topicName, timeout);
}
private final ConcurrentHashMap<String, AtomicLong> retryDelays = new ConcurrentHashMap<>();
- /** Pending backoff-retry timers, so {@link #closeTopic} can cancel a retry for a topic that
- * dropped out of the match set before the timer fires (otherwise it would resurrect the topic). */
- private final ConcurrentHashMap<String, Timeout> retryTimeouts = new ConcurrentHashMap<>();
private long nextBackoff(String topicName) {
AtomicLong al = retryDelays.computeIfAbsent(topicName, t -> new AtomicLong(100));
@@ -258,19 +272,24 @@ final class MultiTopicStreamConsumer<T> implements StreamConsumer<T> {
*/
private CompletableFuture<Void> closeTopic(String topicName) {
retryDelays.remove(topicName);
- // Cancel any pending backoff retry so a topic that just left the match set can't be
- // re-subscribed when a stale timer fires.
- Timeout retry = retryTimeouts.remove(topicName);
- if (retry != null) {
- retry.cancel();
- }
PerTopic<T> state = perTopic.remove(topicName);
if (state == null) {
return CompletableFuture.completedFuture(null);
}
+ // If it isn't attached yet, stop attaching it: an attempt in flight closes the consumer
+ // it opens, and a pending backoff retry must not re-subscribe a topic that just left
+ // the match set.
+ boolean attaching = state.consumer.cancel(false);
+ Timeout retry = state.retry;
+ if (retry != null) {
+ retry.cancel();
+ }
// Stop tracking this topic's delivery positions. No ack flush — see javadoc.
latestDeliveredPerTopicSegment.remove(topicName);
- return state.consumer.closeAsync()
+ if (attaching) {
+ return CompletableFuture.completedFuture(null);
+ }
+ return state.consumer.thenCompose(ScalableStreamConsumer::closeAsync)
.thenRun(() -> log.info().attr("topic", topicName)
.log("Per-topic stream consumer detached"));
}
@@ -336,12 +355,13 @@ final class MultiTopicStreamConsumer<T> implements StreamConsumer<T> {
}
for (var entry : vector.entrySet()) {
PerTopic<T> state = perTopic.get(entry.getKey());
- if (state == null) {
+ ScalableStreamConsumer<T> consumer = state != null ? state.attachedConsumer() : null;
+ if (consumer == null) {
// Topic left the matching set since this message was enqueued: we've
// detached it and no longer ack removed topics, so skip its slice.
continue;
}
- action.accept(state.consumer, entry.getValue());
+ action.accept(consumer, entry.getValue());
}
}
@@ -369,10 +389,6 @@ final class MultiTopicStreamConsumer<T> implements StreamConsumer<T> {
closed = true;
watcher.close();
mux.close();
- // Cancel pending retries for topics that never finished subscribing (they're not in
- // perTopic, so the closeTopic loop below wouldn't reach them).
- retryTimeouts.values().forEach(Timeout::cancel);
- retryTimeouts.clear();
List<CompletableFuture<Void>> closes = new ArrayList<>();
for (var topic : new HashSet<>(perTopic.keySet())) {
closes.add(closeTopic(topic));
@@ -394,7 +410,7 @@ final class MultiTopicStreamConsumer<T> implements StreamConsumer<T> {
}
for (String t : target) {
if (!current.contains(t)) {
- openTopic(t, /* retry= */ true);
+ openTopic(t);
resetBackoff(t);
}
}
@@ -406,7 +422,7 @@ final class MultiTopicStreamConsumer<T> implements StreamConsumer<T> {
closeTopic(t);
}
for (String t : added) {
- openTopic(t, /* retry= */ true);
+ openTopic(t);
resetBackoff(t);
}
}
@@ -451,16 +467,22 @@ final class MultiTopicStreamConsumer<T> implements StreamConsumer<T> {
// --- Per-topic state ---
/**
- * Per-topic bookkeeping. Messages flow into the shared mux directly via the
- * sink installed on the per-topic consumer at create-time, so there's no
- * pump thread to manage — this is just a holder for ack routing and clean
- * shutdown.
+ * Per-topic bookkeeping, from when the topic joins the matching set until it leaves it.
+ * Messages flow into the shared mux directly via the sink installed on the per-topic
+ * consumer at create-time, so there's no pump thread to manage — this is just a holder
+ * for ack routing and clean shutdown, and for the pending retry while attaching fails.
*/
private static final class PerTopic<T> {
- private final ScalableStreamConsumer<T> consumer;
-
- PerTopic(ScalableStreamConsumer<T> consumer) {
- this.consumer = consumer;
+ /**
+ * Completes with the per-topic consumer once an attempt attaches it. Cancelled by
+ * {@code closeTopic} if the topic leaves the matching set first, which stops the attempts.
+ */
+ private final CompletableFuture<ScalableStreamConsumer<T>> consumer = new CompletableFuture<>();
+ private volatile Timeout retry;
+
+ /** The per-topic consumer, or null while the topic isn't attached. */
+ ScalableStreamConsumer<T> attachedConsumer() {
+ return consumer.isDone() && !consumer.isCompletedExceptionally() ? consumer.getNow(null) : null;
}
}