From aba7060ef525dc5aa99178f40f3ea7410b61bd21 Mon Sep 17 00:00:00 2001 From: Zike Yang Date: Fri, 25 Feb 2022 00:03:55 +0800 Subject: [PATCH 1/6] [Broker] Check `allowAutoSubscriptionCreation` when creating init sub Motivation In #13355, we added support for creating initial subscription when creating the producer. But the broker didn't check if the subscription can be created automatically. The initial subscription will be created even if the `allowAutoSubscriptionCreation` is disabled. Modification * Check `allowAutoSubscriptionCreation` when creating the initial subscription. Signed-off-by: Zike Yang --- .../pulsar/broker/service/ServerCnx.java | 11 +++++++++ .../client/api/ProducerCreationTest.java | 24 +++++++++++++++++++ 2 files changed, 35 insertions(+) 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..79e200972eef2 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,17 @@ 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."; + 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)); + } } From 0203df5467a4ad4394ff6c2ab436241132128686 Mon Sep 17 00:00:00 2001 From: Zike Yang Date: Fri, 25 Feb 2022 17:51:16 +0800 Subject: [PATCH 2/6] complete producerFuture before removing it Signed-off-by: Zike Yang --- .../org/apache/pulsar/broker/service/ServerCnx.java | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) 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 79e200972eef2..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 @@ -1277,10 +1277,13 @@ protected void handleProducer(final CommandProducer cmdProducer) { String msg = "Could not create the initial subscription due to the auto subscription " + "creation is not allowed."; - log.warn("[{}] {} initialSubscriptionName: {}, topic: {}", - remoteAddress, msg, initialSubscriptionName, topicName); - commandSender.sendErrorResponse(requestId, - ServerError.NotAllowedError, msg); + 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; } From 5b349b9237265ae8c465aeda6ffe910d9aa20dcb Mon Sep 17 00:00:00 2001 From: Zike Yang Date: Fri, 25 Feb 2022 17:58:23 +0800 Subject: [PATCH 3/6] Update the doc for codes. Signed-off-by: Zike Yang --- .../java/org/apache/pulsar/client/api/DeadLetterPolicy.java | 3 ++- .../org/apache/pulsar/client/impl/ProducerBuilderImpl.java | 3 ++- pulsar-common/src/main/proto/PulsarApi.proto | 3 ++- 3 files changed, 6 insertions(+), 3 deletions(-) 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..629cfd7d1f0c9 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 @@ -51,7 +51,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 not set or the broker's `allowAutoSubscriptionCreation` is disabled, the initial subscription + * for the dead letter topic will not 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..45b9bcd65a397 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 @@ -327,7 +327,8 @@ 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. + * If this field is not set or the broker's `allowAutoSubscriptionCreation` is disabled, the initial subscription + * will not be created. * This method is limited to internal use * * @param initialSubscriptionName Name of the initial subscription of the topic. diff --git a/pulsar-common/src/main/proto/PulsarApi.proto b/pulsar-common/src/main/proto/PulsarApi.proto index fcb516c6f3917..5ee2541d0585a 100644 --- a/pulsar-common/src/main/proto/PulsarApi.proto +++ b/pulsar-common/src/main/proto/PulsarApi.proto @@ -496,7 +496,8 @@ message CommandProducer { optional bool txn_enabled = 12 [default = false]; // 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 not set or the broker's `allowAutoSubscriptionCreation` + // is disabled, the initial subscription will not be created. optional string initial_subscription_name = 13; } From 7e9a9e726b54a534f551c7d59b66a9a87ce962cb Mon Sep 17 00:00:00 2001 From: Zike Yang Date: Fri, 25 Feb 2022 18:22:33 +0800 Subject: [PATCH 4/6] Update doc. Signed-off-by: Zike Yang --- .../org/apache/pulsar/client/api/DeadLetterPolicy.java | 5 +++-- .../org/apache/pulsar/client/impl/ProducerBuilderImpl.java | 7 ++++--- pulsar-common/src/main/proto/PulsarApi.proto | 5 +++-- site2/docs/concepts-messaging.md | 2 +- 4 files changed, 11 insertions(+), 8 deletions(-) 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 629cfd7d1f0c9..60d5cb1b41a08 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 @@ -51,8 +51,9 @@ public class DeadLetterPolicy { /** * Name of the initial subscription name of the dead letter topic. - * If this field is not set or the broker's `allowAutoSubscriptionCreation` is disabled, the initial subscription - * for the dead letter topic will not be created. + * If this field is not set, the initial subscription for the dead letter topic will not be created. + * If this filed 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 45b9bcd65a397..98eeb89e80b8b 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 @@ -327,9 +327,10 @@ public ProducerBuilder enableLazyStartPartitionedProducers(boolean lazyStartP /** * Use this config to automatically create an initial subscription when creating the topic. - * If this field is not set or the broker's `allowAutoSubscriptionCreation` is disabled, the initial subscription - * will not be created. - * This method is limited to internal use + * If this field is not set, the initial subscription will not be created. + * If this filed is set but the broker's `allowAutoSubscriptionCreation` is disabled, the producer will fail to + * be created. + * This method is limited to internal use. * * @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 5ee2541d0585a..a62f7adb0037f 100644 --- a/pulsar-common/src/main/proto/PulsarApi.proto +++ b/pulsar-common/src/main/proto/PulsarApi.proto @@ -496,8 +496,9 @@ message CommandProducer { optional bool txn_enabled = 12 [default = false]; // Name of the initial subscription of the topic. - // If this field is not set or the broker's `allowAutoSubscriptionCreation` - // is disabled, the initial subscription will not be created. + // If this field is not set, the initial subscription will not be created. + // If this filed 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) From 6351f5ba17ad3376344e156d1aa44cf7b1ee4638 Mon Sep 17 00:00:00 2001 From: Zike Yang Date: Fri, 25 Feb 2022 18:25:35 +0800 Subject: [PATCH 5/6] Update doc. Signed-off-by: Zike Yang --- .../java/org/apache/pulsar/client/api/DeadLetterPolicy.java | 2 +- .../java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java | 2 +- pulsar-common/src/main/proto/PulsarApi.proto | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) 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 60d5cb1b41a08..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,7 +52,7 @@ 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 filed is set but the broker's `allowAutoSubscriptionCreation` is disabled, the DLQ producer will fail + * 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 98eeb89e80b8b..92e71aedbd304 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,7 @@ 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. - * If this filed is set but the broker's `allowAutoSubscriptionCreation` is disabled, the producer will fail to + * 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. * diff --git a/pulsar-common/src/main/proto/PulsarApi.proto b/pulsar-common/src/main/proto/PulsarApi.proto index a62f7adb0037f..337c6bb2cd41a 100644 --- a/pulsar-common/src/main/proto/PulsarApi.proto +++ b/pulsar-common/src/main/proto/PulsarApi.proto @@ -497,7 +497,7 @@ 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 filed is set but the broker's `allowAutoSubscriptionCreation` + // 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; } From 71a5d2cfa3271058a2f7a8f2f599d6054909f2c8 Mon Sep 17 00:00:00 2001 From: Zike Yang Date: Sat, 26 Feb 2022 10:10:37 +0800 Subject: [PATCH 6/6] Update doc. Signed-off-by: Zike Yang --- .../java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 92e71aedbd304..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 @@ -330,7 +330,7 @@ public ProducerBuilder enableLazyStartPartitionedProducers(boolean lazyStartP * 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. - * This method is limited to internal use. + * 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