You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2021/10/12 18:43:26 UTC
[GitHub] [pulsar] merlimat opened a new pull request #12339: Avoid potentially blocking calls to metadata on critical threads
merlimat opened a new pull request #12339:
URL: https://github.com/apache/pulsar/pull/12339
### Motivation
We have several places that are accessing the namespace policies in a synchronous way from the critical IO threads of the broker. These accesses are almost always non-blocking calls since the policies are already cached (given that the topic was just loaded and the policies were checked then).
There are few corner cases though in which we would incur in cache misses and that could have a big impact, up to result in a complete deadlock, if we exhaust all the threads in the ordered executors.
For example this can happen when there are thousands of producers/consumers connecting and there is a change in the policies that triggers a cache invalidation.
We must have no potentially blocking calls in the critical path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on pull request #12339: Avoid potentially blocking calls to metadata on critical threads
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on pull request #12339:
URL: https://github.com/apache/pulsar/pull/12339#issuecomment-941800067
@michaeljmarshall Please help review again
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a change in pull request #12339: Avoid potentially blocking calls to metadata on critical threads
Posted by GitBox <gi...@apache.org>.
michaeljmarshall commented on a change in pull request #12339:
URL: https://github.com/apache/pulsar/pull/12339#discussion_r727528015
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
##########
@@ -209,18 +196,12 @@ protected boolean isConsumersExceededOnTopic() {
Integer maxConsumers = getTopicPolicies().map(TopicPolicies::getMaxConsumerPerTopic).orElse(null);
if (maxConsumers == null) {
Policies policies;
- try {
- // Use getDataIfPresent from zk cache to make the call non-blocking and prevent deadlocks
- policies = brokerService.pulsar().getPulsarResources().getNamespaceResources().getPolicies(
- TopicName.get(topic).getNamespaceObject())
- .orElseGet(() -> new Policies());
+ // Use getDataIfPresent from zk cache to make the call non-blocking and prevent deadlocks
+ policies = brokerService.pulsar().getPulsarResources().getNamespaceResources().getPoliciesIfCached(
+ TopicName.get(topic).getNamespaceObject())
+ .orElseGet(() -> new Policies());
- if (policies == null) {
- policies = new Policies();
- }
- } catch (Exception e) {
- log.warn("[{}] Failed to get namespace policies that include max number of consumers: {}", topic,
- e.getMessage());
+ if (policies == null) {
Review comment:
Can we consider this null check no longer necessary? I recognize that an `Optional` can be null, but it probably _shouldn't_ be null.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
##########
@@ -630,7 +611,7 @@ protected void checkTopicFenced() throws BrokerServiceException {
}
}
- protected void internalAddProducer(Producer producer) throws BrokerServiceException {
+ protected void internalAddProducer(Producer producer) throws BrokerServiceException{
Review comment:
Nit: I think we can leave the space here.
```suggestion
protected void internalAddProducer(Producer producer) throws BrokerServiceException {
```
##########
File path: pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
##########
@@ -322,48 +324,42 @@ public void initialize(ServiceConfiguration conf, PulsarResources pulsarResource
return updateSubscriptionPermissionAsync(namespace, subscriptionName, Collections.singleton(role), true);
}
- private CompletableFuture<Void> updateSubscriptionPermissionAsync(NamespaceName namespace, String subscriptionName, Set<String> roles,
- boolean remove) {
- CompletableFuture<Void> result = new CompletableFuture<>();
-
+ private CompletableFuture<Void> updateSubscriptionPermissionAsync(NamespaceName namespace, String subscriptionName,
+ Set<String> roles,
+ boolean remove) {
try {
validatePoliciesReadOnlyAccess();
} catch (Exception e) {
- result.completeExceptionally(e);
+ return FutureUtil.failedFuture(e);
}
- try {
- Policies policies = pulsarResources.getNamespaceResources().getPolicies(namespace)
- .orElseThrow(() -> new NotFoundException(namespace + " not found"));
- if (remove) {
- if (policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName) != null) {
- policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName).removeAll(roles);
- }else {
- log.info("[{}] Couldn't find role {} while revoking for sub = {}", namespace, subscriptionName, roles);
- result.completeExceptionally(new IllegalArgumentException("couldn't find subscription"));
- return result;
- }
- } else {
- policies.auth_policies.getSubscriptionAuthentication().put(subscriptionName, roles);
- }
- pulsarResources.getNamespaceResources().setPolicies(namespace, (data)->policies);
+ CompletableFuture<Void> future =
+ pulsarResources.getNamespaceResources().setPoliciesAsync(namespace, policies -> {
+ if (remove) {
+ if (policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName) != null) {
+ policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName)
+ .removeAll(roles);
Review comment:
Nit: I think we could simplify this with a local variable. As a result, we'll only retrieve the reference from the map once. What do you think?
```suggestion
Set<String> currentRoles = policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName);
if (currentRoles != null) {
currentRoles.removeAll(roles);
```
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java
##########
@@ -180,7 +180,7 @@ public static boolean isDispatchRateNeeded(BrokerService brokerService, Optional
return true;
}
- policies = policies.isPresent() ? policies : getPolicies(brokerService, topicName);
+ policies = policies.isPresent() ? policies : getPolicies(brokerService, topicName);
Review comment:
Nit: we can remove the extra space here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a change in pull request #12339: Avoid potentially blocking calls to metadata on critical threads
Posted by GitBox <gi...@apache.org>.
michaeljmarshall commented on a change in pull request #12339:
URL: https://github.com/apache/pulsar/pull/12339#discussion_r728240747
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
##########
@@ -802,9 +780,19 @@ private void updatePublishDispatcher(Policies policies) {
return;
}
+ Policies policies;
+ try {
+ policies = brokerService.pulsar().getPulsarResources().getNamespaceResources().getPoliciesIfCached(
+ TopicName.get(topic).getNamespaceObject())
+ .orElseGet(() -> new Policies());
Review comment:
I see this was added in the most recent commit to help with a mocked test. This block doesn't throw any checked exceptions. Do we expect `RuntimeExceptions` from it? If we do, should we wrap other similar code blocks in this PR with try and catch? If we don't, perhaps we should update the test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a change in pull request #12339: Avoid potentially blocking calls to metadata on critical threads
Posted by GitBox <gi...@apache.org>.
michaeljmarshall commented on a change in pull request #12339:
URL: https://github.com/apache/pulsar/pull/12339#discussion_r727528015
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
##########
@@ -209,18 +196,12 @@ protected boolean isConsumersExceededOnTopic() {
Integer maxConsumers = getTopicPolicies().map(TopicPolicies::getMaxConsumerPerTopic).orElse(null);
if (maxConsumers == null) {
Policies policies;
- try {
- // Use getDataIfPresent from zk cache to make the call non-blocking and prevent deadlocks
- policies = brokerService.pulsar().getPulsarResources().getNamespaceResources().getPolicies(
- TopicName.get(topic).getNamespaceObject())
- .orElseGet(() -> new Policies());
+ // Use getDataIfPresent from zk cache to make the call non-blocking and prevent deadlocks
+ policies = brokerService.pulsar().getPulsarResources().getNamespaceResources().getPoliciesIfCached(
+ TopicName.get(topic).getNamespaceObject())
+ .orElseGet(() -> new Policies());
- if (policies == null) {
- policies = new Policies();
- }
- } catch (Exception e) {
- log.warn("[{}] Failed to get namespace policies that include max number of consumers: {}", topic,
- e.getMessage());
+ if (policies == null) {
Review comment:
Can we consider this null check no longer necessary? I recognize that an `Optional` can be null, but it probably _shouldn't_ be null.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
##########
@@ -630,7 +611,7 @@ protected void checkTopicFenced() throws BrokerServiceException {
}
}
- protected void internalAddProducer(Producer producer) throws BrokerServiceException {
+ protected void internalAddProducer(Producer producer) throws BrokerServiceException{
Review comment:
Nit: I think we can leave the space here.
```suggestion
protected void internalAddProducer(Producer producer) throws BrokerServiceException {
```
##########
File path: pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
##########
@@ -322,48 +324,42 @@ public void initialize(ServiceConfiguration conf, PulsarResources pulsarResource
return updateSubscriptionPermissionAsync(namespace, subscriptionName, Collections.singleton(role), true);
}
- private CompletableFuture<Void> updateSubscriptionPermissionAsync(NamespaceName namespace, String subscriptionName, Set<String> roles,
- boolean remove) {
- CompletableFuture<Void> result = new CompletableFuture<>();
-
+ private CompletableFuture<Void> updateSubscriptionPermissionAsync(NamespaceName namespace, String subscriptionName,
+ Set<String> roles,
+ boolean remove) {
try {
validatePoliciesReadOnlyAccess();
} catch (Exception e) {
- result.completeExceptionally(e);
+ return FutureUtil.failedFuture(e);
}
- try {
- Policies policies = pulsarResources.getNamespaceResources().getPolicies(namespace)
- .orElseThrow(() -> new NotFoundException(namespace + " not found"));
- if (remove) {
- if (policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName) != null) {
- policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName).removeAll(roles);
- }else {
- log.info("[{}] Couldn't find role {} while revoking for sub = {}", namespace, subscriptionName, roles);
- result.completeExceptionally(new IllegalArgumentException("couldn't find subscription"));
- return result;
- }
- } else {
- policies.auth_policies.getSubscriptionAuthentication().put(subscriptionName, roles);
- }
- pulsarResources.getNamespaceResources().setPolicies(namespace, (data)->policies);
+ CompletableFuture<Void> future =
+ pulsarResources.getNamespaceResources().setPoliciesAsync(namespace, policies -> {
+ if (remove) {
+ if (policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName) != null) {
+ policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName)
+ .removeAll(roles);
Review comment:
Nit: I think we could simplify this with a local variable. As a result, we'll only retrieve the reference from the map once. What do you think?
```suggestion
Set<String> currentRoles = policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName);
if (currentRoles != null) {
currentRoles.removeAll(roles);
```
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/DispatchRateLimiter.java
##########
@@ -180,7 +180,7 @@ public static boolean isDispatchRateNeeded(BrokerService brokerService, Optional
return true;
}
- policies = policies.isPresent() ? policies : getPolicies(brokerService, topicName);
+ policies = policies.isPresent() ? policies : getPolicies(brokerService, topicName);
Review comment:
Nit: we can remove the extra space here.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
##########
@@ -802,9 +780,19 @@ private void updatePublishDispatcher(Policies policies) {
return;
}
+ Policies policies;
+ try {
+ policies = brokerService.pulsar().getPulsarResources().getNamespaceResources().getPoliciesIfCached(
+ TopicName.get(topic).getNamespaceObject())
+ .orElseGet(() -> new Policies());
Review comment:
I see this was added in the most recent commit to help with a mocked test. This block doesn't throw any checked exceptions. Do we expect `RuntimeExceptions` from it? If we do, should we wrap other similar code blocks in this PR with try and catch? If we don't, perhaps we should update the test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] merlimat commented on a change in pull request #12339: Avoid potentially blocking calls to metadata on critical threads
Posted by GitBox <gi...@apache.org>.
merlimat commented on a change in pull request #12339:
URL: https://github.com/apache/pulsar/pull/12339#discussion_r727515017
##########
File path: pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
##########
@@ -322,48 +324,42 @@ public void initialize(ServiceConfiguration conf, PulsarResources pulsarResource
return updateSubscriptionPermissionAsync(namespace, subscriptionName, Collections.singleton(role), true);
}
- private CompletableFuture<Void> updateSubscriptionPermissionAsync(NamespaceName namespace, String subscriptionName, Set<String> roles,
- boolean remove) {
- CompletableFuture<Void> result = new CompletableFuture<>();
-
+ private CompletableFuture<Void> updateSubscriptionPermissionAsync(NamespaceName namespace, String subscriptionName,
+ Set<String> roles,
+ boolean remove) {
try {
validatePoliciesReadOnlyAccess();
} catch (Exception e) {
- result.completeExceptionally(e);
+ return FutureUtil.failedFuture(e);
}
- try {
- Policies policies = pulsarResources.getNamespaceResources().getPolicies(namespace)
- .orElseThrow(() -> new NotFoundException(namespace + " not found"));
- if (remove) {
- if (policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName) != null) {
- policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName).removeAll(roles);
- }else {
- log.info("[{}] Couldn't find role {} while revoking for sub = {}", namespace, subscriptionName, roles);
- result.completeExceptionally(new IllegalArgumentException("couldn't find subscription"));
- return result;
- }
- } else {
- policies.auth_policies.getSubscriptionAuthentication().put(subscriptionName, roles);
- }
- pulsarResources.getNamespaceResources().setPolicies(namespace, (data)->policies);
+ CompletableFuture<Void> future =
+ pulsarResources.getNamespaceResources().setPoliciesAsync(namespace, policies -> {
+ if (remove) {
+ if (policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName) != null) {
+ policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName)
+ .removeAll(roles);
+ } else {
+ log.info("[{}] Couldn't find role {} while revoking for sub = {}", namespace,
Review comment:
👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] cckellogg commented on a change in pull request #12339: Avoid potentially blocking calls to metadata on critical threads
Posted by GitBox <gi...@apache.org>.
cckellogg commented on a change in pull request #12339:
URL: https://github.com/apache/pulsar/pull/12339#discussion_r727442559
##########
File path: pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
##########
@@ -322,48 +324,42 @@ public void initialize(ServiceConfiguration conf, PulsarResources pulsarResource
return updateSubscriptionPermissionAsync(namespace, subscriptionName, Collections.singleton(role), true);
}
- private CompletableFuture<Void> updateSubscriptionPermissionAsync(NamespaceName namespace, String subscriptionName, Set<String> roles,
- boolean remove) {
- CompletableFuture<Void> result = new CompletableFuture<>();
-
+ private CompletableFuture<Void> updateSubscriptionPermissionAsync(NamespaceName namespace, String subscriptionName,
+ Set<String> roles,
+ boolean remove) {
try {
validatePoliciesReadOnlyAccess();
} catch (Exception e) {
- result.completeExceptionally(e);
+ return FutureUtil.failedFuture(e);
}
- try {
- Policies policies = pulsarResources.getNamespaceResources().getPolicies(namespace)
- .orElseThrow(() -> new NotFoundException(namespace + " not found"));
- if (remove) {
- if (policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName) != null) {
- policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName).removeAll(roles);
- }else {
- log.info("[{}] Couldn't find role {} while revoking for sub = {}", namespace, subscriptionName, roles);
- result.completeExceptionally(new IllegalArgumentException("couldn't find subscription"));
- return result;
- }
- } else {
- policies.auth_policies.getSubscriptionAuthentication().put(subscriptionName, roles);
- }
- pulsarResources.getNamespaceResources().setPolicies(namespace, (data)->policies);
+ CompletableFuture<Void> future =
+ pulsarResources.getNamespaceResources().setPoliciesAsync(namespace, policies -> {
+ if (remove) {
+ if (policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName) != null) {
+ policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName)
+ .removeAll(roles);
+ } else {
+ log.info("[{}] Couldn't find role {} while revoking for sub = {}", namespace,
Review comment:
Looks good to me.
Minor it seems like some of the arguments in the logs are not in order (roles and subscriptionName)?
Should the order be more like this?
```
log.info("[{}] Couldn't find roles={} while revoking for sub={}", namespace, roles, subscriptionName)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] merlimat commented on a change in pull request #12339: Avoid potentially blocking calls to metadata on critical threads
Posted by GitBox <gi...@apache.org>.
merlimat commented on a change in pull request #12339:
URL: https://github.com/apache/pulsar/pull/12339#discussion_r727559908
##########
File path: pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
##########
@@ -322,48 +324,42 @@ public void initialize(ServiceConfiguration conf, PulsarResources pulsarResource
return updateSubscriptionPermissionAsync(namespace, subscriptionName, Collections.singleton(role), true);
}
- private CompletableFuture<Void> updateSubscriptionPermissionAsync(NamespaceName namespace, String subscriptionName, Set<String> roles,
- boolean remove) {
- CompletableFuture<Void> result = new CompletableFuture<>();
-
+ private CompletableFuture<Void> updateSubscriptionPermissionAsync(NamespaceName namespace, String subscriptionName,
+ Set<String> roles,
+ boolean remove) {
try {
validatePoliciesReadOnlyAccess();
} catch (Exception e) {
- result.completeExceptionally(e);
+ return FutureUtil.failedFuture(e);
}
- try {
- Policies policies = pulsarResources.getNamespaceResources().getPolicies(namespace)
- .orElseThrow(() -> new NotFoundException(namespace + " not found"));
- if (remove) {
- if (policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName) != null) {
- policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName).removeAll(roles);
- }else {
- log.info("[{}] Couldn't find role {} while revoking for sub = {}", namespace, subscriptionName, roles);
- result.completeExceptionally(new IllegalArgumentException("couldn't find subscription"));
- return result;
- }
- } else {
- policies.auth_policies.getSubscriptionAuthentication().put(subscriptionName, roles);
- }
- pulsarResources.getNamespaceResources().setPolicies(namespace, (data)->policies);
+ CompletableFuture<Void> future =
+ pulsarResources.getNamespaceResources().setPoliciesAsync(namespace, policies -> {
+ if (remove) {
+ if (policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName) != null) {
+ policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName)
+ .removeAll(roles);
Review comment:
👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] merlimat commented on a change in pull request #12339: Avoid potentially blocking calls to metadata on critical threads
Posted by GitBox <gi...@apache.org>.
merlimat commented on a change in pull request #12339:
URL: https://github.com/apache/pulsar/pull/12339#discussion_r728274526
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
##########
@@ -802,9 +780,19 @@ private void updatePublishDispatcher(Policies policies) {
return;
}
+ Policies policies;
+ try {
+ policies = brokerService.pulsar().getPulsarResources().getNamespaceResources().getPoliciesIfCached(
+ TopicName.get(topic).getNamespaceObject())
+ .orElseGet(() -> new Policies());
Review comment:
Many tests are using mocked versions which would throw exception... then the try/catch was the thing that was making all these tests to work..
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on pull request #12339: Avoid potentially blocking calls to metadata on critical threads
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on pull request #12339:
URL: https://github.com/apache/pulsar/pull/12339#issuecomment-941800067
@michaeljmarshall Please help review again
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] merlimat commented on a change in pull request #12339: Avoid potentially blocking calls to metadata on critical threads
Posted by GitBox <gi...@apache.org>.
merlimat commented on a change in pull request #12339:
URL: https://github.com/apache/pulsar/pull/12339#discussion_r727515017
##########
File path: pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
##########
@@ -322,48 +324,42 @@ public void initialize(ServiceConfiguration conf, PulsarResources pulsarResource
return updateSubscriptionPermissionAsync(namespace, subscriptionName, Collections.singleton(role), true);
}
- private CompletableFuture<Void> updateSubscriptionPermissionAsync(NamespaceName namespace, String subscriptionName, Set<String> roles,
- boolean remove) {
- CompletableFuture<Void> result = new CompletableFuture<>();
-
+ private CompletableFuture<Void> updateSubscriptionPermissionAsync(NamespaceName namespace, String subscriptionName,
+ Set<String> roles,
+ boolean remove) {
try {
validatePoliciesReadOnlyAccess();
} catch (Exception e) {
- result.completeExceptionally(e);
+ return FutureUtil.failedFuture(e);
}
- try {
- Policies policies = pulsarResources.getNamespaceResources().getPolicies(namespace)
- .orElseThrow(() -> new NotFoundException(namespace + " not found"));
- if (remove) {
- if (policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName) != null) {
- policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName).removeAll(roles);
- }else {
- log.info("[{}] Couldn't find role {} while revoking for sub = {}", namespace, subscriptionName, roles);
- result.completeExceptionally(new IllegalArgumentException("couldn't find subscription"));
- return result;
- }
- } else {
- policies.auth_policies.getSubscriptionAuthentication().put(subscriptionName, roles);
- }
- pulsarResources.getNamespaceResources().setPolicies(namespace, (data)->policies);
+ CompletableFuture<Void> future =
+ pulsarResources.getNamespaceResources().setPoliciesAsync(namespace, policies -> {
+ if (remove) {
+ if (policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName) != null) {
+ policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName)
+ .removeAll(roles);
+ } else {
+ log.info("[{}] Couldn't find role {} while revoking for sub = {}", namespace,
Review comment:
👍
##########
File path: pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
##########
@@ -322,48 +324,42 @@ public void initialize(ServiceConfiguration conf, PulsarResources pulsarResource
return updateSubscriptionPermissionAsync(namespace, subscriptionName, Collections.singleton(role), true);
}
- private CompletableFuture<Void> updateSubscriptionPermissionAsync(NamespaceName namespace, String subscriptionName, Set<String> roles,
- boolean remove) {
- CompletableFuture<Void> result = new CompletableFuture<>();
-
+ private CompletableFuture<Void> updateSubscriptionPermissionAsync(NamespaceName namespace, String subscriptionName,
+ Set<String> roles,
+ boolean remove) {
try {
validatePoliciesReadOnlyAccess();
} catch (Exception e) {
- result.completeExceptionally(e);
+ return FutureUtil.failedFuture(e);
}
- try {
- Policies policies = pulsarResources.getNamespaceResources().getPolicies(namespace)
- .orElseThrow(() -> new NotFoundException(namespace + " not found"));
- if (remove) {
- if (policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName) != null) {
- policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName).removeAll(roles);
- }else {
- log.info("[{}] Couldn't find role {} while revoking for sub = {}", namespace, subscriptionName, roles);
- result.completeExceptionally(new IllegalArgumentException("couldn't find subscription"));
- return result;
- }
- } else {
- policies.auth_policies.getSubscriptionAuthentication().put(subscriptionName, roles);
- }
- pulsarResources.getNamespaceResources().setPolicies(namespace, (data)->policies);
+ CompletableFuture<Void> future =
+ pulsarResources.getNamespaceResources().setPoliciesAsync(namespace, policies -> {
+ if (remove) {
+ if (policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName) != null) {
+ policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName)
+ .removeAll(roles);
Review comment:
👍
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
##########
@@ -802,9 +780,19 @@ private void updatePublishDispatcher(Policies policies) {
return;
}
+ Policies policies;
+ try {
+ policies = brokerService.pulsar().getPulsarResources().getNamespaceResources().getPoliciesIfCached(
+ TopicName.get(topic).getNamespaceObject())
+ .orElseGet(() -> new Policies());
Review comment:
Many tests are using mocked versions which would throw exception... then the try/catch was the thing that was making all these tests to work..
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] cckellogg commented on a change in pull request #12339: Avoid potentially blocking calls to metadata on critical threads
Posted by GitBox <gi...@apache.org>.
cckellogg commented on a change in pull request #12339:
URL: https://github.com/apache/pulsar/pull/12339#discussion_r727442559
##########
File path: pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
##########
@@ -322,48 +324,42 @@ public void initialize(ServiceConfiguration conf, PulsarResources pulsarResource
return updateSubscriptionPermissionAsync(namespace, subscriptionName, Collections.singleton(role), true);
}
- private CompletableFuture<Void> updateSubscriptionPermissionAsync(NamespaceName namespace, String subscriptionName, Set<String> roles,
- boolean remove) {
- CompletableFuture<Void> result = new CompletableFuture<>();
-
+ private CompletableFuture<Void> updateSubscriptionPermissionAsync(NamespaceName namespace, String subscriptionName,
+ Set<String> roles,
+ boolean remove) {
try {
validatePoliciesReadOnlyAccess();
} catch (Exception e) {
- result.completeExceptionally(e);
+ return FutureUtil.failedFuture(e);
}
- try {
- Policies policies = pulsarResources.getNamespaceResources().getPolicies(namespace)
- .orElseThrow(() -> new NotFoundException(namespace + " not found"));
- if (remove) {
- if (policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName) != null) {
- policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName).removeAll(roles);
- }else {
- log.info("[{}] Couldn't find role {} while revoking for sub = {}", namespace, subscriptionName, roles);
- result.completeExceptionally(new IllegalArgumentException("couldn't find subscription"));
- return result;
- }
- } else {
- policies.auth_policies.getSubscriptionAuthentication().put(subscriptionName, roles);
- }
- pulsarResources.getNamespaceResources().setPolicies(namespace, (data)->policies);
+ CompletableFuture<Void> future =
+ pulsarResources.getNamespaceResources().setPoliciesAsync(namespace, policies -> {
+ if (remove) {
+ if (policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName) != null) {
+ policies.auth_policies.getSubscriptionAuthentication().get(subscriptionName)
+ .removeAll(roles);
+ } else {
+ log.info("[{}] Couldn't find role {} while revoking for sub = {}", namespace,
Review comment:
Looks good to me.
Minor it seems like some of the arguments in the logs are not in order (roles and subscriptionName)?
Should the order be more like this?
```
log.info("[{}] Couldn't find roles={} while revoking for sub={}", namespace, roles, subscriptionName)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org