[Broker] Check allowAutoSubscriptionCreation when creating init sub

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

@@ -1273,6 +1273,20 @@ protected void handleProducer(final CommandProducer cmdProducer) {
                             if (!Strings.isNullOrEmpty(initialSubscriptionName)
                                     && topic.isPersistent()
                                     && !topic.getSubscriptions().containsKey(initialSubscriptionName)) {
+                                if (!this.getBrokerService().isAllowAutoSubscriptionCreation(topicName)) {
+                                    String msg =
+                                            "Could not create the initial subscription due to the auto subscription "
+                                                    + "creation is not allowed.";
+                                    if (producerFuture.completeExceptionally(
+                                            new BrokerServiceException.NotAllowedException(msg))) {
+                                        log.warn("[{}] {} initialSubscriptionName: {}, topic: {}",
+                                                remoteAddress, msg, initialSubscriptionName, topicName);
+                                        commandSender.sendErrorResponse(requestId,
+                                                ServerError.NotAllowedError, msg);
+                                    }
+                                    producers.remove(producerId, producerFuture);
+                                    return;
+                                }
                                 createInitSubFuture =
                                         topic.createSubscription(initialSubscriptionName, InitialPosition.Earliest,
                                                 false);

pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerCreationTest.java

@@ -18,6 +18,7 @@
  */
 package org.apache.pulsar.client.api;
 
+import static org.testng.Assert.fail;
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.impl.ProducerBuilderImpl;
 import org.apache.pulsar.client.impl.ProducerImpl;
@@ -167,4 +168,27 @@ public void testCreateInitialSubscriptionWhenExisting() throws PulsarClientExcep
 
         Assert.assertTrue(admin.topics().getSubscriptions(topic.toString()).contains(initialSubscriptionName));
     }
+
+    @Test
+    public void testInitialSubscriptionCreationWithAutoCreationDisable()
+            throws PulsarAdminException, PulsarClientException {
+        pulsar.getConfiguration().setAllowAutoSubscriptionCreation(false);
+
+        final TopicName topic =
+                TopicName.get("persistent", "public", "default",
+                        "testInitialSubscriptionCreationWithAutoCreationDisable");
+        final String initialSubscriptionName = "init-sub";
+        admin.topics().createNonPartitionedTopic(topic.toString());
+        try {
+            Producer<byte[]> producer = ((ProducerBuilderImpl<byte[]>) pulsarClient.newProducer())
+                    .initialSubscriptionName(initialSubscriptionName)
+                    .topic(topic.toString())
+                    .create();
+            fail("Should not pass");
+        } catch (PulsarClientException.NotAllowedException exception) {
+            // ok
+        }
+
+        Assert.assertFalse(admin.topics().getSubscriptions(topic.toString()).contains(initialSubscriptionName));
+    }
 }

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/DeadLetterPolicy.java

@@ -52,6 +52,8 @@ public class DeadLetterPolicy {
     /**
      * Name of the initial subscription name of the dead letter topic.
      * If this field is not set, the initial subscription for the dead letter topic will not be created.
+     * If this field is set but the broker's `allowAutoSubscriptionCreation` is disabled, the DLQ producer will fail
+     * to be created.
      */
     private String initialSubscriptionName;
 }

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java

@@ -328,7 +328,9 @@ public ProducerBuilder<T> enableLazyStartPartitionedProducers(boolean lazyStartP
     /**
      * Use this config to automatically create an initial subscription when creating the topic.
      * If this field is not set, the initial subscription will not be created.
-     * This method is limited to internal use
+     * If this field is set but the broker's `allowAutoSubscriptionCreation` is disabled, the producer will fail to
+     * be created.
+     * This method is limited to internal use. This method will only be used when the consumer creates the dlq producer.
      *
      * @param initialSubscriptionName Name of the initial subscription of the topic.
      * @return the producer builder implementation instance

pulsar-common/src/main/proto/PulsarApi.proto

@@ -497,6 +497,8 @@ message CommandProducer {
 
     // Name of the initial subscription of the topic.
     // If this field is not set, the initial subscription will not be created.
+    // If this field is set but the broker's `allowAutoSubscriptionCreation`
+    // is disabled, the producer will fail to be created.
     optional string initial_subscription_name = 13;
 }
 

site2/docs/concepts-messaging.md

@@ -317,7 +317,7 @@ Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES)
 
 ```
 
-By default, there is no subscription during a DLQ topic creation. Without a just-in-time subscription to the DLQ topic, you may lose messages. To automatically create an initial subscription for the DLQ, you can specify the `initialSubscriptionName` parameter.
+By default, there is no subscription during a DLQ topic creation. Without a just-in-time subscription to the DLQ topic, you may lose messages. To automatically create an initial subscription for the DLQ, you can specify the `initialSubscriptionName` parameter. If this parameter is set but the broker's `allowAutoSubscriptionCreation` is disabled, the DLQ producer will fail to be created.
 
 ```java
 Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES)