Allow to grant permissions when the authorization is disabled (apache…

…#7074)


### Motivation

Currently if authorization is not enforced, the AuthZ plugin is not loaded at all. That makes it impossible to switch a cluster from no-authz to authz-enforced with no downtime.

In order to do that, we need to allow users to set the ACLs before the the AuthZ is enforced.

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java

@@ -55,25 +55,20 @@ public class AuthorizationService {
 
     public AuthorizationService(ServiceConfiguration conf, ConfigurationCacheService configCache)
             throws PulsarServerException {
-
         this.conf = conf;
-        if (this.conf.isAuthorizationEnabled()) {
-            try {
-                final String providerClassname = conf.getAuthorizationProvider();
-                if (StringUtils.isNotBlank(providerClassname)) {
-                    provider = (AuthorizationProvider) Class.forName(providerClassname).newInstance();
-                    provider.initialize(conf, configCache);
-                    log.info("{} has been loaded.", providerClassname);
-                } else {
-                    throw new PulsarServerException("No authorization providers are present.");
-                }
-            } catch (PulsarServerException e) {
-                throw e;
-            } catch (Throwable e) {
-                throw new PulsarServerException("Failed to load an authorization provider.", e);
+        try {
+            final String providerClassname = conf.getAuthorizationProvider();
+            if (StringUtils.isNotBlank(providerClassname)) {
+                provider = (AuthorizationProvider) Class.forName(providerClassname).newInstance();
+                provider.initialize(conf, configCache);
+                log.info("{} has been loaded.", providerClassname);
+            } else {
+                throw new PulsarServerException("No authorization providers are present.");
             }
-        } else {
-            log.info("Authorization is disabled");
+        } catch (PulsarServerException e) {
+            throw e;
+        } catch (Throwable e) {
+            throw new PulsarServerException("Failed to load an authorization provider.", e);
         }
     }
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java

@@ -374,11 +374,11 @@ protected void internalGrantPermissionOnNamespace(String role, Set<AuthAction> a
         } catch (ExecutionException e) {
             if (e.getCause() instanceof IllegalArgumentException) {
                 log.warn("[{}] Failed to set permissions for namespace {}: does not exist", clientAppId(),
-                        namespaceName);
+                        namespaceName, e);
                 throw new RestException(Status.NOT_FOUND, "Namespace does not exist");
             } else if (e.getCause() instanceof IllegalStateException) {
-                log.warn("[{}] Failed to set permissions for namespace {}: concurrent modification",
-                        clientAppId(), namespaceName);
+                log.warn("[{}] Failed to set permissions for namespace {}: {}",
+                        clientAppId(), namespaceName, e.getCause().getMessage(), e);
                 throw new RestException(Status.CONFLICT, "Concurrent modification");
             } else {
                 log.error("[{}] Failed to get permissions for namespace {}", clientAppId(), namespaceName, e);

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

@@ -258,10 +258,7 @@ public BrokerService(PulsarService pulsar) throws Exception {
         this.workerGroup = EventLoopUtil.newEventLoopGroup(numThreads, workersThreadFactory);
         this.statsUpdater = Executors
                 .newSingleThreadScheduledExecutor(new DefaultThreadFactory("pulsar-stats-updater"));
-        if (pulsar.getConfiguration().isAuthorizationEnabled()) {
-            this.authorizationService = new AuthorizationService(pulsar.getConfiguration(),
-                    pulsar.getConfigurationCache());
-        }
+        this.authorizationService = new AuthorizationService(pulsar.getConfiguration(), pulsar.getConfigurationCache());
 
         if (pulsar.getConfigurationCache() != null) {
             pulsar.getConfigurationCache().policiesCache().registerListener(this);
@@ -1464,7 +1461,7 @@ public boolean isAuthenticationEnabled() {
     }
 
     public boolean isAuthorizationEnabled() {
-        return authorizationService != null;
+        return pulsar.getConfiguration().isAuthorizationEnabled();
     }
 
     public int getKeepAliveIntervalSeconds() {