[broker]make validateTopicPolicyOperation method async in PulsarWebResource
@@ -1062,22 +1062,41 @@ public abstract class PulsarWebResource {
}
public void validateTopicPolicyOperation(TopicName topicName, PolicyName policy, PolicyOperation operation) {
+ try {
+ int timeout = pulsar().getConfiguration().getZooKeeperOperationTimeoutSeconds();
+ validateTopicPolicyOperationAsync(topicName, policy, operation).get(timeout, SECONDS);
+ } catch (InterruptedException | TimeoutException e) {
+ throw new RestException(e);
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof WebApplicationException){
+ throw (WebApplicationException) cause;
+ } else {
+ throw new RestException(cause);
+ }
+ }
+ }
+
+ public CompletableFuture<Void> validateTopicPolicyOperationAsync(TopicName topicName,
+ PolicyName policy, PolicyOperation operation) {
if (pulsar().getConfiguration().isAuthenticationEnabled()
&& pulsar().getBrokerService().isAuthorizationEnabled()) {
if (!isClientAuthenticated(clientAppId())) {
- throw new RestException(Status.FORBIDDEN, "Need to authenticate to perform the request");
+ return FutureUtil.failedFuture(
+ new RestException(Status.FORBIDDEN, "Need to authenticate to perform the request"));
}
-
- Boolean isAuthorized = pulsar().getBrokerService().getAuthorizationService()
- .allowTopicPolicyOperation(topicName, policy, operation, originalPrincipal(), clientAppId(),
- clientAuthData());
-
+ return pulsar().getBrokerService().getAuthorizationService()
+ .allowTopicPolicyOperationAsync(topicName, policy, operation, originalPrincipal(), clientAppId(),
+ clientAuthData()).thenAccept(isAuthorized -> {
if (!isAuthorized) {
- throw new RestException(Status.FORBIDDEN, String.format("Unauthorized to validateTopicPolicyOperation"
+ throw new RestException(Status.FORBIDDEN,
+ String.format("Unauthorized to validateTopicPolicyOperation"
+ " for operation [%s] on topic [%s] on policy [%s]", operation.toString(),
topicName, policy.toString()));
}
+ });
}
+ return CompletableFuture.completedFuture(null);
}
public void validateTopicOperation(TopicName topicName, TopicOperation operation) {
