[fix][broker] fix wrong method name checkTopicExists. (#24293)

The current method `checkTopicExists` is an asynchronous method but follows a synchronous naming convention (lacking the `Async` suffix). This naming inconsistency can mislead developers into assuming it's a blocking operation, potentially causing misuse in client code. Since this method is `public`, we cannot remove it directly without breaking backward compatibility.

1. **Introduce a new asynchronous method**:
   - Added `checkTopicExistsAsync()` with the correct asynchronous naming convention.
   - Internally delegates to the original `checkTopicExists()` method to retain existing logic.

2. **Deprecate the original method**:
   - Marked `checkTopicExists()` as `@Deprecated` with a note directing users to the new `checkTopicExistsAsync()`.

3. **Refactor internal usages**:
   - Updated all internal calls to use `checkTopicExistsAsync()` instead of the deprecated method.

4. **Documentation updates**:
   - Added Javadoc to `checkTopicExists()` clarifying its deprecated status and replacement.

This approach maintains backward compatibility while aligning method names with their asynchronous behavior.

(cherry picked from commit af24849)

Author: liangyepianzhou

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

@@ -808,7 +808,7 @@ public void updatePropertiesFailed(ManagedLedgerException exception, Object ctx)
     }
 
     protected CompletableFuture<Void> internalCheckTopicExists(TopicName topicName) {
-        return pulsar().getNamespaceService().checkTopicExists(topicName)
+        return pulsar().getNamespaceService().checkTopicExistsAsync(topicName)
                 .thenAccept(info -> {
                     boolean exists = info.isExists();
                     info.recycle();
@@ -5638,7 +5638,7 @@ protected CompletableFuture<Void> validateShadowTopics(List<String> shadowTopics
                     return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
                             "Only persistent topic can be set as shadow topic"));
                 }
-                futures.add(pulsar().getNamespaceService().checkTopicExists(shadowTopicName)
+                futures.add(pulsar().getNamespaceService().checkTopicExistsAsync(shadowTopicName)
                         .thenAccept(info -> {
                             boolean exists = info.isExists();
                             info.recycle();

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

@@ -75,7 +75,7 @@ protected CompletableFuture<LookupData> internalLookupTopicAsync(final TopicName
                         return CompletableFuture.completedFuture(true);
                     }
                     // Case-2: Persistent topic.
-                    return pulsar().getNamespaceService().checkTopicExists(topicName).thenCompose(info -> {
+                    return pulsar().getNamespaceService().checkTopicExistsAsync(topicName).thenCompose(info -> {
                         boolean exists = info.isExists();
                         info.recycle();
                         if (exists) {

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

@@ -1368,8 +1368,21 @@ public CompletableFuture<List<String>> getOwnedTopicListForNamespaceBundle(Names
     }
 
     /***
-     * Check topic exists( partitioned or non-partitioned ).
+     * Checks whether the topic exists( partitioned or non-partitioned ).
      */
+    public CompletableFuture<TopicExistsInfo> checkTopicExistsAsync(TopicName topic) {
+        return checkTopicExists(topic);
+    }
+
+    /**
+     * Checks whether the topic exists( partitioned or non-partitioned ).
+     *
+     * @deprecated This method uses a misleading synchronous name for an asynchronous operation.
+     *             Use {@link #checkTopicExistsAsync(TopicName topic)} instead.
+     *
+     * @see #checkTopicExistsAsync(TopicName topic)
+     */
+    @Deprecated
     public CompletableFuture<TopicExistsInfo> checkTopicExists(TopicName topic) {
         return pulsar.getBrokerService()
             .fetchPartitionedTopicMetadataAsync(TopicName.get(topic.toString()))

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

@@ -3238,7 +3238,7 @@ public CompletableFuture<PartitionedTopicMetadata> fetchPartitionedTopicMetadata
         if (pulsar.getNamespaceService() == null) {
             return FutureUtil.failedFuture(new NamingException("namespace service is not ready"));
         }
-        return pulsar.getNamespaceService().checkTopicExists(topicName).thenComposeAsync(topicExistsInfo -> {
+        return pulsar.getNamespaceService().checkTopicExistsAsync(topicName).thenComposeAsync(topicExistsInfo -> {
             final boolean topicExists = topicExistsInfo.isExists();
             final TopicType topicType = topicExistsInfo.getTopicType();
             final Integer partitions = topicExistsInfo.getPartitions();

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

@@ -593,7 +593,7 @@ protected void handlePartitionMetadataRequest(CommandPartitionedTopicMetadata pa
                                 && brokerAllowAutoCreate;
                         if (!autoCreateIfNotExist) {
                             NamespaceService namespaceService = getBrokerService().getPulsar().getNamespaceService();
-                            namespaceService.checkTopicExists(topicName).thenAccept(topicExistsInfo -> {
+                            namespaceService.checkTopicExistsAsync(topicName).thenAccept(topicExistsInfo -> {
                                 lookupSemaphore.release();
                                 if (!topicExistsInfo.isExists()) {
                                     writeAndFlush(Commands.newPartitionMetadataResponse(

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

@@ -360,7 +360,7 @@ public void testLookUpWithException() throws Exception {
         CompletableFuture existFuture = new CompletableFuture();
         existFuture.complete(TopicExistsInfo.newNonPartitionedTopicExists());
         doReturn(future).when(nameSpaceService).getBrokerServiceUrlAsync(any(), any());
-        doReturn(existFuture).when(nameSpaceService).checkTopicExists(any());
+        doReturn(existFuture).when(nameSpaceService).checkTopicExistsAsync(any());
         CompletableFuture existBooleanFuture = new CompletableFuture();
         existBooleanFuture.complete(false);
         doReturn(existBooleanFuture).when(nameSpaceService).checkNonPartitionedTopicExists(any());
@@ -385,7 +385,7 @@ public void testLookUpTopicNotExist() throws Exception {
         existFuture.complete(TopicExistsInfo.newTopicNotExists());
         CompletableFuture existBooleanFuture = new CompletableFuture();
         existBooleanFuture.complete(false);
-        doReturn(existFuture).when(nameSpaceService).checkTopicExists(any());
+        doReturn(existFuture).when(nameSpaceService).checkTopicExistsAsync(any());
         doReturn(existBooleanFuture).when(nameSpaceService).checkNonPartitionedTopicExists(any());
         doReturn(nameSpaceService).when(pulsar).getNamespaceService();
         AsyncResponse asyncResponse = mock(AsyncResponse.class);

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

@@ -152,7 +152,7 @@ public void testLookupTopicNotExist() throws Exception {
         NamespaceService namespaceService = pulsar.getNamespaceService();
         CompletableFuture<TopicExistsInfo> future = new CompletableFuture<>();
         future.complete(TopicExistsInfo.newTopicNotExists());
-        doReturn(future).when(namespaceService).checkTopicExists(any(TopicName.class));
+        doReturn(future).when(namespaceService).checkTopicExistsAsync(any(TopicName.class));
         CompletableFuture<Boolean> booleanFuture = new CompletableFuture<>();
         booleanFuture.complete(false);
         doReturn(booleanFuture).when(namespaceService).checkNonPartitionedTopicExists(any(TopicName.class));
@@ -266,7 +266,7 @@ public void testValidateReplicationSettingsOnNamespace() throws Exception {
         NamespaceService namespaceService = pulsar.getNamespaceService();
         CompletableFuture<TopicExistsInfo> future = new CompletableFuture<>();
         future.complete(TopicExistsInfo.newTopicNotExists());
-        doReturn(future).when(namespaceService).checkTopicExists(any(TopicName.class));
+        doReturn(future).when(namespaceService).checkTopicExistsAsync(any(TopicName.class));
         CompletableFuture<Boolean> booleanFuture = new CompletableFuture<>();
         booleanFuture.complete(false);
         doReturn(future).when(namespaceService).checkNonPartitionedTopicExists(any(TopicName.class));
@@ -303,7 +303,7 @@ public void topicNotFound() throws Exception {
         NamespaceService namespaceService = pulsar.getNamespaceService();
         CompletableFuture<TopicExistsInfo> future = new CompletableFuture<>();
         future.complete(TopicExistsInfo.newTopicNotExists());
-        doReturn(future).when(namespaceService).checkTopicExists(any(TopicName.class));
+        doReturn(future).when(namespaceService).checkTopicExistsAsync(any(TopicName.class));
 
         // Get the current semaphore first
         Integer state1 = pulsar.getBrokerService().getLookupRequestSemaphore().availablePermits();

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

@@ -220,7 +220,7 @@ public void testRemoveCluster() throws Exception {
             assertFalse(tps.containsKey(topicP1));
             assertFalse(tps.containsKey(topicChangeEvents));
             assertFalse(pulsar1.getNamespaceService()
-                    .checkTopicExists(TopicName.get(topicChangeEvents))
+                    .checkTopicExistsAsync(TopicName.get(topicChangeEvents))
                     .get(5, TimeUnit.SECONDS).isExists());
             // Verify: schema will be removed in local cluster, and remote cluster will not.
             List<CompletableFuture<StoredSchema>> schemaList13

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

@@ -213,10 +213,10 @@ public void testRemoveCluster() throws Exception {
                     = pulsar1.getBrokerService().getTopics();
             assertFalse(tps.containsKey(topic));
             assertFalse(tps.containsKey(topicChangeEvents));
-            assertFalse(pulsar1.getNamespaceService().checkTopicExists(TopicName.get(topic))
+            assertFalse(pulsar1.getNamespaceService().checkTopicExistsAsync(TopicName.get(topic))
                     .get(5, TimeUnit.SECONDS).isExists());
             assertFalse(pulsar1.getNamespaceService()
-                    .checkTopicExists(TopicName.get(topicChangeEvents))
+                    .checkTopicExistsAsync(TopicName.get(topicChangeEvents))
                     .get(5, TimeUnit.SECONDS).isExists());
         });