From 95b4928afc5a427c946bf20054bda8bddc987d27 Mon Sep 17 00:00:00 2001 From: mattison chao Date: Tue, 29 Mar 2022 12:31:46 +0800 Subject: [PATCH 1/5] [fix][broker] Fix auto create partitioned topic when topic name contains ``-partition-`` --- .../pulsar/broker/PulsarServerException.java | 6 +++ .../pulsar/broker/admin/AdminResource.java | 4 ++ .../pulsar/broker/service/BrokerService.java | 50 ++++++++----------- .../persistent/PersistentTopicTest.java | 44 +++++++++++++--- 4 files changed, 68 insertions(+), 36 deletions(-) diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/PulsarServerException.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/PulsarServerException.java index 1fd1d0770213a..40dd53b3bafca 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/PulsarServerException.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/PulsarServerException.java @@ -44,4 +44,10 @@ public NotFoundException(Throwable t) { super(t); } } + public static class InvalidTopicNameException extends PulsarServerException { + + public InvalidTopicNameException(String message) { + super(message); + } + } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java index fe5d1056e18b0..5c4ba54164a2e 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java @@ -38,6 +38,7 @@ import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.client.BookKeeper; import org.apache.bookkeeper.mledger.ManagedLedgerException; +import org.apache.pulsar.broker.PulsarServerException; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.broker.service.BrokerServiceException; @@ -504,6 +505,9 @@ protected static PartitionedTopicMetadata fetchPartitionedTopicMetadataCheckAllo if (e.getCause() instanceof RestException) { throw (RestException) e.getCause(); } + if (e.getCause() instanceof PulsarServerException.InvalidTopicNameException) { + throw new RestException(Status.PRECONDITION_FAILED, e.getMessage()); + } throw new RestException(e); } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java index 83d509cc1c323..246538cf56ba0 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java @@ -2606,40 +2606,30 @@ public CompletableFuture fetchPartitionedTopicMetadata return FutureUtil.failedFuture(new NamingException("namespace service is not ready")); } return pulsar.getNamespaceService().checkTopicExists(topicName) - .thenCompose(topicExists -> { - return fetchPartitionedTopicMetadataAsync(topicName) - .thenCompose(metadata -> { - CompletableFuture future = new CompletableFuture<>(); - - // There are a couple of potentially blocking calls, which we cannot make from the - // MetadataStore callback thread. - pulsar.getExecutor().execute(() -> { - // If topic is already exist, creating partitioned topic is not allowed. - - if (metadata.partitions == 0 - && !topicExists - && !topicName.isPartitioned() - && pulsar.getBrokerService().isAllowAutoTopicCreation(topicName) - && pulsar.getBrokerService().isDefaultTopicTypePartitioned(topicName)) { - - pulsar.getBrokerService().createDefaultPartitionedTopicAsync(topicName) - .thenAccept(md -> future.complete(md)) - .exceptionally(ex -> { - future.completeExceptionally(ex); - return null; - }); - } else { - future.complete(metadata); - } - }); - - return future; - }); - }); + .thenCompose(topicExists -> fetchPartitionedTopicMetadataAsync(topicName) + .thenComposeAsync(metadata -> { + // There are a couple of potentially blocking calls, which we cannot make from the + // MetadataStore callback thread. + // If topic is already exist, creating partitioned topic is not allowed. + if (metadata.partitions == 0 + && !topicExists + && !topicName.isPartitioned() + && pulsar.getBrokerService().isAllowAutoTopicCreation(topicName) + && pulsar.getBrokerService().isDefaultTopicTypePartitioned(topicName)) { + return pulsar.getBrokerService().createDefaultPartitionedTopicAsync(topicName); + } else { + return CompletableFuture.completedFuture(null); + } + }, pulsar.getExecutor()) + ); } @SuppressWarnings("deprecation") private CompletableFuture createDefaultPartitionedTopicAsync(TopicName topicName) { + if (topicName.toString().contains(TopicName.PARTITIONED_TOPIC_SUFFIX)) { + return FutureUtil.failedFuture(new PulsarServerException. + InvalidTopicNameException("Invalid topic name: " + topicName)); + } final int defaultNumPartitions = pulsar.getBrokerService().getDefaultNumPartitions(topicName); final int maxPartitions = pulsar().getConfig().getMaxNumPartitionsPerPartitionedTopic(); checkArgument(defaultNumPartitions > 0, diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java index b906941770413..69b199e27944a 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java @@ -31,26 +31,24 @@ import static org.testng.Assert.assertNull; import java.lang.reflect.Field; +import java.net.URI; import java.util.ArrayList; import java.util.List; import java.util.UUID; import java.util.concurrent.TimeUnit; import com.google.common.collect.Sets; +import lombok.Cleanup; import org.apache.bookkeeper.client.LedgerHandle; import org.apache.bookkeeper.mledger.ManagedLedger; import org.apache.pulsar.broker.service.BrokerTestBase; -import org.apache.pulsar.client.api.Consumer; -import org.apache.pulsar.client.api.Message; -import org.apache.pulsar.client.api.MessageRoutingMode; -import org.apache.pulsar.client.api.Producer; -import org.apache.pulsar.client.api.Schema; -import org.apache.pulsar.client.api.SubscriptionType; +import org.apache.pulsar.client.api.*; import org.apache.pulsar.common.naming.NamespaceBundle; import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.policies.data.Policies; import org.apache.pulsar.common.policies.data.TopicStats; import org.awaitility.Awaitility; +import org.junit.Assert; import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; @@ -271,4 +269,38 @@ public void testPersistentPartitionedTopicUnload() throws Exception { producer.close(); } } + + @Test + public void testAutoCreatePartitionedTopicThatNameIncludePartition() throws Exception { + final String topicName = "persistent://prop/autoNs/failedcreate-partition-abcde"; + final String ns = "prop/autoNs"; + admin.namespaces().createNamespace(ns); + pulsar.getConfiguration().setAllowAutoTopicCreationType("partitioned"); + try { + @Cleanup + Producer producer = pulsarClient.newProducer().topic(topicName) + .create(); + Assert.fail("unexpected operation"); + } catch (PulsarClientException ex) { + Assert.assertTrue(ex.getMessage() + .contains("Invalid topic name")); + } + Assert.assertEquals(admin.topics().getList(ns).size(), 0); + URI tcpLookupUrl = new URI(pulsar.getBrokerServiceUrl()); + PulsarClient client = PulsarClient.builder() + .serviceUrl(tcpLookupUrl.toString()) + .build(); + try { + @Cleanup + Producer producer = client.newProducer() + .topic(topicName) + .create(); + Assert.fail("unexpected operation"); + } catch (PulsarClientException ex) { + Assert.assertTrue(ex.getMessage() + .contains("Invalid topic name")); + } + Assert.assertEquals(admin.topics().getList(ns).size(), 0); + pulsar.getConfiguration().setAllowAutoTopicCreationType("non-partitioned"); + } } From 75977a09256d39f94f19e257bd44204741bcedc8 Mon Sep 17 00:00:00 2001 From: mattison chao Date: Tue, 29 Mar 2022 13:56:05 +0800 Subject: [PATCH 2/5] Apply comments --- .../apache/pulsar/broker/service/BrokerService.java | 2 +- .../service/persistent/PersistentTopicTest.java | 11 ++++++++--- 2 files changed, 9 insertions(+), 4 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java index 246538cf56ba0..adf2006c40e72 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java @@ -2618,7 +2618,7 @@ public CompletableFuture fetchPartitionedTopicMetadata && pulsar.getBrokerService().isDefaultTopicTypePartitioned(topicName)) { return pulsar.getBrokerService().createDefaultPartitionedTopicAsync(topicName); } else { - return CompletableFuture.completedFuture(null); + return CompletableFuture.completedFuture(metadata); } }, pulsar.getExecutor()) ); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java index 69b199e27944a..6d3ec63778582 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java @@ -29,20 +29,25 @@ import static org.testng.Assert.assertFalse; import static org.testng.Assert.assertNotNull; import static org.testng.Assert.assertNull; - import java.lang.reflect.Field; import java.net.URI; import java.util.ArrayList; import java.util.List; import java.util.UUID; import java.util.concurrent.TimeUnit; - import com.google.common.collect.Sets; import lombok.Cleanup; import org.apache.bookkeeper.client.LedgerHandle; import org.apache.bookkeeper.mledger.ManagedLedger; import org.apache.pulsar.broker.service.BrokerTestBase; -import org.apache.pulsar.client.api.*; +import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.MessageRoutingMode; +import org.apache.pulsar.client.api.Producer; +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.common.naming.NamespaceBundle; import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.policies.data.Policies; From 4f9ded4c4067dc9b096cafe016359be42bebc969 Mon Sep 17 00:00:00 2001 From: mattison chao Date: Thu, 31 Mar 2022 11:58:46 +0800 Subject: [PATCH 3/5] Apply comments --- .../pulsar/broker/service/BrokerService.java | 49 ++++++++++++------- 1 file changed, 32 insertions(+), 17 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java index adf2006c40e72..414972a0de526 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java @@ -2606,29 +2606,44 @@ public CompletableFuture fetchPartitionedTopicMetadata return FutureUtil.failedFuture(new NamingException("namespace service is not ready")); } return pulsar.getNamespaceService().checkTopicExists(topicName) - .thenCompose(topicExists -> fetchPartitionedTopicMetadataAsync(topicName) - .thenComposeAsync(metadata -> { - // There are a couple of potentially blocking calls, which we cannot make from the - // MetadataStore callback thread. - // If topic is already exist, creating partitioned topic is not allowed. - if (metadata.partitions == 0 - && !topicExists - && !topicName.isPartitioned() - && pulsar.getBrokerService().isAllowAutoTopicCreation(topicName) - && pulsar.getBrokerService().isDefaultTopicTypePartitioned(topicName)) { - return pulsar.getBrokerService().createDefaultPartitionedTopicAsync(topicName); - } else { - return CompletableFuture.completedFuture(metadata); - } - }, pulsar.getExecutor()) - ); + .thenCompose(topicExists -> { + return fetchPartitionedTopicMetadataAsync(topicName) + .thenCompose(metadata -> { + CompletableFuture future = new CompletableFuture<>(); + + // There are a couple of potentially blocking calls, which we cannot make from the + // MetadataStore callback thread. + pulsar.getExecutor().execute(() -> { + // If topic is already exist, creating partitioned topic is not allowed. + + if (metadata.partitions == 0 + && !topicExists + && !topicName.isPartitioned() + && pulsar.getBrokerService().isAllowAutoTopicCreation(topicName) + && pulsar.getBrokerService().isDefaultTopicTypePartitioned(topicName)) { + + pulsar.getBrokerService().createDefaultPartitionedTopicAsync(topicName) + .thenAccept(md -> future.complete(md)) + .exceptionally(ex -> { + future.completeExceptionally(ex); + return null; + }); + } else { + future.complete(metadata); + } + }); + + return future; + }); + }); } @SuppressWarnings("deprecation") private CompletableFuture createDefaultPartitionedTopicAsync(TopicName topicName) { if (topicName.toString().contains(TopicName.PARTITIONED_TOPIC_SUFFIX)) { return FutureUtil.failedFuture(new PulsarServerException. - InvalidTopicNameException("Invalid topic name: " + topicName)); + InvalidTopicNameException( + String.format("Invalid topic name: %s , should not contain -partition-", topicName))); } final int defaultNumPartitions = pulsar.getBrokerService().getDefaultNumPartitions(topicName); final int maxPartitions = pulsar().getConfig().getMaxNumPartitionsPerPartitionedTopic(); From 8d016ea25d0d020d7cb9e86955e9814e705a65df Mon Sep 17 00:00:00 2001 From: mattison chao Date: Tue, 5 Apr 2022 09:30:29 +0800 Subject: [PATCH 4/5] Apply comments --- .../java/org/apache/pulsar/broker/service/BrokerService.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java index 414972a0de526..f8284de8bb3aa 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java @@ -2640,7 +2640,7 @@ public CompletableFuture fetchPartitionedTopicMetadata @SuppressWarnings("deprecation") private CompletableFuture createDefaultPartitionedTopicAsync(TopicName topicName) { - if (topicName.toString().contains(TopicName.PARTITIONED_TOPIC_SUFFIX)) { + if (topicName.getLocalName().contains(TopicName.PARTITIONED_TOPIC_SUFFIX)) { return FutureUtil.failedFuture(new PulsarServerException. InvalidTopicNameException( String.format("Invalid topic name: %s , should not contain -partition-", topicName))); From 18f99cae6f8dc0ea1c7cc4b8fed43ed6dfb6e67f Mon Sep 17 00:00:00 2001 From: mattison chao Date: Thu, 7 Apr 2022 13:26:48 +0800 Subject: [PATCH 5/5] Apply comments --- .../main/java/org/apache/pulsar/broker/admin/AdminResource.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java index 5c4ba54164a2e..b00226e3a2991 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java @@ -506,7 +506,7 @@ protected static PartitionedTopicMetadata fetchPartitionedTopicMetadataCheckAllo throw (RestException) e.getCause(); } if (e.getCause() instanceof PulsarServerException.InvalidTopicNameException) { - throw new RestException(Status.PRECONDITION_FAILED, e.getMessage()); + throw new RestException(Status.PRECONDITION_FAILED, e.getCause().getMessage()); } throw new RestException(e); }