remove the param assumedNonPartitionedNonPersistentTopicAlwaysExists

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

@@ -724,7 +724,7 @@ protected CompletableFuture<Void> internalCheckTopicExists(TopicName topicName)
     }
 
     protected CompletableFuture<Void> internalCheckNonPartitionedTopicExists(TopicName topicName) {
-        return pulsar().getNamespaceService().checkNonPartitionedTopicExists(topicName, false)
+        return pulsar().getNamespaceService().checkNonPartitionedTopicExists(topicName)
                 .thenAccept(exist -> {
                     if (!exist) {
                         throw new RestException(Status.NOT_FOUND, getTopicNotFoundErrorMessage(topicName.toString()));

pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookupBase.java

@@ -67,10 +67,15 @@ protected CompletableFuture<LookupData> internalLookupTopicAsync(final TopicName
                 .thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(topicName.getNamespaceObject()))
                 .thenCompose(__ -> validateTopicOperationAsync(topicName, TopicOperation.LOOKUP, null))
                 .thenCompose(__ -> {
+                    // Case-1: Non-persistent topic.
                     // Currently, it's hard to check the non-persistent-non-partitioned topic, because it only exists
                     // in the broker, it doesn't have metadata. If the topic is non-persistent and non-partitioned,
-                    // we'll return the true flag.
-                    return pulsar().getNamespaceService().checkTopicExists(topicName, true).thenCompose(info -> {
+                    // we'll return the true flag. So either it is a partitioned topic or not, the result will be true.
+                    if (!topicName.isPersistent()) {
+                        return CompletableFuture.completedFuture(true);
+                    }
+                    // Case-2: Persistent topic.
+                    return pulsar().getNamespaceService().checkTopicExists(topicName).thenCompose(info -> {
                         boolean exists = info.isExists();
                         info.recycle();
                         if (exists) {

pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java

@@ -1403,47 +1403,31 @@ public CompletableFuture<List<String>> getOwnedTopicListForNamespaceBundle(Names
                 });
     }
 
-    public CompletableFuture<TopicExistsInfo> checkTopicExists(TopicName topic) {
-        return checkTopicExists(topic, false);
-    }
-
     /***
      * Check topic exists( partitioned or non-partitioned ).
-     * @param assumedNonPartitionedNonPersistentTopicAlwaysExists Currently, it's hard to check the
-     *        non-persistent-non-partitioned topic, because it only exists in the broker, it doesn't have metadata.
-     *        If the topic is non-persistent and non-partitioned, we'll return the true flag.
      */
-    public CompletableFuture<TopicExistsInfo> checkTopicExists(TopicName topic,
-                                                    boolean assumedNonPartitionedNonPersistentTopicAlwaysExists) {
+    public CompletableFuture<TopicExistsInfo> checkTopicExists(TopicName topic) {
         return pulsar.getBrokerService()
             .fetchPartitionedTopicMetadataAsync(TopicName.get(topic.toString()))
             .thenCompose(metadata -> {
                 if (metadata.partitions > 0) {
                     return CompletableFuture.completedFuture(
                             TopicExistsInfo.newPartitionedTopicExists(metadata.partitions));
                 }
-                return checkNonPartitionedTopicExists(topic, assumedNonPartitionedNonPersistentTopicAlwaysExists)
+                return checkNonPartitionedTopicExists(topic)
                     .thenApply(b -> b ? TopicExistsInfo.newNonPartitionedTopicExists()
                             : TopicExistsInfo.newTopicNotExists());
             });
     }
 
     /***
      * Check non-partitioned topic exists.
-     * @param assumedNonPartitionedNonPersistentTopicAlwaysExists Currently, it's hard to check the
-     *        non-persistent-non-partitioned topic, because it only exists in the broker, it doesn't have metadata.
-     *        If the topic is non-persistent and non-partitioned, we'll return the true flag.
      */
-    public CompletableFuture<Boolean> checkNonPartitionedTopicExists(TopicName topic,
-                                                     boolean assumedNonPartitionedNonPersistentTopicAlwaysExists) {
+    public CompletableFuture<Boolean> checkNonPartitionedTopicExists(TopicName topic) {
         if (topic.isPersistent()) {
             return pulsar.getPulsarResources().getTopicResources().persistentTopicExists(topic);
         } else {
-            if (assumedNonPartitionedNonPersistentTopicAlwaysExists) {
-                return CompletableFuture.completedFuture(true);
-            } else {
-                return checkNonPersistentNonPartitionedTopicExists(topic.toString());
-            }
+            return checkNonPersistentNonPartitionedTopicExists(topic.toString());
         }
     }
 

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicsTest.java

@@ -361,10 +361,9 @@ public void testLookUpWithException() throws Exception {
         existFuture.complete(TopicExistsInfo.newNonPartitionedTopicExists());
         doReturn(future).when(nameSpaceService).getBrokerServiceUrlAsync(any(), any());
         doReturn(existFuture).when(nameSpaceService).checkTopicExists(any());
-        doReturn(existFuture).when(nameSpaceService).checkTopicExists(any(), anyBoolean());
         CompletableFuture existBooleanFuture = new CompletableFuture();
         existBooleanFuture.complete(false);
-        doReturn(existBooleanFuture).when(nameSpaceService).checkNonPartitionedTopicExists(any(), anyBoolean());
+        doReturn(existBooleanFuture).when(nameSpaceService).checkNonPartitionedTopicExists(any());
         doReturn(nameSpaceService).when(pulsar).getNamespaceService();
         AsyncResponse asyncResponse = mock(AsyncResponse.class);
         ProducerMessages producerMessages = new ProducerMessages();
@@ -387,7 +386,7 @@ public void testLookUpTopicNotExist() throws Exception {
         CompletableFuture existBooleanFuture = new CompletableFuture();
         existBooleanFuture.complete(false);
         doReturn(existFuture).when(nameSpaceService).checkTopicExists(any());
-        doReturn(existBooleanFuture).when(nameSpaceService).checkNonPartitionedTopicExists(any(), anyBoolean());
+        doReturn(existBooleanFuture).when(nameSpaceService).checkNonPartitionedTopicExists(any());
         doReturn(nameSpaceService).when(pulsar).getNamespaceService();
         AsyncResponse asyncResponse = mock(AsyncResponse.class);
         ProducerMessages producerMessages = new ProducerMessages();

pulsar-broker/src/test/java/org/apache/pulsar/broker/lookup/http/HttpTopicLookupv2Test.java

@@ -154,11 +154,9 @@ public void testLookupTopicNotExist() throws Exception {
         CompletableFuture<TopicExistsInfo> future = new CompletableFuture<>();
         future.complete(TopicExistsInfo.newTopicNotExists());
         doReturn(future).when(namespaceService).checkTopicExists(any(TopicName.class));
-        doReturn(future).when(namespaceService).checkTopicExists(any(TopicName.class), anyBoolean());
         CompletableFuture<Boolean> booleanFuture = new CompletableFuture<>();
         booleanFuture.complete(false);
-        doReturn(booleanFuture).when(namespaceService).checkNonPartitionedTopicExists(any(TopicName.class),
-                anyBoolean());
+        doReturn(booleanFuture).when(namespaceService).checkNonPartitionedTopicExists(any(TopicName.class));
 
         AsyncResponse asyncResponse1 = mock(AsyncResponse.class);
         destLookup.lookupTopicAsync(asyncResponse1, TopicDomain.persistent.value(), "myprop", "usc", "ns2", "topic_not_exist", false, null, null);
@@ -270,10 +268,9 @@ public void testValidateReplicationSettingsOnNamespace() throws Exception {
         CompletableFuture<TopicExistsInfo> future = new CompletableFuture<>();
         future.complete(TopicExistsInfo.newTopicNotExists());
         doReturn(future).when(namespaceService).checkTopicExists(any(TopicName.class));
-        doReturn(future).when(namespaceService).checkTopicExists(any(TopicName.class), anyBoolean());
         CompletableFuture<Boolean> booleanFuture = new CompletableFuture<>();
         booleanFuture.complete(false);
-        doReturn(future).when(namespaceService).checkNonPartitionedTopicExists(any(TopicName.class), anyBoolean());
+        doReturn(future).when(namespaceService).checkNonPartitionedTopicExists(any(TopicName.class));
         destLookup.lookupTopicAsync(asyncResponse, TopicDomain.persistent.value(), property, cluster, ns2,
                 "invalid-localCluster", false, null, null);
         verify(asyncResponse).resume(arg.capture());