You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by bo...@apache.org on 2022/11/11 07:46:29 UTC
[pulsar] branch branch-2.9 updated: [fix][broker] Pass subscriptionName to auth service (#17123)
This is an automated email from the ASF dual-hosted git repository.
bogong pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.9 by this push:
new 702bbe91e6a [fix][broker] Pass subscriptionName to auth service (#17123)
702bbe91e6a is described below
commit 702bbe91e6a8aa1bda5a086742240d16063d33bd
Author: Michael Marshall <mm...@apache.org>
AuthorDate: Thu Aug 18 10:48:53 2022 -0700
[fix][broker] Pass subscriptionName to auth service (#17123)
(cherry picked from commit e9c3e21c7a39565e15df1671357dc404527c533f)
---
.../pulsar/broker/admin/impl/PersistentTopicsBase.java | 14 +++++++-------
1 file changed, 7 insertions(+), 7 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 0e591dee4ff..0ee757bf9e9 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
@@ -1491,7 +1491,7 @@ public class PersistentTopicsBase extends AdminResource {
private void internalDeleteSubscriptionForNonPartitionedTopic(AsyncResponse asyncResponse,
String subName, boolean authoritative) {
validateTopicOwnershipAsync(topicName, authoritative)
- .thenRun(() -> validateTopicOperation(topicName, TopicOperation.UNSUBSCRIBE))
+ .thenRun(() -> validateTopicOperation(topicName, TopicOperation.UNSUBSCRIBE, subName))
.thenCompose(__ -> {
Topic topic = getTopicReference(topicName);
Subscription sub = topic.getSubscription(subName);
@@ -1591,7 +1591,7 @@ public class PersistentTopicsBase extends AdminResource {
private void internalDeleteSubscriptionForNonPartitionedTopicForcefully(AsyncResponse asyncResponse,
String subName, boolean authoritative) {
validateTopicOwnershipAsync(topicName, authoritative)
- .thenRun(() -> validateTopicOperation(topicName, TopicOperation.UNSUBSCRIBE))
+ .thenRun(() -> validateTopicOperation(topicName, TopicOperation.UNSUBSCRIBE, subName))
.thenCompose(__ -> {
Topic topic = getTopicReference(topicName);
Subscription sub = topic.getSubscription(subName);
@@ -1746,7 +1746,7 @@ public class PersistentTopicsBase extends AdminResource {
}
validateTopicOwnership(topicName, authoritative);
- validateTopicOperation(topicName, TopicOperation.SKIP);
+ validateTopicOperation(topicName, TopicOperation.SKIP, subName);
PersistentTopic topic = (PersistentTopic) getTopicReference(topicName);
try {
@@ -2151,7 +2151,7 @@ public class PersistentTopicsBase extends AdminResource {
validateTopicOwnershipAsync(topicName, authoritative)
.thenCompose(__ -> {
- validateTopicOperation(topicName, TopicOperation.SUBSCRIBE);
+ validateTopicOperation(topicName, TopicOperation.SUBSCRIBE, subscriptionName);
return pulsar().getBrokerService().getTopic(topicName.toString(), isAllowAutoTopicCreation);
}).thenApply(optTopic -> {
if (optTopic.isPresent()) {
@@ -2448,7 +2448,7 @@ public class PersistentTopicsBase extends AdminResource {
}
validateTopicOwnership(topicName, authoritative);
- validateTopicOperation(topicName, TopicOperation.PEEK_MESSAGES);
+ validateTopicOperation(topicName, TopicOperation.PEEK_MESSAGES, subName);
if (!(getTopicReference(topicName) instanceof PersistentTopic)) {
log.error("[{}] Not supported operation of non-persistent topic {} {}", clientAppId(), topicName,
@@ -3294,7 +3294,7 @@ public class PersistentTopicsBase extends AdminResource {
}
validateTopicOwnership(topicName, authoritative);
- validateTopicOperation(topicName, TopicOperation.EXPIRE_MESSAGES);
+ validateTopicOperation(topicName, TopicOperation.EXPIRE_MESSAGES, subName);
if (!(getTopicReference(topicName) instanceof PersistentTopic)) {
log.error("[{}] Not supported operation of non-persistent topic {} {}", clientAppId(), topicName, subName);
@@ -3357,7 +3357,7 @@ public class PersistentTopicsBase extends AdminResource {
}
validateTopicOwnership(topicName, authoritative);
- validateTopicOperation(topicName, TopicOperation.EXPIRE_MESSAGES);
+ validateTopicOperation(topicName, TopicOperation.EXPIRE_MESSAGES, subName);
log.info("[{}][{}] received expire messages on subscription {} to position {}", clientAppId(), topicName,
subName, messageId);