You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2022/07/06 02:35:55 UTC
[pulsar] branch master updated: Fix async method use getTopicReferenceAsync (#16359)
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 <An...@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,