[pulsar-broker] Implement AutoSubscriptionCreation by namespace override

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/Namespaces.java

@@ -28,6 +28,8 @@
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandGetTopicsOfNamespace.Mode;
 import org.apache.pulsar.common.naming.NamespaceBundleSplitAlgorithm;
 import org.apache.pulsar.common.policies.data.AuthAction;
+import org.apache.pulsar.common.policies.data.AutoSubscriptionCreationOverride;
+import org.apache.pulsar.common.policies.data.AutoTopicCreationOverride;
 import org.apache.pulsar.common.policies.data.BacklogQuota;
 import org.apache.pulsar.common.policies.data.BookieAffinityGroupData;
 import org.apache.pulsar.common.policies.data.BacklogQuota.BacklogQuotaType;
@@ -393,6 +395,80 @@ public void modifyDeduplication(@PathParam("property") String property, @PathPar
         internalModifyDeduplication(enableDeduplication);
     }
 
+    @POST
+    @Path("/{property}/{cluster}/{namespace}/autoTopicCreation")
+    @ApiOperation(value = "Override broker's allowAutoTopicCreation setting for a namespace")
+    @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
+            @ApiResponse(code = 404, message = "Tenant or cluster or namespace doesn't exist"),
+            @ApiResponse(code = 400, message = "Invalid autoTopicCreation override") })
+    public void setAutoTopicCreation(@Suspended final AsyncResponse asyncResponse,
+                                     @PathParam("property") String property, @PathParam("cluster") String cluster,
+                                     @PathParam("namespace") String namespace, AutoTopicCreationOverride autoTopicCreationOverride) {
+        try {
+            validateNamespaceName(property, cluster, namespace);
+            internalSetAutoTopicCreation(asyncResponse, autoTopicCreationOverride);
+        } catch (RestException e) {
+            asyncResponse.resume(e);
+        } catch (Exception e ) {
+            asyncResponse.resume(new RestException(e));
+        }
+    }
+
+    @DELETE
+    @Path("/{property}/{cluster}/{namespace}/autoTopicCreation")
+    @ApiOperation(value = "Remove override of broker's allowAutoTopicCreation in a namespace")
+    @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
+            @ApiResponse(code = 404, message = "Tenant or cluster or namespace doesn't exist") })
+    public void removeAutoTopicCreation(@Suspended final AsyncResponse asyncResponse,
+                                        @PathParam("property") String property, @PathParam("cluster") String cluster,
+                                        @PathParam("namespace") String namespace) {
+        try {
+            validateNamespaceName(property, cluster, namespace);
+            internalRemoveAutoTopicCreation(asyncResponse);
+        } catch (RestException e) {
+            asyncResponse.resume(e);
+        } catch (Exception e) {
+            asyncResponse.resume(new RestException(e));
+        }
+    }
+
+    @POST
+    @Path("/{property}/{cluster}/{namespace}/autoSubscriptionCreation")
+    @ApiOperation(value = "Override broker's allowAutoSubscriptionCreation setting for a namespace")
+    @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
+            @ApiResponse(code = 404, message = "Tenant or cluster or namespace doesn't exist"),
+            @ApiResponse(code = 400, message = "Invalid autoSubscriptionCreation override") })
+    public void setAutoSubscriptionCreation(@Suspended final AsyncResponse asyncResponse,
+                                     @PathParam("property") String property, @PathParam("cluster") String cluster,
+                                     @PathParam("namespace") String namespace, AutoSubscriptionCreationOverride autoSubscriptionCreationOverride) {
+        try {
+            validateNamespaceName(property, cluster, namespace);
+            internalSetAutoSubscriptionCreation(asyncResponse, autoSubscriptionCreationOverride);
+        } catch (RestException e) {
+            asyncResponse.resume(e);
+        } catch (Exception e ) {
+            asyncResponse.resume(new RestException(e));
+        }
+    }
+
+    @DELETE
+    @Path("/{property}/{cluster}/{namespace}/autoSubscriptionCreation")
+    @ApiOperation(value = "Remove override of broker's allowAutoSubscriptionCreation in a namespace")
+    @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
+            @ApiResponse(code = 404, message = "Tenant or cluster or namespace doesn't exist") })
+    public void removeAutoSubscriptionCreation(@Suspended final AsyncResponse asyncResponse,
+                                        @PathParam("property") String property, @PathParam("cluster") String cluster,
+                                        @PathParam("namespace") String namespace) {
+        try {
+            validateNamespaceName(property, cluster, namespace);
+            internalRemoveAutoSubscriptionCreation(asyncResponse);
+        } catch (RestException e) {
+            asyncResponse.resume(e);
+        } catch (Exception e) {
+            asyncResponse.resume(new RestException(e));
+        }
+    }
+
     @GET
     @Path("/{property}/{cluster}/{namespace}/bundles")
     @ApiOperation(hidden = true, value = "Get the bundles split data.")

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java

@@ -45,6 +45,7 @@
 import org.apache.pulsar.broker.admin.impl.NamespacesBase;
 import org.apache.pulsar.broker.web.RestException;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandGetTopicsOfNamespace.Mode;
+import org.apache.pulsar.common.policies.data.AutoSubscriptionCreationOverride;
 import org.apache.pulsar.common.policies.data.AutoTopicCreationOverride;
 import org.apache.pulsar.common.policies.data.AuthAction;
 import org.apache.pulsar.common.policies.data.BacklogQuota;
@@ -337,6 +338,43 @@ public void removeAutoTopicCreation(@Suspended final AsyncResponse asyncResponse
         }
     }
 
+    @POST
+    @Path("/{tenant}/{namespace}/autoSubscriptionCreation")
+    @ApiOperation(value = "Override broker's allowAutoSubscriptionCreation setting for a namespace")
+    @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
+            @ApiResponse(code = 404, message = "Tenant or cluster or namespace doesn't exist"),
+            @ApiResponse(code = 400, message = "Invalid autoSubscriptionCreation override") })
+    public void setAutoSubscriptionCreation(
+            @Suspended final AsyncResponse asyncResponse,
+            @PathParam("tenant") String tenant, @PathParam("namespace") String namespace,
+            AutoSubscriptionCreationOverride autoSubscriptionCreationOverride) {
+        try {
+            validateNamespaceName(tenant, namespace);
+            internalSetAutoSubscriptionCreation(asyncResponse, autoSubscriptionCreationOverride);
+        } catch (RestException e) {
+            asyncResponse.resume(e);
+        } catch (Exception e ) {
+            asyncResponse.resume(new RestException(e));
+        }
+    }
+
+    @DELETE
+    @Path("/{tenant}/{namespace}/autoSubscriptionCreation")
+    @ApiOperation(value = "Remove override of broker's allowAutoSubscriptionCreation in a namespace")
+    @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
+            @ApiResponse(code = 404, message = "Tenant or cluster or namespace doesn't exist") })
+    public void removeAutoSubscriptionCreation(@Suspended final AsyncResponse asyncResponse,
+                                        @PathParam("tenant") String tenant, @PathParam("namespace") String namespace) {
+        try {
+            validateNamespaceName(tenant, namespace);
+            internalRemoveAutoSubscriptionCreation(asyncResponse);
+        } catch (RestException e) {
+            asyncResponse.resume(e);
+        } catch (Exception e) {
+            asyncResponse.resume(new RestException(e));
+        }
+    }
+
     @GET
     @Path("/{tenant}/{namespace}/bundles")
     @ApiOperation(value = "Get the bundles split data.")

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

@@ -123,6 +123,7 @@
 import org.apache.pulsar.common.naming.TopicDomain;
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
+import org.apache.pulsar.common.policies.data.AutoSubscriptionCreationOverride;
 import org.apache.pulsar.common.policies.data.AutoTopicCreationOverride;
 import org.apache.pulsar.common.policies.data.ClusterData;
 import org.apache.pulsar.common.policies.data.LocalPolicies;
@@ -2175,4 +2176,35 @@ private AutoTopicCreationOverride getAutoTopicCreationOverride(final TopicName t
         log.warn("No autoTopicCreateOverride policy found for {}", topicName);
         return null;
     }
+
+    public boolean isAllowAutoSubscriptionCreation(final String topic) {
+        TopicName topicName = TopicName.get(topic);
+        return isAllowAutoSubscriptionCreation(topicName);
+    }
+
+    public boolean isAllowAutoSubscriptionCreation(final TopicName topicName) {
+        AutoSubscriptionCreationOverride autoSubscriptionCreationOverride = getAutoSubscriptionCreationOverride(topicName);
+        if (autoSubscriptionCreationOverride != null) {
+            return autoSubscriptionCreationOverride.allowAutoSubscriptionCreation;
+        } else {
+            return pulsar.getConfiguration().isAllowAutoSubscriptionCreation();
+        }
+    }
+
+    private AutoSubscriptionCreationOverride getAutoSubscriptionCreationOverride(final TopicName topicName) {
+        try {
+            Optional<Policies> policies = pulsar.getConfigurationCache().policiesCache()
+                    .get(AdminResource.path(POLICIES, topicName.getNamespace()));
+            // If namespace policies have the field set, it will override the broker-level setting
+            if (policies.isPresent() && policies.get().autoSubscriptionCreationOverride != null) {
+                return policies.get().autoSubscriptionCreationOverride;
+            }
+        } catch (Throwable t) {
+            // Ignoring since if we don't have policies, we fallback on the default
+            log.warn("Got exception when reading autoSubscriptionCreateOverride policy for {}: {};", topicName, t.getMessage(), t);
+            return null;
+        }
+        log.warn("No autoSubscriptionCreateOverride policy found for {}", topicName);
+        return null;
+    }
 }

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

@@ -820,7 +820,7 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
                                     Topic topic = optTopic.get();
 
                                     boolean rejectSubscriptionIfDoesNotExist = isDurable
-                                        && !service.pulsar().getConfig().isAllowAutoSubscriptionCreation()
+                                        && !service.isAllowAutoSubscriptionCreation(topicName.toString())
                                         && !topic.getSubscriptions().containsKey(subscriptionName);
 
                                     if (rejectSubscriptionIfDoesNotExist) {

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

@@ -0,0 +1,127 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service;
+
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.AutoSubscriptionCreationOverride;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
+
+public class BrokerServiceAutoSubscriptionCreationTest extends BrokerTestBase {
+
+    @BeforeClass
+    @Override
+    protected void setup() throws Exception {
+        super.baseSetup();
+    }
+
+    @AfterClass
+    @Override
+    protected void cleanup() throws Exception {
+        super.internalCleanup();
+    }
+
+    @AfterMethod
+    protected void cleanupTest() throws Exception {
+        pulsar.getAdminClient().namespaces().removeAutoSubscriptionCreation("prop/ns-abc");
+    }
+
+    @Test
+    public void testAutoSubscriptionCreationDisable() throws Exception {
+        pulsar.getConfiguration().setAllowAutoSubscriptionCreation(false);
+
+        final String topicName = "persistent://prop/ns-abc/test-subtopic";
+        final String subscriptionName = "test-subtopic-sub";
+
+        admin.topics().createNonPartitionedTopic(topicName);
+
+        try {
+            pulsarClient.newConsumer().topic(topicName).subscriptionName(subscriptionName).subscribe();
+            fail("Subscribe operation should have failed");
+        } catch (Exception e) {
+            assertTrue(e instanceof PulsarClientException);
+        }
+        assertFalse(admin.topics().getSubscriptions(topicName).contains(subscriptionName));
+    }
+
+    @Test
+    public void testSubscriptionCreationWithAutoCreationDisable() throws Exception {
+        pulsar.getConfiguration().setAllowAutoSubscriptionCreation(false);
+
+        final String topicName = "persistent://prop/ns-abc/test-subtopic";
+        final String subscriptionName = "test-subtopic-sub-1";
+
+        admin.topics().createNonPartitionedTopic(topicName);
+        assertFalse(admin.topics().getSubscriptions(topicName).contains(subscriptionName));
+
+        // Create the subscription by PulsarAdmin
+        admin.topics().createSubscription(topicName, subscriptionName, MessageId.earliest);
+        assertTrue(admin.topics().getSubscriptions(topicName).contains(subscriptionName));
+
+        // Subscribe operation should be successful
+        pulsarClient.newConsumer().topic(topicName).subscriptionName(subscriptionName).subscribe();
+    }
+
+    @Test
+    public void testAutoSubscriptionCreationNamespaceAllowOverridesBroker() throws Exception {
+        final String topic = "persistent://prop/ns-abc/test-subtopic";
+        final String subscriptionName = "test-subtopic-sub-2";
+        final TopicName topicName = TopicName.get(topic);
+
+        admin.topics().createNonPartitionedTopic(topicName.toString());
+
+        pulsar.getConfiguration().setAllowAutoSubscriptionCreation(false);
+        pulsar.getAdminClient().namespaces().setAutoSubscriptionCreation(topicName.getNamespace(),
+                new AutoSubscriptionCreationOverride(true));
+
+        // Subscribe operation should be successful
+        pulsarClient.newConsumer().topic(topicName.toString()).subscriptionName(subscriptionName).subscribe();
+        assertTrue(admin.topics().getSubscriptions(topicName.toString()).contains(subscriptionName));
+    }
+
+    @Test
+    public void testAutoSubscriptionCreationNamespaceDisallowOverridesBroker() throws Exception {
+        final String topic = "persistent://prop/ns-abc/test-subtopic";
+        final String subscriptionName = "test-subtopic-sub-3";
+        final TopicName topicName = TopicName.get(topic);
+
+        admin.topics().createNonPartitionedTopic(topicName.toString());
+
+        pulsar.getConfiguration().setAllowAutoSubscriptionCreation(true);
+        pulsar.getAdminClient().namespaces().setAutoSubscriptionCreation(topicName.getNamespace(),
+                new AutoSubscriptionCreationOverride(false));
+
+        try {
+            pulsarClient.newConsumer().topic(topicName.toString()).subscriptionName(subscriptionName).subscribe();
+            fail("Subscribe operation should have failed");
+        } catch (Exception e) {
+            assertTrue(e instanceof PulsarClientException);
+        }
+        assertFalse(admin.topics().getSubscriptions(topicName.toString()).contains(subscriptionName));
+    }
+
+}

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

@@ -141,48 +141,6 @@ public void testAutoTopicCreationDisableIfNonPartitionedTopicAlreadyExist() thro
         assertTrue(admin.namespaces().getTopics("prop/ns-abc").contains(topicString));
     }
 
-    @Test
-    public void testAutoSubscriptionCreationDisable() throws Exception{
-        pulsar.getConfiguration().setAllowAutoSubscriptionCreation(false);
-
-        final String topicName = "persistent://prop/ns-abc/test-subtopic";
-        final String subscriptionName = "test-subtopic-sub";
-
-        admin.topics().createNonPartitionedTopic(topicName);
-
-        try {
-            pulsarClient.newConsumer().topic(topicName).subscriptionName(subscriptionName).subscribe();
-            fail("Subscribe operation should have failed");
-        } catch (Exception e) {
-            assertTrue(e instanceof PulsarClientException);
-        }
-        assertFalse(admin.topics().getSubscriptions(topicName).contains(subscriptionName));
-
-        // Reset to default
-        pulsar.getConfiguration().setAllowAutoSubscriptionCreation(true);
-    }
-
-    @Test
-    public void testSubscriptionCreationWithAutoCreationDisable() throws Exception{
-        pulsar.getConfiguration().setAllowAutoSubscriptionCreation(false);
-
-        final String topicName = "persistent://prop/ns-abc/test-subtopic";
-        final String subscriptionName = "test-subtopic-sub";
-
-        admin.topics().createNonPartitionedTopic(topicName);
-        assertFalse(admin.topics().getSubscriptions(topicName).contains(subscriptionName));
-
-        // Create the subscription by PulsarAdmin
-        admin.topics().createSubscription(topicName, subscriptionName, MessageId.earliest);
-        assertTrue(admin.topics().getSubscriptions(topicName).contains(subscriptionName));
-
-        // Subscribe operation should be successful
-        pulsarClient.newConsumer().topic(topicName).subscriptionName(subscriptionName).subscribe();
-
-        // Reset to default
-        pulsar.getConfiguration().setAllowAutoSubscriptionCreation(true);
-    }
-
     /**
      * CheckAllowAutoCreation's default value is false.
      * So using getPartitionedTopicMetadata() directly will not produce partitioned topic