[ISSUE #2986] Support for multiple ACL files in a fixed directory (#3761)

* acl temp

* acl

* fix test case

* fix code style issues

* add considerations on compatibility to the original one ACL config file and scalability of supporting multiple config files in different directories.

* fix test case testWatch

* 1.fix some issues
2.add a detailed design document

* Add warn log when the accesskey is repeated in multiple ACL files.

* 1.Change the folder of acl configuration to conf/acl
2.Add the logic to check if path is a directory in the method of getAllAclFiles(String path)

* Add a parameter in AclFileWatchService constructor.

* Add logic to determine if path exists in the getAllAclFiles(String path) method in PlainPermissionManager.java and AclFileWatchService.java

* Fix the serialization problem of allAclFileVersion field in clusterAclConfigVersion command

* 1.Fix the serialization problem of allAclFileVersion field in clusterAclConfigVersion command
2.Improve the logic of updateAccessConfig method

acl/src/main/java/org/apache/rocketmq/acl/AccessValidator.java

@@ -18,7 +18,10 @@
 package org.apache.rocketmq.acl;
 
 import java.util.List;
+import java.util.Map;
+
 import org.apache.rocketmq.common.AclConfig;
+import org.apache.rocketmq.common.DataVersion;
 import org.apache.rocketmq.common.PlainAccessConfig;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 
@@ -60,17 +63,27 @@ public interface AccessValidator {
      *
      * @return
      */
+    @Deprecated
     String getAclConfigVersion();
 
     /**
      * Update globalWhiteRemoteAddresses in acl yaml config file
+     *
      * @return
      */
     boolean updateGlobalWhiteAddrsConfig(List<String> globalWhiteAddrsList);
 
     /**
      * get broker cluster acl config information
+     *
      * @return
      */
     AclConfig getAllAclConfig();
+
+    /**
+     * get all access resource config version information
+     *
+     * @return
+     */
+    Map<String, DataVersion> getAllAclConfigVersion();
 }

acl/src/main/java/org/apache/rocketmq/acl/plain/PlainAccessValidator.java

@@ -27,6 +27,7 @@ import org.apache.rocketmq.acl.common.AclUtils;
 import org.apache.rocketmq.acl.common.Permission;
 import org.apache.rocketmq.acl.common.SessionCredentials;
 import org.apache.rocketmq.common.AclConfig;
+import org.apache.rocketmq.common.DataVersion;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.PlainAccessConfig;
 import org.apache.rocketmq.common.protocol.RequestCode;
@@ -127,7 +128,7 @@ public class PlainAccessValidator implements AccessValidator {
         SortedMap<String, String> map = new TreeMap<String, String>();
         for (Map.Entry<String, String> entry : request.getExtFields().entrySet()) {
             if (!SessionCredentials.SIGNATURE.equals(entry.getKey())
-                    && !MixAll.UNIQUE_MSG_QUERY_FLAG.equals(entry.getKey())) {
+                && !MixAll.UNIQUE_MSG_QUERY_FLAG.equals(entry.getKey())) {
                 map.put(entry.getKey(), entry.getValue());
             }
         }
@@ -150,7 +151,7 @@ public class PlainAccessValidator implements AccessValidator {
         return aclPlugEngine.deleteAccessConfig(accesskey);
     }
 
-    @Override public String getAclConfigVersion() {
+    @Override public String  getAclConfigVersion() {
         return aclPlugEngine.getAclConfigDataVersion();
     }
 
@@ -161,4 +162,18 @@ public class PlainAccessValidator implements AccessValidator {
     @Override public AclConfig getAllAclConfig() {
         return aclPlugEngine.getAllAclConfig();
     }
+    
+    public Map<String, Object> createAclAccessConfigMap(Map<String, Object> existedAccountMap,
+        PlainAccessConfig plainAccessConfig) {
+        return aclPlugEngine.createAclAccessConfigMap(existedAccountMap, plainAccessConfig);
+    }
+
+    public Map<String, Object> updateAclConfigFileVersion(Map<String, Object> updateAclConfigMap) {
+        return aclPlugEngine.updateAclConfigFileVersion(updateAclConfigMap);
+    }
+
+    @Override
+    public Map<String, DataVersion> getAllAclConfigVersion() {
+        return aclPlugEngine.getDataVersionMap();
+    }
 }

acl/src/main/java/org/apache/rocketmq/acl/plain/PlainPermissionManager.java

@@ -18,6 +18,19 @@ package org.apache.rocketmq.acl.plain;
 
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
+
 import org.apache.commons.lang3.StringUtils;
 import org.apache.rocketmq.acl.common.AclConstants;
 import org.apache.rocketmq.acl.common.AclException;
@@ -30,51 +43,149 @@ import org.apache.rocketmq.common.PlainAccessConfig;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.logging.InternalLogger;
 import org.apache.rocketmq.logging.InternalLoggerFactory;
-import org.apache.rocketmq.srvutil.FileWatchService;
-
-import java.io.File;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
+import org.apache.rocketmq.srvutil.AclFileWatchService;
 
 public class PlainPermissionManager {
 
     private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME);
 
-    private static final String DEFAULT_PLAIN_ACL_FILE = "/conf/plain_acl.yml";
-
     private String fileHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY,
         System.getenv(MixAll.ROCKETMQ_HOME_ENV));
 
-    private String fileName = System.getProperty("rocketmq.acl.plain.file", DEFAULT_PLAIN_ACL_FILE);
+    private String defaultAclDir = fileHome + File.separator + "conf" + File.separator + "acl";
 
-    private Map<String/** AccessKey **/, PlainAccessResource> plainAccessResourceMap = new HashMap<>();
+    private String defaultAclFile = fileHome + File.separator + System.getProperty("rocketmq.acl.plain.file", "conf/plain_acl.yml");
+
+    private Map<String/** fileFullPath **/, Map<String/** AccessKey **/, PlainAccessResource>> aclPlainAccessResourceMap = new HashMap<>();
+
+    private Map<String/** AccessKey **/, String/** fileFullPath **/> accessKeyTable = new HashMap<>();
 
     private List<RemoteAddressStrategy> globalWhiteRemoteAddressStrategy = new ArrayList<>();
 
     private RemoteAddressStrategyFactory remoteAddressStrategyFactory = new RemoteAddressStrategyFactory();
 
+    private Map<String/** fileFullPath **/, List<RemoteAddressStrategy>> globalWhiteRemoteAddressStrategyMap = new HashMap<>();
+
     private boolean isWatchStart;
 
+    private Map<String/** fileFullPath **/, DataVersion> dataVersionMap = new HashMap<>();
+
+    @Deprecated
     private final DataVersion dataVersion = new DataVersion();
 
+    private List<String> fileList = new ArrayList<>();
+
     public PlainPermissionManager() {
         load();
         watch();
     }
 
-    public void load() {
+    public List<String> getAllAclFiles(String path) {
+        if (!new File(path).exists()) {
+            log.info("The default acl dir {} is not exist", path);
+            return new ArrayList<>();
+        }
+        List<String>  allAclFileFullPath = new ArrayList<>();
+        File file = new File(path);
+        File[] files = file.listFiles();
+        for (int i = 0; i < files.length; i++) {
+            String fileName = files[i].getAbsolutePath();
+            File f = new File(fileName);
+            if (fileName.equals(fileHome + MixAll.ACL_CONF_TOOLS_FILE)) {
+                continue;
+            } else if (fileName.endsWith(".yml") || fileName.endsWith(".yaml")) {
+                allAclFileFullPath.add(fileName);
+            } else if (f.isDirectory()) {
+                allAclFileFullPath.addAll(getAllAclFiles(fileName));
+            }
+        }
+        return allAclFileFullPath;
+    }
 
+    public void load() {
+        if (fileHome == null || fileHome.isEmpty()) {
+            return;
+        }
+
+        Map<String, Map<String, PlainAccessResource>> aclPlainAccessResourceMap = new HashMap<>();
+        Map<String, String> accessKeyTable = new HashMap<>();
+        List<RemoteAddressStrategy> globalWhiteRemoteAddressStrategy = new ArrayList<>();
+        Map<String, List<RemoteAddressStrategy>> globalWhiteRemoteAddressStrategyMap = new HashMap<>();
+        Map<String, DataVersion> dataVersionMap = new HashMap<>();
+
+        fileList = getAllAclFiles(defaultAclDir);
+        if (new File(defaultAclFile).exists() && !fileList.contains(defaultAclFile)) {
+            fileList.add(defaultAclFile);
+        }
+
+        for (int i = 0; i < fileList.size(); i++) {
+            JSONObject plainAclConfData = AclUtils.getYamlDataObject(fileList.get(i),
+                JSONObject.class);
+            if (plainAclConfData == null || plainAclConfData.isEmpty()) {
+                throw new AclException(String.format("%s file is not data", fileList.get(i)));
+            }
+            log.info("Broker plain acl conf data is : ", plainAclConfData.toString());
+
+            List<RemoteAddressStrategy> globalWhiteRemoteAddressStrategyList = new ArrayList<>();
+            JSONArray globalWhiteRemoteAddressesList = plainAclConfData.getJSONArray("globalWhiteRemoteAddresses");
+            if (globalWhiteRemoteAddressesList != null && !globalWhiteRemoteAddressesList.isEmpty()) {
+                for (int j = 0; j < globalWhiteRemoteAddressesList.size(); j++) {
+                    globalWhiteRemoteAddressStrategyList.add(remoteAddressStrategyFactory.
+                        getRemoteAddressStrategy(globalWhiteRemoteAddressesList.getString(j)));
+                }
+            }
+            if (globalWhiteRemoteAddressStrategyList.size() > 0) {
+                globalWhiteRemoteAddressStrategyMap.put(fileList.get(i), globalWhiteRemoteAddressStrategyList);
+                globalWhiteRemoteAddressStrategy.addAll(globalWhiteRemoteAddressStrategyList);
+            }
+
+            JSONArray accounts = plainAclConfData.getJSONArray(AclConstants.CONFIG_ACCOUNTS);
+            Map<String, PlainAccessResource> plainAccessResourceMap = new HashMap<>();
+            if (accounts != null && !accounts.isEmpty()) {
+                List<PlainAccessConfig> plainAccessConfigList = accounts.toJavaList(PlainAccessConfig.class);
+                for (PlainAccessConfig plainAccessConfig : plainAccessConfigList) {
+                    PlainAccessResource plainAccessResource = buildPlainAccessResource(plainAccessConfig);
+                    //AccessKey can not be defined in multiple ACL files
+                    if (accessKeyTable.get(plainAccessResource.getAccessKey()) == null) {
+                        plainAccessResourceMap.put(plainAccessResource.getAccessKey(), plainAccessResource);
+                        accessKeyTable.put(plainAccessResource.getAccessKey(), fileList.get(i));
+                    } else {
+                        log.warn("The accesssKey {} is repeated in multiple ACL files", plainAccessResource.getAccessKey());
+                    }
+                }
+            }
+            if (plainAccessResourceMap.size() > 0) {
+                aclPlainAccessResourceMap.put(fileList.get(i), plainAccessResourceMap);
+            }
+
+            JSONArray tempDataVersion = plainAclConfData.getJSONArray(AclConstants.CONFIG_DATA_VERSION);
+            DataVersion dataVersion = new DataVersion();
+            if (tempDataVersion != null && !tempDataVersion.isEmpty()) {
+                List<DataVersion> dataVersions = tempDataVersion.toJavaList(DataVersion.class);
+                DataVersion firstElement = dataVersions.get(0);
+                dataVersion.assignNewOne(firstElement);
+            }
+            dataVersionMap.put(fileList.get(i), dataVersion);
+        }
+
+        if (dataVersionMap.containsKey(defaultAclFile)) {
+            this.dataVersion.assignNewOne(dataVersionMap.get(defaultAclFile));
+        }
+        this.dataVersionMap = dataVersionMap;
+        this.globalWhiteRemoteAddressStrategyMap = globalWhiteRemoteAddressStrategyMap;
+        this.globalWhiteRemoteAddressStrategy = globalWhiteRemoteAddressStrategy;
+        this.aclPlainAccessResourceMap = aclPlainAccessResourceMap;
+        this.accessKeyTable = accessKeyTable;
+    }
+
+    public void load(String aclFilePath) {
         Map<String, PlainAccessResource> plainAccessResourceMap = new HashMap<>();
         List<RemoteAddressStrategy> globalWhiteRemoteAddressStrategy = new ArrayList<>();
 
-        JSONObject plainAclConfData = AclUtils.getYamlDataObject(fileHome + File.separator + fileName,
+        JSONObject plainAclConfData = AclUtils.getYamlDataObject(aclFilePath,
             JSONObject.class);
         if (plainAclConfData == null || plainAclConfData.isEmpty()) {
-            throw new AclException(String.format("%s file is not data", fileHome + File.separator + fileName));
+            throw new AclException(String.format("%s file is not data", aclFilePath));
         }
         log.info("Broker plain acl conf data is : ", plainAclConfData.toString());
         JSONArray globalWhiteRemoteAddressesList = plainAclConfData.getJSONArray("globalWhiteRemoteAddresses");
@@ -85,43 +196,79 @@ public class PlainPermissionManager {
             }
         }
 
+        this.globalWhiteRemoteAddressStrategy.addAll(globalWhiteRemoteAddressStrategy);
+        if (this.globalWhiteRemoteAddressStrategyMap.get(aclFilePath) != null) {
+            List<RemoteAddressStrategy> remoteAddressStrategyList = this.globalWhiteRemoteAddressStrategyMap.get(aclFilePath);
+            for (int i = 0; i < remoteAddressStrategyList.size(); i++) {
+                this.globalWhiteRemoteAddressStrategy.remove(remoteAddressStrategyList.get(i));
+            }
+            this.globalWhiteRemoteAddressStrategyMap.put(aclFilePath, globalWhiteRemoteAddressStrategy);
+        }
+
+
         JSONArray accounts = plainAclConfData.getJSONArray(AclConstants.CONFIG_ACCOUNTS);
         if (accounts != null && !accounts.isEmpty()) {
             List<PlainAccessConfig> plainAccessConfigList = accounts.toJavaList(PlainAccessConfig.class);
             for (PlainAccessConfig plainAccessConfig : plainAccessConfigList) {
                 PlainAccessResource plainAccessResource = buildPlainAccessResource(plainAccessConfig);
-                plainAccessResourceMap.put(plainAccessResource.getAccessKey(), plainAccessResource);
+                //AccessKey can not be defined in multiple ACL files
+                if (this.accessKeyTable.get(plainAccessResource.getAccessKey()) == null) {
+                    plainAccessResourceMap.put(plainAccessResource.getAccessKey(), plainAccessResource);
+                    this.accessKeyTable.put(plainAccessResource.getAccessKey(), aclFilePath);
+                }
             }
         }
 
         // For loading dataversion part just
         JSONArray tempDataVersion = plainAclConfData.getJSONArray(AclConstants.CONFIG_DATA_VERSION);
+        DataVersion dataVersion = new DataVersion();
         if (tempDataVersion != null && !tempDataVersion.isEmpty()) {
-            List<DataVersion> dataVersion = tempDataVersion.toJavaList(DataVersion.class);
-            DataVersion firstElement = dataVersion.get(0);
-            this.dataVersion.assignNewOne(firstElement);
+            List<DataVersion> dataVersions = tempDataVersion.toJavaList(DataVersion.class);
+            DataVersion firstElement = dataVersions.get(0);
+            dataVersion.assignNewOne(firstElement);
         }
 
-        this.globalWhiteRemoteAddressStrategy = globalWhiteRemoteAddressStrategy;
-        this.plainAccessResourceMap = plainAccessResourceMap;
+        this.aclPlainAccessResourceMap.put(aclFilePath, plainAccessResourceMap);
+        this.dataVersionMap.put(aclFilePath, dataVersion);
+        if (aclFilePath.equals(defaultAclFile)) {
+            this.dataVersion.assignNewOne(dataVersion);
+        }
     }
 
+
+    @Deprecated
     public String getAclConfigDataVersion() {
         return this.dataVersion.toJson();
     }
 
-    private Map<String, Object> updateAclConfigFileVersion(Map<String, Object> updateAclConfigMap) {
+    public Map<String, DataVersion> getDataVersionMap() {
+        return this.dataVersionMap;
+    }
 
+    public Map<String, Object> updateAclConfigFileVersion(Map<String, Object> updateAclConfigMap) {
+
+        Object dataVersions = updateAclConfigMap.get(AclConstants.CONFIG_DATA_VERSION);
+        DataVersion dataVersion = new DataVersion();
+        if (dataVersions != null) {
+            List<Map<String, Object>> dataVersionList = (List<Map<String, Object>>) dataVersions;
+            if (dataVersionList.size() > 0) {
+                dataVersion.setTimestamp((long) dataVersionList.get(0).get("timestamp"));
+                dataVersion.setCounter(new AtomicLong(Long.parseLong(dataVersionList.get(0).get("counter").toString())));
+            }
+        }
         dataVersion.nextVersion();
         List<Map<String, Object>> versionElement = new ArrayList<Map<String, Object>>();
-        Map<String, Object> accountsMap = new LinkedHashMap<String, Object>() {
-            {
-                put(AclConstants.CONFIG_COUNTER, dataVersion.getCounter().longValue());
-                put(AclConstants.CONFIG_TIME_STAMP, dataVersion.getTimestamp());
-            }
-        };
+        Map<String, Object> accountsMap = new LinkedHashMap<String, Object>();
+        accountsMap.put(AclConstants.CONFIG_COUNTER, dataVersion.getCounter().longValue());
+        accountsMap.put(AclConstants.CONFIG_TIME_STAMP, dataVersion.getTimestamp());
+
         versionElement.add(accountsMap);
         updateAclConfigMap.put(AclConstants.CONFIG_DATA_VERSION, versionElement);
+
+        List<Map<String, Object>> accounts = (List<Map<String, Object>>) updateAclConfigMap.get(AclConstants.CONFIG_ACCOUNTS);
+        String accessKey = (String) accounts.get(0).get(AclConstants.CONFIG_ACCESS_KEY);
+        String aclFileName = accessKeyTable.get(accessKey);
+        dataVersionMap.put(aclFileName, dataVersion);
         return updateAclConfigMap;
     }
 
@@ -136,14 +283,11 @@ public class PlainPermissionManager {
         Permission.checkResourcePerms(plainAccessConfig.getTopicPerms());
         Permission.checkResourcePerms(plainAccessConfig.getGroupPerms());
 
-        Map<String, Object> aclAccessConfigMap = AclUtils.getYamlDataObject(fileHome + File.separator + fileName,
-            Map.class);
-        if (aclAccessConfigMap == null || aclAccessConfigMap.isEmpty()) {
-            throw new AclException(String.format("the %s file is not found or empty", fileHome + File.separator + fileName));
-        }
-        List<Map<String, Object>> accounts = (List<Map<String, Object>>) aclAccessConfigMap.get(AclConstants.CONFIG_ACCOUNTS);
-        Map<String, Object> updateAccountMap = null;
-        if (accounts != null) {
+        if (accessKeyTable.containsKey(plainAccessConfig.getAccessKey())) {
+            Map<String, Object> updateAccountMap = null;
+            String aclFileName = accessKeyTable.get(plainAccessConfig.getAccessKey());
+            Map<String, Object> aclAccessConfigMap = AclUtils.getYamlDataObject(aclFileName, Map.class);
+            List<Map<String, Object>> accounts = (List<Map<String, Object>>) aclAccessConfigMap.get(AclConstants.CONFIG_ACCOUNTS);
             for (Map<String, Object> account : accounts) {
                 if (account.get(AclConstants.CONFIG_ACCESS_KEY).equals(plainAccessConfig.getAccessKey())) {
                     // Update acl access config elements
@@ -151,27 +295,62 @@ public class PlainPermissionManager {
                     updateAccountMap = createAclAccessConfigMap(account, plainAccessConfig);
                     accounts.add(updateAccountMap);
                     aclAccessConfigMap.put(AclConstants.CONFIG_ACCOUNTS, accounts);
-
-                    if (AclUtils.writeDataObject(fileHome + File.separator + fileName, updateAclConfigFileVersion(aclAccessConfigMap))) {
-                        return true;
-                    }
-                    return false;
+                    break;
                 }
             }
-            // Create acl access config elements
+            Map<String, PlainAccessResource> accountMap = aclPlainAccessResourceMap.get(aclFileName);
+            if (accountMap == null) {
+                accountMap = new HashMap<String, PlainAccessResource>(1);
+                accountMap.put(plainAccessConfig.getAccessKey(), buildPlainAccessResource(plainAccessConfig));
+            } else if (accountMap.size() == 0) {
+                accountMap.put(plainAccessConfig.getAccessKey(), buildPlainAccessResource(plainAccessConfig));
+            } else {
+                for (Map.Entry<String, PlainAccessResource> entry : accountMap.entrySet()) {
+                    if (entry.getValue().equals(plainAccessConfig.getAccessKey())) {
+                        PlainAccessResource plainAccessResource = buildPlainAccessResource(plainAccessConfig);
+                        accountMap.put(entry.getKey(), plainAccessResource);
+                        break;
+                    }
+                }
+            }
+            aclPlainAccessResourceMap.put(aclFileName, accountMap);
+            return AclUtils.writeDataObject(aclFileName, updateAclConfigFileVersion(aclAccessConfigMap));
+        } else {
+            String fileName = defaultAclFile;
+            //Create acl access config elements on the default acl file
+            if (aclPlainAccessResourceMap.get(defaultAclFile) == null || aclPlainAccessResourceMap.get(defaultAclFile).size() == 0) {
+                try {
+                    File defaultAclFile = new File(fileName);
+                    if (!defaultAclFile.exists()) {
+                        defaultAclFile.createNewFile();
+                    }
+                } catch (IOException e) {
+                    log.warn("create default acl file has exception when update accessConfig. ", e);
+                }
+            }
+            Map<String, Object> aclAccessConfigMap = AclUtils.getYamlDataObject(defaultAclFile, Map.class);
+            if (aclAccessConfigMap == null) {
+                aclAccessConfigMap = new HashMap<>();
+                aclAccessConfigMap.put(AclConstants.CONFIG_ACCOUNTS, new ArrayList<>());
+            }
+            List<Map<String, Object>> accounts = (List<Map<String, Object>>) aclAccessConfigMap.get(AclConstants.CONFIG_ACCOUNTS);
             accounts.add(createAclAccessConfigMap(null, plainAccessConfig));
             aclAccessConfigMap.put(AclConstants.CONFIG_ACCOUNTS, accounts);
-            if (AclUtils.writeDataObject(fileHome + File.separator + fileName, updateAclConfigFileVersion(aclAccessConfigMap))) {
-                return true;
+            accessKeyTable.put(plainAccessConfig.getAccessKey(), fileName);
+            if (aclPlainAccessResourceMap.get(fileName) == null) {
+                Map<String, PlainAccessResource> plainAccessResourceMap = new HashMap<>(1);
+                plainAccessResourceMap.put(plainAccessConfig.getAccessKey(), buildPlainAccessResource(plainAccessConfig));
+                aclPlainAccessResourceMap.put(fileName, plainAccessResourceMap);
+            } else {
+                Map<String, PlainAccessResource> plainAccessResourceMap = aclPlainAccessResourceMap.get(fileName);
+                plainAccessResourceMap.put(plainAccessConfig.getAccessKey(), buildPlainAccessResource(plainAccessConfig));
+                aclPlainAccessResourceMap.put(fileName, plainAccessResourceMap);
             }
-            return false;
+            return AclUtils.writeDataObject(defaultAclFile, updateAclConfigFileVersion(aclAccessConfigMap));
         }
-
-        log.error("Users must ensure that the acl yaml config file has accounts node element");
-        return false;
     }
 
-    private Map<String, Object> createAclAccessConfigMap(Map<String, Object> existedAccountMap,
+    public Map<String, Object> createAclAccessConfigMap(Map<String, Object> existedAccountMap,
         PlainAccessConfig plainAccessConfig) {
 
         Map<String, Object> newAccountsMap = null;
@@ -225,38 +404,37 @@ public class PlainPermissionManager {
             return false;
         }
 
-        Map<String, Object> aclAccessConfigMap = AclUtils.getYamlDataObject(fileHome + File.separator + fileName,
-            Map.class);
-        if (aclAccessConfigMap == null || aclAccessConfigMap.isEmpty()) {
-            throw new AclException(String.format("the %s file is not found or empty", fileHome + File.separator + fileName));
-        }
-        List<Map<String, Object>> accounts = (List<Map<String, Object>>) aclAccessConfigMap.get("accounts");
-        if (accounts != null) {
+        if (accessKeyTable.containsKey(accesskey)) {
+            String aclFileName = accessKeyTable.get(accesskey);
+            Map<String, Object> aclAccessConfigMap = AclUtils.getYamlDataObject(aclFileName,
+                Map.class);
+            if (aclAccessConfigMap == null || aclAccessConfigMap.isEmpty()) {
+                throw new AclException(String.format("the %s file is not found or empty", aclFileName));
+            }
+            List<Map<String, Object>> accounts = (List<Map<String, Object>>) aclAccessConfigMap.get("accounts");
             Iterator<Map<String, Object>> itemIterator = accounts.iterator();
             while (itemIterator.hasNext()) {
-
                 if (itemIterator.next().get(AclConstants.CONFIG_ACCESS_KEY).equals(accesskey)) {
                     // Delete the related acl config element
                     itemIterator.remove();
                     aclAccessConfigMap.put(AclConstants.CONFIG_ACCOUNTS, accounts);
-
-                    if (AclUtils.writeDataObject(fileHome + File.separator + fileName, updateAclConfigFileVersion(aclAccessConfigMap))) {
-                        return true;
-                    }
-                    return false;
+                    return AclUtils.writeDataObject(aclFileName, updateAclConfigFileVersion(aclAccessConfigMap));
                 }
             }
         }
-        log.error("Users must ensure that the acl yaml config file has related acl config elements");
-
         return false;
     }
 
     public boolean updateGlobalWhiteAddrsConfig(List<String> globalWhiteAddrsList) {
-        Map<String, Object> aclAccessConfigMap = AclUtils.getYamlDataObject(fileHome + File.separator + fileName,
-            Map.class);
+
+        if (globalWhiteAddrsList == null) {
+            log.error("Parameter value globalWhiteAddrsList is null,Please check your parameter");
+            return false;
+        }
+
+        Map<String, Object> aclAccessConfigMap = AclUtils.getYamlDataObject(defaultAclFile, Map.class);
         if (aclAccessConfigMap == null || aclAccessConfigMap.isEmpty()) {
-            throw new AclException(String.format("the %s file is not found or empty", fileHome + File.separator + fileName));
+            throw new AclException(String.format("the %s file is not found or empty", defaultAclFile));
         }
         List<String> globalWhiteRemoteAddrList = (List<String>) aclAccessConfigMap.get(AclConstants.CONFIG_GLOBAL_WHITE_ADDRS);
 
@@ -267,13 +445,41 @@ public class PlainPermissionManager {
             }
             // Update globalWhiteRemoteAddr element in memory map firstly
             aclAccessConfigMap.put(AclConstants.CONFIG_GLOBAL_WHITE_ADDRS, globalWhiteRemoteAddrList);
-            if (AclUtils.writeDataObject(fileHome + File.separator + fileName, updateAclConfigFileVersion(aclAccessConfigMap))) {
-                return true;
-            }
+            return AclUtils.writeDataObject(defaultAclFile, updateAclConfigFileVersion(aclAccessConfigMap));
+        }
+
+        log.error("Users must ensure that the acl yaml config file has globalWhiteRemoteAddresses flag in the {} firstly", defaultAclFile);
+        return false;
+    }
+
+    public boolean updateGlobalWhiteAddrsConfig(List<String> globalWhiteAddrsList, String fileName) {
+        if (globalWhiteAddrsList == null) {
+            log.error("Parameter value globalWhiteAddrsList is null,Please check your parameter");
             return false;
         }
 
-        log.error("Users must ensure that the acl yaml config file has globalWhiteRemoteAddresses flag firstly");
+        File file = new File(fileName);
+        if (!file.exists() || file.isDirectory()) {
+            log.error("Parameter value fileName is not exist or is a directory,Please check your parameter");
+            return false;
+        }
+
+        Map<String, Object> aclAccessConfigMap = AclUtils.getYamlDataObject(fileName, Map.class);
+        if (aclAccessConfigMap == null || aclAccessConfigMap.isEmpty()) {
+            throw new AclException(String.format("the %s file is not found or empty", fileName));
+        }
+        List<String> globalWhiteRemoteAddrList = (List<String>) aclAccessConfigMap.get(AclConstants.CONFIG_GLOBAL_WHITE_ADDRS);
+        if (globalWhiteRemoteAddrList != null) {
+            globalWhiteRemoteAddrList.clear();
+            if (globalWhiteAddrsList != null) {
+                globalWhiteRemoteAddrList.addAll(globalWhiteAddrsList);
+            }
+            // Update globalWhiteRemoteAddr element in memory map firstly
+            aclAccessConfigMap.put(AclConstants.CONFIG_GLOBAL_WHITE_ADDRS, globalWhiteRemoteAddrList);
+            return AclUtils.writeDataObject(fileName, updateAclConfigFileVersion(aclAccessConfigMap));
+        }
+
+        log.error("Users must ensure that the acl yaml config file has globalWhiteRemoteAddresses flag in the {} firstly", fileName);
         return false;
     }
 
@@ -281,40 +487,65 @@ public class PlainPermissionManager {
         AclConfig aclConfig = new AclConfig();
         List<PlainAccessConfig> configs = new ArrayList<>();
         List<String> whiteAddrs = new ArrayList<>();
-        JSONObject plainAclConfData = AclUtils.getYamlDataObject(fileHome + File.separator + fileName,
-            JSONObject.class);
-        if (plainAclConfData == null || plainAclConfData.isEmpty()) {
-            throw new AclException(String.format("%s file is not data", fileHome + File.separator + fileName));
+        Set<String> accessKeySets = new HashSet<>();
+
+        for (int i = 0; i < fileList.size(); i++) {
+            String path = fileList.get(i);
+            JSONObject plainAclConfData = AclUtils.getYamlDataObject(path,
+                JSONObject.class);
+            if (plainAclConfData == null || plainAclConfData.isEmpty()) {
+                throw new AclException(String.format("%s file is not data", path));
+            }
+            JSONArray globalWhiteAddrs = plainAclConfData.getJSONArray(AclConstants.CONFIG_GLOBAL_WHITE_ADDRS);
+            if (globalWhiteAddrs != null && !globalWhiteAddrs.isEmpty()) {
+                whiteAddrs.addAll(globalWhiteAddrs.toJavaList(String.class));
+            }
+
+            JSONArray accounts = plainAclConfData.getJSONArray(AclConstants.CONFIG_ACCOUNTS);
+            if (accounts != null && !accounts.isEmpty()) {
+                List<PlainAccessConfig> plainAccessConfigs = accounts.toJavaList(PlainAccessConfig.class);
+                for (int j = 0; j < plainAccessConfigs.size(); j++) {
+                    if (!accessKeySets.contains(plainAccessConfigs.get(j).getAccessKey())) {
+                        accessKeySets.add(plainAccessConfigs.get(j).getAccessKey());
+                        PlainAccessConfig plainAccessConfig = new PlainAccessConfig();
+                        plainAccessConfig.setGroupPerms(plainAccessConfigs.get(j).getGroupPerms());
+                        plainAccessConfig.setDefaultTopicPerm(plainAccessConfigs.get(j).getDefaultTopicPerm());
+                        plainAccessConfig.setDefaultGroupPerm(plainAccessConfigs.get(j).getDefaultGroupPerm());
+                        plainAccessConfig.setAccessKey(plainAccessConfigs.get(j).getAccessKey());
+                        plainAccessConfig.setSecretKey(plainAccessConfigs.get(j).getSecretKey());
+                        plainAccessConfig.setAdmin(plainAccessConfigs.get(j).isAdmin());
+                        plainAccessConfig.setTopicPerms(plainAccessConfigs.get(j).getTopicPerms());
+                        plainAccessConfig.setWhiteRemoteAddress(plainAccessConfigs.get(j).getWhiteRemoteAddress());
+                        configs.add(plainAccessConfig);
+                    }
+                }
+            }
         }
-        JSONArray globalWhiteAddrs = plainAclConfData.getJSONArray(AclConstants.CONFIG_GLOBAL_WHITE_ADDRS);
-        if (globalWhiteAddrs != null && !globalWhiteAddrs.isEmpty()) {
-            whiteAddrs = globalWhiteAddrs.toJavaList(String.class);
-        }
-        JSONArray accounts = plainAclConfData.getJSONArray(AclConstants.CONFIG_ACCOUNTS);
-        if (accounts != null && !accounts.isEmpty()) {
-            configs = accounts.toJavaList(PlainAccessConfig.class);
-        }
-        aclConfig.setGlobalWhiteAddrs(whiteAddrs);
         aclConfig.setPlainAccessConfigs(configs);
+        aclConfig.setGlobalWhiteAddrs(whiteAddrs);
         return aclConfig;
     }
 
     private void watch() {
         try {
-            String watchFilePath = fileHome + fileName;
-            FileWatchService fileWatchService = new FileWatchService(new String[] {watchFilePath}, new FileWatchService.Listener() {
+            AclFileWatchService aclFileWatchService = new AclFileWatchService(defaultAclDir, defaultAclFile, new AclFileWatchService.Listener() {
                 @Override
-                public void onChanged(String path) {
-                    log.info("The plain acl yml changed, reload the context");
+                public void onFileChanged(String aclFileName) {
+                    load(aclFileName);
+                }
+
+                @Override
+                public void onFileNumChanged(String path) {
                     load();
                 }
             });
-            fileWatchService.start();
-            log.info("Succeed to start AclWatcherService");
+            aclFileWatchService.start();
+            log.info("Succeed to start AclFileWatchService");
             this.isWatchStart = true;
         } catch (Exception e) {
             log.error("Failed to start AclWatcherService", e);
         }
+
     }
 
     void checkPerm(PlainAccessResource needCheckedAccess, PlainAccessResource ownedAccess) {
@@ -355,18 +586,19 @@ public class PlainPermissionManager {
     }
 
     void clearPermissionInfo() {
-        this.plainAccessResourceMap.clear();
+        this.aclPlainAccessResourceMap.clear();
+        this.accessKeyTable.clear();
         this.globalWhiteRemoteAddressStrategy.clear();
     }
 
     public void checkPlainAccessConfig(PlainAccessConfig plainAccessConfig) throws AclException {
         if (plainAccessConfig.getAccessKey() == null
-                || plainAccessConfig.getSecretKey() == null
-                || plainAccessConfig.getAccessKey().length() <= AclConstants.ACCESS_KEY_MIN_LENGTH
-                || plainAccessConfig.getSecretKey().length() <= AclConstants.SECRET_KEY_MIN_LENGTH) {
+            || plainAccessConfig.getSecretKey() == null
+            || plainAccessConfig.getAccessKey().length() <= AclConstants.ACCESS_KEY_MIN_LENGTH
+            || plainAccessConfig.getSecretKey().length() <= AclConstants.SECRET_KEY_MIN_LENGTH) {
             throw new AclException(String.format(
-                    "The accessKey=%s and secretKey=%s cannot be null and length should longer than 6",
-                    plainAccessConfig.getAccessKey(), plainAccessConfig.getSecretKey()));
+                "The accessKey=%s and secretKey=%s cannot be null and length should longer than 6",
+                plainAccessConfig.getAccessKey(), plainAccessConfig.getSecretKey()));
         }
     }
 
@@ -404,12 +636,13 @@ public class PlainPermissionManager {
             throw new AclException(String.format("No accessKey is configured"));
         }
 
-        if (!plainAccessResourceMap.containsKey(plainAccessResource.getAccessKey())) {
+        if (!accessKeyTable.containsKey(plainAccessResource.getAccessKey())) {
             throw new AclException(String.format("No acl config for %s", plainAccessResource.getAccessKey()));
         }
 
         // Check the white addr for accesskey
-        PlainAccessResource ownedAccess = plainAccessResourceMap.get(plainAccessResource.getAccessKey());
+        String aclFileName = accessKeyTable.get(plainAccessResource.getAccessKey());
+        PlainAccessResource ownedAccess = aclPlainAccessResourceMap.get(aclFileName).get(plainAccessResource.getAccessKey());
         if (ownedAccess.getRemoteAddressStrategy().match(plainAccessResource)) {
             return;
         }

acl/src/test/java/org/apache/rocketmq/acl/common/AclUtilsTest.java

@@ -294,23 +294,18 @@ public class AclUtilsTest {
         Assert.assertTrue(yamlDataObject == null);
     }
 
-    @Test(expected = Exception.class)
-    public void getYamlDataExceptionTest() {
-
-        AclUtils.getYamlDataObject("src/test/resources/conf/plain_acl_format_error.yml", Map.class);
-    }
 
     @Test
     public void getAclRPCHookTest() {
 
-        RPCHook errorContRPCHook = AclUtils.getAclRPCHook("src/test/resources/conf/plain_acl_format_error.yml");
-        Assert.assertNull(errorContRPCHook);
+        //RPCHook errorContRPCHook = AclUtils.getAclRPCHook("src/test/resources/conf/plain_acl_format_error.yml");
+        //Assert.assertNull(errorContRPCHook);
 
         RPCHook noFileRPCHook = AclUtils.getAclRPCHook("src/test/resources/plain_acl_format_error1.yml");
         Assert.assertNull(noFileRPCHook);
 
-        RPCHook emptyContRPCHook = AclUtils.getAclRPCHook("src/test/resources/conf/plain_acl_null.yml");
-        Assert.assertNull(emptyContRPCHook);
+        //RPCHook emptyContRPCHook = AclUtils.getAclRPCHook("src/test/resources/conf/plain_acl_null.yml");
+        //Assert.assertNull(emptyContRPCHook);
 
         RPCHook incompleteContRPCHook = AclUtils.getAclRPCHook("src/test/resources/conf/plain_acl_incomplete.yml");
         Assert.assertNull(incompleteContRPCHook);

acl/src/test/java/org/apache/rocketmq/acl/plain/PlainAccessValidatorTest.java

@@ -16,9 +16,13 @@
  */
 package org.apache.rocketmq.acl.plain;
 
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -29,6 +33,7 @@ import org.apache.rocketmq.acl.common.AclException;
 import org.apache.rocketmq.acl.common.AclUtils;
 import org.apache.rocketmq.acl.common.SessionCredentials;
 import org.apache.rocketmq.common.AclConfig;
+import org.apache.rocketmq.common.DataVersion;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.PlainAccessConfig;
 import org.apache.rocketmq.common.protocol.RequestCode;
@@ -56,10 +61,11 @@ public class PlainAccessValidatorTest {
     private AclClientRPCHook aclClient;
     private SessionCredentials sessionCredentials;
 
+
     @Before
     public void init() {
-        System.setProperty("rocketmq.home.dir", "src/test/resources");
-        System.setProperty("rocketmq.acl.plain.file", "/conf/plain_acl.yml");
+        File file = new File("src/test/resources");
+        System.setProperty("rocketmq.home.dir", file.getAbsolutePath());
         plainAccessValidator = new PlainAccessValidator();
         sessionCredentials = new SessionCredentials();
         sessionCredentials.setAccessKey("RocketMQ");
@@ -420,16 +426,14 @@ public class PlainAccessValidatorTest {
     }
 
     @Test
-    public void updateAccessAclYamlConfigNormalTest() {
-        System.setProperty("rocketmq.home.dir", "src/test/resources");
-        System.setProperty("rocketmq.acl.plain.file", "/conf/plain_acl_update_create.yml");
-
-        String targetFileName = "src/test/resources/conf/plain_acl_update_create.yml";
+    public void addAccessAclYamlConfigTest() throws InterruptedException {
+        String targetFileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/plain_acl.yml";
         Map<String, Object> backUpAclConfigMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
 
         PlainAccessConfig plainAccessConfig = new PlainAccessConfig();
-        plainAccessConfig.setAccessKey("RocketMQ");
+        plainAccessConfig.setAccessKey("rocketmq3");
         plainAccessConfig.setSecretKey("1234567890");
+        plainAccessConfig.setWhiteRemoteAddress("192.168.0.*");
         plainAccessConfig.setDefaultGroupPerm("PUB");
         plainAccessConfig.setDefaultTopicPerm("SUB");
         List<String> topicPerms = new ArrayList<String>();
@@ -442,16 +446,21 @@ public class PlainAccessValidatorTest {
         plainAccessConfig.setGroupPerms(groupPerms);
 
         PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
-        // Update acl access yaml config file
         plainAccessValidator.updateAccessConfig(plainAccessConfig);
+        Thread.sleep(10000);
 
-        Map<String, Object> readableMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
-        List<Map<String, Object>> accounts = (List<Map<String, Object>>) readableMap.get("accounts");
-        Map<String, Object> verifyMap = null;
-        for (Map<String, Object> account : accounts) {
-            if (account.get("accessKey").equals(plainAccessConfig.getAccessKey())) {
-                verifyMap = account;
-                break;
+        Map<String, Object> verifyMap = new HashMap<>();
+        AclConfig aclConfig = plainAccessValidator.getAllAclConfig();
+        List<PlainAccessConfig> plainAccessConfigs = aclConfig.getPlainAccessConfigs();
+        for (PlainAccessConfig plainAccessConfig1 : plainAccessConfigs) {
+            if (plainAccessConfig1.getAccessKey().equals(plainAccessConfig.getAccessKey())) {
+                verifyMap.put(AclConstants.CONFIG_SECRET_KEY, plainAccessConfig1.getSecretKey());
+                verifyMap.put(AclConstants.CONFIG_DEFAULT_TOPIC_PERM, plainAccessConfig1.getDefaultTopicPerm());
+                verifyMap.put(AclConstants.CONFIG_DEFAULT_GROUP_PERM, plainAccessConfig1.getDefaultGroupPerm());
+                verifyMap.put(AclConstants.CONFIG_ADMIN_ROLE, plainAccessConfig1.isAdmin());
+                verifyMap.put(AclConstants.CONFIG_WHITE_ADDR, plainAccessConfig1.getWhiteRemoteAddress());
+                verifyMap.put(AclConstants.CONFIG_TOPIC_PERMS, plainAccessConfig1.getTopicPerms());
+                verifyMap.put(AclConstants.CONFIG_GROUP_PERMS, plainAccessConfig1.getGroupPerms());
             }
         }
 
@@ -463,58 +472,50 @@ public class PlainAccessValidatorTest {
         Assert.assertEquals(((List) verifyMap.get(AclConstants.CONFIG_TOPIC_PERMS)).size(), 2);
         Assert.assertEquals(((List) verifyMap.get(AclConstants.CONFIG_GROUP_PERMS)).size(), 2);
 
-        // Verify the dateversion element is correct or not
-        List<Map<String, Object>> dataVersions = (List<Map<String, Object>>) readableMap.get("dataVersion");
-        Assert.assertEquals(1, dataVersions.get(0).get("counter"));
+        String aclFileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/plain_acl.yml";
+        Map<String, Object> readableMap = AclUtils.getYamlDataObject(aclFileName, Map.class);
+        List<Map<String, Object>> dataVersions = (List<Map<String, Object>>) readableMap.get(AclConstants.CONFIG_DATA_VERSION);
+        Assert.assertEquals(1, dataVersions.get(0).get(AclConstants.CONFIG_COUNTER));
 
-        // Restore the backup file and flush to yaml file
         AclUtils.writeDataObject(targetFileName, backUpAclConfigMap);
     }
 
     @Test
-    public void updateAccessAclYamlConfigTest() {
-        System.setProperty("rocketmq.home.dir", "src/test/resources");
-        System.setProperty("rocketmq.acl.plain.file", "/conf/plain_acl_update_create.yml");
-
-        String targetFileName = "src/test/resources/conf/plain_acl_update_create.yml";
-        Map<String, Object> backUpAclConfigMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
-
-        PlainAccessConfig plainAccessConfig = new PlainAccessConfig();
-        plainAccessConfig.setAccessKey("RocketMQ");
-        plainAccessConfig.setSecretKey("123456789111");
-
+    public void getAccessAclYamlConfigTest() {
+        String accessKey = "rocketmq2";
         PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
-        // Update element in the acl access yaml config file
-        plainAccessValidator.updateAccessConfig(plainAccessConfig);
-
-        Map<String, Object> readableMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
-        List<Map<String, Object>> accounts = (List<Map<String, Object>>) readableMap.get(AclConstants.CONFIG_ACCOUNTS);
-        Map<String, Object> verifyMap = null;
-        for (Map<String, Object> account : accounts) {
-            if (account.get(AclConstants.CONFIG_ACCESS_KEY).equals(plainAccessConfig.getAccessKey())) {
-                verifyMap = account;
-                break;
+        AclConfig aclConfig = plainAccessValidator.getAllAclConfig();
+        List<PlainAccessConfig> plainAccessConfigs = aclConfig.getPlainAccessConfigs();
+        Map<String, Object> verifyMap = new HashMap<>();
+        for (PlainAccessConfig plainAccessConfig : plainAccessConfigs) {
+            if (plainAccessConfig.getAccessKey().equals(accessKey)) {
+                verifyMap.put(AclConstants.CONFIG_SECRET_KEY, plainAccessConfig.getSecretKey());
+                verifyMap.put(AclConstants.CONFIG_ADMIN_ROLE, plainAccessConfig.isAdmin());
+                verifyMap.put(AclConstants.CONFIG_WHITE_ADDR, plainAccessConfig.getWhiteRemoteAddress());
             }
         }
-        Assert.assertEquals(verifyMap.get(AclConstants.CONFIG_SECRET_KEY), "123456789111");
 
-        // Restore the backup file and flush to yaml file
-        AclUtils.writeDataObject(targetFileName, backUpAclConfigMap);
+        Assert.assertEquals(verifyMap.get(AclConstants.CONFIG_SECRET_KEY), "12345678");
+        Assert.assertEquals(verifyMap.get(AclConstants.CONFIG_ADMIN_ROLE), true);
+        Assert.assertEquals(verifyMap.get(AclConstants.CONFIG_WHITE_ADDR), "192.168.1.*");
+
+        String aclFileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/acl/plain_acl.yml";
+        Map<String, DataVersion> dataVersionMap = plainAccessValidator.getAllAclConfigVersion();
+        DataVersion dataVersion = dataVersionMap.get(aclFileName);
+        Assert.assertEquals(0, dataVersion.getCounter().get());
     }
 
     @Test
-    public void createAndUpdateAccessAclYamlConfigNormalTest() {
-        System.setProperty("rocketmq.home.dir", "src/test/resources");
-        System.setProperty("rocketmq.acl.plain.file", "/conf/plain_acl_update_create.yml");
-
-        String targetFileName = "src/test/resources/conf/plain_acl_update_create.yml";
+    public void updateAccessAclYamlConfigTest() throws InterruptedException{
+        String targetFileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/plain_acl.yml";
         Map<String, Object> backUpAclConfigMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
 
         PlainAccessConfig plainAccessConfig = new PlainAccessConfig();
-        plainAccessConfig.setAccessKey("RocketMQ33");
-        plainAccessConfig.setSecretKey("123456789111");
+        plainAccessConfig.setAccessKey("rocketmq3");
+        plainAccessConfig.setSecretKey("1234567890");
+        plainAccessConfig.setWhiteRemoteAddress("192.168.0.*");
         plainAccessConfig.setDefaultGroupPerm("PUB");
-        plainAccessConfig.setDefaultTopicPerm("DENY");
+        plainAccessConfig.setDefaultTopicPerm("SUB");
         List<String> topicPerms = new ArrayList<String>();
         topicPerms.add("topicC=PUB|SUB");
         topicPerms.add("topicB=PUB");
@@ -525,194 +526,332 @@ public class PlainAccessValidatorTest {
         plainAccessConfig.setGroupPerms(groupPerms);
 
         PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
-        // Create element in the acl access yaml config file
         plainAccessValidator.updateAccessConfig(plainAccessConfig);
 
-        Map<String, Object> readableMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
-        List<Map<String, Object>> accounts = (List<Map<String, Object>>) readableMap.get(AclConstants.CONFIG_ACCOUNTS);
-        Map<String, Object> verifyMap = null;
-        for (Map<String, Object> account : accounts) {
-            if (account.get(AclConstants.CONFIG_ACCESS_KEY).equals(plainAccessConfig.getAccessKey())) {
-                verifyMap = account;
-                break;
+        Thread.sleep(10000);
+
+        PlainAccessConfig plainAccessConfig1 = new PlainAccessConfig();
+        plainAccessConfig1.setAccessKey("rocketmq3");
+        plainAccessConfig1.setSecretKey("1234567891");
+        plainAccessConfig1.setWhiteRemoteAddress("192.168.0.*");
+        plainAccessConfig1.setDefaultGroupPerm("PUB");
+        plainAccessConfig1.setDefaultTopicPerm("SUB");
+        List<String> topicPerms1 = new ArrayList<String>();
+        topicPerms1.add("topicC=PUB|SUB");
+        topicPerms1.add("topicB=PUB");
+        plainAccessConfig1.setTopicPerms(topicPerms1);
+        List<String> groupPerms1 = new ArrayList<String>();
+        groupPerms1.add("groupB=PUB|SUB");
+        groupPerms1.add("groupC=DENY");
+        plainAccessConfig1.setGroupPerms(groupPerms1);
+
+        plainAccessValidator.updateAccessConfig(plainAccessConfig1);
+
+        Thread.sleep(10000);
+
+        Map<String, Object> verifyMap = new HashMap<>();
+        AclConfig aclConfig = plainAccessValidator.getAllAclConfig();
+        List<PlainAccessConfig> plainAccessConfigs = aclConfig.getPlainAccessConfigs();
+        for (PlainAccessConfig plainAccessConfig2 : plainAccessConfigs) {
+            if (plainAccessConfig2.getAccessKey().equals(plainAccessConfig1.getAccessKey())) {
+                verifyMap.put(AclConstants.CONFIG_SECRET_KEY, plainAccessConfig2.getSecretKey());
+                verifyMap.put(AclConstants.CONFIG_DEFAULT_TOPIC_PERM, plainAccessConfig2.getDefaultTopicPerm());
+                verifyMap.put(AclConstants.CONFIG_DEFAULT_GROUP_PERM, plainAccessConfig2.getDefaultGroupPerm());
+                verifyMap.put(AclConstants.CONFIG_ADMIN_ROLE, plainAccessConfig2.isAdmin());
+                verifyMap.put(AclConstants.CONFIG_WHITE_ADDR, plainAccessConfig2.getWhiteRemoteAddress());
+                verifyMap.put(AclConstants.CONFIG_TOPIC_PERMS, plainAccessConfig2.getTopicPerms());
+                verifyMap.put(AclConstants.CONFIG_GROUP_PERMS, plainAccessConfig2.getGroupPerms());
             }
         }
-        Assert.assertEquals(verifyMap.get(AclConstants.CONFIG_SECRET_KEY), "123456789111");
-        Assert.assertEquals(verifyMap.get(AclConstants.CONFIG_DEFAULT_TOPIC_PERM), "DENY");
+
+        Assert.assertEquals(verifyMap.get(AclConstants.CONFIG_SECRET_KEY), "1234567891");
+        Assert.assertEquals(verifyMap.get(AclConstants.CONFIG_DEFAULT_TOPIC_PERM), "SUB");
         Assert.assertEquals(verifyMap.get(AclConstants.CONFIG_DEFAULT_GROUP_PERM), "PUB");
+        Assert.assertEquals(verifyMap.get(AclConstants.CONFIG_ADMIN_ROLE), false);
+        Assert.assertEquals(verifyMap.get(AclConstants.CONFIG_WHITE_ADDR), "192.168.0.*");
         Assert.assertEquals(((List) verifyMap.get(AclConstants.CONFIG_TOPIC_PERMS)).size(), 2);
         Assert.assertEquals(((List) verifyMap.get(AclConstants.CONFIG_GROUP_PERMS)).size(), 2);
-        Assert.assertTrue(((List) verifyMap.get(AclConstants.CONFIG_TOPIC_PERMS)).contains("topicC=PUB|SUB"));
-        Assert.assertTrue(((List) verifyMap.get(AclConstants.CONFIG_TOPIC_PERMS)).contains("topicB=PUB"));
-        Assert.assertTrue(((List) verifyMap.get(AclConstants.CONFIG_GROUP_PERMS)).contains("groupB=PUB|SUB"));
-        Assert.assertTrue(((List) verifyMap.get(AclConstants.CONFIG_GROUP_PERMS)).contains("groupC=DENY"));
 
-        // Verify the dateversion element is correct or not
+        String aclFileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/plain_acl.yml";
+        Map<String, Object> readableMap = AclUtils.getYamlDataObject(aclFileName, Map.class);
         List<Map<String, Object>> dataVersions = (List<Map<String, Object>>) readableMap.get(AclConstants.CONFIG_DATA_VERSION);
-        Assert.assertEquals(1, dataVersions.get(0).get(AclConstants.CONFIG_COUNTER));
+        Assert.assertEquals(2, dataVersions.get(0).get(AclConstants.CONFIG_COUNTER));
 
-        // Update element in the acl config yaml file
-        PlainAccessConfig plainAccessConfig2 = new PlainAccessConfig();
-        plainAccessConfig2.setAccessKey("rocketmq2");
-        plainAccessConfig2.setSecretKey("1234567890123");
-
-        // Update acl access yaml config file secondly
-        plainAccessValidator.updateAccessConfig(plainAccessConfig2);
-
-        Map<String, Object> readableMap2 = AclUtils.getYamlDataObject(targetFileName, Map.class);
-        List<Map<String, Object>> accounts2 = (List<Map<String, Object>>) readableMap2.get(AclConstants.CONFIG_ACCOUNTS);
-        Map<String, Object> verifyMap2 = null;
-        for (Map<String, Object> account : accounts2) {
-            if (account.get(AclConstants.CONFIG_ACCESS_KEY).equals(plainAccessConfig2.getAccessKey())) {
-                verifyMap2 = account;
-                break;
-            }
-        }
-
-        // Verify the dateversion element after updating is correct or not
-        List<Map<String, Object>> dataVersions2 = (List<Map<String, Object>>) readableMap2.get(AclConstants.CONFIG_DATA_VERSION);
-        Assert.assertEquals(2, dataVersions2.get(0).get(AclConstants.CONFIG_COUNTER));
-        Assert.assertEquals(verifyMap2.get(AclConstants.CONFIG_SECRET_KEY), "1234567890123");
-
-        // Restore the backup file and flush to yaml file
-        AclUtils.writeDataObject(targetFileName, backUpAclConfigMap);
-    }
-
-    @Test(expected = AclException.class)
-    public void updateAccessAclYamlConfigExceptionTest() {
-        System.setProperty("rocketmq.home.dir", "src/test/resources");
-        System.setProperty("rocketmq.acl.plain.file", "/conf/plain_acl_update_create.yml");
-
-        PlainAccessConfig plainAccessConfig = new PlainAccessConfig();
-        plainAccessConfig.setAccessKey("RocketMQ");
-        plainAccessConfig.setSecretKey("12345");
-
-        PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
-        // Update acl access yaml config file
-        plainAccessValidator.updateAccessConfig(plainAccessConfig);
-    }
-
-    @Test
-    public void deleteAccessAclYamlConfigNormalTest() {
-        System.setProperty("rocketmq.home.dir", "src/test/resources");
-        System.setProperty("rocketmq.acl.plain.file", "/conf/plain_acl_delete.yml");
-
-        String targetFileName = "src/test/resources/conf/plain_acl_delete.yml";
-        Map<String, Object> backUpAclConfigMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
-
-        String accessKey = "rocketmq2";
-        PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
-        plainAccessValidator.deleteAccessConfig(accessKey);
-
-        Map<String, Object> readableMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
-        List<Map<String, Object>> accounts = (List<Map<String, Object>>) readableMap.get(AclConstants.CONFIG_ACCOUNTS);
-        Map<String, Object> verifyMap = null;
-        for (Map<String, Object> account : accounts) {
-            if (account.get(AclConstants.CONFIG_ACCESS_KEY).equals(accessKey)) {
-                verifyMap = account;
-                break;
-            }
-        }
-
-        // Verify the specified element is removed or not
-        Assert.assertEquals(verifyMap, null);
-        // Verify the dateversion element is correct or not
-        List<Map<String, Object>> dataVersions = (List<Map<String, Object>>) readableMap.get(AclConstants.CONFIG_DATA_VERSION);
-        Assert.assertEquals(1, dataVersions.get(0).get(AclConstants.CONFIG_COUNTER));
-
-        // Restore the backup file and flush to yaml file
         AclUtils.writeDataObject(targetFileName, backUpAclConfigMap);
     }
 
     @Test
-    public void updateAccessAclYamlConfigWithNoAccoutsExceptionTest() {
-        System.setProperty("rocketmq.home.dir", "src/test/resources");
-        System.setProperty("rocketmq.acl.plain.file", "/conf/plain_acl_with_no_accouts.yml");
-
-        String targetFileName = "src/test/resources/conf/plain_acl_with_no_accouts.yml";
+    public void deleteAccessAclYamlConfigTest() throws InterruptedException {
+        String targetFileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/plain_acl.yml";
         Map<String, Object> backUpAclConfigMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
 
         PlainAccessConfig plainAccessConfig = new PlainAccessConfig();
-        plainAccessConfig.setAccessKey("RocketMQ");
+        plainAccessConfig.setAccessKey("rocketmq3");
         plainAccessConfig.setSecretKey("1234567890");
-
-        PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
-        // Update acl access yaml config file and verify the return value is true
-        Assert.assertEquals(plainAccessValidator.updateAccessConfig(plainAccessConfig), false);
-    }
-
-    @Test(expected = AclException.class)
-    public void createAndUpdateAccessAclYamlConfigExceptionTest() {
-        System.setProperty("rocketmq.home.dir", "src/test/resources");
-        System.setProperty("rocketmq.acl.plain.file", "/conf/plain_acl_update_create.yml");
-
-        PlainAccessConfig plainAccessConfig = new PlainAccessConfig();
-        plainAccessConfig.setAccessKey("RocketMQ33");
-        plainAccessConfig.setSecretKey("123456789111");
+        plainAccessConfig.setWhiteRemoteAddress("192.168.0.*");
+        plainAccessConfig.setDefaultGroupPerm("PUB");
+        plainAccessConfig.setDefaultTopicPerm("SUB");
         List<String> topicPerms = new ArrayList<String>();
+        topicPerms.add("topicC=PUB|SUB");
         topicPerms.add("topicB=PUB");
         plainAccessConfig.setTopicPerms(topicPerms);
         List<String> groupPerms = new ArrayList<String>();
-        groupPerms.add("groupC=DENY1");
+        groupPerms.add("groupB=PUB|SUB");
+        groupPerms.add("groupC=DENY");
         plainAccessConfig.setGroupPerms(groupPerms);
 
         PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
-        // Create element in the acl access yaml config file
         plainAccessValidator.updateAccessConfig(plainAccessConfig);
+
+        String accessKey = "rocketmq3";
+        plainAccessValidator.deleteAccessConfig(accessKey);
+        Thread.sleep(10000);
+
+        Map<String, Object> verifyMap = new HashMap<>();
+        AclConfig aclConfig = plainAccessValidator.getAllAclConfig();
+        List<PlainAccessConfig> plainAccessConfigs = aclConfig.getPlainAccessConfigs();
+        for (PlainAccessConfig plainAccessConfig1 : plainAccessConfigs) {
+            if (plainAccessConfig1.getAccessKey().equals(accessKey)) {
+                verifyMap.put(AclConstants.CONFIG_SECRET_KEY, plainAccessConfig1.getSecretKey());
+                verifyMap.put(AclConstants.CONFIG_DEFAULT_TOPIC_PERM, plainAccessConfig1.getDefaultTopicPerm());
+                verifyMap.put(AclConstants.CONFIG_DEFAULT_GROUP_PERM, plainAccessConfig1.getDefaultGroupPerm());
+                verifyMap.put(AclConstants.CONFIG_ADMIN_ROLE, plainAccessConfig1.isAdmin());
+                verifyMap.put(AclConstants.CONFIG_WHITE_ADDR, plainAccessConfig1.getWhiteRemoteAddress());
+                verifyMap.put(AclConstants.CONFIG_TOPIC_PERMS, plainAccessConfig1.getTopicPerms());
+                verifyMap.put(AclConstants.CONFIG_GROUP_PERMS, plainAccessConfig1.getGroupPerms());
+            }
+        }
+
+        Assert.assertEquals(verifyMap.size(), 0);
+
+        AclUtils.writeDataObject(targetFileName, backUpAclConfigMap);
+    }
+
+    @Test
+    public void updateGlobalWhiteRemoteAddressesTest() throws InterruptedException {
+        String targetFileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/plain_acl.yml";
+        Map<String, Object> backUpAclConfigMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
+
+        List<String> globalWhiteAddrsList = new ArrayList<>();
+        globalWhiteAddrsList.add("192.168.1.*");
+
+        PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
+        Assert.assertEquals(plainAccessValidator.updateGlobalWhiteAddrsConfig(globalWhiteAddrsList), true);
+
+        String aclFileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/plain_acl.yml";
+        Map<String, Object> readableMap = AclUtils.getYamlDataObject(aclFileName, Map.class);
+        List<Map<String, Object>> dataVersions = (List<Map<String, Object>>) readableMap.get(AclConstants.CONFIG_DATA_VERSION);
+        Assert.assertEquals(1, dataVersions.get(0).get(AclConstants.CONFIG_COUNTER));
+        AclUtils.writeDataObject(targetFileName, backUpAclConfigMap);
+    }
+
+    @Test
+    public void addYamlConfigTest() throws IOException, InterruptedException {
+        String fileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/acl/plain_acl_test.yml";
+        File transport = new File(fileName);
+        transport.delete();
+        transport.createNewFile();
+        FileWriter writer = new FileWriter(transport);
+        writer.write("accounts:\r\n");
+        writer.write("- accessKey: watchrocketmqx\r\n");
+        writer.write("  secretKey: 12345678\r\n");
+        writer.write("  whiteRemoteAddress: 127.0.0.1\r\n");
+        writer.write("  admin: true\r\n");
+        writer.flush();
+        writer.close();
+
+        Thread.sleep(1000);
+
+        PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
+        AclConfig aclConfig = plainAccessValidator.getAllAclConfig();
+        List<PlainAccessConfig> plainAccessConfigs = aclConfig.getPlainAccessConfigs();
+        Map<String, Object> verifyMap = new HashMap<>();
+        for (PlainAccessConfig plainAccessConfig : plainAccessConfigs) {
+            if (plainAccessConfig.getAccessKey().equals("watchrocketmqx")) {
+                verifyMap.put(AclConstants.CONFIG_SECRET_KEY, plainAccessConfig.getSecretKey());
+                verifyMap.put(AclConstants.CONFIG_WHITE_ADDR, plainAccessConfig.getWhiteRemoteAddress());
+                verifyMap.put(AclConstants.CONFIG_ADMIN_ROLE, plainAccessConfig.isAdmin());
+            }
+        }
+
+        Assert.assertEquals(verifyMap.get(AclConstants.CONFIG_SECRET_KEY), "12345678");
+        Assert.assertEquals(verifyMap.get(AclConstants.CONFIG_WHITE_ADDR), "127.0.0.1");
+        Assert.assertEquals(verifyMap.get(AclConstants.CONFIG_ADMIN_ROLE), true);
+
+        Map<String, DataVersion> dataVersionMap = plainAccessValidator.getAllAclConfigVersion();
+        DataVersion dataVersion = dataVersionMap.get(fileName);
+        Assert.assertEquals(0, dataVersion.getCounter().get());
+
+        transport.delete();
+    }
+
+    @Test
+    public void updateAccessAnotherAclYamlConfigTest() throws IOException, InterruptedException {
+        String fileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/acl/plain_acl_test.yml";
+        File transport = new File(fileName);
+        transport.delete();
+        transport.createNewFile();
+        FileWriter writer = new FileWriter(transport);
+        writer.write("accounts:\r\n");
+        writer.write("- accessKey: watchrocketmqy\r\n");
+        writer.write("  secretKey: 12345678\r\n");
+        writer.write("  whiteRemoteAddress: 127.0.0.1\r\n");
+        writer.write("  admin: true\r\n");
+        writer.write("- accessKey: watchrocketmqx\r\n");
+        writer.write("  secretKey: 123456781\r\n");
+        writer.write("  whiteRemoteAddress: 127.0.0.1\r\n");
+        writer.write("  admin: true\r\n");
+        writer.flush();
+        writer.close();
+
+        Thread.sleep(1000);
+
+        PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
+
+        PlainAccessConfig plainAccessConfig = new PlainAccessConfig();
+        plainAccessConfig.setAccessKey("watchrocketmqy");
+        plainAccessConfig.setSecretKey("1234567890");
+        plainAccessConfig.setWhiteRemoteAddress("127.0.0.1");
+        plainAccessConfig.setAdmin(false);
+
+        plainAccessValidator.updateAccessConfig(plainAccessConfig);
+
+        Thread.sleep(1000);
+
+        AclConfig aclConfig = plainAccessValidator.getAllAclConfig();
+        List<PlainAccessConfig> plainAccessConfigs = aclConfig.getPlainAccessConfigs();
+        Map<String, Object> verifyMap = new HashMap<>();
+        for (PlainAccessConfig plainAccessConfig1 : plainAccessConfigs) {
+            if (plainAccessConfig1.getAccessKey().equals("watchrocketmqy")) {
+                verifyMap.put(AclConstants.CONFIG_SECRET_KEY, plainAccessConfig1.getSecretKey());
+                verifyMap.put(AclConstants.CONFIG_WHITE_ADDR, plainAccessConfig1.getWhiteRemoteAddress());
+                verifyMap.put(AclConstants.CONFIG_ADMIN_ROLE, plainAccessConfig1.isAdmin());
+            }
+        }
+
+        Assert.assertEquals(verifyMap.get(AclConstants.CONFIG_SECRET_KEY), "1234567890");
+        Assert.assertEquals(verifyMap.get(AclConstants.CONFIG_WHITE_ADDR), "127.0.0.1");
+        Assert.assertEquals(verifyMap.get(AclConstants.CONFIG_ADMIN_ROLE), false);
+
+        Map<String, DataVersion> dataVersionMap = plainAccessValidator.getAllAclConfigVersion();
+        DataVersion dataVersion = dataVersionMap.get(fileName);
+        Assert.assertEquals(1, dataVersion.getCounter().get());
+
+        transport.delete();
+
     }
 
     @Test(expected = AclException.class)
     public void createAndUpdateAccessAclNullSkExceptionTest() {
+        String targetFileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/acl/plain_acl.yml";
+        Map<String, Object> backUpAclConfigMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
+
         PlainAccessConfig plainAccessConfig = new PlainAccessConfig();
         plainAccessConfig.setAccessKey("RocketMQ33");
         // secret key is null
 
         PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
         plainAccessValidator.updateAccessConfig(plainAccessConfig);
+
+        AclUtils.writeDataObject(targetFileName, backUpAclConfigMap);
     }
 
     @Test
-    public void updateGlobalWhiteAddrsNormalTest() {
-        System.setProperty("rocketmq.home.dir", "src/test/resources");
-        System.setProperty("rocketmq.acl.plain.file", "/conf/plain_acl_global_white_addrs.yml");
-
-        String targetFileName = "src/test/resources/conf/plain_acl_global_white_addrs.yml";
+    public void addAccessDefaultAclYamlConfigTest() throws InterruptedException {
+        PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
+        String targetFileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/plain_acl.yml";
         Map<String, Object> backUpAclConfigMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
 
-        PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
-        // Update global white remote addr value list in the acl access yaml config file
+        PlainAccessConfig plainAccessConfig = new PlainAccessConfig();
+        plainAccessConfig.setAccessKey("watchrocketmqh");
+        plainAccessConfig.setSecretKey("1234567890");
+        plainAccessConfig.setWhiteRemoteAddress("127.0.0.1");
+        plainAccessConfig.setAdmin(false);
 
-        List<String> globalWhiteAddrsList = new ArrayList<String>();
-        globalWhiteAddrsList.add("10.10.154.1");
-        globalWhiteAddrsList.add("10.10.154.2");
-        globalWhiteAddrsList.add("10.10.154.3");
-        plainAccessValidator.updateGlobalWhiteAddrsConfig(globalWhiteAddrsList);
+        plainAccessValidator.updateAccessConfig(plainAccessConfig);
+
+        Thread.sleep(10000);
+
+        AclConfig aclConfig = plainAccessValidator.getAllAclConfig();
+        List<PlainAccessConfig> plainAccessConfigs = aclConfig.getPlainAccessConfigs();
+        Map<String, Object> verifyMap = new HashMap<>();
+        for (PlainAccessConfig plainAccessConfig1 : plainAccessConfigs) {
+            if (plainAccessConfig1.getAccessKey().equals("watchrocketmqh")) {
+                verifyMap.put(AclConstants.CONFIG_SECRET_KEY, plainAccessConfig1.getSecretKey());
+                verifyMap.put(AclConstants.CONFIG_WHITE_ADDR, plainAccessConfig1.getWhiteRemoteAddress());
+                verifyMap.put(AclConstants.CONFIG_ADMIN_ROLE, plainAccessConfig1.isAdmin());
+            }
+        }
+
+        Assert.assertEquals(verifyMap.get(AclConstants.CONFIG_SECRET_KEY), "1234567890");
+        Assert.assertEquals(verifyMap.get(AclConstants.CONFIG_WHITE_ADDR), "127.0.0.1");
+        Assert.assertEquals(verifyMap.get(AclConstants.CONFIG_ADMIN_ROLE), false);
 
         Map<String, Object> readableMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
-
-        List<String> globalWhiteAddrList = (List<String>) readableMap.get(AclConstants.CONFIG_GLOBAL_WHITE_ADDRS);
-        Assert.assertTrue(globalWhiteAddrList.contains("10.10.154.1"));
-        Assert.assertTrue(globalWhiteAddrList.contains("10.10.154.2"));
-        Assert.assertTrue(globalWhiteAddrList.contains("10.10.154.3"));
-
-        // Verify the dateversion element is correct or not
         List<Map<String, Object>> dataVersions = (List<Map<String, Object>>) readableMap.get(AclConstants.CONFIG_DATA_VERSION);
         Assert.assertEquals(1, dataVersions.get(0).get(AclConstants.CONFIG_COUNTER));
 
-        // Restore the backup file and flush to yaml file
         AclUtils.writeDataObject(targetFileName, backUpAclConfigMap);
     }
 
+    @Test
+    public void deleteAccessAnotherAclYamlConfigTest() throws IOException, InterruptedException {
+        String fileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/acl/plain_acl_test.yml";
+        File transport = new File(fileName);
+        transport.delete();
+        transport.createNewFile();
+        FileWriter writer = new FileWriter(transport);
+        writer.write("accounts:\r\n");
+        writer.write("- accessKey: watchrocketmqx\r\n");
+        writer.write("  secretKey: 12345678\r\n");
+        writer.write("  whiteRemoteAddress: 127.0.0.1\r\n");
+        writer.write("  admin: true\r\n");
+        writer.write("- accessKey: watchrocketmqy\r\n");
+        writer.write("  secretKey: 1234567890\r\n");
+        writer.write("  whiteRemoteAddress: 127.0.0.1\r\n");
+        writer.write("  admin: false\r\n");
+        writer.flush();
+        writer.close();
+
+        Thread.sleep(1000);
+
+        PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
+        plainAccessValidator.deleteAccessConfig("watchrocketmqx");
+        Thread.sleep(10000);
+
+        Map<String, Object> verifyMap = new HashMap<>();
+        AclConfig aclConfig = plainAccessValidator.getAllAclConfig();
+        List<PlainAccessConfig> plainAccessConfigs = aclConfig.getPlainAccessConfigs();
+        for (PlainAccessConfig plainAccessConfig : plainAccessConfigs) {
+            if (plainAccessConfig.getAccessKey().equals("watchrocketmqx")) {
+                verifyMap.put(AclConstants.CONFIG_SECRET_KEY, plainAccessConfig.getSecretKey());
+                verifyMap.put(AclConstants.CONFIG_DEFAULT_TOPIC_PERM, plainAccessConfig.getDefaultTopicPerm());
+                verifyMap.put(AclConstants.CONFIG_DEFAULT_GROUP_PERM, plainAccessConfig.getDefaultGroupPerm());
+                verifyMap.put(AclConstants.CONFIG_ADMIN_ROLE, plainAccessConfig.isAdmin());
+                verifyMap.put(AclConstants.CONFIG_WHITE_ADDR, plainAccessConfig.getWhiteRemoteAddress());
+                verifyMap.put(AclConstants.CONFIG_TOPIC_PERMS, plainAccessConfig.getTopicPerms());
+                verifyMap.put(AclConstants.CONFIG_GROUP_PERMS, plainAccessConfig.getGroupPerms());
+            }
+        }
+
+        Assert.assertEquals(verifyMap.size(), 0);
+
+        transport.delete();
+    }
+
     @Test
     public void getAllAclConfigTest() {
         PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
         AclConfig aclConfig = plainAccessValidator.getAllAclConfig();
-        Assert.assertEquals(aclConfig.getGlobalWhiteAddrs().size(), 2);
+        Assert.assertEquals(aclConfig.getGlobalWhiteAddrs().size(), 4);
         Assert.assertEquals(aclConfig.getPlainAccessConfigs().size(), 2);
     }
 
     @Test
     public void updateAccessConfigEmptyPermListTest() {
+        String targetFileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/plain_acl.yml";
+        Map<String, Object> backUpAclConfigMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
+
         PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
         PlainAccessConfig plainAccessConfig = new PlainAccessConfig();
         String accessKey = "updateAccessConfigEmptyPerm";
@@ -724,15 +863,23 @@ public class PlainAccessValidatorTest {
         plainAccessConfig.setTopicPerms(new ArrayList<>());
         plainAccessValidator.updateAccessConfig(plainAccessConfig);
 
-        PlainAccessConfig result = plainAccessValidator.getAllAclConfig().getPlainAccessConfigs()
-            .stream().filter(c -> c.getAccessKey().equals(accessKey)).findFirst().orElse(null);
-        Assert.assertEquals(0, result.getTopicPerms().size());
+        List<PlainAccessConfig> plainAccessConfigs = plainAccessValidator.getAllAclConfig().getPlainAccessConfigs();
+        for (int i = 0; i < plainAccessConfigs.size(); i++) {
+            PlainAccessConfig plainAccessConfig1 = plainAccessConfigs.get(i);
+            if (plainAccessConfig1.getAccessKey() == accessKey) {
+                Assert.assertEquals(0, plainAccessConfig1.getTopicPerms().size());
+            }
+        }
 
         plainAccessValidator.deleteAccessConfig(accessKey);
+        AclUtils.writeDataObject(targetFileName, backUpAclConfigMap);
     }
 
     @Test
     public void updateAccessConfigEmptyWhiteRemoteAddressTest() {
+        String targetFileName = System.getProperty("rocketmq.home.dir") + File.separator + "conf/plain_acl.yml";
+        Map<String, Object> backUpAclConfigMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
+
         PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
         PlainAccessConfig plainAccessConfig = new PlainAccessConfig();
         String accessKey = "updateAccessConfigEmptyWhiteRemoteAddress";
@@ -744,10 +891,15 @@ public class PlainAccessValidatorTest {
         plainAccessConfig.setWhiteRemoteAddress("");
         plainAccessValidator.updateAccessConfig(plainAccessConfig);
 
-        PlainAccessConfig result = plainAccessValidator.getAllAclConfig().getPlainAccessConfigs()
-            .stream().filter(c -> c.getAccessKey().equals(accessKey)).findFirst().orElse(null);
-        Assert.assertEquals("", result.getWhiteRemoteAddress());
+        List<PlainAccessConfig> plainAccessConfigs = plainAccessValidator.getAllAclConfig().getPlainAccessConfigs();
+        for (int i = 0; i < plainAccessConfigs.size(); i++) {
+            PlainAccessConfig plainAccessConfig1 = plainAccessConfigs.get(i);
+            if (plainAccessConfig1.getAccessKey() == accessKey) {
+                Assert.assertEquals("", plainAccessConfig1.getWhiteRemoteAddress());
+            }
+        }
 
         plainAccessValidator.deleteAccessConfig(accessKey);
+        AclUtils.writeDataObject(targetFileName, backUpAclConfigMap);
     }
 }

acl/src/test/java/org/apache/rocketmq/acl/plain/PlainPermissionManagerTest.java

@@ -62,9 +62,9 @@ public class PlainPermissionManagerTest {
         ANYPlainAccessResource = clonePlainAccessResource(Permission.ANY);
         DENYPlainAccessResource = clonePlainAccessResource(Permission.DENY);
 
-        System.setProperty("rocketmq.home.dir", "src/test/resources");
-        System.setProperty("rocketmq.acl.plain.file", "/conf/plain_acl.yml");
-        
+        File file = new File("src/test/resources");
+        System.setProperty("rocketmq.home.dir", file.getAbsolutePath());
+
         plainPermissionManager = new PlainPermissionManager();
 
     }
@@ -117,7 +117,7 @@ public class PlainPermissionManagerTest {
         Assert.assertEquals(resourcePermMap.size(), 3);
 
         Assert.assertEquals(resourcePermMap.get(PlainAccessResource.getRetryTopic("groupA")).byteValue(), Permission.DENY);
-        Assert.assertEquals(resourcePermMap.get(PlainAccessResource.getRetryTopic("groupB")).byteValue(), Permission.PUB|Permission.SUB);
+        Assert.assertEquals(resourcePermMap.get(PlainAccessResource.getRetryTopic("groupB")).byteValue(), Permission.PUB | Permission.SUB);
         Assert.assertEquals(resourcePermMap.get(PlainAccessResource.getRetryTopic("groupC")).byteValue(), Permission.PUB);
 
         List<String> topics = new ArrayList<String>();
@@ -130,7 +130,7 @@ public class PlainPermissionManagerTest {
         Assert.assertEquals(resourcePermMap.size(), 6);
 
         Assert.assertEquals(resourcePermMap.get("topicA").byteValue(), Permission.DENY);
-        Assert.assertEquals(resourcePermMap.get("topicB").byteValue(), Permission.PUB|Permission.SUB);
+        Assert.assertEquals(resourcePermMap.get("topicB").byteValue(), Permission.PUB | Permission.SUB);
         Assert.assertEquals(resourcePermMap.get("topicC").byteValue(), Permission.PUB);
     }
 
@@ -157,6 +157,7 @@ public class PlainPermissionManagerTest {
         plainPermissionManager.checkPerm(plainAccessResource, ANYPlainAccessResource);
 
     }
+
     @Test(expected = AclException.class)
     public void checkErrorPermDefaultValueNotMatch() {
 
@@ -164,6 +165,7 @@ public class PlainPermissionManagerTest {
         plainAccessResource.addResourceAndPerm("topicF", Permission.PUB);
         plainPermissionManager.checkPerm(plainAccessResource, SUBPlainAccessResource);
     }
+
     @Test(expected = AclException.class)
     public void accountNullTest() {
         plainAccessConfig.setAccessKey(null);
@@ -184,25 +186,20 @@ public class PlainPermissionManagerTest {
 
     @Test(expected = AclException.class)
     public void passWordThanTest() {
-        plainAccessConfig.setAccessKey("123");
+        plainAccessConfig.setSecretKey("123");
         plainPermissionManager.buildPlainAccessResource(plainAccessConfig);
     }
 
-    @Test(expected = AclException.class)
-    public void testPlainAclPlugEngineInit() {
-        System.setProperty("rocketmq.home.dir", "");
-        new PlainPermissionManager().load();
-    }
 
     @SuppressWarnings("unchecked")
     @Test
     public void cleanAuthenticationInfoTest() throws IllegalAccessException {
         // PlainPermissionManager.addPlainAccessResource(plainAccessResource);
-        Map<String, List<PlainAccessResource>> plainAccessResourceMap = (Map<String, List<PlainAccessResource>>) FieldUtils.readDeclaredField(plainPermissionManager, "plainAccessResourceMap", true);
+        Map<String, Map<String, PlainAccessResource>> plainAccessResourceMap = (Map<String, Map<String, PlainAccessResource>>) FieldUtils.readDeclaredField(plainPermissionManager, "aclPlainAccessResourceMap", true);
         Assert.assertFalse(plainAccessResourceMap.isEmpty());
 
         plainPermissionManager.clearPermissionInfo();
-        plainAccessResourceMap = (Map<String, List<PlainAccessResource>>) FieldUtils.readDeclaredField(plainPermissionManager, "plainAccessResourceMap", true);
+        plainAccessResourceMap = (Map<String, Map<String, PlainAccessResource>>) FieldUtils.readDeclaredField(plainPermissionManager, "aclPlainAccessResourceMap", true);
         Assert.assertTrue(plainAccessResourceMap.isEmpty());
         // RemoveDataVersionFromYamlFile("src/test/resources/conf/plain_acl.yml");
     }
@@ -213,34 +210,36 @@ public class PlainPermissionManagerTest {
         PlainPermissionManager plainPermissionManager = new PlainPermissionManager();
         Assert.assertTrue(plainPermissionManager.isWatchStart());
         // RemoveDataVersionFromYamlFile("src/test/resources/conf/plain_acl.yml");
-
     }
 
-
     @Test
-    public void testWatch() throws IOException, IllegalAccessException ,InterruptedException{
-        System.setProperty("rocketmq.home.dir", "src/test/resources");
-        System.setProperty("rocketmq.acl.plain.file", "/conf/plain_acl-test.yml");
-        String fileName =System.getProperty("rocketmq.home.dir", "src/test/resources")+System.getProperty("rocketmq.acl.plain.file", "/conf/plain_acl.yml");
+    public void testWatch() throws IOException, IllegalAccessException, InterruptedException {
+        File file = new File("src/test/resources");
+        System.setProperty("rocketmq.home.dir", file.getAbsolutePath());
+
+        String fileName = System.getProperty("rocketmq.home.dir") + File.separator + "/conf/acl/plain_acl_test.yml";
         File transport = new File(fileName);
         transport.delete();
         transport.createNewFile();
         FileWriter writer = new FileWriter(transport);
         writer.write("accounts:\r\n");
-        writer.write("- accessKey: watchrocketmq\r\n");
+        writer.write("- accessKey: watchrocketmqx\r\n");
         writer.write("  secretKey: 12345678\r\n");
         writer.write("  whiteRemoteAddress: 127.0.0.1\r\n");
         writer.write("  admin: true\r\n");
         writer.flush();
         writer.close();
 
+        Thread.sleep(1000);
 
         PlainPermissionManager plainPermissionManager = new PlainPermissionManager();
         Assert.assertTrue(plainPermissionManager.isWatchStart());
 
+        Map<String, String> accessKeyTable = (Map<String, String>) FieldUtils.readDeclaredField(plainPermissionManager, "accessKeyTable", true);
+        String aclFileName = accessKeyTable.get("watchrocketmqx");
         {
-            Map<String, PlainAccessResource> plainAccessResourceMap = (Map<String, PlainAccessResource>) FieldUtils.readDeclaredField(plainPermissionManager, "plainAccessResourceMap", true);
-            PlainAccessResource accessResource = plainAccessResourceMap.get("watchrocketmq");
+            Map<String, Map<String, PlainAccessResource>> plainAccessResourceMap = (Map<String, Map<String, PlainAccessResource>>) FieldUtils.readDeclaredField(plainPermissionManager, "aclPlainAccessResourceMap", true);
+            PlainAccessResource accessResource = plainAccessResourceMap.get(aclFileName).get("watchrocketmqx");
             Assert.assertNotNull(accessResource);
             Assert.assertEquals(accessResource.getSecretKey(), "12345678");
             Assert.assertTrue(accessResource.isAdmin());
@@ -251,16 +250,16 @@ public class PlainPermissionManagerTest {
         List<Map<String, Object>> accounts = (List<Map<String, Object>>) updatedMap.get("accounts");
         accounts.get(0).remove("accessKey");
         accounts.get(0).remove("secretKey");
-        accounts.get(0).put("accessKey", "watchrocketmq1");
+        accounts.get(0).put("accessKey", "watchrocketmq1y");
         accounts.get(0).put("secretKey", "88888888");
         accounts.get(0).put("admin", "false");
         // Update file and flush to yaml file
         AclUtils.writeDataObject(fileName, updatedMap);
 
-        Thread.sleep(1000);
+        Thread.sleep(10000);
         {
-            Map<String, PlainAccessResource> plainAccessResourceMap = (Map<String, PlainAccessResource>) FieldUtils.readDeclaredField(plainPermissionManager, "plainAccessResourceMap", true);
-            PlainAccessResource accessResource = plainAccessResourceMap.get("watchrocketmq1");
+            Map<String, Map<String, PlainAccessResource>> plainAccessResourceMap = (Map<String, Map<String, PlainAccessResource>>) FieldUtils.readDeclaredField(plainPermissionManager, "aclPlainAccessResourceMap", true);
+            PlainAccessResource accessResource = plainAccessResourceMap.get(aclFileName).get("watchrocketmq1y");
             Assert.assertNotNull(accessResource);
             Assert.assertEquals(accessResource.getSecretKey(), "88888888");
             Assert.assertFalse(accessResource.isAdmin());
@@ -268,13 +267,5 @@ public class PlainPermissionManagerTest {
         }
         transport.delete();
         System.setProperty("rocketmq.home.dir", "src/test/resources");
-        System.setProperty("rocketmq.acl.plain.file", "/conf/plain_acl.yml");
-    }
-
-    @Test(expected = AclException.class)
-    public void initializeTest() {
-        System.setProperty("rocketmq.acl.plain.file", "/conf/plain_acl_null.yml");
-        new PlainPermissionManager();
-
     }
 }

acl/src/test/resources/conf/acl/plain_acl.yml

@@ -15,12 +15,30 @@
 
 ## suggested format
 
-date 2015-02-01
-accounts:
-  - name: Jai
-accounts:
-- accessKey: RocketMQ
-  secretKey: 12345678
-  whiteRemoteAddress: 192.168.0.*
-  admin: false
+globalWhiteRemoteAddresses:
+  - 10.10.103.*
+  - 192.168.0.*
+
+accounts:
+  - accessKey: RocketMQ
+    secretKey: 12345678
+    whiteRemoteAddress: 192.168.0.*
+    admin: false
+    defaultTopicPerm: DENY
+    defaultGroupPerm: SUB
+    topicPerms:
+      - topicA=DENY
+      - topicB=PUB|SUB
+      - topicC=SUB
+    groupPerms:
+      # the group should convert to retry topic
+      - groupA=DENY
+      - groupB=SUB
+      - groupC=SUB
+
+  - accessKey: rocketmq2
+    secretKey: 12345678
+    whiteRemoteAddress: 192.168.1.*
+    # if it is admin, it could access all resources
+    admin: true
 

acl/src/test/resources/conf/plain_acl_null.yml

@@ -1,18 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License.  You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-#  Unless required by applicable law or agreed to in writing, software
-#  distributed under the License is distributed on an "AS IS" BASIS,
-#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-#  See the License for the specific language governing permissions and
-#  limitations under the License.
-
-## suggested format
-
-

broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java

@@ -429,6 +429,7 @@ public class AdminBrokerProcessor extends AsyncNettyRequestProcessor implements
         try {
             AccessValidator accessValidator = this.brokerController.getAccessValidatorMap().get(PlainAccessValidator.class);
 
+            responseHeader.setAllAclFileVersion(JSON.toJSONString(accessValidator.getAllAclConfigVersion()));
             responseHeader.setVersion(accessValidator.getAclConfigVersion());
             responseHeader.setBrokerAddr(this.brokerController.getBrokerAddr());
             responseHeader.setBrokerName(this.brokerController.getBrokerConfig().getBrokerName());

client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java

@@ -166,6 +166,7 @@ import org.apache.rocketmq.remoting.netty.ResponseFuture;
 import org.apache.rocketmq.remoting.protocol.LanguageCode;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+import com.alibaba.fastjson.JSON;
 
 public class MQClientAPIImpl {
 
@@ -387,6 +388,12 @@ public class MQClientAPIImpl {
                 clusterAclVersionInfo.setBrokerName(responseHeader.getBrokerName());
                 clusterAclVersionInfo.setBrokerAddr(responseHeader.getBrokerAddr());
                 clusterAclVersionInfo.setAclConfigDataVersion(DataVersion.fromJson(responseHeader.getVersion(), DataVersion.class));
+                HashMap<String, Object> dataVersionMap = JSON.parseObject(responseHeader.getAllAclFileVersion(), HashMap.class);
+                Map<String, DataVersion> allAclConfigDataVersion = new HashMap<String, DataVersion>();
+                for (Map.Entry<String, Object> entry : dataVersionMap.entrySet()) {
+                    allAclConfigDataVersion.put(entry.getKey(),DataVersion.fromJson(JSON.toJSONString(entry.getValue()), DataVersion.class));
+                }
+                clusterAclVersionInfo.setAllAclConfigDataVersion(allAclConfigDataVersion);
                 return clusterAclVersionInfo;
             }
             default:

common/src/main/java/org/apache/rocketmq/common/protocol/body/ClusterAclVersionInfo.java

@@ -19,14 +19,19 @@ package org.apache.rocketmq.common.protocol.body;
 import org.apache.rocketmq.common.DataVersion;
 import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
 
+import java.util.Map;
+
 public class ClusterAclVersionInfo extends RemotingSerializable {
 
     private String brokerName;
 
     private String brokerAddr;
 
+    @Deprecated
     private DataVersion aclConfigDataVersion;
 
+    private Map<String, DataVersion> allAclConfigDataVersion;
+
     private String clusterName;
 
     public String getBrokerName() {
@@ -45,7 +50,6 @@ public class ClusterAclVersionInfo extends RemotingSerializable {
         this.brokerAddr = brokerAddr;
     }
 
-
     public String getClusterName() {
         return clusterName;
     }
@@ -61,4 +65,13 @@ public class ClusterAclVersionInfo extends RemotingSerializable {
     public void setAclConfigDataVersion(DataVersion aclConfigDataVersion) {
         this.aclConfigDataVersion = aclConfigDataVersion;
     }
+
+    public Map<String, DataVersion> getAllAclConfigDataVersion() {
+        return allAclConfigDataVersion;
+    }
+
+    public void setAllAclConfigDataVersion(
+        Map<String, DataVersion> allAclConfigDataVersion) {
+        this.allAclConfigDataVersion = allAclConfigDataVersion;
+    }
 }

common/src/main/java/org/apache/rocketmq/common/protocol/header/GetBrokerAclConfigResponseHeader.java

@@ -25,6 +25,8 @@ public class GetBrokerAclConfigResponseHeader implements CommandCustomHeader {
     @CFNotNull
     private String version;
 
+    private String allAclFileVersion;
+
     @CFNotNull
     private String brokerName;
 
@@ -68,4 +70,12 @@ public class GetBrokerAclConfigResponseHeader implements CommandCustomHeader {
     public void setClusterName(String clusterName) {
         this.clusterName = clusterName;
     }
+
+    public String getAllAclFileVersion() {
+        return allAclFileVersion;
+    }
+
+    public void setAllAclFileVersion(String allAclFileVersion) {
+        this.allAclFileVersion = allAclFileVersion;
+    }
 }

distribution/conf/acl/plain_acl.yml

@@ -14,29 +14,29 @@
 #  limitations under the License.
 
 globalWhiteRemoteAddresses:
-- 10.10.103.*
-- 192.168.0.*
+  - 10.10.103.*
+  - 192.168.0.*
 
 accounts:
-- accessKey: RocketMQ
-  secretKey: 12345678
-  whiteRemoteAddress:
-  admin: false
-  defaultTopicPerm: DENY
-  defaultGroupPerm: SUB
-  topicPerms:
-  - topicA=DENY
-  - topicB=PUB|SUB
-  - topicC=SUB
-  groupPerms:
-  # the group should convert to retry topic
-  - groupA=DENY
-  - groupB=PUB|SUB
-  - groupC=SUB
+  - accessKey: RocketMQ
+    secretKey: 12345678
+    whiteRemoteAddress:
+    admin: false
+    defaultTopicPerm: DENY
+    defaultGroupPerm: SUB
+    topicPerms:
+      - topicA=DENY
+      - topicB=PUB|SUB
+      - topicC=SUB
+    groupPerms:
+      # the group should convert to retry topic
+      - groupA=DENY
+      - groupB=PUB|SUB
+      - groupC=SUB
 
-- accessKey: rocketmq2
-  secretKey: 12345678
-  whiteRemoteAddress: 192.168.1.*
-  # if it is admin, it could access all resources
-  admin: true
+  - accessKey: rocketmq2
+    secretKey: 12345678
+    whiteRemoteAddress: 192.168.1.*
+    # if it is admin, it could access all resources
+    admin: true
     

docs/cn/acl/RocketMQ_Multiple_ACL_Files_设计.md

@@ -0,0 +1,137 @@
+# Version记录
+| 时间 | 主要内容 | 作者 |
+| --- | --- | --- |
+| 2022-01-27 | 初版，包括需求背景、兼容性影响、重要业务逻辑和后续扩展性考虑 | sunxi92 |
+
+中文文档在描述特定专业术语时，仍然使用英文。
+# 需求背景
+RocketMQ ACL特性目前只支持单个ACL配置文件，当存在很多用户时该配置文件会非常大，因此提出支持多ACL配置文件的想法。
+如果支持该特性那么也方便对RocketMQ用户进行分类。
+
+# 兼容性影响
+当前在支持多ACL配置文件特性的设计上是向前兼容的。
+
+# 重要业务逻辑
+## 1. ACL配置文件存储路径
+ACL配置文件夹是在RocketMQ安装目录下的conf/acl目录中，也可以在该路径新建子目录并在子目录中新建ACL配置文件，同时也保留了之前默认的配置文件conf/plain_acl.yml。
+注意：目前用户还不能自定义配置文件目录。
+## 2. ACL配置文件更新
+热感知：当检测到ACL配置文件改动会自动刷新数据，判断ACL配置文件是否发生变化的依据是文件的修改时间是否发生变化
+## 3. RocketMQ Broker缓存ACL配置信息数据结构设计
+- aclPlainAccessResourceMap
+
+aclPlainAccessResourceMap是个Map类型，用来缓存所有ACL配置文件的权限数据，其中key表示ACL配置文件的绝对路径，
+value表示相应配置文件中的权限数据，需要注意的是value也是一个Map类型，其中key是String类型表示AccessKey，value是PlainAccessResource类型。
+- accessKeyTable
+
+accessKeyTable是个Map类型，用来缓存AccessKey和ACL配置文件的映射关系，其中key表示AccessKey，value表示ACL配置文件的绝对路径。
+- globalWhiteRemoteAddressStrategy
+
+globalWhiteRemoteAddressStrategy用来缓存所有ACL配置文件的全局白名单。
+- globalWhiteRemoteAddressStrategyMap
+
+globalWhiteRemoteAddressStrategyMap是个Map类型，用来缓存ACL配置文件和全局白名单的映射关系
+- dataVersionMap
+
+dataVersionMap是个Map类型，用来缓存所有ACL配置文件的DataVersion，其中key表示ACL配置文件的绝对路径，value表示该配置文件对应的DataVersion。
+## 4.加载和监控ACL配置文件
+### 4.1 加载ACL配置文件
+- load()
+
+load()方法会获取"RocketMQ安装目录/conf"目录（包括该目录的子目录）和"rocketmq.acl.plain.file"下所有ACL配置文件，然后遍历这些文件读取权限数据和全局白名单。
+- load(String aclFilePath)
+
+load(String aclFilePath)方法完成加载指定ACL配置文件内容的功能，将配置文件中的全局白名单globalWhiteRemoteAddresses和用户权限accounts加载到缓存中，
+这里需要注意以下几点：
+
+（1）判断缓存中该配置文件的全局白名单globalWhiteRemoteAddresses和用户权限accounts数据是否为空，如果不为空则需要注意删除文件原有数据
+
+（2）相同的accessKey只允许存在在一个ACL配置文件中
+### 4.2 监控ACL配置文件
+watch()方法用来监控"RocketMQ安装目录/conf"目录下所有ACL配置文件和"rocketmq.acl.plain.file"是否发生变化，变化考虑两种情况：一种是ACL配置文件的数量发生变化，
+此时会调用load()方法重新加载所有配置文件的数据；一种是配置文件的内容发生变化；具体完成监控ACL配置文件变化的是AclFileWatchService服务，
+该服务是一个线程，当启动该服务后它会以WATCH_INTERVAL（该参数目前设置为5秒，目前还不能在Broker配置文件中设置）的时间间隔来执行其核心逻辑。在该服务中会记录其监控的ACL配置文件目录aclPath、
+ACL配置文件的数量aclFilesNum、所有ACL配置文件绝对路径fileList以及每个ACL配置文件最近一次修改的时间fileLastModifiedTime
+（Map类型，key为ACL配置文件的绝对路径，value为其最近一次修改时间）。
+该服务的核心逻辑如下：
+获取ACL配置文件数量并和aclFilesNum进行比较是否相等，如果不相等则更新aclFilesNum和fileList并调用load()方法重新加载所有配置文件；
+如果相等则遍历每个ACL配置文件，获取其最近一次修改的时间，并将该时间与fileLastModifiedTime中记录的时间进行比较，如果不相等则表示该文件发生过修改，
+此时调用load(String aclFilePath)方法重新加载该配置文件。
+
+## 5. 权限数据相关操作修改
+（1） updateAclConfigFileVersion(Map<String, Object> updateAclConfigMap)
+
+添加对缓存dataVersionMap的修改
+
+（2）updateAccessConfig(PlainAccessConfig plainAccessConfig)
+
+将该方法原有的逻辑修改为：首先判断accessKeyTable中是否包含待修改的accessKey，如果包含则根据accessKey来获取其对应的ACL配置文件绝对路径，
+再根据该路径更新aclPlainAccessResourceMap中缓存的数据，最后将该ACL配置文件中的数据写回原文件；如果不包含则会将数据写到"rocketmq.acl.plain.file"配置文件中，
+然后更新accessKeyTable和aclPlainAccessResourceMap，最后最后将该ACL配置文件中的数据写回原文件。
+
+（3）deleteAccessConfig(String accesskey)
+
+将该方法原有的逻辑修改为：判断accessKeyTable中是否存在accesskey，如果不存在则返回false，否则将其删除并将修改后的数据写回原文件。
+
+（4）getAllAclConfig()
+
+fileList中存储了所有ACL配置文件的绝对路径，遍历fileList分别从各ACL配置文件中读取数据并组装返回
+
+（5）updateGlobalWhiteAddrsConfig(List<String> globalWhiteAddrsList, String fileName)
+
+该方法是新增的，完成功能是修改指定ACL配置文件的全局白名单，为后续添加相关运维命令做准备
+## 6. ACL相关运维命令修改
+（1）ClusterAclConfigVersionListSubCommand
+
+将printClusterBaseInfo(final DefaultMQAdminExt defaultMQAdminExt, final String addr)方法原有的逻辑修改为：
+获取全部的ACL配置文件的DataVersion并输出。注意：获取的全部ACL配置文件的DataVersion集合可能为空，这里需要添加判断
+
+（2）GetBrokerAclConfigResponseHeader
+
+在GetBrokerAclConfigResponseHeader中新增allAclFileVersion字段，它是个Map类型，其key表示ACL配置文件的绝对路径，value表示对应ACL配置文件的DataVersion
+
+（3）ClusterAclVersionInfo
+
+在ClusterAclVersionInfo中废弃了aclConfigDataVersion属性，增加了allAclConfigDataVersion属性，该属性是个Map类型，用来存储所有ACL配置文件的版本数据，
+其中key表示ACL配置文件的绝对路径，value表示对应ACL配置文件的DataVersion
+
+## 7. 关于ACL配置文件DataVersion存储修改
+
+在原来版本中ACL权限数据存储在一个配置文件中，所以只记录了该配置文件的DataVersion，而现在需要支持多个配置文件特性，每个配置文件都有自己的DataVersion，
+为了能够准确记录所有配置文件的DataVersion，需要调整相关类型的属性、接口及方法。
+
+（1）PlainPermissionManager
+
+对PlainPermissionManager属性的修改具体如下：
+
+- 废弃dataVersion属性，该属性在历史版本中是用来存来存储默认ACL配置文件的DataVersion
+
+- 新增dataVersionMap属性用来缓存所有ACL配置文件的DataVersion，它是一个Map类型，其key表示ACL配置文件的绝对路径，value表示对应配置文件的DataVersion
+
+（2）AccessValidator
+
+对AccessValidator的修改如下：
+
+- 废弃String getAclConfigVersion();，该接口原来是获取ACL配置文件文件的版本数据
+
+- 新增Map<String, DataVersion> getAllAclConfigVersion();该接口是用来获取所有ACL配置文件的版本数据，接口会返回一个Map类型数据，
+key表示各ACL配置文件的绝对路径，value表示对应配置文件的版本数据
+
+（3）PlainAccessValidator
+
+由于PlainAccessValidator实现了AccessValidator接口，所以相应地增加了getAllAclConfigVersion()方法
+
+# 后续扩展性考虑
+1.目前的修改只支持ACL配置文件存储在"RocketMQ安装目录/conf"目录下，后续可以考虑支持多目录；
+
+2.目前ACL配置文件路径是不支持让用户指定，后续可以考虑让用户指定指定ACL配置文件的存储路径
+
+3.当前updateGlobalWhiteAddrsConfig命令只支持修改"rocketmq.acl.plain.file"文件中全局白名单，
+后续可以扩展为修改指定ACL配置文件的全局白名单（如果参数中没有传ACL配置文件则会修改"rocketmq.acl.plain.file"文件）
+
+4.目前ACL数据中的secretKey是以明文形式存储在文件中，在一些对此类信息敏感的行业是不允许以明文落地，后续可以考虑安全性问题
+
+5.目前ACL数据存储只支持文件形式存储，后续可以考虑增加数据库存储
+
+
+

srvutil/src/main/java/org/apache/rocketmq/srvutil/AclFileWatchService.java

@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.srvutil;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.rocketmq.common.ServiceThread;
+import org.apache.rocketmq.common.UtilAll;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.security.MessageDigest;
+import java.util.HashMap;
+import java.util.Map;
+
+public class AclFileWatchService extends ServiceThread {
+    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME);
+
+    private final String aclPath;
+    private int aclFilesNum;
+    @Deprecated
+    private final Map<String, String> fileCurrentHash;
+    private Map<String, Long> fileLastModifiedTime;
+    private List<String/**absolute pathname **/> fileList = new ArrayList<>();
+    private final AclFileWatchService.Listener listener;
+    private static final int WATCH_INTERVAL = 5000;
+    private MessageDigest md = MessageDigest.getInstance("MD5");
+    private String defaultAclFile;
+
+    public AclFileWatchService(String path, String defaultAclFile, final AclFileWatchService.Listener listener) throws Exception {
+        this.aclPath = path;
+        this.defaultAclFile = defaultAclFile;
+        this.fileCurrentHash = new HashMap<>();
+        this.fileLastModifiedTime = new HashMap<>();
+        this.listener = listener;
+
+        getAllAclFiles(path);
+        if (new File(this.defaultAclFile).exists() && !fileList.contains(this.defaultAclFile)) {
+            fileList.add(this.defaultAclFile);
+        }
+        this.aclFilesNum = fileList.size();
+        for (int i = 0; i < aclFilesNum; i++) {
+            String fileAbsolutePath = fileList.get(i);
+            this.fileLastModifiedTime.put(fileAbsolutePath, new File(fileAbsolutePath).lastModified());
+        }
+
+    }
+
+    public void getAllAclFiles(String path) {
+        File file = new File(path);
+        if (!file.exists()) {
+            log.info("The default acl dir {} is not exist", path);
+            return;
+        }
+        File[] files = file.listFiles();
+        for (int i = 0; i < files.length; i++) {
+            String fileName = files[i].getAbsolutePath();
+            File f = new File(fileName);
+            if (fileName.equals(aclPath + File.separator + "tools.yml")) {
+                continue;
+            } else if (fileName.endsWith(".yml") || fileName.endsWith(".yaml")) {
+                fileList.add(fileName);
+            } else if (f.isDirectory()) {
+                getAllAclFiles(fileName);
+            }
+        }
+    }
+
+    @Override
+    public String getServiceName() {
+        return "AclFileWatchService";
+    }
+
+    @Override
+    public void run() {
+        log.info(this.getServiceName() + " service started");
+
+        while (!this.isStopped()) {
+            try {
+                this.waitForRunning(WATCH_INTERVAL);
+
+                if (fileList.size() > 0) {
+                    fileList.clear();
+                }
+                getAllAclFiles(aclPath);
+                if (new File(defaultAclFile).exists() && !fileList.contains(defaultAclFile)) {
+                    fileList.add(defaultAclFile);
+                }
+                int realAclFilesNum = fileList.size();
+
+                if (aclFilesNum != realAclFilesNum) {
+                    log.info("aclFilesNum: " + aclFilesNum + "  realAclFilesNum: " + realAclFilesNum);
+                    aclFilesNum = realAclFilesNum;
+                    log.info("aclFilesNum: " + aclFilesNum + "  realAclFilesNum: " + realAclFilesNum);
+                    Map<String, Long> fileLastModifiedTime = new HashMap<>(realAclFilesNum);
+                    for (int i = 0; i < realAclFilesNum; i++) {
+                        String fileAbsolutePath = fileList.get(i);
+                        fileLastModifiedTime.put(fileAbsolutePath, new File(fileAbsolutePath).lastModified());
+                    }
+                    this.fileLastModifiedTime = fileLastModifiedTime;
+                    listener.onFileNumChanged(aclPath);
+                } else {
+                    for (int i = 0; i < aclFilesNum; i++) {
+                        String fileName = fileList.get(i);
+                        Long newLastModifiedTime = new File(fileName).lastModified();
+                        if (!newLastModifiedTime.equals(fileLastModifiedTime.get(fileName))) {
+                            fileLastModifiedTime.put(fileName, newLastModifiedTime);
+                            listener.onFileChanged(fileName);
+                        }
+                    }
+                }
+            } catch (Exception e) {
+                log.warn(this.getServiceName() + " service has exception. ", e);
+            }
+        }
+        log.info(this.getServiceName() + " service end");
+    }
+
+    @Deprecated
+    private String hash(String filePath) throws IOException {
+        Path path = Paths.get(filePath);
+        md.update(Files.readAllBytes(path));
+        byte[] hash = md.digest();
+        return UtilAll.bytes2string(hash);
+    }
+
+    public interface Listener {
+        /**
+         * Will be called when the target file is changed
+         *
+         * @param aclFileName the changed file absolute path
+         */
+        void onFileChanged(String aclFileName);
+
+        /**
+         * Will be called when the number of the acl file is changed
+         *
+         * @param path the path of the acl dir
+         */
+        void onFileNumChanged(String path);
+    }
+}

tools/src/main/java/org/apache/rocketmq/tools/command/acl/ClusterAclConfigVersionListSubCommand.java

@@ -19,6 +19,7 @@ package org.apache.rocketmq.tools.command.acl;
 import java.sql.Timestamp;
 import java.text.DateFormat;
 import java.text.SimpleDateFormat;
+import java.util.Map;
 import java.util.Set;
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.Option;
@@ -58,7 +59,7 @@ public class ClusterAclConfigVersionListSubCommand implements SubCommand {
 
         optionGroup.setRequired(true);
         options.addOptionGroup(optionGroup);
-        
+
         return options;
     }
 
@@ -85,10 +86,11 @@ public class ClusterAclConfigVersionListSubCommand implements SubCommand {
 
                 Set<String> masterSet =
                     CommandUtil.fetchMasterAddrByClusterName(defaultMQAdminExt, clusterName);
-                System.out.printf("%-16s  %-22s  %-22s  %-20s  %-22s%n",
+                System.out.printf("%-16s  %-22s  %-22s  %-20s  %-22s  %-22s%n",
                     "#Cluster Name",
                     "#Broker Name",
                     "#Broker Addr",
+                    "#AclFilePath",
                     "#AclConfigVersionNum",
                     "#AclLastUpdateTime"
                 );
@@ -112,20 +114,20 @@ public class ClusterAclConfigVersionListSubCommand implements SubCommand {
         final DefaultMQAdminExt defaultMQAdminExt, final String addr) throws
         InterruptedException, MQBrokerException, RemotingException, MQClientException {
 
-
         ClusterAclVersionInfo clusterAclVersionInfo = defaultMQAdminExt.examineBrokerClusterAclVersionInfo(addr);
-        DataVersion aclDataVersion = clusterAclVersionInfo.getAclConfigDataVersion();
-        String versionNum = String.valueOf(aclDataVersion.getCounter());
-
+        Map<String, DataVersion> aclDataVersion = clusterAclVersionInfo.getAllAclConfigDataVersion();
         DateFormat sdf = new SimpleDateFormat(UtilAll.YYYY_MM_DD_HH_MM_SS);
-        String timeStampStr = sdf.format(new Timestamp(aclDataVersion.getTimestamp()));
-
-        System.out.printf("%-16s  %-22s  %-22s  %-20s  %-22s%n",
-            clusterAclVersionInfo.getClusterName(),
-            clusterAclVersionInfo.getBrokerName(),
-            clusterAclVersionInfo.getBrokerAddr(),
-            versionNum,
-            timeStampStr
-        );
+        if (aclDataVersion.size() > 0) {
+            for (Map.Entry<String, DataVersion> entry : aclDataVersion.entrySet()) {
+                System.out.printf("%-16s  %-22s  %-22s  %-20s  %-22s  %-22s%n",
+                    clusterAclVersionInfo.getClusterName(),
+                    clusterAclVersionInfo.getBrokerName(),
+                    clusterAclVersionInfo.getBrokerAddr(),
+                    entry.getKey(),
+                    String.valueOf(entry.getValue().getCounter()),
+                    sdf.format(new Timestamp(entry.getValue().getTimestamp()))
+                );
+            }
+        }
     }
 }