Merge branch 'master' into master

README.md

@@ -1,29 +1,35 @@
-## Apache RocketMQ [![Build Status](https://travis-ci.org/apache/rocketmq.svg?branch=master)](https://travis-ci.org/apache/rocketmq) [![Coverage Status](https://coveralls.io/repos/github/apache/rocketmq/badge.svg?branch=master)](https://coveralls.io/github/apache/rocketmq?branch=master)
+## Apache RocketMQ 
+[![Build Status](https://travis-ci.org/apache/rocketmq.svg?branch=master)](https://travis-ci.org/apache/rocketmq) [![Coverage Status](https://coveralls.io/repos/github/apache/rocketmq/badge.svg?branch=master)](https://coveralls.io/github/apache/rocketmq?branch=master)
 [![Maven Central](https://maven-badges.herokuapp.com/maven-central/org.apache.rocketmq/rocketmq-all/badge.svg)](http://search.maven.org/#search%7Cga%7C1%7Corg.apache.rocketmq)
 [![GitHub release](https://img.shields.io/badge/release-download-orange.svg)](https://rocketmq.apache.org/dowloading/releases)
 [![License](https://img.shields.io/badge/license-Apache%202-4EB1BA.svg)](https://www.apache.org/licenses/LICENSE-2.0.html)
+[![Average time to resolve an issue](http://isitmaintained.com/badge/resolution/apache/rocketmq.svg)](http://isitmaintained.com/project/apache/rocketmq "Average time to resolve an issue")
+[![Percentage of issues still open](http://isitmaintained.com/badge/open/apache/rocketmq.svg)](http://isitmaintained.com/project/apache/rocketmq "Percentage of issues still open")
+![Twitter Follow](https://img.shields.io/twitter/follow/ApacheRocketMQ?style=social)
 
 **[Apache RocketMQ](https://rocketmq.apache.org) is a distributed messaging and streaming platform with low latency, high performance and reliability, trillion-level capacity and flexible scalability.**
 
 It offers a variety of features:
 
-* Pub/Sub messaging model
+* Messageing patterns including publish/subscribe, request/reply and streaming
 * Financial grade transactional message
+* Built-in fault tolerance and high availability configuration options base on [DLedger](https://github.com/openmessaging/openmessaging-storage-dledger)
 * A variety of cross language clients, such as Java, C/C++, Python, Go
 * Pluggable transport protocols, such as TCP, SSL, AIO
-* Inbuilt message tracing capability, also support opentracing
+* Built-in message tracing capability, also support opentracing
 * Versatile big-data and streaming ecosytem integration
 * Message retroactivity by time or offset
 * Reliable FIFO and strict ordered messaging in the same queue
-* Efficient pull&push consumption model
+* Efficient pull and push consumption model
 * Million-level message accumulation capacity in a single queue
 * Multiple messaging protocols like JMS and OpenMessaging
 * Flexible distributed scale-out deployment architecture
 * Lightning-fast batch message exchange system
 * Various message filter mechanics such as SQL and Tag
 * Docker images for isolated testing and cloud isolated clusters
 * Feature-rich administrative dashboard for configuration, metrics and monitoring
-* Authentication and authorisation
+* Authentication and authorization
+* Free open source connectors, for both sources and sinks
 
 ----------
 

acl/pom.xml

@@ -13,7 +13,7 @@
     <parent>
         <groupId>org.apache.rocketmq</groupId>
         <artifactId>rocketmq-all</artifactId>
-        <version>4.5.2</version>
+        <version>4.6.1</version>
     </parent>
     <artifactId>rocketmq-acl</artifactId>
     <name>rocketmq-acl ${project.version}</name>
@@ -67,6 +67,10 @@
             <artifactId>logback-core</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>commons-validator</groupId>
+            <artifactId>commons-validator</artifactId>
+        </dependency>
 
     </dependencies>
 </project>

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

@@ -18,6 +18,7 @@
 package org.apache.rocketmq.acl;
 
 import java.util.List;
+import org.apache.rocketmq.common.AclConfig;
 import org.apache.rocketmq.common.PlainAccessConfig;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 
@@ -66,4 +67,10 @@ public interface AccessValidator {
      * @return
      */
     boolean updateGlobalWhiteAddrsConfig(List<String> globalWhiteAddrsList);
+
+    /**
+     * get broker cluster acl config information
+     * @return
+     */
+    AclConfig getAllAclConfig();
 }

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

@@ -23,6 +23,7 @@
 import java.io.FileWriter;
 import java.io.IOException;
 import java.io.PrintWriter;
+import java.util.ArrayList;
 import java.util.Map;
 import java.util.SortedMap;
 import org.apache.commons.lang3.StringUtils;
@@ -69,24 +70,75 @@ public static String calSignature(byte[] data, String secretKey) {
         return signature;
     }
 
+    public static void IPv6AddressCheck(String netaddress) {
+        if (isAsterisk(netaddress) || isMinus(netaddress)) {
+            int asterisk = netaddress.indexOf("*");
+            int minus = netaddress.indexOf("-");
+//            '*' must be the end of netaddress if it exists
+            if (asterisk > -1 && asterisk != netaddress.length() - 1) {
+                throw new AclException(String.format("Netaddress examine scope Exception netaddress is %s", netaddress));
+            }
+
+//            format like "2::ac5:78:1-200:*" or "2::ac5:78:1-200" is legal
+            if (minus > -1) {
+                if (asterisk == -1) {
+                    if (minus <= netaddress.lastIndexOf(":")) {
+                        throw new AclException(String.format("Netaddress examine scope Exception netaddress is %s", netaddress));
+                    }
+                } else {
+                    if (minus <= netaddress.lastIndexOf(":", netaddress.lastIndexOf(":") - 1)) {
+                        throw new AclException(String.format("Netaddress examine scope Exception netaddress is %s", netaddress));
+                    }
+                }
+            }
+        }
+    }
+
+    public static String v6ipProcess(String netaddress, String[] strArray, int index) {
+        int part;
+        String subAddress;
+        boolean isAsterisk = isAsterisk(netaddress);
+        boolean isMinus = isMinus(netaddress);
+        if (isAsterisk && isMinus) {
+            part = 6;
+            int lastColon = netaddress.lastIndexOf(':');
+            int secondLastColon = netaddress.substring(0, lastColon).lastIndexOf(':');
+            subAddress = netaddress.substring(0, secondLastColon);
+        } else if (!isAsterisk && !isMinus) {
+            part = 8;
+            subAddress = netaddress;
+        } else {
+            part = 7;
+            subAddress = netaddress.substring(0, netaddress.lastIndexOf(':'));
+        }
+        return expandIP(subAddress, part);
+    }
+
     public static void verify(String netaddress, int index) {
         if (!AclUtils.isScope(netaddress, index)) {
             throw new AclException(String.format("Netaddress examine scope Exception netaddress is %s", netaddress));
         }
     }
 
-    public static String[] getAddreeStrArray(String netaddress, String four) {
-        String[] fourStrArray = StringUtils.split(four.substring(1, four.length() - 1), ",");
+    public static String[] getAddreeStrArray(String netaddress, String partialAddress) {
+        String[] parAddStrArray = StringUtils.split(partialAddress.substring(1, partialAddress.length() - 1), ",");
         String address = netaddress.substring(0, netaddress.indexOf("{"));
-        String[] addreeStrArray = new String[fourStrArray.length];
-        for (int i = 0; i < fourStrArray.length; i++) {
-            addreeStrArray[i] = address + fourStrArray[i];
+        String[] addreeStrArray = new String[parAddStrArray.length];
+        for (int i = 0; i < parAddStrArray.length; i++) {
+            addreeStrArray[i] = address + parAddStrArray[i];
         }
         return addreeStrArray;
     }
 
-    public static boolean isScope(String num, int index) {
-        String[] strArray = StringUtils.split(num, ".");
+    public static boolean isScope(String netaddress, int index) {
+//        IPv6 Address
+        if (isColon(netaddress)) {
+            netaddress = expandIP(netaddress, 8);
+            String[] strArray = StringUtils.split(netaddress, ":");
+            return isIPv6Scope(strArray, index);
+        }
+
+        String[] strArray = StringUtils.split(netaddress, ".");
         if (strArray.length != 4) {
             return false;
         }
@@ -107,6 +159,10 @@ public static boolean isScope(String[] num, int index) {
 
     }
 
+    public static boolean isColon(String netaddress) {
+        return netaddress.indexOf(':') > -1;
+    }
+
     public static boolean isScope(String num) {
         return isScope(Integer.valueOf(num.trim()));
     }
@@ -119,7 +175,7 @@ public static boolean isAsterisk(String asterisk) {
         return asterisk.indexOf('*') > -1;
     }
 
-    public static boolean isColon(String colon) {
+    public static boolean isComma(String colon) {
         return colon.indexOf(',') > -1;
     }
 
@@ -128,6 +184,88 @@ public static boolean isMinus(String minus) {
 
     }
 
+    public static boolean isIPv6Scope(String[] num, int index) {
+        for (int i = 0; i < index; i++) {
+            int value;
+            try {
+                value = Integer.parseInt(num[i], 16);
+            } catch (NumberFormatException e) {
+                return false;
+            }
+            if (!isIPv6Scope(value)) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    public static boolean isIPv6Scope(int num) {
+        int min = Integer.parseInt("0", 16);
+        int max = Integer.parseInt("ffff", 16);
+        return num >= min && num <= max;
+    }
+
+    public static String expandIP(String netaddress, int part) {
+        boolean compress = false;
+        int compressIndex = -1;
+        String[] strArray = StringUtils.split(netaddress, ":");
+        ArrayList<Integer> indexes = new ArrayList<>();
+        for (int i = 0; i < netaddress.length(); i++) {
+            if (netaddress.charAt(i) == ':') {
+                if (indexes.size() > 0 && i - indexes.get(indexes.size() - 1) == 1) {
+                    compressIndex = i;
+                    compress = true;
+                }
+                indexes.add(i);
+            }
+        }
+
+        for (int i = 0; i < strArray.length; i++) {
+            if (strArray[i].length() < 4) {
+                strArray[i] = "0000".substring(0, 4 - strArray[i].length()) + strArray[i];
+            }
+        }
+
+        StringBuilder sb = new StringBuilder();
+        if (compress) {
+            int pos = indexes.indexOf(compressIndex);
+            int index = 0;
+            if (!netaddress.startsWith(":")) {
+                for (int i = 0; i < pos; i++) {
+                    sb.append(strArray[index]).append(":");
+                    index += 1;
+                }
+            }
+            int zeroNum = part - strArray.length;
+            if (netaddress.endsWith(":")) {
+                for (int i = 0; i < zeroNum; i++) {
+                    sb.append("0000");
+                    if (i != zeroNum - 1) {
+                        sb.append(":");
+                    }
+                }
+            } else {
+                for (int i = 0; i < zeroNum; i++) {
+                    sb.append("0000").append(":");
+                }
+                for (int i = index; i < strArray.length; i++) {
+                    sb.append(strArray[i]);
+                    if (i != strArray.length - 1) {
+                        sb.append(":");
+                    }
+                }
+            }
+        } else {
+            for (int i = 0; i < strArray.length; i++) {
+                sb.append(strArray[i]);
+                if (i != strArray.length - 1) {
+                    sb.append(":");
+                }
+            }
+        }
+        return sb.toString().toUpperCase();
+    }
+
     public static <T> T getYamlDataObject(String path, Class<T> clazz) {
         Yaml yaml = new Yaml();
         FileInputStream fis = null;
@@ -148,7 +286,7 @@ public static <T> T getYamlDataObject(String path, Class<T> clazz) {
         }
     }
 
-    public static boolean writeDataObject(String path, Map<String,Object> dataMap) {
+    public static boolean writeDataObject(String path, Map<String, Object> dataMap) {
         Yaml yaml = new Yaml();
         PrintWriter pw = null;
         try {
@@ -172,15 +310,15 @@ public static RPCHook getAclRPCHook(String fileName) {
             yamlDataObject = AclUtils.getYamlDataObject(fileName,
                 JSONObject.class);
         } catch (Exception e) {
-            log.error("Convert yaml file to data object error, ",e);
+            log.error("Convert yaml file to data object error, ", e);
             return null;
         }
 
         if (yamlDataObject == null || yamlDataObject.isEmpty()) {
-            log.warn("Cannot find conf file :{}, acl isn't be enabled." ,fileName);
+            log.warn("Cannot find conf file :{}, acl isn't be enabled.", fileName);
             return null;
         }
-        
+
         String accessKey = yamlDataObject.getString(AclConstants.CONFIG_ACCESS_KEY);
         String secretKey = yamlDataObject.getString(AclConstants.CONFIG_SECRET_KEY);
 
@@ -189,7 +327,7 @@ public static RPCHook getAclRPCHook(String fileName) {
 
             return null;
         }
-        return new AclClientRPCHook(new SessionCredentials(accessKey,secretKey));
+        return new AclClientRPCHook(new SessionCredentials(accessKey, secretKey));
     }
 
 }

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

@@ -26,6 +26,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.PlainAccessConfig;
 import org.apache.rocketmq.common.protocol.RequestCode;
 import org.apache.rocketmq.common.protocol.header.GetConsumerListByGroupRequestHeader;
@@ -50,7 +51,7 @@ public PlainAccessValidator() {
     public AccessResource parse(RemotingCommand request, String remoteAddr) {
         PlainAccessResource accessResource = new PlainAccessResource();
         if (remoteAddr != null && remoteAddr.contains(":")) {
-            accessResource.setWhiteRemoteAddress(remoteAddr.split(":")[0]);
+            accessResource.setWhiteRemoteAddress(remoteAddr.substring(0, remoteAddr.lastIndexOf(':')));
         } else {
             accessResource.setWhiteRemoteAddress(remoteAddr);
         }
@@ -155,4 +156,7 @@ public boolean deleteAccessConfig(String accesskey) {
         return aclPlugEngine.updateGlobalWhiteAddrsConfig(globalWhiteAddrsList);
     }
 
+    @Override public AclConfig getAllAclConfig() {
+        return aclPlugEngine.getAllAclConfig();
+    }
 }

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

@@ -30,6 +30,7 @@
 import org.apache.rocketmq.acl.common.AclException;
 import org.apache.rocketmq.acl.common.AclUtils;
 import org.apache.rocketmq.acl.common.Permission;
+import org.apache.rocketmq.common.AclConfig;
 import org.apache.rocketmq.common.DataVersion;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.PlainAccessConfig;
@@ -270,6 +271,28 @@ public boolean updateGlobalWhiteAddrsConfig(List<String> globalWhiteAddrsList) {
         return false;
     }
 
+    public AclConfig getAllAclConfig() {
+        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));
+        }
+        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);
+        return aclConfig;
+    }
+
     private void watch() {
         try {
             String watchFilePath = fileHome + fileName;