[pulsar] branch master updated: Fix async method use getTopicReferenceAsync (#16359)

[technoboy]

This is an automated email from the ASF dual-hosted git repository.

technoboy pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new 59d6bb40105 Fix async method use getTopicReferenceAsync (#16359)
59d6bb40105 is described below

commit 59d6bb40105011930bfa6587574bb264814c2271
Author: Xiaoyu Hou <anonhx...@gmail.com>
AuthorDate: Wed Jul 6 10:35:49 2022 +0800

    Fix async method use getTopicReferenceAsync (#16359)
---
 .../pulsar/broker/admin/impl/PersistentTopicsBase.java       | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

diff --git 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
index 5815e31d535..908874843ac 100644
--- 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
+++ 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -1541,8 +1541,8 @@ public class PersistentTopicsBase extends AdminResource {
                                                                   String 
subName, boolean authoritative) {
         validateTopicOwnershipAsync(topicName, authoritative)
             .thenRun(() -> validateTopicOperation(topicName, 
TopicOperation.UNSUBSCRIBE))
-            .thenCompose(__ -> {
-                Topic topic = getTopicReference(topicName);
+            .thenCompose(__ -> getTopicReferenceAsync(topicName))
+            .thenCompose(topic -> {
                 Subscription sub = topic.getSubscription(subName);
                 if (sub == null) {
                     throw new RestException(Status.NOT_FOUND,
@@ -1575,8 +1575,8 @@ public class PersistentTopicsBase extends AdminResource {
                                       boolean authoritative) {
         validateTopicOwnershipAsync(topicName, authoritative)
                 .thenRun(() -> validateTopicOperation(topicName, 
TopicOperation.CONSUME))
-                .thenCompose(__ -> {
-                    Topic topic = getTopicReference(topicName);
+                .thenCompose(__ -> getTopicReferenceAsync(topicName))
+                .thenCompose(topic -> {
                     Subscription sub = topic.getSubscription(subName);
                     if (sub == null) {
                         throw new RestException(Status.NOT_FOUND,
@@ -1705,8 +1705,8 @@ public class PersistentTopicsBase extends AdminResource {
                                                                             
String subName, boolean authoritative) {
         validateTopicOwnershipAsync(topicName, authoritative)
                 .thenRun(() -> validateTopicOperation(topicName, 
TopicOperation.UNSUBSCRIBE))
-                .thenCompose(__ -> {
-                    Topic topic = getTopicReference(topicName);
+                .thenCompose(__ -> getTopicReferenceAsync(topicName))
+                .thenCompose(topic -> {
                     Subscription sub = topic.getSubscription(subName);
                     if (sub == null) {
                         throw new RestException(Status.NOT_FOUND,