diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java index f21690206772f..73e63fd92155f 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java +++ b/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); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerCreationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerCreationTest.java index 3e43e72f47ec2..bd5ce7b1746e6 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerCreationTest.java +++ b/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 producer = ((ProducerBuilderImpl) 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)); + } } diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/DeadLetterPolicy.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/DeadLetterPolicy.java index 351768f0e7bfd..e998238139045 100644 --- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/DeadLetterPolicy.java +++ b/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; } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java index bd533fa9f576a..9ad42f77ea868 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java @@ -328,7 +328,9 @@ public ProducerBuilder 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 diff --git a/pulsar-common/src/main/proto/PulsarApi.proto b/pulsar-common/src/main/proto/PulsarApi.proto index fcb516c6f3917..337c6bb2cd41a 100644 --- a/pulsar-common/src/main/proto/PulsarApi.proto +++ b/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; } diff --git a/site2/docs/concepts-messaging.md b/site2/docs/concepts-messaging.md index 389c622a165ea..595000f3c3027 100644 --- a/site2/docs/concepts-messaging.md +++ b/site2/docs/concepts-messaging.md @@ -317,7 +317,7 @@ Consumer 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 consumer = pulsarClient.newConsumer(Schema.BYTES)