Fix system topic can not auto created (#9272)

Fixes #9265 ### Motivation If disabled the topic auto-creation, the system topic under the namespace also not able to be created automatically, this will result in the topic level policy not able to use. ### Modifications Make the system topic be automatically created at any time
apache · Jan 26, 2021 · 60c570b · 60c570b
1 parent b9fe5d1
commit 60c570b
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 0 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
@@ -127,6 +127,7 @@
 import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
 import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
 import org.apache.pulsar.common.configuration.FieldContext;
+import org.apache.pulsar.common.events.EventsTopicNames;
 import org.apache.pulsar.common.intercept.AppendIndexMetadataInterceptor;
 import org.apache.pulsar.common.intercept.BrokerEntryMetadataInterceptor;
 import org.apache.pulsar.common.intercept.BrokerEntryMetadataUtils;
@@ -2420,6 +2421,11 @@ public boolean isAllowAutoTopicCreation(final String topic) {
     }
 
     public boolean isAllowAutoTopicCreation(final TopicName topicName) {
+        //System topic can always be created automatically
+        if (EventsTopicNames.NAMESPACE_EVENTS_LOCAL_NAME.equals(topicName.getLocalName())
+                && pulsar.getConfiguration().isSystemTopicEnabled()) {
+            return true;
+        }
         AutoTopicCreationOverride autoTopicCreationOverride = getAutoTopicCreationOverride(topicName);
         if (autoTopicCreationOverride != null) {
             return autoTopicCreationOverride.allowAutoTopicCreation;

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
@@ -1433,4 +1433,18 @@ public void testReplicatorRateApi() throws Exception {
                 -> assertNull(admin.topics().getReplicatorDispatchRate(topic)));
     }
 
+    @Test(timeOut = 30000)
+    public void testAutoCreationDisabled() throws Exception {
+        cleanup();
+        conf.setAllowAutoTopicCreation(false);
+        setup();
+        final String topic = testTopic + UUID.randomUUID();
+        admin.topics().createPartitionedTopic(topic, 3);
+        pulsarClient.newProducer().topic(topic).create().close();
+        Awaitility.await().atMost(5, TimeUnit.SECONDS)
+                .until(() -> pulsar.getTopicPoliciesService().cacheIsInitialized(TopicName.get(topic)));
+        //should not fail
+        assertNull(admin.topics().getMessageTTL(topic));
+    }
+
 }