[ISSUE#403]fix some bugs and Optimization code for rocketmq's acl feature.

acl/src/main/java/org/apache/rocketmq/acl/common/SessionCredentials.java

@@ -30,7 +30,7 @@ public class SessionCredentials {
     public static final String SECURITY_TOKEN = "SecurityToken";
 
     public static final String KEY_FILE = System.getProperty("rocketmq.client.keyFile",
-        System.getProperty("user.home") + File.separator + "onskey");
+        System.getProperty("user.home") + File.separator + "key");
 
     private String accessKey;
     private String secretKey;

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

@@ -81,8 +81,8 @@ public class PlainPermissionLoader {
         }
 
         JSONArray accounts = accessControlTransport.getJSONArray("accounts");
-        List<PlainAccessConfig> plainAccessList = accounts.toJavaList(PlainAccessConfig.class);
-        if (plainAccessList != null && !plainAccessList.isEmpty()) {
+        if (accounts != null && !accounts.isEmpty()) {
+            List<PlainAccessConfig> plainAccessList = accounts.toJavaList(PlainAccessConfig.class);
             for (PlainAccessConfig plainAccess : plainAccessList) {
                 this.addPlainAccessResource(getPlainAccessResource(plainAccess));
             }
@@ -168,6 +168,11 @@ public class PlainPermissionLoader {
         Map<String, Byte> needCheckedPermMap = needCheckedAccess.getResourcePermMap();
         Map<String, Byte> ownedPermMap = ownedAccess.getResourcePermMap();
 
+        if (needCheckedPermMap == null) {
+            //if the needCheckedPermMap is null,then return
+            return;
+        }
+
         for (Map.Entry<String, Byte> needCheckedEntry : needCheckedPermMap.entrySet()) {
             String resource = needCheckedEntry.getKey();
             Byte neededPerm = needCheckedEntry.getValue();
@@ -223,16 +228,14 @@ public class PlainPermissionLoader {
     public void validate(PlainAccessResource plainAccessResource) {
 
         //Step 1, check the global white remote addr
-        if (plainAccessResource.getAccessKey() == null) {
-            if (globalWhiteRemoteAddressStrategy.isEmpty()) {
-                throw new AclException(String.format("No accessKey is configured and no global white remote addr is configured"));
+        for (RemoteAddressStrategy remoteAddressStrategy : globalWhiteRemoteAddressStrategy) {
+            if (remoteAddressStrategy.match(plainAccessResource)) {
+                return;
             }
-            for (RemoteAddressStrategy remoteAddressStrategy : globalWhiteRemoteAddressStrategy) {
-                if (remoteAddressStrategy.match(plainAccessResource)) {
-                    return;
-                }
-            }
-            throw new AclException(String.format("No accessKey is configured and no global white remote addr is matched"));
+        }
+
+        if (plainAccessResource.getAccessKey() == null) {
+            throw new AclException(String.format("No accessKey is configured"));
         }
 
         if (!plainAccessResourceMap.containsKey(plainAccessResource.getAccessKey())) {

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

@@ -21,19 +21,26 @@ import java.util.Set;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.rocketmq.acl.common.AclException;
 import org.apache.rocketmq.acl.common.AclUtils;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
 
 public class RemoteAddressStrategyFactory {
 
+    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.ACL_PLUG_LOGGER_NAME);
+
     public static final NullRemoteAddressStrategy NULL_NET_ADDRESS_STRATEGY = new NullRemoteAddressStrategy();
 
+    public static final BlankRemoteAddressStrategy BLANK_NET_ADDRESS_STRATEGY = new BlankRemoteAddressStrategy();
+
     public RemoteAddressStrategy getRemoteAddressStrategy(PlainAccessResource plainAccessResource) {
         return getRemoteAddressStrategy(plainAccessResource.getWhiteRemoteAddress());
-
     }
 
     public RemoteAddressStrategy getRemoteAddressStrategy(String remoteAddr) {
         if (StringUtils.isBlank(remoteAddr)) {
-            throw new AclException("Must fill in the white list address");
+            log.warn("white list address is null");
+            return BLANK_NET_ADDRESS_STRATEGY;
         }
         if ("*".equals(remoteAddr)) {
             return NULL_NET_ADDRESS_STRATEGY;
@@ -62,6 +69,14 @@ public class RemoteAddressStrategyFactory {
 
     }
 
+    public static class BlankRemoteAddressStrategy implements RemoteAddressStrategy {
+        @Override
+        public boolean match(PlainAccessResource plainAccessResource) {
+            return false;
+        }
+
+    }
+
     public static class MultipleRemoteAddressStrategy implements RemoteAddressStrategy {
 
         private final Set<String> multipleSet = new HashSet<>();

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

@@ -227,6 +227,7 @@ public class PlainPermissionLoaderTest {
         File file = new File("src/test/resources/watch/conf");
         file.mkdirs();
         File transport = new File("src/test/resources/watch/conf/plain_acl.yml");
+        transport.delete();
         transport.createNewFile();
 
         FileWriter writer = new FileWriter(transport);
@@ -258,11 +259,6 @@ public class PlainPermissionLoaderTest {
         plainAccessResourceMap = (Map<String, List<PlainAccessResource>>) FieldUtils.readDeclaredField(plainPermissionLoader, "plainAccessResourceMap", true);
         Assert.assertNotNull(plainAccessResourceMap.get("rokcet1"));
 
-        transport.delete();
-        file.delete();
-        file = new File("src/test/resources/watch");
-        file.delete();
-
     }
 
     @Test(expected = AclException.class)

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

@@ -24,10 +24,12 @@ public class RemoteAddressStrategyTest {
 
     RemoteAddressStrategyFactory remoteAddressStrategyFactory = new RemoteAddressStrategyFactory();
 
-    @Test(expected = AclException.class)
+    @Test
     public void netaddressStrategyFactoryExceptionTest() {
         PlainAccessResource plainAccessResource = new PlainAccessResource();
         remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+        Assert.assertEquals(remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource).getClass(),
+            RemoteAddressStrategyFactory.BlankRemoteAddressStrategy.class);
     }
 
     @Test
@@ -61,6 +63,10 @@ public class RemoteAddressStrategyTest {
         plainAccessResource.setWhiteRemoteAddress("127.0.1-20.*");
         remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
         Assert.assertEquals(remoteAddressStrategy.getClass(), RemoteAddressStrategyFactory.RangeRemoteAddressStrategy.class);
+
+        plainAccessResource.setWhiteRemoteAddress("");
+        remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+        Assert.assertEquals(remoteAddressStrategy.getClass(), RemoteAddressStrategyFactory.BlankRemoteAddressStrategy.class);
     }
 
     @Test(expected = AclException.class)
@@ -78,6 +84,12 @@ public class RemoteAddressStrategyTest {
         Assert.assertTrue(isMatch);
     }
 
+    @Test
+    public void blankNetaddressStrategyTest() {
+        boolean isMatch = RemoteAddressStrategyFactory.BLANK_NET_ADDRESS_STRATEGY.match(new PlainAccessResource());
+        Assert.assertFalse(isMatch);
+    }
+
     public void oneNetaddressStrategyTest() {
         PlainAccessResource plainAccessResource = new PlainAccessResource();
         plainAccessResource.setWhiteRemoteAddress("127.0.0.1");

broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java

@@ -499,6 +499,7 @@ public class BrokerController {
 
         List<AccessValidator> accessValidators = ServiceProvider.load(ServiceProvider.ACL_VALIDATOR_ID, AccessValidator.class);
         if (accessValidators == null || accessValidators.isEmpty()) {
+            log.info("The broker dose not load the AccessValidator");
             return;
         }
 

broker/src/main/resources/META-INF/service/org.apache.rocketmq.acl.AccessValidator

+org.apache.rocketmq.acl.plain.PlainAccessValidator
\ No newline at end of file

broker/src/test/java/org/apache/rocketmq/broker/BrokerControllerTest.java

@@ -42,6 +42,21 @@ public class BrokerControllerTest {
         brokerController.shutdown();
     }
 
+    @Test
+    public void testBrokerStartAclEnabled() throws Exception {
+        BrokerConfig brokerConfigAclEnabled = new BrokerConfig();
+        brokerConfigAclEnabled.setEnableAcl(true);
+        
+        BrokerController brokerController = new BrokerController(
+            brokerConfigAclEnabled,
+            new NettyServerConfig(),
+            new NettyClientConfig(),
+            new MessageStoreConfig());
+        assertThat(brokerController.initialize());
+        brokerController.start();
+        brokerController.shutdown();
+    }
+
     @After
     public void destroy() {
         UtilAll.deleteFile(new File(new MessageStoreConfig().getStorePathRootDir()));

broker/src/test/resources/META-INF/service/org.apache.rocketmq.acl.AccessValidator

-org.apache.rocketmq.acl.DefaultAclRemotingServiceImpl
+org.apache.rocketmq.acl.plain.PlainAccessValidator
\ No newline at end of file

common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java

@@ -171,7 +171,11 @@ public class BrokerConfig {
     @ImportantField
     private long transactionCheckInterval = 60 * 1000;
 
-    private boolean enableAcl;
+    /**
+     * Acl feature switch
+     */
+    @ImportantField
+    private boolean enableAcl = false;
 
 
     public static String localHostName() {