[fix][broker] Fix potential deadlock when creating partitioned topic (#24313)

### Motivation

There is still a call to the synchronous method in `internalCreatePartitionedTopic`, which can lead to a deadlock under certain conditions.

### Modifications

- Replaced synchronous call to `getNamespacePolicies()` with `getNamespacePoliciesAsync()`.
- Replaced synchronous call to `getTopicPartitionList()` with `getTopicPartitionListAsync()`.
- Updated `internalCreatePartitionedTopic` to propagate the async flow properly.

(cherry picked from commit 7ca8c0e)

Author: nodece

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java

@@ -305,6 +305,10 @@ protected WorkerService validateAndGetWorkerService() {
         }
     }
 
+    /**
+     * @deprecated Use {@link #getNamespacePoliciesAsync(NamespaceName)} instead.
+     */
+    @Deprecated
     protected Policies getNamespacePolicies(NamespaceName namespaceName) {
         try {
             Policies policies = namespaceResources().getPolicies(namespaceName)
@@ -541,6 +545,10 @@ protected CompletableFuture<List<String>> getPartitionedTopicListAsync(TopicDoma
                 .listPartitionedTopicsAsync(namespaceName, topicDomain);
     }
 
+    /**
+     * @deprecated Use {@link #getTopicPartitionListAsync()} instead.
+     */
+    @Deprecated
     protected List<String> getTopicPartitionList(TopicDomain topicDomain) {
         try {
             return getPulsarResources().getTopicResources().getExistingPartitions(topicName)
@@ -552,6 +560,10 @@ protected List<String> getTopicPartitionList(TopicDomain topicDomain) {
         }
     }
 
+    protected CompletableFuture<List<String>> getTopicPartitionListAsync() {
+        return getPulsarResources().getTopicResources().getExistingPartitions(topicName);
+    }
+
     protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int numPartitions,
                                                   boolean createLocalTopicOnly) {
         internalCreatePartitionedTopic(asyncResponse, numPartitions, createLocalTopicOnly, null);
@@ -571,33 +583,36 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
             return;
         }
         validateNamespaceOperationAsync(topicName.getNamespaceObject(), NamespaceOperation.CREATE_TOPIC)
-                .thenRun(() -> {
-                    Policies policies = null;
-                    try {
-                        policies = getNamespacePolicies(namespaceName);
-                    } catch (RestException e) {
-                        if (e.getResponse().getStatus() != Status.NOT_FOUND.getStatusCode()) {
-                            throw e;
+                .thenCompose((__) -> getNamespacePoliciesAsync(namespaceName).exceptionally(ex -> {
+                    Throwable unwrapped = FutureUtil.unwrapCompletionException(ex);
+                    if (unwrapped instanceof RestException re) {
+                        if (re.getResponse().getStatus() == Status.NOT_FOUND.getStatusCode()) {
+                            return null;
                         }
                     }
-
+                    throw FutureUtil.wrapToCompletionException(ex);
+                }))
+                .thenCompose(policies -> {
                     int maxTopicsPerNamespace = policies != null && policies.max_topics_per_namespace != null
                             ? policies.max_topics_per_namespace : pulsar().getConfig().getMaxTopicsPerNamespace();
 
                     // new create check
                     if (maxTopicsPerNamespace > 0 && !pulsar().getBrokerService().isSystemTopic(topicName)) {
-                        List<String> partitionedTopics = getTopicPartitionList(TopicDomain.persistent);
-                        // exclude created system topic
-                        long topicsCount =
-                                partitionedTopics.stream().filter(t ->
-                                        !pulsar().getBrokerService().isSystemTopic(TopicName.get(t))).count();
-                        if (topicsCount + numPartitions > maxTopicsPerNamespace) {
-                            log.error("[{}] Failed to create partitioned topic {}, "
-                                    + "exceed maximum number of topics in namespace", clientAppId(), topicName);
-                            throw new RestException(Status.PRECONDITION_FAILED,
-                                    "Exceed maximum number of topics in namespace.");
-                        }
+                        return getTopicPartitionListAsync().thenAccept(partitionedTopics -> {
+                            // exclude created system topic
+                            long topicsCount = partitionedTopics.stream()
+                                    .filter(t -> !pulsar().getBrokerService().isSystemTopic(TopicName.get(t)))
+                                    .count();
+                            if (topicsCount + numPartitions > maxTopicsPerNamespace) {
+                                log.error("[{}] Failed to create partitioned topic {}, "
+                                                + "exceed maximum number of topics in namespace", clientAppId(),
+                                        topicName);
+                                throw new RestException(Status.PRECONDITION_FAILED,
+                                        "Exceed maximum number of topics in namespace.");
+                            }
+                        });
                     }
+                    return CompletableFuture.completedFuture(null);
                 })
                 .thenCompose(__ -> checkTopicExistsAsync(topicName))
                 .thenAccept(exists -> {