[Issuse 13640][broker] Fix non persistent topic subscription error. (#…

…13685)

Fixes #13640

If pulsar broker started with `allowAutoSubscriptionCreation=false`, there are no way to subscribe non persistent topic.

Add `isPersistent` check in `ServerCnx`.

(cherry picked from commit 3dbe418)

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

@@ -605,7 +605,7 @@ public void createSubscription(@Suspended final AsyncResponse asyncResponse, @Pa
         try {
             validateTopicName(property, cluster, namespace, topic);
             if (!topicName.isPersistent()) {
-                throw new RestException(Response.Status.BAD_REQUEST, "Create subscription on non-persistent topic"
+                throw new RestException(Response.Status.BAD_REQUEST, "Create subscription on non-persistent topic "
                         + "can only be done through client");
             }
             internalCreateSubscription(asyncResponse, decode(encodedSubName), messageId, authoritative, replicated);

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

@@ -1326,7 +1326,7 @@ public void createSubscription(
         try {
             validateTopicName(tenant, namespace, topic);
             if (!topicName.isPersistent()) {
-                throw new RestException(Response.Status.BAD_REQUEST, "Create subscription on non-persistent topic"
+                throw new RestException(Response.Status.BAD_REQUEST, "Create subscription on non-persistent topic "
                         + "can only be done through client");
             }
             internalCreateSubscription(asyncResponse, decode(encodedSubName), messageId, authoritative, replicated);

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

@@ -1014,7 +1014,8 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
 
                                     boolean rejectSubscriptionIfDoesNotExist = isDurable
                                         && !service.isAllowAutoSubscriptionCreation(topicName.toString())
-                                        && !topic.getSubscriptions().containsKey(subscriptionName);
+                                        && !topic.getSubscriptions().containsKey(subscriptionName)
+                                        && topic.isPersistent();
 
                                     if (rejectSubscriptionIfDoesNotExist) {
                                         return FutureUtil

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

@@ -244,6 +244,8 @@ default boolean isSystemTopic() {
         return false;
     }
 
+    boolean isPersistent();
+
     /* ------ Transaction related ------ */
 
     /**

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java

@@ -1075,4 +1075,9 @@ public CompletableFuture<Void> truncate() {
     protected boolean isTerminated() {
         return false;
     }
+
+    @Override
+    public boolean isPersistent() {
+        return false;
+    }
 }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

@@ -2927,6 +2927,11 @@ public boolean isSystemTopic() {
         return false;
     }
 
+    @Override
+    public boolean isPersistent() {
+        return true;
+    }
+
     private synchronized void fence() {
         isFenced = true;
         ScheduledFuture<?> monitoringTask = this.fencedTopicMonitoringTask;

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

@@ -131,4 +131,18 @@ public void testAutoSubscriptionCreationNamespaceDisallowOverridesBroker() throw
         assertFalse(admin.topics().getSubscriptions(topicName.toString()).contains(subscriptionName));
     }
 
+    @Test
+    public void testNonPersistentTopicSubscriptionCreationWithAutoCreationDisable() throws Exception {
+        pulsar.getConfiguration().setAllowAutoSubscriptionCreation(false);
+
+        final String topicName = "non-persistent://prop/ns-abc/test-subtopic-" + testId.getAndIncrement();
+        final String subscriptionName = "test-subtopic-sub";
+
+        admin.topics().createNonPartitionedTopic(topicName);
+
+        // Subscribe operation should be successful
+        pulsarClient.newConsumer().topic(topicName).subscriptionName(subscriptionName).subscribe();
+        assertTrue(admin.topics().getSubscriptions(topicName).contains(subscriptionName));
+    }
+
 }