Merge pull request #1413 from apache/IPV6

[RIP-15]Add Ipv6 support for RocketMQ (#1352)

acl/pom.xml

@@ -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/common/AclUtils.java

@@ -23,6 +23,7 @@ import java.io.FileNotFoundException;
 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 class AclUtils {
         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 class AclUtils {
 
     }
 
+    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 class AclUtils {
         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 class AclUtils {
 
     }
 
+    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 class AclUtils {
         }
     }
 
-    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 class AclUtils {
             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 class AclUtils {
 
             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

@@ -51,7 +51,7 @@ public class PlainAccessValidator implements AccessValidator {
     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);
         }

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

@@ -19,6 +19,7 @@ package org.apache.rocketmq.acl.plain;
 import java.util.HashSet;
 import java.util.Set;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.validator.routines.InetAddressValidator;
 import org.apache.rocketmq.acl.common.AclException;
 import org.apache.rocketmq.acl.common.AclUtils;
 import org.apache.rocketmq.common.constant.LoggerName;
@@ -41,17 +42,26 @@ public class RemoteAddressStrategyFactory {
         if (StringUtils.isBlank(remoteAddr)) {
             return BLANK_NET_ADDRESS_STRATEGY;
         }
-        if ("*".equals(remoteAddr) || "*.*.*.*".equals(remoteAddr)) {
+        if ("*".equals(remoteAddr) || "*.*.*.*".equals(remoteAddr) || "*:*:*:*:*:*:*:*".equals(remoteAddr)) {
             return NULL_NET_ADDRESS_STRATEGY;
         }
         if (remoteAddr.endsWith("}")) {
-            String[] strArray = StringUtils.split(remoteAddr, ".");
-            String four = strArray[3];
-            if (!four.startsWith("{")) {
-                throw new AclException(String.format("MultipleRemoteAddressStrategy netaddress examine scope Exception netaddress", remoteAddr));
+            if (AclUtils.isColon(remoteAddr)) {
+                String[] strArray = StringUtils.split(remoteAddr, ":");
+                String last = strArray[strArray.length - 1];
+                if (!last.startsWith("{")) {
+                    throw new AclException(String.format("MultipleRemoteAddressStrategy netaddress examine scope Exception netaddress", remoteAddr));
+                }
+                return new MultipleRemoteAddressStrategy(AclUtils.getAddreeStrArray(remoteAddr, last));
+            } else {
+                String[] strArray = StringUtils.split(remoteAddr, ".");
+                String four = strArray[3];
+                if (!four.startsWith("{")) {
+                    throw new AclException(String.format("MultipleRemoteAddressStrategy netaddress examine scope Exception netaddress", remoteAddr));
+                }
+                return new MultipleRemoteAddressStrategy(AclUtils.getAddreeStrArray(remoteAddr, four));
             }
-            return new MultipleRemoteAddressStrategy(AclUtils.getAddreeStrArray(remoteAddr, four));
-        } else if (AclUtils.isColon(remoteAddr)) {
+        } else if (AclUtils.isComma(remoteAddr)) {
             return new MultipleRemoteAddressStrategy(StringUtils.split(remoteAddr, ","));
         } else if (AclUtils.isAsterisk(remoteAddr) || AclUtils.isMinus(remoteAddr)) {
             return new RangeRemoteAddressStrategy(remoteAddr);
@@ -81,15 +91,26 @@ public class RemoteAddressStrategyFactory {
         private final Set<String> multipleSet = new HashSet<>();
 
         public MultipleRemoteAddressStrategy(String[] strArray) {
+            InetAddressValidator validator = InetAddressValidator.getInstance();
             for (String netaddress : strArray) {
-                AclUtils.verify(netaddress, 4);
-                multipleSet.add(netaddress);
+                if (validator.isValidInet4Address(netaddress)) {
+                    multipleSet.add(netaddress);
+                } else if (validator.isValidInet6Address(netaddress)) {
+                    multipleSet.add(AclUtils.expandIP(netaddress, 8));
+                } else {
+                    throw new AclException(String.format("Netaddress examine Exception netaddress is %s", netaddress));
+                }
             }
         }
 
         @Override
         public boolean match(PlainAccessResource plainAccessResource) {
-            return multipleSet.contains(plainAccessResource.getWhiteRemoteAddress());
+            InetAddressValidator validator = InetAddressValidator.getInstance();
+            String whiteRemoteAddress = plainAccessResource.getWhiteRemoteAddress();
+            if (validator.isValidInet6Address(whiteRemoteAddress)) {
+                whiteRemoteAddress = AclUtils.expandIP(whiteRemoteAddress, 8);
+            }
+            return multipleSet.contains(whiteRemoteAddress);
         }
 
     }
@@ -100,12 +121,16 @@ public class RemoteAddressStrategyFactory {
 
         public OneRemoteAddressStrategy(String netaddress) {
             this.netaddress = netaddress;
-            AclUtils.verify(netaddress, 4);
+            InetAddressValidator validator = InetAddressValidator.getInstance();
+            if (!(validator.isValidInet4Address(netaddress) || validator.isValidInet6Address(netaddress))) {
+                throw new AclException(String.format("Netaddress examine Exception netaddress is %s", netaddress));
+            }
         }
 
         @Override
         public boolean match(PlainAccessResource plainAccessResource) {
-            return netaddress.equals(plainAccessResource.getWhiteRemoteAddress());
+            String writeRemoteAddress = AclUtils.expandIP(plainAccessResource.getWhiteRemoteAddress(), 8).toUpperCase();
+            return AclUtils.expandIP(netaddress, 8).toUpperCase().equals(writeRemoteAddress);
         }
 
     }
@@ -121,14 +146,29 @@ public class RemoteAddressStrategyFactory {
         private int index;
 
         public RangeRemoteAddressStrategy(String remoteAddr) {
-            String[] strArray = StringUtils.split(remoteAddr, ".");
-            if (analysis(strArray, 1) || analysis(strArray, 2) || analysis(strArray, 3)) {
-                AclUtils.verify(remoteAddr, index - 1);
-                StringBuffer sb = new StringBuffer().append(strArray[0].trim()).append(".").append(strArray[1].trim()).append(".");
-                if (index == 3) {
-                    sb.append(strArray[2].trim()).append(".");
+//            IPv6 Address
+            if (AclUtils.isColon(remoteAddr)) {
+                AclUtils.IPv6AddressCheck(remoteAddr);
+                String[] strArray = StringUtils.split(remoteAddr, ":");
+                for (int i = 1; i < strArray.length; i++) {
+                    if (ipv6Analysis(strArray, i)) {
+                        AclUtils.verify(remoteAddr, index - 1);
+                        String preAddress = AclUtils.v6ipProcess(remoteAddr, strArray, index);
+                        this.index = StringUtils.split(preAddress, ":").length;
+                        this.head = preAddress;
+                        break;
+                    }
+                }
+            } else {
+                String[] strArray = StringUtils.split(remoteAddr, ".");
+                if (analysis(strArray, 1) || analysis(strArray, 2) || analysis(strArray, 3)) {
+                    AclUtils.verify(remoteAddr, index - 1);
+                    StringBuffer sb = new StringBuffer();
+                    for (int j = 0; j < index; j++) {
+                        sb.append(strArray[j].trim()).append(".");
+                    }
+                    this.head = sb.toString();
                 }
-                this.head = sb.toString();
             }
         }
 
@@ -152,6 +192,27 @@ public class RemoteAddressStrategyFactory {
             return this.end > 0 ? true : false;
         }
 
+        private boolean ipv6Analysis(String[] strArray, int index) {
+            String value = strArray[index].trim();
+            this.index = index;
+            if ("*".equals(value)) {
+                int min = Integer.parseInt("0", 16);
+                int max = Integer.parseInt("ffff", 16);
+                setValue(min, max);
+            } else if (AclUtils.isMinus(value)) {
+                if (value.indexOf("-") == 0) {
+                    throw new AclException(String.format("RangeRemoteAddressStrategy netaddress examine scope Exception value %s ", value));
+                }
+                String[] valueArray = StringUtils.split(value, "-");
+                this.start = Integer.parseInt(valueArray[0], 16);
+                this.end = Integer.parseInt(valueArray[1], 16);
+                if (!(AclUtils.isIPv6Scope(end) && AclUtils.isIPv6Scope(start) && start <= end)) {
+                    throw new AclException(String.format("RangeRemoteAddressStrategy netaddress examine scope Exception start is %s , end is %s", start, end));
+                }
+            }
+            return this.end > 0 ? true : false;
+        }
+
         private void setValue(int start, int end) {
             this.start = start;
             this.end = end;
@@ -160,21 +221,33 @@ public class RemoteAddressStrategyFactory {
         @Override
         public boolean match(PlainAccessResource plainAccessResource) {
             String netAddress = plainAccessResource.getWhiteRemoteAddress();
-            if (netAddress.startsWith(this.head)) {
-                String value;
-                if (index == 3) {
-                    value = netAddress.substring(this.head.length());
-                } else {
-                    value = netAddress.substring(this.head.length(), netAddress.lastIndexOf('.'));
+            InetAddressValidator validator = InetAddressValidator.getInstance();
+            if (validator.isValidInet4Address(netAddress)) {
+                if (netAddress.startsWith(this.head)) {
+                    String value;
+                    if (index == 3) {
+                        value = netAddress.substring(this.head.length());
+                    } else if (index == 2) {
+                        value = netAddress.substring(this.head.length(), netAddress.lastIndexOf('.'));
+                    } else {
+                        value = netAddress.substring(this.head.length(), netAddress.lastIndexOf('.', netAddress.lastIndexOf('.') - 1));
+                    }
+                    Integer address = Integer.valueOf(value);
+                    if (address >= this.start && address <= this.end) {
+                        return true;
+                    }
                 }
-                Integer address = Integer.valueOf(value);
-                if (address >= this.start && address <= this.end) {
-                    return true;
+            } else if (validator.isValidInet6Address(netAddress)) {
+                netAddress = AclUtils.expandIP(netAddress, 8).toUpperCase();
+                if (netAddress.startsWith(this.head)) {
+                    String value = netAddress.substring(5 * index, 5 * index + 4);
+                    Integer address = Integer.parseInt(value, 16);
+                    if (address >= this.start && address <= this.end) {
+                        return true;
+                    }
                 }
             }
             return false;
         }
-
     }
-
 }

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

@@ -46,20 +46,35 @@ public class AclUtilsTest {
         addressList.add("1.1.1.3");
         addressList.add("1.1.1.4");
         Assert.assertEquals(newAddressList, addressList);
+
+//        IPv6 test
+        String ipv6Address = "1:ac41:9987::bb22:666:{1,2,3,4}";
+        String[] ipv6AddressArray = AclUtils.getAddreeStrArray(ipv6Address, "{1,2,3,4}");
+        List<String> newIPv6AddressList = new ArrayList<>();
+        for (String a : ipv6AddressArray) {
+            newIPv6AddressList.add(a);
+        }
+
+        List<String> ipv6AddressList = new ArrayList<>();
+        ipv6AddressList.add("1:ac41:9987::bb22:666:1");
+        ipv6AddressList.add("1:ac41:9987::bb22:666:2");
+        ipv6AddressList.add("1:ac41:9987::bb22:666:3");
+        ipv6AddressList.add("1:ac41:9987::bb22:666:4");
+        Assert.assertEquals(newIPv6AddressList, ipv6AddressList);
     }
 
     @Test
     public void isScopeStringArray() {
-        String adderss = "12";
+        String address = "12";
 
         for (int i = 0; i < 6; i++) {
-            boolean isScope = AclUtils.isScope(adderss, 4);
+            boolean isScope = AclUtils.isScope(address, 4);
             if (i == 3) {
                 Assert.assertTrue(isScope);
             } else {
                 Assert.assertFalse(isScope);
             }
-            adderss = adderss + ".12";
+            address = address + ".12";
         }
     }
 
@@ -77,6 +92,25 @@ public class AclUtilsTest {
         isScope = AclUtils.isScope(adderss, 3);
         Assert.assertFalse(isScope);
 
+//        IPv6 test
+        adderss = StringUtils.split("1050:0000:0000:0000:0005:0600:300c:326b", ":");
+        isScope = AclUtils.isIPv6Scope(adderss, 8);
+        Assert.assertTrue(isScope);
+        isScope = AclUtils.isIPv6Scope(adderss, 4);
+        Assert.assertTrue(isScope);
+
+        adderss = StringUtils.split("1050:9876:0000:0000:0005:akkg:300c:326b", ":");
+        isScope = AclUtils.isIPv6Scope(adderss, 8);
+        Assert.assertFalse(isScope);
+        isScope = AclUtils.isIPv6Scope(adderss, 4);
+        Assert.assertTrue(isScope);
+
+        adderss = StringUtils.split(AclUtils.expandIP("1050::0005:akkg:300c:326b", 8), ":");
+        isScope = AclUtils.isIPv6Scope(adderss, 8);
+        Assert.assertFalse(isScope);
+        isScope = AclUtils.isIPv6Scope(adderss, 4);
+        Assert.assertTrue(isScope);
+
     }
 
     @Test
@@ -102,6 +136,18 @@ public class AclUtilsTest {
         isScope = AclUtils.isScope(256);
         Assert.assertFalse(isScope);
 
+        //        IPv6 test
+        int min = Integer.parseInt("0", 16);
+        int max = Integer.parseInt("ffff", 16);
+        for (int i = min; i < max + 1; i++) {
+            isScope = AclUtils.isIPv6Scope(i);
+            Assert.assertTrue(isScope);
+        }
+        isScope = AclUtils.isIPv6Scope(-1);
+        Assert.assertFalse(isScope);
+        isScope = AclUtils.isIPv6Scope(max + 1);
+        Assert.assertFalse(isScope);
+
     }
 
     @Test
@@ -115,10 +161,10 @@ public class AclUtilsTest {
 
     @Test
     public void isColonTest() {
-        boolean isColon = AclUtils.isColon(",");
+        boolean isColon = AclUtils.isComma(",");
         Assert.assertTrue(isColon);
 
-        isColon = AclUtils.isColon("-");
+        isColon = AclUtils.isComma("-");
         Assert.assertFalse(isColon);
     }
 
@@ -131,6 +177,36 @@ public class AclUtilsTest {
         Assert.assertFalse(isMinus);
     }
 
+    @Test
+    public void v6ipProcessTest() {
+        String remoteAddr = "5::7:6:1-200:*";
+        String[] strArray = StringUtils.split(remoteAddr, ":");
+        Assert.assertEquals(AclUtils.v6ipProcess(remoteAddr, strArray, 3), "0005:0000:0000:0000:0007:0006");
+
+        remoteAddr = "5::7:6:1-200";
+        strArray = StringUtils.split(remoteAddr, ":");
+        Assert.assertEquals(AclUtils.v6ipProcess(remoteAddr, strArray, 3), "0005:0000:0000:0000:0000:0007:0006");
+
+        remoteAddr = "5::7:6:*";
+        strArray = StringUtils.split(remoteAddr, ":");
+        Assert.assertEquals(AclUtils.v6ipProcess(remoteAddr, strArray, 3), "0005:0000:0000:0000:0000:0007:0006");
+
+        remoteAddr = "5:7:6:*";
+        strArray = StringUtils.split(remoteAddr, ":");
+        Assert.assertEquals(AclUtils.v6ipProcess(remoteAddr, strArray, 3), "0005:0007:0006");
+    }
+
+    @Test
+    public void expandIPTest() {
+        Assert.assertEquals(AclUtils.expandIP("::1", 8), "0000:0000:0000:0000:0000:0000:0000:0001");
+        Assert.assertEquals(AclUtils.expandIP("3::", 8), "0003:0000:0000:0000:0000:0000:0000:0000");
+        Assert.assertEquals(AclUtils.expandIP("2::2", 8), "0002:0000:0000:0000:0000:0000:0000:0002");
+        Assert.assertEquals(AclUtils.expandIP("4::aac4:92", 8), "0004:0000:0000:0000:0000:0000:AAC4:0092");
+        Assert.assertEquals(AclUtils.expandIP("ab23:56:901a::cc6:765:bb:9011", 8), "AB23:0056:901A:0000:0CC6:0765:00BB:9011");
+        Assert.assertEquals(AclUtils.expandIP("ab23:56:901a:1:cc6:765:bb:9011", 8), "AB23:0056:901A:0001:0CC6:0765:00BB:9011");
+        Assert.assertEquals(AclUtils.expandIP("5::7:6", 6), "0005:0000:0000:0000:0007:0006");
+    }
+
     @SuppressWarnings("unchecked")
     @Test
     public void getYamlDataObjectTest() {
@@ -140,7 +216,7 @@ public class AclUtilsTest {
     }
 
     @Test
-    public void writeDataObject2YamlFileTest() throws IOException{
+    public void writeDataObject2YamlFileTest() throws IOException {
 
         String targetFileName = "src/test/resources/conf/plain_write_acl.yml";
         File transport = new File(targetFileName);
@@ -153,7 +229,7 @@ public class AclUtilsTest {
         List<String> globalWhiteRemoteAddrs = new ArrayList<String>();
         globalWhiteRemoteAddrs.add("10.10.103.*");
         globalWhiteRemoteAddrs.add("192.168.0.*");
-        aclYamlMap.put("globalWhiteRemoteAddrs",globalWhiteRemoteAddrs);
+        aclYamlMap.put("globalWhiteRemoteAddrs", globalWhiteRemoteAddrs);
 
         // For accounts element in acl yaml config file
         List<Map<String, Object>> accounts = new ArrayList<Map<String, Object>>();
@@ -166,14 +242,14 @@ public class AclUtilsTest {
             }
         };
         accounts.add(accountsMap);
-        aclYamlMap.put("accounts",accounts);
+        aclYamlMap.put("accounts", accounts);
         Assert.assertTrue(AclUtils.writeDataObject(targetFileName, aclYamlMap));
 
         transport.delete();
     }
 
     @Test
-    public void updateExistedYamlFileTest()  throws IOException{
+    public void updateExistedYamlFileTest() throws IOException {
 
         String targetFileName = "src/test/resources/conf/plain_update_acl.yml";
         File transport = new File(targetFileName);
@@ -186,7 +262,7 @@ public class AclUtilsTest {
         List<String> globalWhiteRemoteAddrs = new ArrayList<String>();
         globalWhiteRemoteAddrs.add("10.10.103.*");
         globalWhiteRemoteAddrs.add("192.168.0.*");
-        aclYamlMap.put("globalWhiteRemoteAddrs",globalWhiteRemoteAddrs);
+        aclYamlMap.put("globalWhiteRemoteAddrs", globalWhiteRemoteAddrs);
 
         // Write file to yaml file
         AclUtils.writeDataObject(targetFileName, aclYamlMap);
@@ -201,7 +277,7 @@ public class AclUtilsTest {
 
         Map<String, Object> readableMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
         List<String> updatedGlobalWhiteRemoteAddrs = (List<String>) readableMap.get("globalWhiteRemoteAddrs");
-        Assert.assertEquals("192.168.1.2",updatedGlobalWhiteRemoteAddrs.get(0));
+        Assert.assertEquals("192.168.1.2", updatedGlobalWhiteRemoteAddrs.get(0));
 
         transport.delete();
     }
@@ -235,5 +311,4 @@ public class AclUtilsTest {
         Assert.assertNull(incompleteContRPCHook);
     }
 
-
 }

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

@@ -39,7 +39,7 @@ public class RemoteAddressStrategyTest {
         plainAccessResource.setWhiteRemoteAddress("*");
         RemoteAddressStrategy remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
         Assert.assertEquals(remoteAddressStrategy, RemoteAddressStrategyFactory.NULL_NET_ADDRESS_STRATEGY);
-        
+
         plainAccessResource.setWhiteRemoteAddress("*.*.*.*");
         remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
         Assert.assertEquals(remoteAddressStrategy, RemoteAddressStrategyFactory.NULL_NET_ADDRESS_STRATEGY);
@@ -71,6 +71,35 @@ public class RemoteAddressStrategyTest {
         plainAccessResource.setWhiteRemoteAddress("");
         remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
         Assert.assertEquals(remoteAddressStrategy.getClass(), RemoteAddressStrategyFactory.BlankRemoteAddressStrategy.class);
+
+//        IPv6 test
+        plainAccessResource.setWhiteRemoteAddress("*:*:*:*:*:*:*:*");
+        remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+        Assert.assertEquals(remoteAddressStrategy, RemoteAddressStrategyFactory.NULL_NET_ADDRESS_STRATEGY);
+
+        plainAccessResource.setWhiteRemoteAddress("1050:0000:0000:0000:0005:0600:300c:326b");
+        remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+        Assert.assertEquals(remoteAddressStrategy.getClass(), RemoteAddressStrategyFactory.OneRemoteAddressStrategy.class);
+
+        plainAccessResource.setWhiteRemoteAddress("1050::0005:0600:300c:3261,1050::0005:0600:300c:3262,1050::0005:0600:300c:3263");
+        remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+        Assert.assertEquals(remoteAddressStrategy.getClass(), RemoteAddressStrategyFactory.MultipleRemoteAddressStrategy.class);
+
+        plainAccessResource.setWhiteRemoteAddress("1050::0005:0600:300c:3261:{1,2,3}");
+        remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+        Assert.assertEquals(remoteAddressStrategy.getClass(), RemoteAddressStrategyFactory.MultipleRemoteAddressStrategy.class);
+
+        plainAccessResource.setWhiteRemoteAddress("1050::0005:0600:300c:3261:1-200");
+        remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+        Assert.assertEquals(remoteAddressStrategy.getClass(), RemoteAddressStrategyFactory.RangeRemoteAddressStrategy.class);
+
+        plainAccessResource.setWhiteRemoteAddress("1050:0005:0600:300c:3261:*");
+        remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+        Assert.assertEquals(remoteAddressStrategy.getClass(), RemoteAddressStrategyFactory.RangeRemoteAddressStrategy.class);
+
+        plainAccessResource.setWhiteRemoteAddress("1050::0005:0600:300c:3261:1-20:*");
+        remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+        Assert.assertEquals(remoteAddressStrategy.getClass(), RemoteAddressStrategyFactory.RangeRemoteAddressStrategy.class);
     }
 
     @Test(expected = AclException.class)
@@ -80,6 +109,8 @@ public class RemoteAddressStrategyTest {
         remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
         plainAccessResource.setWhiteRemoteAddress("256.0.0.1");
         remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+        plainAccessResource.setWhiteRemoteAddress("::1ggg");
+        remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
     }
 
     @Test
@@ -94,6 +125,7 @@ public class RemoteAddressStrategyTest {
         Assert.assertFalse(isMatch);
     }
 
+    @Test
     public void oneNetaddressStrategyTest() {
         PlainAccessResource plainAccessResource = new PlainAccessResource();
         plainAccessResource.setWhiteRemoteAddress("127.0.0.1");
@@ -109,6 +141,26 @@ public class RemoteAddressStrategyTest {
         plainAccessResource.setWhiteRemoteAddress("127.0.0.1");
         match = remoteAddressStrategy.match(plainAccessResource);
         Assert.assertTrue(match);
+
+//        Ipv6 test
+        plainAccessResource = new PlainAccessResource();
+        plainAccessResource.setWhiteRemoteAddress("::1");
+        remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+        plainAccessResource.setWhiteRemoteAddress("");
+        match = remoteAddressStrategy.match(plainAccessResource);
+        Assert.assertFalse(match);
+
+        plainAccessResource.setWhiteRemoteAddress("::2");
+        match = remoteAddressStrategy.match(plainAccessResource);
+        Assert.assertFalse(match);
+
+        plainAccessResource.setWhiteRemoteAddress("::1");
+        match = remoteAddressStrategy.match(plainAccessResource);
+        Assert.assertTrue(match);
+
+        plainAccessResource.setWhiteRemoteAddress("0000:0000:0000:0000:0000:0000:0000:0001");
+        match = remoteAddressStrategy.match(plainAccessResource);
+        Assert.assertTrue(match);
     }
 
     @Test
@@ -122,6 +174,21 @@ public class RemoteAddressStrategyTest {
         remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
         multipleNetaddressStrategyTest(remoteAddressStrategy);
 
+        plainAccessResource.setWhiteRemoteAddress("192.100-150.*.*");
+        remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+        plainAccessResource.setWhiteRemoteAddress("192.130.0.2");
+        boolean match = remoteAddressStrategy.match(plainAccessResource);
+        Assert.assertTrue(match);
+
+        plainAccessResource = new PlainAccessResource();
+        plainAccessResource.setWhiteRemoteAddress("1050::0005:0600:300c:1,1050::0005:0600:300c:2,1050::0005:0600:300c:3");
+        remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+        multipleIPv6NetaddressStrategyTest(remoteAddressStrategy);
+
+        plainAccessResource.setWhiteRemoteAddress("1050::0005:0600:300c:{1,2,3}");
+        remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+        multipleIPv6NetaddressStrategyTest(remoteAddressStrategy);
+
     }
 
     @Test(expected = AclException.class)
@@ -129,6 +196,8 @@ public class RemoteAddressStrategyTest {
         PlainAccessResource plainAccessResource = new PlainAccessResource();
         plainAccessResource.setWhiteRemoteAddress("127.0.0.1,2,3}");
         remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+        plainAccessResource.setWhiteRemoteAddress("::1,2,3}");
+        remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
     }
 
     private void multipleNetaddressStrategyTest(RemoteAddressStrategy remoteAddressStrategy) {
@@ -155,6 +224,30 @@ public class RemoteAddressStrategyTest {
 
     }
 
+    private void multipleIPv6NetaddressStrategyTest(RemoteAddressStrategy remoteAddressStrategy) {
+        PlainAccessResource plainAccessResource = new PlainAccessResource();
+        plainAccessResource.setWhiteRemoteAddress("1050:0000:0000:0000:0005:0600:300c:1");
+        boolean match = remoteAddressStrategy.match(plainAccessResource);
+        Assert.assertTrue(match);
+
+        plainAccessResource.setWhiteRemoteAddress("1050:0000:0000:0000:0005:0600:300c:2");
+        match = remoteAddressStrategy.match(plainAccessResource);
+        Assert.assertTrue(match);
+
+        plainAccessResource.setWhiteRemoteAddress("1050:0000:0000:0000:0005:0600:300c:3");
+        match = remoteAddressStrategy.match(plainAccessResource);
+        Assert.assertTrue(match);
+
+        plainAccessResource.setWhiteRemoteAddress("1050:0000:0000:0000:0005:0600:300c:4");
+        match = remoteAddressStrategy.match(plainAccessResource);
+        Assert.assertFalse(match);
+
+        plainAccessResource.setWhiteRemoteAddress("1050:0000:0000:0000:0005:0600:300c:0");
+        match = remoteAddressStrategy.match(plainAccessResource);
+        Assert.assertFalse(match);
+
+    }
+
     @Test
     public void rangeNetaddressStrategyTest() {
         String head = "127.0.0.";
@@ -162,6 +255,7 @@ public class RemoteAddressStrategyTest {
         plainAccessResource.setWhiteRemoteAddress("127.0.0.1-200");
         RemoteAddressStrategy remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
         rangeNetaddressStrategyTest(remoteAddressStrategy, head, 1, 200, true);
+
         plainAccessResource.setWhiteRemoteAddress("127.0.0.*");
         remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
         rangeNetaddressStrategyTest(remoteAddressStrategy, head, 0, 255, true);
@@ -169,14 +263,40 @@ public class RemoteAddressStrategyTest {
         plainAccessResource.setWhiteRemoteAddress("127.0.1-200.*");
         remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
         rangeNetaddressStrategyThirdlyTest(remoteAddressStrategy, head, 1, 200);
-        
+
         plainAccessResource.setWhiteRemoteAddress("127.*.*.*");
         remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
-        rangeNetaddressStrategyThirdlyTest(remoteAddressStrategy, head, 1, 200);
-        
+        rangeNetaddressStrategyTest(remoteAddressStrategy, head, 0, 255, true);
+
         plainAccessResource.setWhiteRemoteAddress("127.1-150.*.*");
         remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
         rangeNetaddressStrategyThirdlyTest(remoteAddressStrategy, head, 1, 200);
+
+//        IPv6 test
+        head = "1050::0005:0600:300c:";
+        plainAccessResource = new PlainAccessResource();
+        plainAccessResource.setWhiteRemoteAddress("1050::0005:0600:300c:1-200");
+        remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+        rangeIPv6NetaddressStrategyTest(remoteAddressStrategy, head, "1", "200", true);
+
+        plainAccessResource.setWhiteRemoteAddress("1050::0005:0600:300c:*");
+        remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+        rangeIPv6NetaddressStrategyTest(remoteAddressStrategy, head, "0", "ffff", true);
+
+        plainAccessResource.setWhiteRemoteAddress("1050::0005:0600:3001:*");
+        remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+        rangeIPv6NetaddressStrategyTest(remoteAddressStrategy, head, "0", "ffff", false);
+
+        head = "1050::0005:0600:300c:1:";
+        plainAccessResource.setWhiteRemoteAddress("1050::0005:0600:300c:1-200:*");
+        remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+        rangeIPv6NetaddressStrategyTest(remoteAddressStrategy, head, "0", "ffff", true);
+
+        head = "1050::0005:0600:300c:201:";
+        plainAccessResource.setWhiteRemoteAddress("1050::0005:0600:300c:1-200:*");
+        remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+        rangeIPv6NetaddressStrategyTest(remoteAddressStrategy, head, "0", "ffff", false);
+
     }
 
     private void rangeNetaddressStrategyTest(RemoteAddressStrategy remoteAddressStrategy, String head, int start,
@@ -206,6 +326,25 @@ public class RemoteAddressStrategyTest {
         }
     }
 
+    private void rangeIPv6NetaddressStrategyTest(RemoteAddressStrategy remoteAddressStrategy, String head, String start,
+        String end,
+        boolean isFalse) {
+        PlainAccessResource plainAccessResource = new PlainAccessResource();
+        for (int i = -10; i < 65536 + 100; i++) {
+            String hex = Integer.toHexString(i);
+            plainAccessResource.setWhiteRemoteAddress(head + hex);
+            boolean match = remoteAddressStrategy.match(plainAccessResource);
+            int startNum = Integer.parseInt(start, 16);
+            int endNum = Integer.parseInt(end, 16);
+            if (isFalse && i >= startNum && i <= endNum) {
+                Assert.assertTrue(match);
+                continue;
+            }
+            Assert.assertFalse(match);
+
+        }
+    }
+
     @Test(expected = AclException.class)
     public void rangeNetaddressStrategyExceptionStartGreaterEndTest() {
         rangeNetaddressStrategyExceptionTest("127.0.0.2-1");

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

@@ -41,8 +41,6 @@ import org.apache.rocketmq.common.constant.PermName;
 import org.apache.rocketmq.common.filter.ExpressionType;
 import org.apache.rocketmq.common.filter.FilterAPI;
 import org.apache.rocketmq.common.help.FAQUrl;
-import org.apache.rocketmq.logging.InternalLogger;
-import org.apache.rocketmq.logging.InternalLoggerFactory;
 import org.apache.rocketmq.common.message.MessageDecoder;
 import org.apache.rocketmq.common.message.MessageQueue;
 import org.apache.rocketmq.common.protocol.ResponseCode;
@@ -52,7 +50,10 @@ import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
 import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
 import org.apache.rocketmq.common.protocol.topic.OffsetMovedEvent;
 import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
+import org.apache.rocketmq.common.sysflag.MessageSysFlag;
 import org.apache.rocketmq.common.sysflag.PullSysFlag;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
 import org.apache.rocketmq.remoting.common.RemotingHelper;
 import org.apache.rocketmq.remoting.common.RemotingUtil;
 import org.apache.rocketmq.remoting.exception.RemotingCommandException;
@@ -494,7 +495,21 @@ public class PullMessageProcessor implements NettyRequestProcessor {
             for (ByteBuffer bb : messageBufferList) {
 
                 byteBuffer.put(bb);
-                storeTimestamp = bb.getLong(MessageDecoder.MESSAGE_STORE_TIMESTAMP_POSTION);
+                int sysFlag = bb.getInt(MessageDecoder.SYSFLAG_POSITION);
+//                bornhost has the IPv4 ip if the MessageSysFlag.BORNHOST_V6_FLAG bit of sysFlag is 0
+//                IPv4 host = ip(4 byte) + port(4 byte); IPv6 host = ip(16 byte) + port(4 byte)
+                int bornhostLength = (sysFlag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 8 : 20;
+                int msgStoreTimePos = 4 // 1 TOTALSIZE
+                    + 4 // 2 MAGICCODE
+                    + 4 // 3 BODYCRC
+                    + 4 // 4 QUEUEID
+                    + 4 // 5 FLAG
+                    + 8 // 6 QUEUEOFFSET
+                    + 8 // 7 PHYSICALOFFSET
+                    + 4 // 8 SYSFLAG
+                    + 8 // 9 BORNTIMESTAMP
+                    + bornhostLength; // 10 BORNHOST
+                storeTimestamp = bb.getLong(msgStoreTimePos);
             }
         } finally {
             getMessageResult.release();

common/pom.xml

@@ -41,5 +41,9 @@
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-lang3</artifactId>
         </dependency>
+        <dependency>
+            <groupId>commons-validator</groupId>
+            <artifactId>commons-validator</artifactId>
+        </dependency>
     </dependencies>
 </project>

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

@@ -124,8 +124,10 @@ public class MixAll {
 
     public static String brokerVIPChannel(final boolean isChange, final String brokerAddr) {
         if (isChange) {
-            String[] ipAndPort = brokerAddr.split(":");
-            String brokerAddrNew = ipAndPort[0] + ":" + (Integer.parseInt(ipAndPort[1]) - 2);
+            int split = brokerAddr.lastIndexOf(":");
+            String ip = brokerAddr.substring(0, split);
+            String port = brokerAddr.substring(split + 1);
+            String brokerAddrNew = ip + ":" + (Integer.parseInt(port) - 2);
             return brokerAddrNew;
         } else {
             return brokerAddr;

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

@@ -23,6 +23,7 @@ import java.io.IOException;
 import java.lang.management.ManagementFactory;
 import java.lang.management.RuntimeMXBean;
 import java.net.Inet4Address;
+import java.net.Inet6Address;
 import java.net.InetAddress;
 import java.net.NetworkInterface;
 import java.text.NumberFormat;
@@ -39,6 +40,7 @@ import java.util.zip.CRC32;
 import java.util.zip.DeflaterOutputStream;
 import java.util.zip.InflaterInputStream;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.validator.routines.InetAddressValidator;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.logging.InternalLogger;
 import org.apache.rocketmq.logging.InternalLoggerFactory;
@@ -438,6 +440,18 @@ public class UtilAll {
         return false;
     }
 
+    public static boolean isInternalV6IP(byte[] ip) {
+        if (ip.length != 16) {
+            throw new RuntimeException("illegal ipv6 bytes");
+        }
+
+        //FEC0:0000:0000:0000:0000:0000:0000:0000/10
+        if (ip[0] == (byte) 254 && ip[1] >= (byte) 192) {
+            return true;
+        }
+        return false;
+    }
+
     private static boolean ipCheck(byte[] ip) {
         if (ip.length != 4) {
             throw new RuntimeException("illegal ipv4 bytes");
@@ -474,6 +488,15 @@ public class UtilAll {
         return false;
     }
 
+    private static boolean ipV6Check(byte[] ip) {
+        if (ip.length != 16) {
+            throw new RuntimeException("illegal ipv6 bytes");
+        }
+
+        InetAddressValidator validator = InetAddressValidator.getInstance();
+        return validator.isValidInet6Address(ipToIPv6Str(ip));
+    }
+
     public static String ipToIPv4Str(byte[] ip) {
         if (ip.length != 4) {
             return null;
@@ -483,6 +506,25 @@ public class UtilAll {
             .append(".").append(ip[3] & 0xFF).toString();
     }
 
+    public static String ipToIPv6Str(byte[] ip) {
+        if (ip.length != 16) {
+            return null;
+        }
+
+        StringBuilder sb = new StringBuilder();
+        for (int i = 0; i < ip.length; i++) {
+            String hex = Integer.toHexString(ip[i] & 0xFF);
+            if (hex.length() < 2) {
+                sb.append(0);
+            }
+            sb.append(hex);
+            if (i % 2 == 1 && i < ip.length - 1) {
+                sb.append(":");
+            }
+        }
+        return sb.toString();
+    }
+
     public static byte[] getIP() {
         try {
             Enumeration allNetInterfaces = NetworkInterface.getNetworkInterfaces();
@@ -504,6 +546,17 @@ public class UtilAll {
                                 }
                             }
                         }
+                    } else if (ip != null && ip instanceof Inet6Address) {
+                        byte[] ipByte = ip.getAddress();
+                        if (ipByte.length == 16) {
+                            if (ipV6Check(ipByte)) {
+                                if (!isInternalV6IP(ipByte)) {
+                                    return ipByte;
+                                } else if (internalIP == null) {
+                                    internalIP = ipByte;
+                                }
+                            }
+                        }
                     }
                 }
             }
@@ -532,12 +585,12 @@ public class UtilAll {
         }
     }
 
-    public static String List2String(List<String> list,String splitor) {
+    public static String List2String(List<String> list, String splitor) {
         if (list == null || list.size() == 0) {
             return null;
         }
         StringBuffer str = new StringBuffer();
-        for (int i = 0;i < list.size();i++) {
+        for (int i = 0; i < list.size(); i++) {
             str.append(list.get(i));
             if (i == list.size() - 1) {
                 continue;
@@ -547,7 +600,7 @@ public class UtilAll {
         return str.toString();
     }
 
-    public static List<String> String2List(String str,String splitor) {
+    public static List<String> String2List(String str, String splitor) {
         if (StringUtils.isEmpty(str)) {
             return null;
         }

common/src/main/java/org/apache/rocketmq/common/message/MessageClientIDSetter.java

@@ -31,17 +31,19 @@ public class MessageClientIDSetter {
     private static long nextStartTime;
 
     static {
-        LEN = 4 + 2 + 4 + 4 + 2;
-        ByteBuffer tempBuffer = ByteBuffer.allocate(10);
-        tempBuffer.position(2);
-        tempBuffer.putInt(UtilAll.getPid());
-        tempBuffer.position(0);
+        byte[] ip;
         try {
-            tempBuffer.put(UtilAll.getIP());
+            ip = UtilAll.getIP();
         } catch (Exception e) {
-            tempBuffer.put(createFakeIP());
+            ip = createFakeIP();
         }
-        tempBuffer.position(6);
+        LEN = ip.length + 2 + 4 + 4 + 2;
+        ByteBuffer tempBuffer = ByteBuffer.allocate(ip.length + 2 + 4);
+        tempBuffer.position(0);
+        tempBuffer.put(ip);
+        tempBuffer.position(ip.length);
+        tempBuffer.putInt(UtilAll.getPid());
+        tempBuffer.position(ip.length + 2);
         tempBuffer.putInt(MessageClientIDSetter.class.getClassLoader().hashCode());
         FIX_STRING = UtilAll.bytes2string(tempBuffer.array());
         setStartTime(System.currentTimeMillis());
@@ -64,11 +66,12 @@ public class MessageClientIDSetter {
     public static Date getNearlyTimeFromID(String msgID) {
         ByteBuffer buf = ByteBuffer.allocate(8);
         byte[] bytes = UtilAll.string2bytes(msgID);
+        int ipLength = bytes.length == 28 ? 16 : 4;
         buf.put((byte) 0);
         buf.put((byte) 0);
         buf.put((byte) 0);
         buf.put((byte) 0);
-        buf.put(bytes, 10, 4);
+        buf.put(bytes, ipLength + 2 + 4, 4);
         buf.position(0);
         long spanMS = buf.getLong();
         Calendar cal = Calendar.getInstance();
@@ -89,13 +92,18 @@ public class MessageClientIDSetter {
 
     public static String getIPStrFromID(String msgID) {
         byte[] ipBytes = getIPFromID(msgID);
-        return UtilAll.ipToIPv4Str(ipBytes);
+        if (ipBytes.length == 16) {
+            return UtilAll.ipToIPv6Str(ipBytes);
+        } else {
+            return UtilAll.ipToIPv4Str(ipBytes);
+        }
     }
 
     public static byte[] getIPFromID(String msgID) {
-        byte[] result = new byte[4];
         byte[] bytes = UtilAll.string2bytes(msgID);
-        System.arraycopy(bytes, 0, result, 0, 4);
+        int ipLength = bytes.length == 28 ? 16 : 4;
+        byte[] result = new byte[ipLength];
+        System.arraycopy(bytes, 0, result, 0, ipLength);
         return result;
     }
 

common/src/main/java/org/apache/rocketmq/common/message/MessageDecoder.java

@@ -16,9 +16,7 @@
  */
 package org.apache.rocketmq.common.message;
 
-import org.apache.rocketmq.common.UtilAll;
-import org.apache.rocketmq.common.sysflag.MessageSysFlag;
-
+import java.net.Inet4Address;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
@@ -29,37 +27,41 @@ import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import org.apache.rocketmq.common.UtilAll;
+import org.apache.rocketmq.common.sysflag.MessageSysFlag;
 
 public class MessageDecoder {
-    public final static int MSG_ID_LENGTH = 8 + 8;
+//    public final static int MSG_ID_LENGTH = 8 + 8;
 
     public final static Charset CHARSET_UTF8 = Charset.forName("UTF-8");
     public final static int MESSAGE_MAGIC_CODE_POSTION = 4;
     public final static int MESSAGE_FLAG_POSTION = 16;
     public final static int MESSAGE_PHYSIC_OFFSET_POSTION = 28;
-    public final static int MESSAGE_STORE_TIMESTAMP_POSTION = 56;
+    //    public final static int MESSAGE_STORE_TIMESTAMP_POSTION = 56;
     public final static int MESSAGE_MAGIC_CODE = -626843481;
     public static final char NAME_VALUE_SEPARATOR = 1;
     public static final char PROPERTY_SEPARATOR = 2;
-    public static final int PHY_POS_POSITION =  4 + 4 + 4 + 4 + 4 + 8;
-    public static final int BODY_SIZE_POSITION = 4 // 1 TOTALSIZE
-        + 4 // 2 MAGICCODE
-        + 4 // 3 BODYCRC
-        + 4 // 4 QUEUEID
-        + 4 // 5 FLAG
-        + 8 // 6 QUEUEOFFSET
-        + 8 // 7 PHYSICALOFFSET
-        + 4 // 8 SYSFLAG
-        + 8 // 9 BORNTIMESTAMP
-        + 8 // 10 BORNHOST
-        + 8 // 11 STORETIMESTAMP
-        + 8 // 12 STOREHOSTADDRESS
-        + 4 // 13 RECONSUMETIMES
-        + 8; // 14 Prepared Transaction Offset
+    public static final int PHY_POS_POSITION = 4 + 4 + 4 + 4 + 4 + 8;
+    public static final int SYSFLAG_POSITION = 4 + 4 + 4 + 4 + 4 + 8 + 8;
+//    public static final int BODY_SIZE_POSITION = 4 // 1 TOTALSIZE
+//        + 4 // 2 MAGICCODE
+//        + 4 // 3 BODYCRC
+//        + 4 // 4 QUEUEID
+//        + 4 // 5 FLAG
+//        + 8 // 6 QUEUEOFFSET
+//        + 8 // 7 PHYSICALOFFSET
+//        + 4 // 8 SYSFLAG
+//        + 8 // 9 BORNTIMESTAMP
+//        + 8 // 10 BORNHOST
+//        + 8 // 11 STORETIMESTAMP
+//        + 8 // 12 STOREHOSTADDRESS
+//        + 4 // 13 RECONSUMETIMES
+//        + 8; // 14 Prepared Transaction Offset
 
     public static String createMessageId(final ByteBuffer input, final ByteBuffer addr, final long offset) {
         input.flip();
-        input.limit(MessageDecoder.MSG_ID_LENGTH);
+        int msgIDLength = addr.limit() == 8 ? 16 : 28;
+        input.limit(msgIDLength);
 
         input.put(addr);
         input.putLong(offset);
@@ -68,8 +70,9 @@ public class MessageDecoder {
     }
 
     public static String createMessageId(SocketAddress socketAddress, long transactionIdhashCode) {
-        ByteBuffer byteBuffer = ByteBuffer.allocate(MessageDecoder.MSG_ID_LENGTH);
         InetSocketAddress inetSocketAddress = (InetSocketAddress) socketAddress;
+        int msgIDLength = inetSocketAddress.getAddress() instanceof Inet4Address ? 16 : 28;
+        ByteBuffer byteBuffer = ByteBuffer.allocate(msgIDLength);
         byteBuffer.put(inetSocketAddress.getAddress().getAddress());
         byteBuffer.putInt(inetSocketAddress.getPort());
         byteBuffer.putLong(transactionIdhashCode);
@@ -80,15 +83,16 @@ public class MessageDecoder {
     public static MessageId decodeMessageId(final String msgId) throws UnknownHostException {
         SocketAddress address;
         long offset;
+        int ipLength = msgId.length() == 32 ? 4 * 2 : 16 * 2;
 
-        byte[] ip = UtilAll.string2bytes(msgId.substring(0, 8));
-        byte[] port = UtilAll.string2bytes(msgId.substring(8, 16));
+        byte[] ip = UtilAll.string2bytes(msgId.substring(0, ipLength));
+        byte[] port = UtilAll.string2bytes(msgId.substring(ipLength, ipLength + 8));
         ByteBuffer bb = ByteBuffer.wrap(port);
         int portInt = bb.getInt(0);
         address = new InetSocketAddress(InetAddress.getByAddress(ip), portInt);
 
         // offset
-        byte[] data = UtilAll.string2bytes(msgId.substring(16, 32));
+        byte[] data = UtilAll.string2bytes(msgId.substring(ipLength + 8, ipLength + 8 + 16));
         bb = ByteBuffer.wrap(data);
         offset = bb.getLong(0);
 
@@ -101,7 +105,24 @@ public class MessageDecoder {
      * @param byteBuffer msg commit log buffer.
      */
     public static Map<String, String> decodeProperties(java.nio.ByteBuffer byteBuffer) {
-        int topicLengthPosition = BODY_SIZE_POSITION + 4 + byteBuffer.getInt(BODY_SIZE_POSITION);
+        int sysFlag = byteBuffer.getInt(SYSFLAG_POSITION);
+        int bornhostLength = (sysFlag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 8 : 20;
+        int storehostAddressLength = (sysFlag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 8 : 20;
+        int bodySizePosition = 4 // 1 TOTALSIZE
+            + 4 // 2 MAGICCODE
+            + 4 // 3 BODYCRC
+            + 4 // 4 QUEUEID
+            + 4 // 5 FLAG
+            + 8 // 6 QUEUEOFFSET
+            + 8 // 7 PHYSICALOFFSET
+            + 4 // 8 SYSFLAG
+            + 8 // 9 BORNTIMESTAMP
+            + bornhostLength // 10 BORNHOST
+            + 8 // 11 STORETIMESTAMP
+            + storehostAddressLength // 12 STOREHOSTADDRESS
+            + 4 // 13 RECONSUMETIMES
+            + 8; // 14 Prepared Transaction Offset
+        int topicLengthPosition = bodySizePosition + 4 + byteBuffer.getInt(bodySizePosition);
 
         byte topicLength = byteBuffer.get(topicLengthPosition);
 
@@ -139,6 +160,8 @@ public class MessageDecoder {
         byte[] propertiesBytes = properties.getBytes(CHARSET_UTF8);
         short propertiesLength = (short) propertiesBytes.length;
         int sysFlag = messageExt.getSysFlag();
+        int bornhostLength = (sysFlag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 8 : 20;
+        int storehostAddressLength = (sysFlag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 8 : 20;
         byte[] newBody = messageExt.getBody();
         if (needCompress && (sysFlag & MessageSysFlag.COMPRESSED_FLAG) == MessageSysFlag.COMPRESSED_FLAG) {
             newBody = UtilAll.compress(body, 5);
@@ -158,9 +181,9 @@ public class MessageDecoder {
                 + 8 // 7 PHYSICALOFFSET
                 + 4 // 8 SYSFLAG
                 + 8 // 9 BORNTIMESTAMP
-                + 8 // 10 BORNHOST
+                + bornhostLength // 10 BORNHOST
                 + 8 // 11 STORETIMESTAMP
-                + 8 // 12 STOREHOSTADDRESS
+                + storehostAddressLength // 12 STOREHOSTADDRESS
                 + 4 // 13 RECONSUMETIMES
                 + 8 // 14 Prepared Transaction Offset
                 + 4 + bodyLength // 14 BODY
@@ -291,8 +314,9 @@ public class MessageDecoder {
             msgExt.setBornTimestamp(bornTimeStamp);
 
             // 10 BORNHOST
-            byte[] bornHost = new byte[4];
-            byteBuffer.get(bornHost, 0, 4);
+            int bornhostIPLength = (sysFlag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 4 : 16;
+            byte[] bornHost = new byte[bornhostIPLength];
+            byteBuffer.get(bornHost, 0, bornhostIPLength);
             int port = byteBuffer.getInt();
             msgExt.setBornHost(new InetSocketAddress(InetAddress.getByAddress(bornHost), port));
 
@@ -301,8 +325,9 @@ public class MessageDecoder {
             msgExt.setStoreTimestamp(storeTimestamp);
 
             // 12 STOREHOST
-            byte[] storeHost = new byte[4];
-            byteBuffer.get(storeHost, 0, 4);
+            int storehostIPLength = (sysFlag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 : 16;
+            byte[] storeHost = new byte[storehostIPLength];
+            byteBuffer.get(storeHost, 0, storehostIPLength);
             port = byteBuffer.getInt();
             msgExt.setStoreHost(new InetSocketAddress(InetAddress.getByAddress(storeHost), port));
 
@@ -348,7 +373,8 @@ public class MessageDecoder {
                 msgExt.setProperties(map);
             }
 
-            ByteBuffer byteBufferMsgId = ByteBuffer.allocate(MSG_ID_LENGTH);
+            int msgIDLength = storehostIPLength + 4 + 8;
+            ByteBuffer byteBufferMsgId = ByteBuffer.allocate(msgIDLength);
             String msgId = createMessageId(byteBufferMsgId, msgExt.getStoreHostBytes(), msgExt.getCommitLogOffset());
             msgExt.setMsgId(msgId);
 

common/src/main/java/org/apache/rocketmq/common/message/MessageExt.java

@@ -16,6 +16,8 @@
  */
 package org.apache.rocketmq.common.message;
 
+import java.net.Inet4Address;
+import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.nio.ByteBuffer;
@@ -66,14 +68,26 @@ public class MessageExt extends Message {
 
     public static ByteBuffer socketAddress2ByteBuffer(final SocketAddress socketAddress, final ByteBuffer byteBuffer) {
         InetSocketAddress inetSocketAddress = (InetSocketAddress) socketAddress;
-        byteBuffer.put(inetSocketAddress.getAddress().getAddress(), 0, 4);
+        InetAddress address = inetSocketAddress.getAddress();
+        if (address instanceof Inet4Address) {
+            byteBuffer.put(inetSocketAddress.getAddress().getAddress(), 0, 4);
+        } else {
+            byteBuffer.put(inetSocketAddress.getAddress().getAddress(), 0, 16);
+        }
         byteBuffer.putInt(inetSocketAddress.getPort());
         byteBuffer.flip();
         return byteBuffer;
     }
 
     public static ByteBuffer socketAddress2ByteBuffer(SocketAddress socketAddress) {
-        ByteBuffer byteBuffer = ByteBuffer.allocate(8);
+        InetSocketAddress inetSocketAddress = (InetSocketAddress) socketAddress;
+        InetAddress address = inetSocketAddress.getAddress();
+        ByteBuffer byteBuffer;
+        if (address instanceof Inet4Address) {
+            byteBuffer = ByteBuffer.allocate(4 + 4);
+        } else {
+            byteBuffer = ByteBuffer.allocate(16 + 4);
+        }
         return socketAddress2ByteBuffer(socketAddress, byteBuffer);
     }
 
@@ -167,6 +181,10 @@ public class MessageExt extends Message {
         this.sysFlag = sysFlag;
     }
 
+    public void setStoreHostAddressV6Flag() { this.sysFlag = this.sysFlag | MessageSysFlag.STOREHOSTADDRESS_V6_FLAG; }
+
+    public void setBornHostV6Flag() { this.sysFlag = this.sysFlag | MessageSysFlag.BORNHOST_V6_FLAG; }
+
     public int getBodyCRC() {
         return bodyCRC;
     }

common/src/main/java/org/apache/rocketmq/common/sysflag/MessageSysFlag.java

@@ -23,6 +23,8 @@ public class MessageSysFlag {
     public final static int TRANSACTION_PREPARED_TYPE = 0x1 << 2;
     public final static int TRANSACTION_COMMIT_TYPE = 0x2 << 2;
     public final static int TRANSACTION_ROLLBACK_TYPE = 0x3 << 2;
+    public final static int BORNHOST_V6_FLAG = 0x1 << 4;
+    public final static int STOREHOSTADDRESS_V6_FLAG = 0x1 << 5;
 
     public static int getTransactionValue(final int flag) {
         return flag & TRANSACTION_ROLLBACK_TYPE;
@@ -35,4 +37,5 @@ public class MessageSysFlag {
     public static int clearCompressedFlag(final int flag) {
         return flag & (~COMPRESSED_FLAG);
     }
+
 }

common/src/test/java/org/apache/rocketmq/common/UtilAllTest.java

@@ -98,6 +98,15 @@ public class UtilAllTest {
         assertThat(UtilAll.isBlank("Hello")).isFalse();
     }
 
+    @Test
+    public void testIPv6Check() {
+        byte[] nonInternalIp = UtilAll.string2bytes("24084004018081003FAA1DDE2B3F898A");
+        byte[] internalIp = UtilAll.string2bytes("FEC0000000000000000000000000FFFF");
+        assertThat(UtilAll.isInternalV6IP(nonInternalIp)).isFalse();
+        assertThat(UtilAll.isInternalV6IP(internalIp)).isTrue();
+        assertThat(UtilAll.ipToIPv6Str(nonInternalIp).toUpperCase()).isEqualTo("2408:4004:0180:8100:3FAA:1DDE:2B3F:898A");
+    }
+
     static class DemoConfig {
         private int demoWidth = 0;
         private int demoLength = 0;

common/src/test/java/org/apache/rocketmq/common/message/MessageClientIDSetterTest.java

+/*
+ * 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.common.message;
+
+import java.util.Calendar;
+import java.util.Date;
+import org.junit.Test;
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class MessageClientIDSetterTest {
+
+    @Test
+    public void testGetIPStrFromID() {
+        String ipv4HostMsgId = "C0A803CA00002A9F0000000000031367";
+        String ipv6HostMsgId = "24084004018081003FAA1DDE2B3F898A00002A9F0000000000000CA0";
+        String v4Ip = "192.168.3.202";
+        String v6Ip = "2408:4004:0180:8100:3faa:1dde:2b3f:898a";
+        assertThat(MessageClientIDSetter.getIPStrFromID(ipv4HostMsgId)).isEqualTo(v4Ip);
+        assertThat(MessageClientIDSetter.getIPStrFromID(ipv6HostMsgId)).isEqualTo(v6Ip);
+    }
+
+    @Test
+    public void testGetNearlyTimeFromID() {
+        String ipv4HostMsgId = "C0A803CA00002A9F0000000000031367";
+        String ipv6HostMsgId = "24084004018081003FAA1DDE2B3F898A00002A9F0000000000000CA0";
+        Calendar cal = Calendar.getInstance();
+        cal.set(2019, Calendar.AUGUST, 1);
+        Date date = cal.getTime();
+        assertThat(MessageClientIDSetter.getNearlyTimeFromID(ipv4HostMsgId)).isCloseTo(date, 24*60*60*1000);
+        assertThat(MessageClientIDSetter.getNearlyTimeFromID(ipv6HostMsgId)).isCloseTo(date, 24*60*60*1000);
+    }
+}

common/src/test/java/org/apache/rocketmq/common/message/MessageDecoderTest.java

@@ -25,6 +25,7 @@ import java.net.UnknownHostException;
 import java.nio.ByteBuffer;
 import java.util.Map;
 
+import static org.apache.rocketmq.common.message.MessageDecoder.createMessageId;
 import static org.assertj.core.api.Assertions.assertThat;
 
 public class MessageDecoderTest {
@@ -77,4 +78,172 @@ public class MessageDecoderTest {
         assertThat("hello").isEqualTo(properties.get("b"));
         assertThat("3.14").isEqualTo(properties.get("c"));
     }
+
+    @Test
+    public void testDecodePropertiesOnIPv6Host() {
+        MessageExt messageExt = new MessageExt();
+
+        messageExt.setMsgId("24084004018081003FAA1DDE2B3F898A00002A9F0000000000000CA0");
+        messageExt.setBornHostV6Flag();
+        messageExt.setStoreHostAddressV6Flag();
+        messageExt.setTopic("abc");
+        messageExt.setBody("hello!q!".getBytes());
+        try {
+            messageExt.setBornHost(new InetSocketAddress(InetAddress.getByName("1050:0000:0000:0000:0005:0600:300c:326b"), 0));
+        } catch (UnknownHostException e) {
+            e.printStackTrace();
+            assertThat(Boolean.FALSE).isTrue();
+        }
+        messageExt.setBornTimestamp(System.currentTimeMillis());
+        messageExt.setCommitLogOffset(123456);
+        messageExt.setPreparedTransactionOffset(0);
+        messageExt.setQueueId(0);
+        messageExt.setQueueOffset(123);
+        messageExt.setReconsumeTimes(0);
+        try {
+            messageExt.setStoreHost(new InetSocketAddress(InetAddress.getByName("::1"), 0));
+        } catch (UnknownHostException e) {
+            e.printStackTrace();
+            assertThat(Boolean.FALSE).isTrue();
+        }
+
+        messageExt.putUserProperty("a", "123");
+        messageExt.putUserProperty("b", "hello");
+        messageExt.putUserProperty("c", "3.14");
+
+        byte[] msgBytes = new byte[0];
+        try {
+            msgBytes = MessageDecoder.encode(messageExt, false);
+        } catch (Exception e) {
+            e.printStackTrace();
+            assertThat(Boolean.FALSE).isTrue();
+        }
+
+        ByteBuffer byteBuffer = ByteBuffer.allocate(msgBytes.length);
+        byteBuffer.put(msgBytes);
+
+        Map<String, String> properties = MessageDecoder.decodeProperties(byteBuffer);
+
+        assertThat(properties).isNotNull();
+        assertThat("123").isEqualTo(properties.get("a"));
+        assertThat("hello").isEqualTo(properties.get("b"));
+        assertThat("3.14").isEqualTo(properties.get("c"));
+    }
+
+    @Test
+    public void testEncodeAndDecode() {
+        MessageExt messageExt = new MessageExt();
+
+        messageExt.setMsgId("645100FA00002A9F000000489A3AA09E");
+        messageExt.setTopic("abc");
+        messageExt.setBody("hello!q!".getBytes());
+        try {
+            messageExt.setBornHost(new InetSocketAddress(InetAddress.getByName("127.0.0.1"), 0));
+        } catch (UnknownHostException e) {
+            e.printStackTrace();
+            assertThat(Boolean.FALSE).isTrue();
+        }
+        messageExt.setBornTimestamp(System.currentTimeMillis());
+        messageExt.setCommitLogOffset(123456);
+        messageExt.setPreparedTransactionOffset(0);
+        messageExt.setQueueId(1);
+        messageExt.setQueueOffset(123);
+        messageExt.setReconsumeTimes(0);
+        try {
+            messageExt.setStoreHost(new InetSocketAddress(InetAddress.getLocalHost(), 0));
+        } catch (UnknownHostException e) {
+            e.printStackTrace();
+            assertThat(Boolean.FALSE).isTrue();
+        }
+
+        messageExt.putUserProperty("a", "123");
+        messageExt.putUserProperty("b", "hello");
+        messageExt.putUserProperty("c", "3.14");
+
+        byte[] msgBytes = new byte[0];
+        try {
+            msgBytes = MessageDecoder.encode(messageExt, false);
+        } catch (Exception e) {
+            e.printStackTrace();
+            assertThat(Boolean.FALSE).isTrue();
+        }
+
+        ByteBuffer byteBuffer = ByteBuffer.allocate(msgBytes.length);
+        byteBuffer.put(msgBytes);
+
+        byteBuffer.clear();
+        MessageExt decodedMsg = MessageDecoder.decode(byteBuffer);
+
+        assertThat(decodedMsg).isNotNull();
+        assertThat(1).isEqualTo(decodedMsg.getQueueId());
+        assertThat(123456L).isEqualTo(decodedMsg.getCommitLogOffset());
+        assertThat("hello!q!".getBytes()).isEqualTo(decodedMsg.getBody());
+
+        int msgIDLength = 4 + 4 + 8;
+        ByteBuffer byteBufferMsgId = ByteBuffer.allocate(msgIDLength);
+        String msgId = createMessageId(byteBufferMsgId, messageExt.getStoreHostBytes(), messageExt.getCommitLogOffset());
+        assertThat(msgId).isEqualTo(decodedMsg.getMsgId());
+
+        assertThat("abc").isEqualTo(decodedMsg.getTopic());
+    }
+
+    @Test
+    public void testEncodeAndDecodeOnIPv6Host() {
+        MessageExt messageExt = new MessageExt();
+
+        messageExt.setMsgId("24084004018081003FAA1DDE2B3F898A00002A9F0000000000000CA0");
+        messageExt.setBornHostV6Flag();
+        messageExt.setStoreHostAddressV6Flag();
+        messageExt.setTopic("abc");
+        messageExt.setBody("hello!q!".getBytes());
+        try {
+            messageExt.setBornHost(new InetSocketAddress(InetAddress.getByName("1050:0000:0000:0000:0005:0600:300c:326b"), 0));
+        } catch (UnknownHostException e) {
+            e.printStackTrace();
+            assertThat(Boolean.FALSE).isTrue();
+        }
+        messageExt.setBornTimestamp(System.currentTimeMillis());
+        messageExt.setCommitLogOffset(123456);
+        messageExt.setPreparedTransactionOffset(0);
+        messageExt.setQueueId(1);
+        messageExt.setQueueOffset(123);
+        messageExt.setReconsumeTimes(0);
+        try {
+            messageExt.setStoreHost(new InetSocketAddress(InetAddress.getByName("::1"), 0));
+        } catch (UnknownHostException e) {
+            e.printStackTrace();
+            assertThat(Boolean.FALSE).isTrue();
+        }
+
+        messageExt.putUserProperty("a", "123");
+        messageExt.putUserProperty("b", "hello");
+        messageExt.putUserProperty("c", "3.14");
+
+        byte[] msgBytes = new byte[0];
+        try {
+            msgBytes = MessageDecoder.encode(messageExt, false);
+        } catch (Exception e) {
+            e.printStackTrace();
+            assertThat(Boolean.FALSE).isTrue();
+        }
+
+        ByteBuffer byteBuffer = ByteBuffer.allocate(msgBytes.length);
+        byteBuffer.put(msgBytes);
+
+        byteBuffer.clear();
+        MessageExt decodedMsg = MessageDecoder.decode(byteBuffer);
+
+        assertThat(decodedMsg).isNotNull();
+        assertThat(1).isEqualTo(decodedMsg.getQueueId());
+        assertThat(123456L).isEqualTo(decodedMsg.getCommitLogOffset());
+        assertThat("hello!q!".getBytes()).isEqualTo(decodedMsg.getBody());
+        assertThat(48).isEqualTo(decodedMsg.getSysFlag());
+
+        int msgIDLength = 16 + 4 + 8;
+        ByteBuffer byteBufferMsgId = ByteBuffer.allocate(msgIDLength);
+        String msgId = createMessageId(byteBufferMsgId, messageExt.getStoreHostBytes(), messageExt.getCommitLogOffset());
+        assertThat(msgId).isEqualTo(decodedMsg.getMsgId());
+
+        assertThat("abc").isEqualTo(decodedMsg.getTopic());
+    }
 }

pom.xml

@@ -16,7 +16,8 @@
   limitations under the License.
   -->
 
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
 
     <parent>
         <groupId>org.apache</groupId>
@@ -159,7 +160,7 @@
                 </executions>
                 <configuration>
                     <rules>
-                        <banCircularDependencies />
+                        <banCircularDependencies/>
                     </rules>
                     <fail>true</fail>
                 </configuration>
@@ -620,6 +621,13 @@
                 <artifactId>log4j-slf4j-impl</artifactId>
                 <version>2.7</version>
             </dependency>
+            <dependency>
+                <groupId>commons-validator</groupId>
+                <artifactId>commons-validator</artifactId>
+                <version>1.6</version>
+            </dependency>
+
+
         </dependencies>
     </dependencyManagement>
 </project>

remoting/src/main/java/org/apache/rocketmq/remoting/common/RemotingHelper.java

@@ -17,18 +17,17 @@
 package org.apache.rocketmq.remoting.common;
 
 import io.netty.channel.Channel;
-import org.apache.rocketmq.remoting.exception.RemotingConnectException;
-import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
-import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
-import org.apache.rocketmq.logging.InternalLogger;
-import org.apache.rocketmq.logging.InternalLoggerFactory;
-import org.apache.rocketmq.remoting.protocol.RemotingCommand;
-
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.nio.ByteBuffer;
 import java.nio.channels.SocketChannel;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
+import org.apache.rocketmq.remoting.exception.RemotingConnectException;
+import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
+import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 
 public class RemotingHelper {
     public static final String ROCKETMQ_REMOTING = "RocketmqRemoting";
@@ -53,8 +52,10 @@ public class RemotingHelper {
     }
 
     public static SocketAddress string2SocketAddress(final String addr) {
-        String[] s = addr.split(":");
-        InetSocketAddress isa = new InetSocketAddress(s[0], Integer.parseInt(s[1]));
+        int split = addr.lastIndexOf(":");
+        String host = addr.substring(0, split);
+        String port = addr.substring(split + 1);
+        InetSocketAddress isa = new InetSocketAddress(host, Integer.parseInt(port));
         return isa;
     }
 

remoting/src/main/java/org/apache/rocketmq/remoting/common/RemotingUtil.java

@@ -31,7 +31,6 @@ import java.nio.channels.SocketChannel;
 import java.nio.channels.spi.SelectorProvider;
 import java.util.ArrayList;
 import java.util.Enumeration;
-
 import org.apache.rocketmq.logging.InternalLogger;
 import org.apache.rocketmq.logging.InternalLoggerFactory;
 
@@ -145,8 +144,10 @@ public class RemotingUtil {
     }
 
     public static SocketAddress string2SocketAddress(final String addr) {
-        String[] s = addr.split(":");
-        InetSocketAddress isa = new InetSocketAddress(s[0], Integer.parseInt(s[1]));
+        int split = addr.lastIndexOf(":");
+        String host = addr.substring(0, split);
+        String port = addr.substring(split + 1);
+        InetSocketAddress isa = new InetSocketAddress(host, Integer.parseInt(port));
         return isa;
     }
 

store/src/main/java/org/apache/rocketmq/store/CommitLog.java

@@ -16,6 +16,8 @@
  */
 package org.apache.rocketmq.store;
 
+import java.net.Inet6Address;
+import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -26,14 +28,14 @@ import java.util.concurrent.TimeUnit;
 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 org.apache.rocketmq.common.message.MessageAccessor;
 import org.apache.rocketmq.common.message.MessageConst;
 import org.apache.rocketmq.common.message.MessageDecoder;
 import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.common.message.MessageExtBatch;
 import org.apache.rocketmq.common.sysflag.MessageSysFlag;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
 import org.apache.rocketmq.store.config.BrokerRole;
 import org.apache.rocketmq.store.config.FlushDiskType;
 import org.apache.rocketmq.store.ha.HAService;
@@ -270,11 +272,21 @@ public class CommitLog {
 
             long bornTimeStamp = byteBuffer.getLong();
 
-            ByteBuffer byteBuffer1 = byteBuffer.get(bytesContent, 0, 8);
+            ByteBuffer byteBuffer1;
+            if ((sysFlag & MessageSysFlag.BORNHOST_V6_FLAG) == 0) {
+                byteBuffer1 = byteBuffer.get(bytesContent, 0, 4 + 4);
+            } else {
+                byteBuffer1 = byteBuffer.get(bytesContent, 0, 16 + 4);
+            }
 
             long storeTimestamp = byteBuffer.getLong();
 
-            ByteBuffer byteBuffer2 = byteBuffer.get(bytesContent, 0, 8);
+            ByteBuffer byteBuffer2;
+            if ((sysFlag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0) {
+                byteBuffer2 = byteBuffer.get(bytesContent, 0, 4 + 4);
+            } else {
+                byteBuffer2 = byteBuffer.get(bytesContent, 0, 16 + 4);
+            }
 
             int reconsumeTimes = byteBuffer.getInt();
 
@@ -339,7 +351,7 @@ public class CommitLog {
                 }
             }
 
-            int readLength = calMsgLength(bodyLen, topicLen, propertiesLength);
+            int readLength = calMsgLength(sysFlag, bodyLen, topicLen, propertiesLength);
             if (totalSize != readLength) {
                 doNothingForDeadCode(reconsumeTimes);
                 doNothingForDeadCode(flag);
@@ -372,7 +384,9 @@ public class CommitLog {
         return new DispatchRequest(-1, false /* success */);
     }
 
-    protected static int calMsgLength(int bodyLength, int topicLength, int propertiesLength) {
+    protected static int calMsgLength(int sysFlag, int bodyLength, int topicLength, int propertiesLength) {
+        int bornhostLength = (sysFlag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 8 : 20;
+        int storehostAddressLength = (sysFlag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 8 : 20;
         final int msgLen = 4 //TOTALSIZE
             + 4 //MAGICCODE
             + 4 //BODYCRC
@@ -382,9 +396,9 @@ public class CommitLog {
             + 8 //PHYSICALOFFSET
             + 4 //SYSFLAG
             + 8 //BORNTIMESTAMP
-            + 8 //BORNHOST
+            + bornhostLength //BORNHOST
             + 8 //STORETIMESTAMP
-            + 8 //STOREHOSTADDRESS
+            + storehostAddressLength //STOREHOSTADDRESS
             + 4 //RECONSUMETIMES
             + 8 //Prepared Transaction Offset
             + 4 + (bodyLength > 0 ? bodyLength : 0) //BODY
@@ -496,7 +510,10 @@ public class CommitLog {
             return false;
         }
 
-        long storeTimestamp = byteBuffer.getLong(MessageDecoder.MESSAGE_STORE_TIMESTAMP_POSTION);
+        int sysFlag = byteBuffer.getInt(MessageDecoder.SYSFLAG_POSITION);
+        int bornhostLength = (sysFlag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 8 : 20;
+        int msgStoreTimePos = 4 + 4 + 4 + 4 + 4 + 8 + 8 + 4 + 8 + bornhostLength;
+        long storeTimestamp = byteBuffer.getLong(msgStoreTimePos);
         if (0 == storeTimestamp) {
             return false;
         }
@@ -569,7 +586,18 @@ public class CommitLog {
             }
         }
 
-        long elapsedTimeInLock = 0;
+        InetSocketAddress bornSocketAddress = (InetSocketAddress) msg.getBornHost();
+        if (bornSocketAddress.getAddress() instanceof Inet6Address) {
+            msg.setBornHostV6Flag();
+        }
+
+        InetSocketAddress storeSocketAddress = (InetSocketAddress) msg.getStoreHost();
+        if (storeSocketAddress.getAddress() instanceof Inet6Address) {
+            msg.setStoreHostAddressV6Flag();
+        }
+
+        long eclipsedTimeInLock = 0;
+
         MappedFile unlockMappedFile = null;
         MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
 
@@ -619,14 +647,14 @@ public class CommitLog {
                     return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result);
             }
 
-            elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
+            eclipsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
             beginTimeInLock = 0;
         } finally {
             putMessageLock.unlock();
         }
 
-        if (elapsedTimeInLock > 500) {
-            log.warn("[NOTIFYME]putMessage in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", elapsedTimeInLock, msg.getBody().length, result);
+        if (eclipsedTimeInLock > 500) {
+            log.warn("[NOTIFYME]putMessage in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", eclipsedTimeInLock, msg.getBody().length, result);
         }
 
         if (null != unlockMappedFile && this.defaultMessageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {
@@ -714,7 +742,17 @@ public class CommitLog {
             return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, null);
         }
 
-        long elapsedTimeInLock = 0;
+        InetSocketAddress bornSocketAddress = (InetSocketAddress) messageExtBatch.getBornHost();
+        if (bornSocketAddress.getAddress() instanceof Inet6Address) {
+            messageExtBatch.setBornHostV6Flag();
+        }
+
+        InetSocketAddress storeSocketAddress = (InetSocketAddress) messageExtBatch.getStoreHost();
+        if (storeSocketAddress.getAddress() instanceof Inet6Address) {
+            messageExtBatch.setStoreHostAddressV6Flag();
+        }
+
+        long eclipsedTimeInLock = 0;
         MappedFile unlockMappedFile = null;
         MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
 
@@ -769,14 +807,14 @@ public class CommitLog {
                     return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result);
             }
 
-            elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
+            eclipsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
             beginTimeInLock = 0;
         } finally {
             putMessageLock.unlock();
         }
 
-        if (elapsedTimeInLock > 500) {
-            log.warn("[NOTIFYME]putMessages in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", elapsedTimeInLock, messageExtBatch.getBody().length, result);
+        if (eclipsedTimeInLock > 500) {
+            log.warn("[NOTIFYME]putMessages in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", eclipsedTimeInLock, messageExtBatch.getBody().length, result);
         }
 
         if (null != unlockMappedFile && this.defaultMessageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {
@@ -804,7 +842,10 @@ public class CommitLog {
             SelectMappedBufferResult result = this.getMessage(offset, size);
             if (null != result) {
                 try {
-                    return result.getByteBuffer().getLong(MessageDecoder.MESSAGE_STORE_TIMESTAMP_POSTION);
+                    int sysFlag = result.getByteBuffer().getInt(MessageDecoder.SYSFLAG_POSITION);
+                    int bornhostLength = (sysFlag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 8 : 20;
+                    int msgStoreTimePos = 4 + 4 + 4 + 4 + 4 + 8 + 8 + 4 + 8 + bornhostLength;
+                    return result.getByteBuffer().getLong(msgStoreTimePos);
                 } finally {
                     result.release();
                 }
@@ -1170,6 +1211,7 @@ public class CommitLog {
         // File at the end of the minimum fixed length empty
         private static final int END_FILE_MIN_BLANK_LENGTH = 4 + 4;
         private final ByteBuffer msgIdMemory;
+        private final ByteBuffer msgIdV6Memory;
         // Store the message content
         private final ByteBuffer msgStoreItemMemory;
         // The maximum length of the message
@@ -1179,10 +1221,9 @@ public class CommitLog {
 
         private final StringBuilder msgIdBuilder = new StringBuilder();
 
-        private final ByteBuffer hostHolder = ByteBuffer.allocate(8);
-
         DefaultAppendMessageCallback(final int size) {
-            this.msgIdMemory = ByteBuffer.allocate(MessageDecoder.MSG_ID_LENGTH);
+            this.msgIdMemory = ByteBuffer.allocate(4 + 4 + 8);
+            this.msgIdV6Memory = ByteBuffer.allocate(16 + 4 + 8);
             this.msgStoreItemMemory = ByteBuffer.allocate(size + END_FILE_MIN_BLANK_LENGTH);
             this.maxMessageSize = size;
         }
@@ -1198,8 +1239,20 @@ public class CommitLog {
             // PHY OFFSET
             long wroteOffset = fileFromOffset + byteBuffer.position();
 
-            this.resetByteBuffer(hostHolder, 8);
-            String msgId = MessageDecoder.createMessageId(this.msgIdMemory, msgInner.getStoreHostBytes(hostHolder), wroteOffset);
+            int sysflag = msgInner.getSysFlag();
+
+            int bornHostLength = (sysflag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;
+            int storeHostLength = (sysflag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;
+            ByteBuffer bornHostHolder = ByteBuffer.allocate(bornHostLength);
+            ByteBuffer storeHostHolder = ByteBuffer.allocate(storeHostLength);
+
+            this.resetByteBuffer(storeHostHolder, storeHostLength);
+            String msgId;
+            if ((sysflag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0) {
+                msgId = MessageDecoder.createMessageId(this.msgIdMemory, msgInner.getStoreHostBytes(storeHostHolder), wroteOffset);
+            } else {
+                msgId = MessageDecoder.createMessageId(this.msgIdV6Memory, msgInner.getStoreHostBytes(storeHostHolder), wroteOffset);
+            }
 
             // Record ConsumeQueue information
             keyBuilder.setLength(0);
@@ -1246,7 +1299,7 @@ public class CommitLog {
 
             final int bodyLength = msgInner.getBody() == null ? 0 : msgInner.getBody().length;
 
-            final int msgLen = calMsgLength(bodyLength, topicLength, propertiesLength);
+            final int msgLen = calMsgLength(msgInner.getSysFlag(), bodyLength, topicLength, propertiesLength);
 
             // Exceeds the maximum message
             if (msgLen > this.maxMessageSize) {
@@ -1291,14 +1344,13 @@ public class CommitLog {
             // 9 BORNTIMESTAMP
             this.msgStoreItemMemory.putLong(msgInner.getBornTimestamp());
             // 10 BORNHOST
-            this.resetByteBuffer(hostHolder, 8);
-            this.msgStoreItemMemory.put(msgInner.getBornHostBytes(hostHolder));
+            this.resetByteBuffer(bornHostHolder, bornHostLength);
+            this.msgStoreItemMemory.put(msgInner.getBornHostBytes(bornHostHolder));
             // 11 STORETIMESTAMP
             this.msgStoreItemMemory.putLong(msgInner.getStoreTimestamp());
             // 12 STOREHOSTADDRESS
-            this.resetByteBuffer(hostHolder, 8);
-            this.msgStoreItemMemory.put(msgInner.getStoreHostBytes(hostHolder));
-            //this.msgBatchMemory.put(msgInner.getStoreHostBytes());
+            this.resetByteBuffer(storeHostHolder, storeHostLength);
+            this.msgStoreItemMemory.put(msgInner.getStoreHostBytes(storeHostHolder));
             // 13 RECONSUMETIMES
             this.msgStoreItemMemory.putInt(msgInner.getReconsumeTimes());
             // 14 Prepared Transaction Offset
@@ -1359,8 +1411,13 @@ public class CommitLog {
             msgIdBuilder.setLength(0);
             final long beginTimeMills = CommitLog.this.defaultMessageStore.now();
             ByteBuffer messagesByteBuff = messageExtBatch.getEncodedBuff();
-            this.resetByteBuffer(hostHolder, 8);
-            ByteBuffer storeHostBytes = messageExtBatch.getStoreHostBytes(hostHolder);
+
+            int sysFlag = messageExtBatch.getSysFlag();
+            int storeHostLength = (sysFlag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;
+            ByteBuffer storeHostHolder = ByteBuffer.allocate(storeHostLength);
+
+            this.resetByteBuffer(storeHostHolder, storeHostLength);
+            ByteBuffer storeHostBytes = messageExtBatch.getStoreHostBytes(storeHostHolder);
             messagesByteBuff.mark();
             while (messagesByteBuff.hasRemaining()) {
                 // 1 TOTALSIZE
@@ -1396,7 +1453,13 @@ public class CommitLog {
                 messagesByteBuff.putLong(wroteOffset + totalMsgLen - msgLen);
 
                 storeHostBytes.rewind();
-                String msgId = MessageDecoder.createMessageId(this.msgIdMemory, storeHostBytes, wroteOffset + totalMsgLen - msgLen);
+                String msgId;
+                if ((sysFlag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0) {
+                    msgId = MessageDecoder.createMessageId(this.msgIdMemory, storeHostBytes, wroteOffset + totalMsgLen - msgLen);
+                } else {
+                    msgId = MessageDecoder.createMessageId(this.msgIdV6Memory, storeHostBytes, wroteOffset + totalMsgLen - msgLen);
+                }
+
                 if (msgIdBuilder.length() > 0) {
                     msgIdBuilder.append(',').append(msgId);
                 } else {
@@ -1432,8 +1495,6 @@ public class CommitLog {
         // The maximum length of the message
         private final int maxMessageSize;
 
-        private final ByteBuffer hostHolder = ByteBuffer.allocate(8);
-
         MessageExtBatchEncoder(final int size) {
             this.msgBatchMemory = ByteBuffer.allocateDirect(size);
             this.maxMessageSize = size;
@@ -1443,6 +1504,13 @@ public class CommitLog {
             msgBatchMemory.clear(); //not thread-safe
             int totalMsgLen = 0;
             ByteBuffer messagesByteBuff = messageExtBatch.wrap();
+
+            int sysFlag = messageExtBatch.getSysFlag();
+            int bornHostLength = (sysFlag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;
+            int storeHostLength = (sysFlag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;
+            ByteBuffer bornHostHolder = ByteBuffer.allocate(bornHostLength);
+            ByteBuffer storeHostHolder = ByteBuffer.allocate(storeHostLength);
+
             while (messagesByteBuff.hasRemaining()) {
                 // 1 TOTALSIZE
                 messagesByteBuff.getInt();
@@ -1466,7 +1534,7 @@ public class CommitLog {
 
                 final int topicLength = topicData.length;
 
-                final int msgLen = calMsgLength(bodyLen, topicLength, propertiesLen);
+                final int msgLen = calMsgLength(messageExtBatch.getSysFlag(), bodyLen, topicLength, propertiesLen);
 
                 // Exceeds the maximum message
                 if (msgLen > this.maxMessageSize) {
@@ -1500,13 +1568,13 @@ public class CommitLog {
                 // 9 BORNTIMESTAMP
                 this.msgBatchMemory.putLong(messageExtBatch.getBornTimestamp());
                 // 10 BORNHOST
-                this.resetByteBuffer(hostHolder, 8);
-                this.msgBatchMemory.put(messageExtBatch.getBornHostBytes(hostHolder));
+                this.resetByteBuffer(bornHostHolder, bornHostLength);
+                this.msgBatchMemory.put(messageExtBatch.getBornHostBytes(bornHostHolder));
                 // 11 STORETIMESTAMP
                 this.msgBatchMemory.putLong(messageExtBatch.getStoreTimestamp());
                 // 12 STOREHOSTADDRESS
-                this.resetByteBuffer(hostHolder, 8);
-                this.msgBatchMemory.put(messageExtBatch.getStoreHostBytes(hostHolder));
+                this.resetByteBuffer(storeHostHolder, storeHostLength);
+                this.msgBatchMemory.put(messageExtBatch.getStoreHostBytes(storeHostHolder));
                 // 13 RECONSUMETIMES
                 this.msgBatchMemory.putInt(messageExtBatch.getReconsumeTimes());
                 // 14 Prepared Transaction Offset, batch does not support transaction

store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java

@@ -19,6 +19,8 @@ package org.apache.rocketmq.store;
 import java.io.File;
 import java.io.IOException;
 import java.io.RandomAccessFile;
+import java.net.Inet6Address;
+import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.nio.ByteBuffer;
 import java.nio.channels.FileLock;
@@ -285,8 +287,6 @@ public class DefaultMessageStore implements MessageStore {
         this.shutdown = false;
     }
 
-
-
     public void shutdown() {
         if (!this.shutdown) {
             this.shutdown = true;
@@ -469,7 +469,7 @@ public class DefaultMessageStore implements MessageStore {
         long diff = this.systemClock.now() - begin;
 
         return diff < 10000000
-                && diff > this.messageStoreConfig.getOsPageCacheBusyTimeOutMills();
+            && diff > this.messageStoreConfig.getOsPageCacheBusyTimeOutMills();
     }
 
     @Override
@@ -1042,7 +1042,9 @@ public class DefaultMessageStore implements MessageStore {
                         int i = 0;
                         for (; i < bufferConsumeQueue.getSize(); i += ConsumeQueue.CQ_STORE_UNIT_SIZE) {
                             long offsetPy = bufferConsumeQueue.getByteBuffer().getLong();
-                            final ByteBuffer msgIdMemory = ByteBuffer.allocate(MessageDecoder.MSG_ID_LENGTH);
+                            InetSocketAddress inetSocketAddress = (InetSocketAddress) storeHost;
+                            int msgIdLength = (inetSocketAddress.getAddress() instanceof Inet6Address) ? 16 + 4 + 8 : 4 + 4 + 8;
+                            final ByteBuffer msgIdMemory = ByteBuffer.allocate(msgIdLength);
                             String msgId =
                                 MessageDecoder.createMessageId(msgIdMemory, MessageExt.socketAddress2ByteBuffer(storeHost), offsetPy);
                             messageIds.put(msgId, nextOffset++);

store/src/main/java/org/apache/rocketmq/store/dledger/DLedgerCommitLog.java

@@ -68,12 +68,11 @@ public class DLedgerCommitLog extends CommitLog {
     //This offset separate the old commitlog from dledger commitlog
     private long dividedCommitlogOffset = -1;
 
-
     private boolean isInrecoveringOldCommitlog = false;
 
     public DLedgerCommitLog(final DefaultMessageStore defaultMessageStore) {
         super(defaultMessageStore);
-        dLedgerConfig =  new DLedgerConfig();
+        dLedgerConfig = new DLedgerConfig();
         dLedgerConfig.setEnableDiskForceClean(defaultMessageStore.getMessageStoreConfig().isCleanFileForciblyEnable());
         dLedgerConfig.setStoreType(DLedgerConfig.FILE);
         dLedgerConfig.setSelfId(defaultMessageStore.getMessageStoreConfig().getdLegerSelfId());
@@ -158,8 +157,6 @@ public class DLedgerCommitLog extends CommitLog {
         log.warn("Should not set confirm offset {} for dleger commitlog", phyOffset);
     }
 
-
-
     @Override
     public long remainHowManyDataToCommit() {
         return dLedgerFileList.remainHowManyDataToCommit();
@@ -180,7 +177,7 @@ public class DLedgerCommitLog extends CommitLog {
         if (mappedFileQueue.getMappedFiles().isEmpty()) {
             refreshConfig();
             //To prevent too much log in defaultMessageStore
-            return  Integer.MAX_VALUE;
+            return Integer.MAX_VALUE;
         } else {
             disableDeleteDledger();
         }
@@ -201,7 +198,6 @@ public class DLedgerCommitLog extends CommitLog {
         return 1;
     }
 
-
     public SelectMappedBufferResult convertSbr(SelectMmapBufferResult sbr) {
         if (sbr == null) {
             return null;
@@ -232,7 +228,6 @@ public class DLedgerCommitLog extends CommitLog {
         return this.getData(offset, offset == 0);
     }
 
-
     @Override
     public SelectMappedBufferResult getData(final long offset, final boolean returnFirstOnNotFound) {
         if (offset < dividedCommitlogOffset) {
@@ -246,7 +241,7 @@ public class DLedgerCommitLog extends CommitLog {
         if (mappedFile != null) {
             int pos = (int) (offset % mappedFileSize);
             SelectMmapBufferResult sbr = mappedFile.selectMappedBuffer(pos);
-            return  convertSbr(truncate(sbr));
+            return convertSbr(truncate(sbr));
         }
 
         return null;
@@ -278,7 +273,7 @@ public class DLedgerCommitLog extends CommitLog {
         if (mappedFile == null) {
             return;
         }
-        ByteBuffer byteBuffer =  mappedFile.sliceByteBuffer();
+        ByteBuffer byteBuffer = mappedFile.sliceByteBuffer();
         byteBuffer.position(mappedFile.getWrotePosition());
         boolean needWriteMagicCode = true;
         // 1 TOTAL SIZE
@@ -311,7 +306,7 @@ public class DLedgerCommitLog extends CommitLog {
     }
 
     @Override
-    public void recoverAbnormally(long maxPhyOffsetOfConsumeQueue)  {
+    public void recoverAbnormally(long maxPhyOffsetOfConsumeQueue) {
         recover(maxPhyOffsetOfConsumeQueue);
     }
 
@@ -329,9 +324,9 @@ public class DLedgerCommitLog extends CommitLog {
         try {
             int bodyOffset = DLedgerEntry.BODY_OFFSET;
             int pos = byteBuffer.position();
-            int magic =  byteBuffer.getInt();
+            int magic = byteBuffer.getInt();
             //In dledger, this field is size, it must be gt 0, so it could prevent collision
-            int magicOld =  byteBuffer.getInt();
+            int magicOld = byteBuffer.getInt();
             if (magicOld == CommitLog.BLANK_MAGIC_CODE || magicOld == CommitLog.MESSAGE_MAGIC_CODE) {
                 byteBuffer.position(pos);
                 return super.checkMessageAndReturnSize(byteBuffer, checkCRC, readBody);
@@ -409,10 +404,10 @@ public class DLedgerCommitLog extends CommitLog {
         long elapsedTimeInLock;
         long queueOffset;
         try {
-            beginTimeInDledgerLock =  this.defaultMessageStore.getSystemClock().now();
+            beginTimeInDledgerLock = this.defaultMessageStore.getSystemClock().now();
             encodeResult = this.messageSerializer.serialize(msg);
             queueOffset = topicQueueTable.get(encodeResult.queueOffsetKey);
-            if (encodeResult.status  != AppendMessageStatus.PUT_OK) {
+            if (encodeResult.status != AppendMessageStatus.PUT_OK) {
                 return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, new AppendMessageResult(encodeResult.status));
             }
             AppendEntryRequest request = new AppendEntryRequest();
@@ -423,8 +418,11 @@ public class DLedgerCommitLog extends CommitLog {
             if (dledgerFuture.getPos() == -1) {
                 return new PutMessageResult(PutMessageStatus.OS_PAGECACHE_BUSY, new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR));
             }
-            long wroteOffset =  dledgerFuture.getPos() + DLedgerEntry.BODY_OFFSET;
-            ByteBuffer buffer = ByteBuffer.allocate(MessageDecoder.MSG_ID_LENGTH);
+            long wroteOffset = dledgerFuture.getPos() + DLedgerEntry.BODY_OFFSET;
+
+            int msgIdLength = (msg.getSysFlag() & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 + 4 + 8 : 16 + 4 + 8;
+            ByteBuffer buffer = ByteBuffer.allocate(msgIdLength);
+
             String msgId = MessageDecoder.createMessageId(buffer, msg.getStoreHostBytes(), wroteOffset);
             elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginTimeInDledgerLock;
             appendResult = new AppendMessageResult(AppendMessageStatus.PUT_OK, wroteOffset, encodeResult.data.length, msgId, System.currentTimeMillis(), queueOffset, elapsedTimeInLock);
@@ -491,8 +489,6 @@ public class DLedgerCommitLog extends CommitLog {
         return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, null);
     }
 
-
-
     @Override
     public SelectMappedBufferResult getMessage(final long offset, final int size) {
         if (offset < dividedCommitlogOffset) {
@@ -502,7 +498,7 @@ public class DLedgerCommitLog extends CommitLog {
         MmapFile mappedFile = this.dLedgerFileList.findMappedFileByOffset(offset, offset == 0);
         if (mappedFile != null) {
             int pos = (int) (offset % mappedFileSize);
-            return  convertSbr(mappedFile.selectMappedBuffer(pos, size));
+            return convertSbr(mappedFile.selectMappedBuffer(pos, size));
         }
         return null;
     }
@@ -559,6 +555,7 @@ public class DLedgerCommitLog extends CommitLog {
         private String queueOffsetKey;
         private byte[] data;
         private AppendMessageStatus status;
+
         public EncodeResult(AppendMessageStatus status, byte[] data, String queueOffsetKey) {
             this.data = data;
             this.status = status;
@@ -570,6 +567,7 @@ public class DLedgerCommitLog extends CommitLog {
         // File at the end of the minimum fixed length empty
         private static final int END_FILE_MIN_BLANK_LENGTH = 4 + 4;
         private final ByteBuffer msgIdMemory;
+        private final ByteBuffer msgIdV6Memory;
         // Store the message content
         private final ByteBuffer msgStoreItemMemory;
         // The maximum length of the message
@@ -579,10 +577,11 @@ public class DLedgerCommitLog extends CommitLog {
 
         private final StringBuilder msgIdBuilder = new StringBuilder();
 
-        private final ByteBuffer hostHolder = ByteBuffer.allocate(8);
+//        private final ByteBuffer hostHolder = ByteBuffer.allocate(8);
 
         MessageSerializer(final int size) {
-            this.msgIdMemory = ByteBuffer.allocate(MessageDecoder.MSG_ID_LENGTH);
+            this.msgIdMemory = ByteBuffer.allocate(4 + 4 + 8);
+            this.msgIdV6Memory = ByteBuffer.allocate(16 + 4 + 8);
             this.msgStoreItemMemory = ByteBuffer.allocate(size + END_FILE_MIN_BLANK_LENGTH);
             this.maxMessageSize = size;
         }
@@ -597,7 +596,13 @@ public class DLedgerCommitLog extends CommitLog {
             // PHY OFFSET
             long wroteOffset = 0;
 
-            this.resetByteBuffer(hostHolder, 8);
+            int sysflag = msgInner.getSysFlag();
+
+            int bornHostLength = (sysflag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;
+            int storeHostLength = (sysflag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;
+            ByteBuffer bornHostHolder = ByteBuffer.allocate(bornHostLength);
+            ByteBuffer storeHostHolder = ByteBuffer.allocate(storeHostLength);
+
             // Record ConsumeQueue information
             keyBuilder.setLength(0);
             keyBuilder.append(msgInner.getTopic());
@@ -644,7 +649,7 @@ public class DLedgerCommitLog extends CommitLog {
 
             final int bodyLength = msgInner.getBody() == null ? 0 : msgInner.getBody().length;
 
-            final int msgLen = calMsgLength(bodyLength, topicLength, propertiesLength);
+            final int msgLen = calMsgLength(msgInner.getSysFlag(), bodyLength, topicLength, propertiesLength);
 
             // Exceeds the maximum message
             if (msgLen > this.maxMessageSize) {
@@ -673,13 +678,13 @@ public class DLedgerCommitLog extends CommitLog {
             // 9 BORNTIMESTAMP
             this.msgStoreItemMemory.putLong(msgInner.getBornTimestamp());
             // 10 BORNHOST
-            this.resetByteBuffer(hostHolder, 8);
-            this.msgStoreItemMemory.put(msgInner.getBornHostBytes(hostHolder));
+            this.resetByteBuffer(bornHostHolder, bornHostLength);
+            this.msgStoreItemMemory.put(msgInner.getBornHostBytes(bornHostHolder));
             // 11 STORETIMESTAMP
             this.msgStoreItemMemory.putLong(msgInner.getStoreTimestamp());
             // 12 STOREHOSTADDRESS
-            this.resetByteBuffer(hostHolder, 8);
-            this.msgStoreItemMemory.put(msgInner.getStoreHostBytes(hostHolder));
+            this.resetByteBuffer(storeHostHolder, storeHostLength);
+            this.msgStoreItemMemory.put(msgInner.getStoreHostBytes(storeHostHolder));
             //this.msgBatchMemory.put(msgInner.getStoreHostBytes());
             // 13 RECONSUMETIMES
             this.msgStoreItemMemory.putInt(msgInner.getReconsumeTimes());
@@ -714,6 +719,7 @@ public class DLedgerCommitLog extends CommitLog {
     public static class DLedgerSelectMappedBufferResult extends SelectMappedBufferResult {
 
         private SelectMmapBufferResult sbr;
+
         public DLedgerSelectMappedBufferResult(SelectMmapBufferResult sbr) {
             super(sbr.getStartOffset(), sbr.getByteBuffer(), sbr.getSize(), null);
             this.sbr = sbr;

store/src/test/java/org/apache/rocketmq/store/AppendCallbackTest.java

@@ -97,6 +97,43 @@ public class AppendCallbackTest {
         assertTrue(result.getMsgId().length() > 0); //should have already constructed some message ids
     }
 
+    @Test
+    public void testAppendIPv6HostMessageBatchEndOfFile() throws Exception {
+        List<Message> messages = new ArrayList<>();
+        String topic = "test-topic";
+        int queue = 0;
+        for (int i = 0; i < 10; i++) {
+            Message msg = new Message();
+            msg.setBody("body".getBytes());
+            msg.setTopic(topic);
+            msg.setTags("abc");
+            messages.add(msg);
+        }
+        MessageExtBatch messageExtBatch = new MessageExtBatch();
+        messageExtBatch.setTopic(topic);
+        messageExtBatch.setQueueId(queue);
+        messageExtBatch.setBornTimestamp(System.currentTimeMillis());
+        messageExtBatch.setMsgId("24084004018081003FAA1DDE2B3F898A00002A9F0000000000000CA0");
+        messageExtBatch.setSysFlag(0);
+        messageExtBatch.setBornHostV6Flag();
+        messageExtBatch.setStoreHostAddressV6Flag();
+        messageExtBatch.setBornHost(new InetSocketAddress("1050:0000:0000:0000:0005:0600:300c:326b", 123));
+        messageExtBatch.setStoreHost(new InetSocketAddress("::1", 124));
+        messageExtBatch.setBody(MessageDecoder.encodeMessages(messages));
+
+        messageExtBatch.setEncodedBuff(batchEncoder.encode(messageExtBatch));
+        ByteBuffer buff = ByteBuffer.allocate(1024 * 10);
+        //encounter end of file when append half of the data
+        AppendMessageResult result = callback.doAppend(0, buff, 1000, messageExtBatch);
+        assertEquals(AppendMessageStatus.END_OF_FILE, result.getStatus());
+        assertEquals(0, result.getWroteOffset());
+        assertEquals(0, result.getLogicsOffset());
+        assertEquals(1000, result.getWroteBytes());
+        assertEquals(8, buff.position()); //write blank size and magic value
+
+        assertTrue(result.getMsgId().length() > 0); //should have already constructed some message ids
+    }
+
     @Test
     public void testAppendMessageBatchSucc() throws Exception {
         List<Message> messages = new ArrayList<>();
@@ -153,4 +190,64 @@ public class AppendCallbackTest {
 
     }
 
+    @Test
+    public void testAppendIPv6HostMessageBatchSucc() throws Exception {
+        List<Message> messages = new ArrayList<>();
+        String topic = "test-topic";
+        int queue = 0;
+        for (int i = 0; i < 10; i++) {
+            Message msg = new Message();
+            msg.setBody("body".getBytes());
+            msg.setTopic(topic);
+            msg.setTags("abc");
+            messages.add(msg);
+        }
+        MessageExtBatch messageExtBatch = new MessageExtBatch();
+        messageExtBatch.setTopic(topic);
+        messageExtBatch.setQueueId(queue);
+        messageExtBatch.setBornTimestamp(System.currentTimeMillis());
+        messageExtBatch.setMsgId("24084004018081003FAA1DDE2B3F898A00002A9F0000000000000CA0");
+        messageExtBatch.setSysFlag(0);
+        messageExtBatch.setBornHostV6Flag();
+        messageExtBatch.setStoreHostAddressV6Flag();
+        messageExtBatch.setBornHost(new InetSocketAddress("1050:0000:0000:0000:0005:0600:300c:326b", 123));
+        messageExtBatch.setStoreHost(new InetSocketAddress("::1", 124));
+        messageExtBatch.setBody(MessageDecoder.encodeMessages(messages));
+
+        messageExtBatch.setEncodedBuff(batchEncoder.encode(messageExtBatch));
+        ByteBuffer buff = ByteBuffer.allocate(1024 * 10);
+        AppendMessageResult allresult = callback.doAppend(0, buff, 1024 * 10, messageExtBatch);
+
+        assertEquals(AppendMessageStatus.PUT_OK, allresult.getStatus());
+        assertEquals(0, allresult.getWroteOffset());
+        assertEquals(0, allresult.getLogicsOffset());
+        assertEquals(buff.position(), allresult.getWroteBytes());
+
+        assertEquals(messages.size(), allresult.getMsgNum());
+
+        Set<String> msgIds = new HashSet<>();
+        for (String msgId : allresult.getMsgId().split(",")) {
+            assertEquals(56, msgId.length());
+            msgIds.add(msgId);
+        }
+        assertEquals(messages.size(), msgIds.size());
+
+        List<MessageExt> decodeMsgs = MessageDecoder.decodes((ByteBuffer) buff.flip());
+        assertEquals(decodeMsgs.size(), decodeMsgs.size());
+        long queueOffset = decodeMsgs.get(0).getQueueOffset();
+        long storeTimeStamp = decodeMsgs.get(0).getStoreTimestamp();
+        for (int i = 0; i < messages.size(); i++) {
+            assertEquals(messages.get(i).getTopic(), decodeMsgs.get(i).getTopic());
+            assertEquals(new String(messages.get(i).getBody()), new String(decodeMsgs.get(i).getBody()));
+            assertEquals(messages.get(i).getTags(), decodeMsgs.get(i).getTags());
+
+            assertEquals(messageExtBatch.getBornHostNameString(), decodeMsgs.get(i).getBornHostNameString());
+
+            assertEquals(messageExtBatch.getBornTimestamp(), decodeMsgs.get(i).getBornTimestamp());
+            assertEquals(storeTimeStamp, decodeMsgs.get(i).getStoreTimestamp());
+            assertEquals(queueOffset++, decodeMsgs.get(i).getQueueOffset());
+        }
+
+    }
+
 }

store/src/test/java/org/apache/rocketmq/store/BatchPutMessageTest.java

@@ -23,6 +23,7 @@ import org.apache.rocketmq.common.message.Message;
 import org.apache.rocketmq.common.message.MessageDecoder;
 import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.common.message.MessageExtBatch;
+import org.apache.rocketmq.common.sysflag.MessageSysFlag;
 import org.apache.rocketmq.store.config.FlushDiskType;
 import org.apache.rocketmq.store.config.MessageStoreConfig;
 import org.apache.rocketmq.store.stats.BrokerStatsManager;
@@ -125,6 +126,58 @@ public class BatchPutMessageTest {
 
     }
 
+    @Test
+    public void testPutIPv6HostMessages() throws Exception {
+        List<Message> messages = new ArrayList<>();
+        String topic = "batch-write-topic";
+        int queue = 0;
+        int[] msgLengthArr = new int[11];
+        msgLengthArr[0] = 0;
+        int j = 1;
+        for (int i = 0; i < 10; i++) {
+            Message msg = new Message();
+            msg.setBody(("body" + i).getBytes());
+            msg.setTopic(topic);
+            msg.setTags("TAG1");
+            msg.setKeys(String.valueOf(System.currentTimeMillis()));
+            messages.add(msg);
+            String properties = messageProperties2String(msg.getProperties());
+            byte[] propertiesBytes = properties.getBytes(CHARSET_UTF8);
+            short propertiesLength = (short) propertiesBytes.length;
+            final byte[] topicData = msg.getTopic().getBytes(MessageDecoder.CHARSET_UTF8);
+            final int topicLength = topicData.length;
+            msgLengthArr[j] = calIPv6HostMsgLength(msg.getBody().length, topicLength, propertiesLength) + msgLengthArr[j - 1];
+            j++;
+        }
+        byte[] batchMessageBody = MessageDecoder.encodeMessages(messages);
+        MessageExtBatch messageExtBatch = new MessageExtBatch();
+        messageExtBatch.setTopic(topic);
+        messageExtBatch.setQueueId(queue);
+        messageExtBatch.setBody(batchMessageBody);
+        messageExtBatch.setMsgId("24084004018081003FAA1DDE2B3F898A00002A9F0000000000000CA0");
+        messageExtBatch.setBornTimestamp(System.currentTimeMillis());
+        messageExtBatch.setSysFlag(0);
+        messageExtBatch.setBornHostV6Flag();
+        messageExtBatch.setStoreHostAddressV6Flag();
+        messageExtBatch.setStoreHost(new InetSocketAddress("1050:0000:0000:0000:0005:0600:300c:326b", 125));
+        messageExtBatch.setBornHost(new InetSocketAddress("::1", 126));
+
+        PutMessageResult putMessageResult = messageStore.putMessages(messageExtBatch);
+        assertThat(putMessageResult.isOk()).isTrue();
+
+        Thread.sleep(3 * 1000);
+
+        for (long i = 0; i < 10; i++) {
+            MessageExt messageExt = messageStore.lookMessageByOffset(msgLengthArr[(int) i]);
+            assertThat(messageExt).isNotNull();
+            GetMessageResult result = messageStore.getMessage("batch_write_group", topic, queue, i, 1024 * 1024, null);
+            assertThat(result).isNotNull();
+            assertThat(result.getStatus()).isEqualTo(GetMessageStatus.FOUND);
+            result.release();
+        }
+
+    }
+
     private int calMsgLength(int bodyLength, int topicLength, int propertiesLength) {
         final int msgLen = 4 //TOTALSIZE
                 + 4 //MAGICCODE
@@ -147,6 +200,28 @@ public class BatchPutMessageTest {
         return msgLen;
     }
 
+    private int calIPv6HostMsgLength(int bodyLength, int topicLength, int propertiesLength) {
+        final int msgLen = 4 //TOTALSIZE
+            + 4 //MAGICCODE
+            + 4 //BODYCRC
+            + 4 //QUEUEID
+            + 4 //FLAG
+            + 8 //QUEUEOFFSET
+            + 8 //PHYSICALOFFSET
+            + 4 //SYSFLAG
+            + 8 //BORNTIMESTAMP
+            + 20 //BORNHOST
+            + 8 //STORETIMESTAMP
+            + 20 //STOREHOSTADDRESS
+            + 4 //RECONSUMETIMES
+            + 8 //Prepared Transaction Offset
+            + 4 + (bodyLength > 0 ? bodyLength : 0) //BODY
+            + 1 + topicLength //TOPIC
+            + 2 + (propertiesLength > 0 ? propertiesLength : 0) //propertiesLength
+            + 0;
+        return msgLen;
+    }
+
     public String messageProperties2String(Map<String, String> properties) {
         StringBuilder sb = new StringBuilder();
         if (properties != null) {

store/src/test/java/org/apache/rocketmq/store/ConsumeQueueTest.java

@@ -83,6 +83,29 @@ public class ConsumeQueueTest {
         return msg;
     }
 
+    public MessageExtBrokerInner buildIPv6HostMessage() {
+        MessageExtBrokerInner msg = new MessageExtBrokerInner();
+        msg.setTopic(topic);
+        msg.setTags("TAG1");
+        msg.setKeys("Hello");
+        msg.setBody(msgBody);
+        msg.setMsgId("24084004018081003FAA1DDE2B3F898A00002A9F0000000000000CA0");
+        msg.setKeys(String.valueOf(System.currentTimeMillis()));
+        msg.setQueueId(queueId);
+        msg.setSysFlag(0);
+        msg.setBornHostV6Flag();
+        msg.setStoreHostAddressV6Flag();
+        msg.setBornTimestamp(System.currentTimeMillis());
+        msg.setBornHost(new InetSocketAddress("1050:0000:0000:0000:0005:0600:300c:326b", 123));
+        msg.setStoreHost(new InetSocketAddress("::1", 124));
+        for (int i = 0; i < 1; i++) {
+            msg.putUserProperty(String.valueOf(i), "imagoodperson" + i);
+        }
+        msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));
+
+        return msg;
+    }
+
     public MessageStoreConfig buildStoreConfig(int commitLogFileSize, int cqFileSize,
         boolean enableCqExt, int cqExtFileSize) {
         MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
@@ -127,7 +150,11 @@ public class ConsumeQueueTest {
         long totalMsgs = 200;
 
         for (long i = 0; i < totalMsgs; i++) {
-            master.putMessage(buildMessage());
+            if (i < totalMsgs / 2) {
+                master.putMessage(buildMessage());
+            } else {
+                master.putMessage(buildIPv6HostMessage());
+            }
         }
     }
 

store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java

@@ -24,6 +24,7 @@ import java.lang.reflect.Method;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
+import java.net.UnknownHostException;
 import java.nio.ByteBuffer;
 import java.nio.MappedByteBuffer;
 import java.nio.channels.FileChannel;
@@ -116,13 +117,19 @@ public class DefaultMessageStoreTest {
 
     @Test
     public void testWriteAndRead() {
-        long totalMsgs = 10;
+        long ipv4HostMsgs = 10;
+        long ipv6HostMsgs = 10;
+        long totalMsgs = ipv4HostMsgs + ipv6HostMsgs;
         QUEUE_TOTAL = 1;
         MessageBody = StoreMessage.getBytes();
-        for (long i = 0; i < totalMsgs; i++) {
+        for (long i = 0; i < ipv4HostMsgs; i++) {
             messageStore.putMessage(buildMessage());
         }
 
+        for (long i = 0; i < ipv6HostMsgs; i++) {
+            messageStore.putMessage(buildIPv6HostMessage());
+        }
+
         StoreTestUtil.waitCommitLogReput((DefaultMessageStore) messageStore);
 
         for (long i = 0; i < totalMsgs; i++) {
@@ -134,7 +141,7 @@ public class DefaultMessageStoreTest {
     }
 
     @Test
-    public void should_look_message_successfully_when_offset_is_first() {
+    public void testLookMessageByOffset_OffsetIsFirst() {
         final int totalCount = 10;
         int queueId = new Random().nextInt(10);
         String topic = "FooBar";
@@ -150,7 +157,7 @@ public class DefaultMessageStoreTest {
     }
 
     @Test
-    public void should_look_message_successfully_when_offset_is_last() {
+    public void testLookMessageByOffset_OffsetIsLast() {
         final int totalCount = 10;
         int queueId = new Random().nextInt(10);
         String topic = "FooBar";
@@ -164,7 +171,7 @@ public class DefaultMessageStoreTest {
     }
 
     @Test
-    public void should_look_message_failed_and_return_null_when_offset_is_out_of_bound() {
+    public void testLookMessageByOffset_OffsetIsOutOfBound() {
         final int totalCount = 10;
         int queueId = new Random().nextInt(10);
         String topic = "FooBar";
@@ -177,7 +184,7 @@ public class DefaultMessageStoreTest {
     }
 
     @Test
-    public void should_get_consume_queue_offset_successfully_when_incomming_by_timestamp() throws InterruptedException {
+    public void testGetOffsetInQueueByTime() {
         final int totalCount = 10;
         int queueId = 0;
         String topic = "FooBar";
@@ -196,7 +203,7 @@ public class DefaultMessageStoreTest {
     }
 
     @Test
-    public void should_get_consume_queue_offset_successfully_when_timestamp_is_skewing() throws InterruptedException {
+    public void testGetOffsetInQueueByTime_TimestampIsSkewing() {
         final int totalCount = 10;
         int queueId = 0;
         String topic = "FooBar";
@@ -221,7 +228,7 @@ public class DefaultMessageStoreTest {
     }
 
     @Test
-    public void should_get_min_of_max_consume_queue_offset_when_timestamp_s_skewing_is_large() throws InterruptedException {
+    public void testGetOffsetInQueueByTime_TimestampSkewingIsLarge() {
         final int totalCount = 10;
         int queueId = 0;
         String topic = "FooBar";
@@ -247,7 +254,7 @@ public class DefaultMessageStoreTest {
     }
 
     @Test
-    public void should_return_zero_when_consume_queue_not_found() throws InterruptedException {
+    public void testGetOffsetInQueueByTime_ConsumeQueueNotFound1() {
         final int totalCount = 10;
         int queueId = 0;
         int wrongQueueId = 1;
@@ -263,7 +270,7 @@ public class DefaultMessageStoreTest {
     }
 
     @Test
-    public void should_return_negative_one_when_invoke_getMessageStoreTimeStamp_if_consume_queue_not_found() throws InterruptedException {
+    public void testGetOffsetInQueueByTime_ConsumeQueueNotFound2() {
         final int totalCount = 10;
         int queueId = 0;
         int wrongQueueId = 1;
@@ -278,7 +285,7 @@ public class DefaultMessageStoreTest {
     }
 
     @Test
-    public void should_return_negative_one_when_invoke_getMessageStoreTimeStamp_if_consumeQueueOffset_not_exist() throws InterruptedException {
+    public void testGetOffsetInQueueByTime_ConsumeQueueOffsetNotExist() {
         final int totalCount = 10;
         int queueId = 0;
         int wrongQueueId = 1;
@@ -293,9 +300,8 @@ public class DefaultMessageStoreTest {
         assertThat(messageStoreTimeStamp).isEqualTo(-1);
     }
 
-
     @Test
-    public void should_get_message_store_timestamp_successfully_when_incomming_by_topic_queueId_and_consumeQueueOffset() throws InterruptedException {
+    public void testGetMessageStoreTimeStamp() {
         final int totalCount = 10;
         int queueId = 0;
         String topic = "FooBar";
@@ -304,7 +310,7 @@ public class DefaultMessageStoreTest {
         StoreTestUtil.waitCommitLogReput((DefaultMessageStore) messageStore);
 
         ConsumeQueue consumeQueue = getDefaultMessageStore().findConsumeQueue(topic, queueId);
-        int minOffsetInQueue = (int)consumeQueue.getMinOffsetInQueue();
+        int minOffsetInQueue = (int) consumeQueue.getMinOffsetInQueue();
         for (int i = minOffsetInQueue; i < consumeQueue.getMaxOffsetInQueue(); i++) {
             long messageStoreTimeStamp = messageStore.getMessageStoreTimeStamp(topic, queueId, i);
             assertThat(messageStoreTimeStamp).isEqualTo(appendMessageResults[i].getStoreTimestamp());
@@ -312,14 +318,14 @@ public class DefaultMessageStoreTest {
     }
 
     @Test
-    public void should_return_negative_one_when_invoke_getStoreTime_if_incomming_param_is_null() {
+    public void testGetStoreTime_ParamIsNull() {
         long storeTime = getStoreTime(null);
 
         assertThat(storeTime).isEqualTo(-1);
     }
 
     @Test
-    public void should_get_store_time_successfully_when_invoke_getStoreTime_if_everything_is_ok() throws InterruptedException {
+    public void testGetStoreTime_EverythingIsOk() {
         final int totalCount = 10;
         int queueId = 0;
         String topic = "FooBar";
@@ -337,7 +343,7 @@ public class DefaultMessageStoreTest {
     }
 
     @Test
-    public void should_return_negative_one_when_invoke_getStoreTime_if_phyOffset_is_less_than_commitLog_s_minOffset() {
+    public void testGetStoreTime_PhyOffsetIsLessThanCommitLogMinOffset() {
         long phyOffset = -10;
         int size = 138;
         ByteBuffer byteBuffer = ByteBuffer.allocate(100);
@@ -354,7 +360,7 @@ public class DefaultMessageStoreTest {
     }
 
     private DefaultMessageStore getDefaultMessageStore() {
-        return (DefaultMessageStore)this.messageStore;
+        return (DefaultMessageStore) this.messageStore;
     }
 
     private AppendMessageResult[] putMessages(int totalCount, String topic, int queueId) {
@@ -365,7 +371,9 @@ public class DefaultMessageStoreTest {
         AppendMessageResult[] appendMessageResultArray = new AppendMessageResult[totalCount];
         for (int i = 0; i < totalCount; i++) {
             String messageBody = buildMessageBodyByOffset(StoreMessage, i);
-            MessageExtBrokerInner msgInner = buildMessage(messageBody.getBytes(), topic);
+
+            MessageExtBrokerInner msgInner =
+                i < totalCount / 2 ? buildMessage(messageBody.getBytes(), topic) : buildIPv6HostMessage(messageBody.getBytes(), topic);
             msgInner.setQueueId(queueId);
             PutMessageResult result = messageStore.putMessage(msgInner);
             appendMessageResultArray[i] = result.getAppendMessageResult();
@@ -374,7 +382,7 @@ public class DefaultMessageStoreTest {
                 try {
                     Thread.sleep(10);
                 } catch (InterruptedException e) {
-                    throw  new RuntimeException("Thread sleep ERROR");
+                    throw new RuntimeException("Thread sleep ERROR");
                 }
             }
         }
@@ -397,7 +405,7 @@ public class DefaultMessageStoreTest {
         try {
             Method getStoreTime = getDefaultMessageStore().getClass().getDeclaredMethod("getStoreTime", SelectMappedBufferResult.class);
             getStoreTime.setAccessible(true);
-            return (long)getStoreTime.invoke(getDefaultMessageStore(), result);
+            return (long) getStoreTime.invoke(getDefaultMessageStore(), result);
         } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
             throw new RuntimeException(e);
         }
@@ -418,10 +426,43 @@ public class DefaultMessageStoreTest {
         return msg;
     }
 
+    private MessageExtBrokerInner buildIPv6HostMessage(byte[] messageBody, String topic) {
+        MessageExtBrokerInner msg = new MessageExtBrokerInner();
+        msg.setTopic(topic);
+        msg.setTags("TAG1");
+        msg.setKeys("Hello");
+        msg.setBody(messageBody);
+        msg.setMsgId("24084004018081003FAA1DDE2B3F898A00002A9F0000000000000CA0");
+        msg.setKeys(String.valueOf(System.currentTimeMillis()));
+        msg.setQueueId(Math.abs(QueueId.getAndIncrement()) % QUEUE_TOTAL);
+        msg.setSysFlag(0);
+        msg.setBornHostV6Flag();
+        msg.setStoreHostAddressV6Flag();
+        msg.setBornTimestamp(System.currentTimeMillis());
+        try {
+            msg.setBornHost(new InetSocketAddress(InetAddress.getByName("1050:0000:0000:0000:0005:0600:300c:326b"), 0));
+        } catch (UnknownHostException e) {
+            e.printStackTrace();
+            assertThat(Boolean.FALSE).isTrue();
+        }
+
+        try {
+            msg.setStoreHost(new InetSocketAddress(InetAddress.getByName("::1"), 0));
+        } catch (UnknownHostException e) {
+            e.printStackTrace();
+            assertThat(Boolean.FALSE).isTrue();
+        }
+        return msg;
+    }
+
     private MessageExtBrokerInner buildMessage() {
         return buildMessage(MessageBody, "FooBar");
     }
 
+    private MessageExtBrokerInner buildIPv6HostMessage() {
+        return buildIPv6HostMessage(MessageBody, "FooBar");
+    }
+
     private void verifyThatMasterIsFunctional(long totalMsgs, MessageStore master) {
         for (long i = 0; i < totalMsgs; i++) {
             master.putMessage(buildMessage());
@@ -477,7 +518,7 @@ public class DefaultMessageStoreTest {
             messageStore.putMessage(messageExtBrokerInner);
         }
 
-       // Thread.sleep(100);//wait for build consumer queue
+        // Thread.sleep(100);//wait for build consumer queue
         StoreTestUtil.waitCommitLogReput((DefaultMessageStore) messageStore);
 
         long maxPhyOffset = messageStore.getMaxPhyOffset();
@@ -587,7 +628,7 @@ public class DefaultMessageStoreTest {
     private class MyMessageArrivingListener implements MessageArrivingListener {
         @Override
         public void arriving(String topic, int queueId, long logicOffset, long tagsCode, long msgStoreTime,
-                             byte[] filterBitMap, Map<String, String> properties) {
+            byte[] filterBitMap, Map<String, String> properties) {
         }
     }
 }

store/src/test/java/org/apache/rocketmq/store/StoreTestBase.java

@@ -20,12 +20,12 @@ import java.io.File;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
+import java.net.UnknownHostException;
 import java.util.HashSet;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.rocketmq.common.UtilAll;
-import org.apache.rocketmq.store.config.StorePathConfigHelper;
 import org.junit.After;
 
 public class StoreTestBase {
@@ -59,6 +59,33 @@ public class StoreTestBase {
         return msg;
     }
 
+    protected MessageExtBrokerInner buildIPv6HostMessage() {
+        MessageExtBrokerInner msg = new MessageExtBrokerInner();
+        msg.setTopic("StoreTest");
+        msg.setTags("TAG1");
+        msg.setKeys("Hello");
+        msg.setBody(MessageBody);
+        msg.setMsgId("24084004018081003FAA1DDE2B3F898A00002A9F0000000000000CA0");
+        msg.setKeys(String.valueOf(System.currentTimeMillis()));
+        msg.setQueueId(Math.abs(QueueId.getAndIncrement()) % QUEUE_TOTAL);
+        msg.setSysFlag(0);
+        msg.setBornHostV6Flag();
+        msg.setStoreHostAddressV6Flag();
+        msg.setBornTimestamp(System.currentTimeMillis());
+        try {
+            msg.setBornHost(new InetSocketAddress(InetAddress.getByName("1050:0000:0000:0000:0005:0600:300c:326b"), 8123));
+        } catch (UnknownHostException e) {
+            e.printStackTrace();
+        }
+
+        try {
+            msg.setStoreHost(new InetSocketAddress(InetAddress.getByName("::1"), 8123));
+        } catch (UnknownHostException e) {
+            e.printStackTrace();
+        }
+        return msg;
+    }
+
     public static String createBaseDir() {
         String baseDir = System.getProperty("user.home") + File.separator + "unitteststore" + File.separator + UUID.randomUUID();
         final File file = new File(baseDir);
@@ -74,7 +101,6 @@ public class StoreTestBase {
         return file.createNewFile();
     }
 
-
     public static void deleteFile(String fileName) {
         deleteFile(new File(fileName));
     }

store/src/test/java/org/apache/rocketmq/store/dledger/DLedgerCommitlogTest.java

@@ -145,7 +145,8 @@ public class DLedgerCommitlogTest extends MessageStoreTestBase {
 
         List<PutMessageResult> results = new ArrayList<>();
         for (int i = 0; i < 10; i++) {
-            MessageExtBrokerInner msgInner =  buildMessage();
+            MessageExtBrokerInner msgInner =
+                i < 5 ? buildMessage() : buildIPv6HostMessage();
             msgInner.setTopic(topic);
             msgInner.setQueueId(0);
             PutMessageResult putMessageResult = messageStore.putMessage(msgInner);
@@ -209,5 +210,39 @@ public class DLedgerCommitlogTest extends MessageStoreTestBase {
         followerStore.shutdown();
     }
 
+    @Test
+    public void testIPv6HostMsgCommittedPos() throws Exception {
+        String peers = String.format("n0-localhost:%d;n1-localhost:%d", nextPort(), nextPort());
+        String group = UUID.randomUUID().toString();
+        DefaultMessageStore leaderStore = createDledgerMessageStore(createBaseDir(), group,"n0", peers, "n0", false, 0);
+
+        String topic = UUID.randomUUID().toString();
+        MessageExtBrokerInner msgInner =  buildIPv6HostMessage();
+        msgInner.setTopic(topic);
+        msgInner.setQueueId(0);
+        PutMessageResult putMessageResult = leaderStore.putMessage(msgInner);
+        Assert.assertEquals(PutMessageStatus.OS_PAGECACHE_BUSY, putMessageResult.getPutMessageStatus());
+
+        Thread.sleep(1000);
+
+        Assert.assertEquals(0, leaderStore.getCommitLog().getMaxOffset());
+        Assert.assertEquals(0, leaderStore.getMaxOffsetInQueue(topic, 0));
+
+
+        DefaultMessageStore followerStore = createDledgerMessageStore(createBaseDir(), group,"n1", peers, "n0", false, 0);
+        Thread.sleep(2000);
+
+        Assert.assertEquals(1, leaderStore.getMaxOffsetInQueue(topic, 0));
+        Assert.assertEquals(1, followerStore.getMaxOffsetInQueue(topic, 0));
+        Assert.assertTrue(leaderStore.getCommitLog().getMaxOffset() > 0);
+
+
+        leaderStore.destroy();
+        followerStore.destroy();
+
+        leaderStore.shutdown();
+        followerStore.shutdown();
+    }
+
 
 }

store/src/test/java/org/apache/rocketmq/store/dledger/MessageStoreTestBase.java

@@ -19,6 +19,7 @@ package org.apache.rocketmq.store.dledger;
 import io.openmessaging.storage.dledger.DLedgerConfig;
 import io.openmessaging.storage.dledger.DLedgerServer;
 import java.io.File;
+import java.net.UnknownHostException;
 import java.util.Arrays;
 import org.apache.rocketmq.common.BrokerConfig;
 import org.apache.rocketmq.common.message.MessageDecoder;
@@ -118,7 +119,7 @@ public class MessageStoreTestBase extends StoreTestBase {
         return defaultMessageStore;
     }
 
-    protected void doPutMessages(MessageStore messageStore, String topic, int queueId, int num, long beginLogicsOffset) {
+    protected void doPutMessages(MessageStore messageStore, String topic, int queueId, int num, long beginLogicsOffset) throws UnknownHostException {
         for (int i = 0; i < num; i++) {
             MessageExtBrokerInner msgInner = buildMessage();
             msgInner.setTopic(topic);