[ISSUE #7955] Don't set default auth metadata provider

auth/src/main/java/org/apache/rocketmq/auth/authentication/chain/DefaultAuthenticationHandler.java

@@ -45,6 +45,9 @@ public CompletableFuture<Void> handle(DefaultAuthenticationContext context,
     }
 
     protected CompletableFuture<User> getUser(DefaultAuthenticationContext context) {
+        if (this.authenticationMetadataProvider == null) {
+            throw new AuthenticationException("The authenticationMetadataProvider is not configured");
+        }
         if (StringUtils.isEmpty(context.getUsername())) {
             throw new AuthenticationException("username cannot be null.");
         }

auth/src/main/java/org/apache/rocketmq/auth/authentication/factory/AuthenticationFactory.java

@@ -31,7 +31,6 @@
 import org.apache.rocketmq.auth.authentication.provider.AuthenticationMetadataProvider;
 import org.apache.rocketmq.auth.authentication.provider.AuthenticationProvider;
 import org.apache.rocketmq.auth.authentication.provider.DefaultAuthenticationProvider;
-import org.apache.rocketmq.auth.authentication.provider.LocalAuthenticationMetadataProvider;
 import org.apache.rocketmq.auth.authentication.strategy.AuthenticationStrategy;
 import org.apache.rocketmq.auth.authentication.strategy.StatelessAuthenticationStrategy;
 import org.apache.rocketmq.auth.config.AuthConfig;
@@ -78,10 +77,11 @@ public static AuthenticationMetadataProvider getMetadataProvider(AuthConfig conf
         }
         return computeIfAbsent(METADATA_PROVIDER_PREFIX + config.getConfigName(), key -> {
             try {
-                Class<? extends AuthenticationMetadataProvider> clazz = LocalAuthenticationMetadataProvider.class;
-                if (StringUtils.isNotBlank(config.getAuthenticationMetadataProvider())) {
-                    clazz = (Class<? extends AuthenticationMetadataProvider>) Class.forName(config.getAuthenticationMetadataProvider());
+                if (StringUtils.isBlank(config.getAuthenticationMetadataProvider())) {
+                    return null;
                 }
+                Class<? extends AuthenticationMetadataProvider> clazz = (Class<? extends AuthenticationMetadataProvider>)
+                    Class.forName(config.getAuthenticationMetadataProvider());
                 AuthenticationMetadataProvider result = clazz.getDeclaredConstructor().newInstance();
                 result.initialize(config, metadataService);
                 return result;
@@ -142,7 +142,9 @@ private static <V> V computeIfAbsent(String key, Function<String, ? extends V> f
                 }
                 if (result == null) {
                     result = function.apply(key);
-                    INSTANCE_MAP.put(key, result);
+                    if (result != null) {
+                        INSTANCE_MAP.put(key, result);
+                    }
                 }
             }
         }

auth/src/main/java/org/apache/rocketmq/auth/authentication/manager/AuthenticationMetadataManagerImpl.java

@@ -206,17 +206,17 @@ private void handleException(Exception e, CompletableFuture<?> result) {
         result.completeExceptionally(throwable);
     }
 
-    private AuthorizationMetadataProvider getAuthorizationMetadataProvider() {
-        if (authenticationMetadataProvider == null) {
+    private AuthenticationMetadataProvider getAuthenticationMetadataProvider() {
+        if (authorizationMetadataProvider == null) {
             throw new IllegalStateException("The authenticationMetadataProvider is not configured");
         }
-        return authorizationMetadataProvider;
+        return authenticationMetadataProvider;
     }
 
-    private AuthenticationMetadataProvider getAuthenticationMetadataProvider() {
-        if (authorizationMetadataProvider == null) {
+    private AuthorizationMetadataProvider getAuthorizationMetadataProvider() {
+        if (authenticationMetadataProvider == null) {
             throw new IllegalStateException("The authorizationMetadataProvider is not configured");
         }
-        return authenticationMetadataProvider;
+        return authorizationMetadataProvider;
     }
 }

auth/src/main/java/org/apache/rocketmq/auth/authorization/chain/AclAuthorizationHandler.java

@@ -54,7 +54,10 @@ public AclAuthorizationHandler(AuthConfig config, Supplier<?> metadataService) {
     @Override
     public CompletableFuture<Void> handle(DefaultAuthorizationContext context,
         HandlerChain<DefaultAuthorizationContext, CompletableFuture<Void>> chain) {
-        return authorizationMetadataProvider.getAcl(context.getSubject()).thenAccept(acl -> {
+        if (this.authorizationMetadataProvider == null) {
+            throw new AuthorizationException("The authorizationMetadataProvider is not configured");
+        }
+        return this.authorizationMetadataProvider.getAcl(context.getSubject()).thenAccept(acl -> {
             if (acl == null) {
                 throwException(context, "no matched policies.");
             }

auth/src/main/java/org/apache/rocketmq/auth/authorization/chain/UserAuthorizationHandler.java

@@ -54,6 +54,9 @@ public CompletableFuture<Void> handle(DefaultAuthorizationContext context, Handl
     }
 
     private CompletableFuture<User> getUser(Subject subject) {
+        if (this.authenticationMetadataProvider == null) {
+            throw new AuthorizationException("The authenticationMetadataProvider is not configured");
+        }
         User user = (User) subject;
         return authenticationMetadataProvider.getUser(user.getUsername()).thenApply(result -> {
             if (result == null) {

auth/src/main/java/org/apache/rocketmq/auth/authorization/factory/AuthorizationFactory.java

@@ -19,9 +19,9 @@
 import com.google.protobuf.GeneratedMessageV3;
 import io.grpc.Metadata;
 import io.netty.channel.ChannelHandlerContext;
+import java.util.HashMap;
 import java.util.List;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
+import java.util.Map;
 import java.util.function.Function;
 import java.util.function.Supplier;
 import org.apache.commons.lang3.StringUtils;
@@ -32,15 +32,14 @@
 import org.apache.rocketmq.auth.authorization.provider.AuthorizationMetadataProvider;
 import org.apache.rocketmq.auth.authorization.provider.AuthorizationProvider;
 import org.apache.rocketmq.auth.authorization.provider.DefaultAuthorizationProvider;
-import org.apache.rocketmq.auth.authorization.provider.LocalAuthorizationMetadataProvider;
 import org.apache.rocketmq.auth.authorization.strategy.AuthorizationStrategy;
 import org.apache.rocketmq.auth.authorization.strategy.StatelessAuthorizationStrategy;
 import org.apache.rocketmq.auth.config.AuthConfig;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 
 public class AuthorizationFactory {
 
-    private static final ConcurrentMap<String, Object> INSTANCE_MAP = new ConcurrentHashMap<>();
+    private static final Map<String, Object> INSTANCE_MAP = new HashMap<>();
     private static final String PROVIDER_PREFIX = "PROVIDER_";
     private static final String METADATA_PROVIDER_PREFIX = "METADATA_PROVIDER_";
     private static final String EVALUATOR_PREFIX = "EVALUATOR_";
@@ -80,10 +79,11 @@ public static AuthorizationMetadataProvider getMetadataProvider(AuthConfig confi
         }
         return computeIfAbsent(METADATA_PROVIDER_PREFIX + config.getConfigName(), key -> {
             try {
-                Class<? extends AuthorizationMetadataProvider> clazz = LocalAuthorizationMetadataProvider.class;
-                if (StringUtils.isNotBlank(config.getAuthorizationMetadataProvider())) {
-                    clazz = (Class<? extends AuthorizationMetadataProvider>) Class.forName(config.getAuthorizationMetadataProvider());
+                if (StringUtils.isBlank(config.getAuthorizationMetadataProvider())) {
+                    return null;
                 }
+                Class<? extends AuthorizationMetadataProvider> clazz = (Class<? extends AuthorizationMetadataProvider>)
+                    Class.forName(config.getAuthorizationMetadataProvider());
                 AuthorizationMetadataProvider result = clazz.getDeclaredConstructor().newInstance();
                 result.initialize(config, metadataService);
                 return result;
@@ -145,7 +145,9 @@ private static <V> V computeIfAbsent(String key, Function<String, ? extends V> f
                 }
                 if (result == null) {
                     result = function.apply(key);
-                    INSTANCE_MAP.put(key, result);
+                    if (result != null) {
+                        INSTANCE_MAP.put(key, result);
+                    }
                 }
             }
         }

auth/src/main/java/org/apache/rocketmq/auth/authorization/manager/AuthorizationMetadataManagerImpl.java

@@ -268,17 +268,17 @@ private <T> CompletableFuture<T> handleException(Exception e) {
         return result;
     }
 
-    private AuthorizationMetadataProvider getAuthorizationMetadataProvider() {
-        if (authenticationMetadataProvider == null) {
+    private AuthenticationMetadataProvider getAuthenticationMetadataProvider() {
+        if (authorizationMetadataProvider == null) {
             throw new IllegalStateException("The authenticationMetadataProvider is not configured.");
         }
-        return authorizationMetadataProvider;
+        return authenticationMetadataProvider;
     }
 
-    private AuthenticationMetadataProvider getAuthenticationMetadataProvider() {
-        if (authorizationMetadataProvider == null) {
-            throw new IllegalStateException("The authorizationMetadataProvider is not configured.");
+    private AuthorizationMetadataProvider getAuthorizationMetadataProvider() {
+        if (authenticationMetadataProvider == null) {
+            throw new IllegalStateException("The authenticationMetadataProvider is not configured.");
         }
-        return authenticationMetadataProvider;
+        return authorizationMetadataProvider;
     }
 }