You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2026/10/12 01:11:53 UTC
(pulsar) branch master updated: [fix][client] Stop V5 namespace stream consumers from recreating deleted topics (#26851)
This is an automated email from the ASF dual-hosted git repository.
codelipenghui 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 77ceda70ec4e [fix][client] Stop V5 namespace stream consumers from recreating deleted topics (#26851)
77ceda70ec4e is described below
commit 77ceda70ec4e6932965b3591d74e847692755ae1
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Sun Oct 11 18:11:44 2026 -0700
[fix][client] Stop V5 namespace stream consumers from recreating deleted topics (#26851)
---
.../api/v5/V5ScalableTopicAutoCreateTest.java | 15 ++
...V5NamespaceStreamConsumerTopicDeletionTest.java | 175 +++++++++++++++++++++
.../client/impl/v5/MultiTopicStreamConsumer.java | 7 +-
.../client/impl/v5/ScalableConsumerClient.java | 16 +-
4 files changed, 210 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5ScalableTopicAutoCreateTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5ScalableTopicAutoCreateTest.java
index b6cbbbe032f6..a9c31ebdde9c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5ScalableTopicAutoCreateTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5ScalableTopicAutoCreateTest.java
@@ -72,6 +72,21 @@ public class V5ScalableTopicAutoCreateTest extends V5ClientBaseTest {
assertEquals(md.getSegments().size(), 1);
}
+ @Test
+ public void testStreamConsumerAutoCreatesScalableTopic() throws Exception {
+ String topic = freshTopic();
+
+ @Cleanup
+ StreamConsumer<String> consumer = v5Client.newStreamConsumer(Schema.string())
+ .topic(topic)
+ .subscriptionName("sub")
+ .subscribe();
+
+ ScalableTopicMetadata md = admin.scalableTopics().getMetadata(topic);
+ assertNotNull(md);
+ assertEquals(md.getSegments().size(), 1);
+ }
+
@Test
public void testNoAutoCreateWhenNamespaceDisallows() throws Exception {
// Turn auto-topic-creation off for this namespace; a lookup of a non-existent
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5NamespaceStreamConsumerTopicDeletionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5NamespaceStreamConsumerTopicDeletionTest.java
new file mode 100644
index 000000000000..83df689d9dd5
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5NamespaceStreamConsumerTopicDeletionTest.java
@@ -0,0 +1,175 @@
+/*
+ * 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.assertThatThrownBy;
+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.atomic.AtomicInteger;
+import lombok.Cleanup;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.v5.StreamConsumer;
+import org.apache.pulsar.client.api.v5.V5ClientBaseTest;
+import org.apache.pulsar.client.api.v5.schema.Schema;
+import org.apache.pulsar.client.impl.ClientCnx;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.awaitility.Awaitility;
+import org.testng.annotations.Test;
+
+/**
+ * A namespace (multi-topic) stream consumer attaches only to the topics the namespace watch reports, so it must
+ * never auto-create a topic, even though the broker allows auto topic creation: a topic deleted while the consumer
+ * looks it up stays deleted. The consumer looks a topic up for the first attach of each topic of the initial
+ * snapshot, for the attach of each topic added later, and whenever the session of an attached topic reconnects.
+ *
+ * <p>Each test holds back the consumer's lookups of a topic, deletes the topic, and then lets the lookups go.
+ */
+public class V5NamespaceStreamConsumerTopicDeletionTest extends V5ClientBaseTest {
+
+ /**
+ * A v4 client that holds back the lookups of the scalable topics the test picks until the test releases them,
+ * and counts the lookups of each scalable topic.
+ */
+ private static final class LookupHoldingClient extends PulsarClientImpl {
+
+ private final Set<String> heldTopics = ConcurrentHashMap.newKeySet();
+ private final CompletableFuture<Void> lookupsReleased = new CompletableFuture<>();
+ private final Map<String, AtomicInteger> lookups = new ConcurrentHashMap<>();
+
+ LookupHoldingClient(ClientConfigurationData conf) throws PulsarClientException {
+ super(conf);
+ }
+
+ void holdLookups(String topic) {
+ heldTopics.add(topic);
+ }
+
+ void releaseLookups() {
+ lookupsReleased.complete(null);
+ }
+
+ /** How many times the client looked up the scalable topic, held or not. */
+ int lookups(String topic) {
+ AtomicInteger count = lookups.get(topic);
+ return count == null ? 0 : count.get();
+ }
+
+ /** Close every connection of the client, as a network failure would. */
+ void closeConnections() {
+ getCnxPool().getConnections().forEach(cnx -> cnx.thenAccept(c -> c.ctx().channel().close()));
+ }
+
+ // The V5 client looks up a scalable topic only through this method.
+ @Override
+ public CompletableFuture<ClientCnx> getConnection(String topic) {
+ lookups.computeIfAbsent(topic, __ -> new AtomicInteger()).incrementAndGet();
+ if (heldTopics.contains(topic)) {
+ return lookupsReleased.thenCompose(__ -> super.getConnection(topic));
+ }
+ return super.getConnection(topic);
+ }
+ }
+
+ @Test
+ public void testInitialTopicDeletedBeforeFirstAttachIsNotRecreated() throws Exception {
+ String topic = newScalableTopic(1);
+ PulsarClientV5 client = newLookupHoldingClient();
+ LookupHoldingClient lookupClient = (LookupHoldingClient) client.v4Client();
+ lookupClient.holdLookups(topic);
+
+ CompletableFuture<StreamConsumer<String>> subscribe = client.newStreamConsumer(Schema.string())
+ .namespace(getNamespace())
+ .subscriptionName("sub")
+ .subscribeAsync();
+ Awaitility.await().until(() -> lookupClient.lookups(topic) > 0);
+ admin.scalableTopics().deleteScalableTopic(topic, true);
+ lookupClient.releaseLookups();
+ @Cleanup
+ StreamConsumer<String> consumer = subscribe.get(30, SECONDS);
+
+ assertStaysDeleted(topic);
+ }
+
+ @Test
+ public void testTopicDeletedWhileAttachingIsNotRecreated() throws Exception {
+ PulsarClientV5 client = newLookupHoldingClient();
+ LookupHoldingClient lookupClient = (LookupHoldingClient) client.v4Client();
+ @Cleanup
+ StreamConsumer<String> consumer = client.newStreamConsumer(Schema.string())
+ .namespace(getNamespace())
+ .subscriptionName("sub")
+ .subscribe();
+
+ String topic = newScalableTopicName();
+ lookupClient.holdLookups(topic);
+ admin.scalableTopics().createScalableTopic(topic, 1);
+ Awaitility.await().until(() -> lookupClient.lookups(topic) > 0);
+ admin.scalableTopics().deleteScalableTopic(topic, true);
+ lookupClient.releaseLookups();
+
+ assertStaysDeleted(topic);
+ }
+
+ @Test
+ public void testTopicDeletedWhileReconnectingIsNotRecreated() throws Exception {
+ String topic = newScalableTopic(1);
+ PulsarClientV5 client = newLookupHoldingClient();
+ LookupHoldingClient lookupClient = (LookupHoldingClient) client.v4Client();
+ @Cleanup
+ StreamConsumer<String> consumer = client.newStreamConsumer(Schema.string())
+ .namespace(getNamespace())
+ .subscriptionName("sub")
+ .subscribe();
+
+ int lookupsBeforeReconnect = lookupClient.lookups(topic);
+ lookupClient.holdLookups(topic);
+ lookupClient.closeConnections();
+ Awaitility.await().until(() -> lookupClient.lookups(topic) > lookupsBeforeReconnect);
+ admin.scalableTopics().deleteScalableTopic(topic, true);
+ lookupClient.releaseLookups();
+
+ assertStaysDeleted(topic);
+ }
+
+ // --- Helpers ---
+
+ private PulsarClientV5 newLookupHoldingClient() throws Exception {
+ ClientConfigurationData conf = new ClientConfigurationData();
+ conf.setServiceUrl(getBrokerServiceUrl());
+ return track(new PulsarClientV5(new LookupHoldingClient(conf), "lookup-holding-client", null));
+ }
+
+ private String newScalableTopicName() {
+ return "topic://" + getNamespace() + "/scalable-" + UUID.randomUUID().toString().substring(0, 8);
+ }
+
+ /** Held for a few seconds: the released lookups reach the broker well within that. */
+ private void assertStaysDeleted(String topic) {
+ Awaitility.await().atMost(15, SECONDS).during(3, SECONDS).untilAsserted(() ->
+ assertThatThrownBy(() -> admin.scalableTopics().getMetadata(topic))
+ .as("the deleted topic stays deleted")
+ .isInstanceOf(PulsarAdminException.NotFoundException.class));
+ }
+}
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 cf637a129227..4b6d40c5eac8 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
@@ -178,12 +178,15 @@ final class MultiTopicStreamConsumer<T> implements StreamConsumer<T> {
/** 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.
+ // One ScalableConsumerClient session per topic, same as the single-topic builder. 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 an attach or a reconnect that looks it up).
ScalableConsumerClient session = new ScalableConsumerClient(
client.v4Client(), topic,
consumerConf.getSubscriptionName(),
perTopicConsumerName(topicName),
- ScalableConsumerType.STREAM);
+ ScalableConsumerType.STREAM,
+ /* createIfMissing= */ false);
// Per-topic message sink: each delivered message arrives with its
// single-topic positionVector (computed by ScalableStreamConsumer). Update
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 af2c7242b3b2..540a2b2b6d82 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
@@ -75,6 +75,10 @@ final class ScalableConsumerClient implements ScalableConsumerSession, AutoClose
private final String consumerName;
private final long consumerId;
private final ScalableConsumerType consumerType;
+ /** When false, the DAG lookup must not auto-create the scalable topic if it's missing. Namespace
+ * (multi-topic) consumers set this false so a deleted topic isn't resurrected by a per-topic
+ * attach or reconnect. */
+ private final boolean createIfMissing;
private final Backoff reconnectBackoff;
private final AtomicReference<List<ActiveSegment>> currentAssignment =
@@ -100,12 +104,22 @@ final class ScalableConsumerClient implements ScalableConsumerSession, AutoClose
String subscription,
String consumerName,
ScalableConsumerType consumerType) {
+ this(v4Client, topicName, subscription, consumerName, consumerType, true);
+ }
+
+ ScalableConsumerClient(PulsarClientImpl v4Client,
+ TopicName topicName,
+ String subscription,
+ String consumerName,
+ ScalableConsumerType consumerType,
+ boolean createIfMissing) {
this.v4Client = v4Client;
this.topicName = topicName;
this.subscription = subscription;
this.consumerName = consumerName;
this.consumerId = v4Client.newConsumerId();
this.consumerType = consumerType;
+ this.createIfMissing = createIfMissing;
this.reconnectBackoff = Backoff.builder()
.initialDelay(Duration.ofMillis(100))
.maxBackoff(Duration.ofSeconds(30))
@@ -156,7 +170,7 @@ final class ScalableConsumerClient implements ScalableConsumerSession, AutoClose
inFlightSubscribes.add(result);
result.whenComplete((__, ___) -> inFlightSubscribes.remove(result));
- DagWatchClient watch = new DagWatchClient(v4Client, topicName);
+ DagWatchClient watch = new DagWatchClient(v4Client, topicName, createIfMissing);
watch.start()
.thenCompose(layout -> {
if (watch.isUsingProxy()) {