[feat][broker] Prevent auto-creation of topics using legacy cluster-b…

…ased naming scheme (#23620)

Co-authored-by: zjxxzjwang <[email protected]>

conf/broker.conf

@@ -198,6 +198,10 @@ allowAutoTopicCreation=true
 # The type of topic that is allowed to be automatically created.(partitioned/non-partitioned)
 allowAutoTopicCreationType=non-partitioned
 
+# If 'allowAutoTopicCreation' is true and the name of the topic contains 'cluster',
+# the topic cannot be automatically created.
+allowAutoTopicCreationWithLegacyNamingScheme=true
+
 # Enable subscription auto creation if new consumer connected (disable auto creation with value false)
 allowAutoSubscriptionCreation=true
 

conf/standalone.conf

@@ -1181,6 +1181,10 @@ allowAutoTopicCreation=true
 # The type of topic that is allowed to be automatically created.(partitioned/non-partitioned)
 allowAutoTopicCreationType=non-partitioned
 
+# If 'allowAutoTopicCreation' is true and the name of the topic contains 'cluster',
+# the topic cannot be automatically created.
+allowAutoTopicCreationWithLegacyNamingScheme=true
+
 # Enable subscription auto creation if new consumer connected (disable auto creation with value false)
 allowAutoSubscriptionCreation=true
 

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -2154,6 +2154,11 @@ The max allowed delay for delayed delivery (in milliseconds). If the broker rece
             doc = "The type of topic that is allowed to be automatically created.(partitioned/non-partitioned)"
     )
     private TopicType allowAutoTopicCreationType = TopicType.NON_PARTITIONED;
+    @FieldContext(category = CATEGORY_SERVER, dynamic = true,
+            doc = "If 'allowAutoTopicCreation' is true and the name of the topic contains 'cluster',"
+                    + "the topic cannot be automatically created."
+    )
+    private boolean allowAutoTopicCreationWithLegacyNamingScheme = true;
     @FieldContext(
         category = CATEGORY_STORAGE_ML,
         dynamic = true,

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

@@ -3536,6 +3536,13 @@ private CompletableFuture<Boolean> isAllowAutoTopicCreationAsync(final TopicName
             return CompletableFuture.completedFuture(true);
         }
 
+        //If 'allowAutoTopicCreation' is true, and the name of the topic contains 'cluster',
+        //the topic cannot be automatically created.
+        if (!pulsar.getConfiguration().isAllowAutoTopicCreationWithLegacyNamingScheme()
+                && StringUtils.isNotBlank(topicName.getCluster())) {
+            return CompletableFuture.completedFuture(false);
+        }
+
         final boolean allowed;
         AutoTopicCreationOverride autoTopicCreationOverride = getAutoTopicCreationOverride(topicName, policies);
         if (autoTopicCreationOverride != null) {

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java

@@ -587,4 +587,23 @@ public void testExtensibleLoadManagerImplInternalTopicAutoCreations()
 
     }
 
+    @Test
+    public void testAutoPartitionedTopicNameWithClusterName() throws Exception {
+        pulsar.getConfiguration().setAllowAutoTopicCreation(true);
+        pulsar.getConfiguration().setAllowAutoTopicCreationType(TopicType.PARTITIONED);
+        pulsar.getConfiguration().setDefaultNumPartitions(3);
+
+        final String topicString = "persistent://prop/ns-abc/testTopic/1";
+        // When allowAutoTopicCreationWithLegacyNamingScheme as the default value is false,
+        // four-paragraph topic cannot be created.
+        pulsar.getConfiguration().setAllowAutoTopicCreationWithLegacyNamingScheme(false);
+        Assert.assertThrows(PulsarClientException.NotFoundException.class,
+                () -> pulsarClient.newProducer().topic(topicString).create());
+
+        pulsar.getConfiguration().setAllowAutoTopicCreationWithLegacyNamingScheme(true);
+        Producer producer = pulsarClient.newProducer().topic(topicString).create();
+        Assert.assertEquals(producer.getTopic(), topicString);
+        producer.close();
+    }
+
 }