diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java index 89004b888ba411..c52258fe1cc1b7 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java @@ -2306,7 +2306,7 @@ protected void internalCreateSubscription(AsyncResponse asyncResponse, String su .thenCompose(allowAutoTopicCreation -> getPartitionedTopicMetadataAsync(topicName, authoritative, allowAutoTopicCreation).thenAccept(partitionMetadata -> { final int numPartitions = partitionMetadata.partitions; - if (numPartitions > 0) { + if (partitionMetadata.partitions > 0 && !isUnexpectedTopicName(partitionMetadata)) { final CompletableFuture future = new CompletableFuture<>(); final AtomicInteger count = new AtomicInteger(numPartitions); final AtomicInteger failureCount = new AtomicInteger(0); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicNameForInfiniteHttpCallGetSubscriptionsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicNameForInfiniteHttpCallGetSubscriptionsTest.java new file mode 100644 index 00000000000000..96f57a59bda13e --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicNameForInfiniteHttpCallGetSubscriptionsTest.java @@ -0,0 +1,146 @@ +/* + * 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; + +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import lombok.AllArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.common.naming.TopicName; +import org.apache.pulsar.common.policies.data.TopicType; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +@Slf4j +@Test(groups = "broker") +public class TopicNameForInfiniteHttpCallGetSubscriptionsTest extends ProducerConsumerBase { + + @BeforeMethod(alwaysRun = true) + @Override + protected void setup() throws Exception { + super.internalSetup(); + super.producerBaseSetup(); + } + + @Override + protected void doInitConf() throws Exception { + super.doInitConf(); + conf.setAllowAutoTopicCreationType(TopicType.PARTITIONED); + conf.setDefaultNumPartitions(1); + } + + @AfterMethod(alwaysRun = true) + @Override + protected void cleanup() throws Exception { + super.internalCleanup(); + } + + @Test + public void testInfiniteHttpCallGetOrCreateSubscriptions() throws Exception { + final String randomStr = UUID.randomUUID().toString().replaceAll("-", ""); + final String partitionedTopicName = "persistent://my-property/my-ns/tp1_" + randomStr; + final String topic_p0 = partitionedTopicName + TopicName.PARTITIONED_TOPIC_SUFFIX + "0"; + final String subscriptionName = "sub1"; + final String topicDLQ = topic_p0 + "-" + subscriptionName + "-DLQ"; + + admin.topics().createPartitionedTopic(partitionedTopicName, 2); + + // Do test. + ProducerAndConsumerEntry pcEntry = triggerDLQCreated(topic_p0, topicDLQ, subscriptionName); + admin.topics().getSubscriptions(topicDLQ); + admin.topics().createSubscription(topicDLQ, "s1", MessageId.earliest); + + // cleanup. + pcEntry.consumer.close(); + pcEntry.producer.close(); + admin.topics().deletePartitionedTopic(partitionedTopicName); + } + + @Test + public void testInfiniteHttpCallGetOrCreateSubscriptions2() throws Exception { + final String randomStr = UUID.randomUUID().toString().replaceAll("-", ""); + final String topicName = "persistent://my-property/my-ns/tp1_" + randomStr + "-partition-0-abc"; + Producer producer = pulsarClient.newProducer(Schema.STRING) + .topic(topicName) + .create(); + + // Do test. + admin.topics().getSubscriptions(topicName); + admin.topics().createSubscription(topicName, "s1", MessageId.earliest); + + // cleanup. + producer.close(); + } + + @Test + public void testInfiniteHttpCallGetOrCreateSubscriptions3() throws Exception { + final String randomStr = UUID.randomUUID().toString().replaceAll("-", ""); + final String topicName = "persistent://my-property/my-ns/tp1_" + randomStr + "-partition-0"; + Producer producer = pulsarClient.newProducer(Schema.STRING) + .topic(topicName) + .create(); + + // Do test. + admin.topics().getSubscriptions(topicName); + admin.topics().createSubscription(topicName, "s1", MessageId.earliest); + + // cleanup. + producer.close(); + } + + @AllArgsConstructor + private static class ProducerAndConsumerEntry { + private Producer producer; + private Consumer consumer; + } + + private ProducerAndConsumerEntry triggerDLQCreated(String topicName, String DLQName, String subscriptionName) throws Exception { + Consumer consumer = pulsarClient.newConsumer(Schema.STRING) + .topic(topicName) + .subscriptionName(subscriptionName) + .subscriptionType(SubscriptionType.Shared) + .enableRetry(true) + .deadLetterPolicy(DeadLetterPolicy.builder().deadLetterTopic(DLQName).maxRedeliverCount(2).build()) + .receiverQueueSize(100) + .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest) + .subscribe(); + Producer producer = pulsarClient.newProducer(Schema.STRING) + .topic(topicName) + .create(); + // send messages. + for (int i = 0; i < 5; i++) { + producer.newMessage() + .value("value-" + i) + .sendAsync(); + } + producer.flush(); + // trigger the DLQ created. + for (int i = 0; i < 20; i++) { + Message msg = consumer.receive(5, TimeUnit.SECONDS); + if (msg != null) { + consumer.reconsumeLater(msg, 1, TimeUnit.SECONDS); + } else { + break; + } + } + + return new ProducerAndConsumerEntry(producer, consumer); + } +} \ No newline at end of file