make validateNamespacePolicyOperation method async
@@ -945,23 +945,44 @@ public abstract class PulsarWebResource {
public void validateNamespacePolicyOperation(NamespaceName namespaceName,
PolicyName policy,
PolicyOperation operation) {
+ try {
+ int timeout = pulsar().getConfiguration().getZooKeeperOperationTimeoutSeconds();
+ validateNamespacePolicyOperationAsync(namespaceName, 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> validateNamespacePolicyOperationAsync(NamespaceName namespaceName,
+ 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()
- .allowNamespacePolicyOperation(namespaceName, policy, operation,
- originalPrincipal(), clientAppId(), clientAuthData());
-
+ return pulsar().getBrokerService().getAuthorizationService()
+ .allowNamespacePolicyOperationAsync(namespaceName, policy, operation,
+ originalPrincipal(), clientAppId(), clientAuthData())
+ .thenAccept(isAuthorized -> {
if (!isAuthorized) {
throw new RestException(Status.FORBIDDEN,
String.format("Unauthorized to validateNamespacePolicyOperation for"
+ " operation [%s] on namespace [%s] on policy [%s]",
operation.toString(), namespaceName, policy.toString()));
}
+ });
}
+ return CompletableFuture.completedFuture(null);
}
protected PulsarResources getPulsarResources() {
