diff --git a/acl/pom.xml b/acl/pom.xml
index d9e1fcf93cabfa55b9cbbd9c56980caf7595a63b..ec5ab0493fb588101213d8105a632f5db49bd0e6 100644
--- a/acl/pom.xml
+++ b/acl/pom.xml
@@ -67,6 +67,10 @@
logback-core
test
+
+ commons-validator
+ commons-validator
+
diff --git a/acl/src/main/java/org/apache/rocketmq/acl/common/AclUtils.java b/acl/src/main/java/org/apache/rocketmq/acl/common/AclUtils.java
index 20e1cfa26b90916aecfec39e2669179c85854e20..8973320237c684c68864c29ff0c4a90d75bfdd3e 100644
--- a/acl/src/main/java/org/apache/rocketmq/acl/common/AclUtils.java
+++ b/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 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 getYamlDataObject(String path, Class clazz) {
Yaml yaml = new Yaml();
FileInputStream fis = null;
@@ -148,7 +286,7 @@ public class AclUtils {
}
}
- public static boolean writeDataObject(String path, Map dataMap) {
+ public static boolean writeDataObject(String path, Map 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));
}
}
diff --git a/acl/src/main/java/org/apache/rocketmq/acl/plain/PlainAccessValidator.java b/acl/src/main/java/org/apache/rocketmq/acl/plain/PlainAccessValidator.java
index 7e3b3cf35deee99e59366df9642cc2433ce6fb26..7d96d41548c8ebe5b8a410d5c35b255641386b65 100644
--- a/acl/src/main/java/org/apache/rocketmq/acl/plain/PlainAccessValidator.java
+++ b/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);
}
diff --git a/acl/src/main/java/org/apache/rocketmq/acl/plain/RemoteAddressStrategyFactory.java b/acl/src/main/java/org/apache/rocketmq/acl/plain/RemoteAddressStrategyFactory.java
index cc2dcee77f9b71d5a516b2eb68a10b2744d6364e..6931eb7c426aa3511769f5e9fa8a4abc6c260bb3 100644
--- a/acl/src/main/java/org/apache/rocketmq/acl/plain/RemoteAddressStrategyFactory.java
+++ b/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 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;
}
-
}
-
}
diff --git a/acl/src/test/java/org/apache/rocketmq/acl/common/AclUtilsTest.java b/acl/src/test/java/org/apache/rocketmq/acl/common/AclUtilsTest.java
index 5b2627de85dc66f13f1868b7b0c5e014aba3117f..5705b745a0e7472ca7a215468ae20775a53babfb 100644
--- a/acl/src/test/java/org/apache/rocketmq/acl/common/AclUtilsTest.java
+++ b/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 newIPv6AddressList = new ArrayList<>();
+ for (String a : ipv6AddressArray) {
+ newIPv6AddressList.add(a);
+ }
+
+ List 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 globalWhiteRemoteAddrs = new ArrayList();
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> accounts = new ArrayList>();
@@ -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 globalWhiteRemoteAddrs = new ArrayList();
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 readableMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
List updatedGlobalWhiteRemoteAddrs = (List) 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);
}
-
}
diff --git a/acl/src/test/java/org/apache/rocketmq/acl/plain/RemoteAddressStrategyTest.java b/acl/src/test/java/org/apache/rocketmq/acl/plain/RemoteAddressStrategyTest.java
index 2c2e76bb6fdb328be9c15c9f13efa40ff76c04e2..8998dd993ccbcf75c9788fd8cf314c4c93f58771 100644
--- a/acl/src/test/java/org/apache/rocketmq/acl/plain/RemoteAddressStrategyTest.java
+++ b/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");
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/PullMessageProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/PullMessageProcessor.java
index 10c0112feb4a48dfbc483a75069f5022625489fc..b4f6daa05eb8fa6a5474a57515e5145c08f91e5f 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/processor/PullMessageProcessor.java
+++ b/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();
diff --git a/common/pom.xml b/common/pom.xml
index 74ea467ca1ae518f2969869f37e92dd09483f542..11df833bff0a596a7e030ac2bba6f0afc3c532e9 100644
--- a/common/pom.xml
+++ b/common/pom.xml
@@ -41,5 +41,9 @@
org.apache.commons
commons-lang3
+
+ commons-validator
+ commons-validator
+
diff --git a/common/src/main/java/org/apache/rocketmq/common/MixAll.java b/common/src/main/java/org/apache/rocketmq/common/MixAll.java
index 0af65dff0f11b43e31ae73a8747adf1f72156a0d..bba8b36ff5e172098573024e2acdb019c4e03f12 100644
--- a/common/src/main/java/org/apache/rocketmq/common/MixAll.java
+++ b/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;
diff --git a/common/src/main/java/org/apache/rocketmq/common/UtilAll.java b/common/src/main/java/org/apache/rocketmq/common/UtilAll.java
index 68d93f3af3821008591a6385ff16f7ca579df93f..85a4cde121d8a3be2349e34b96b86e4c047d6f50 100644
--- a/common/src/main/java/org/apache/rocketmq/common/UtilAll.java
+++ b/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 list,String splitor) {
+ public static String List2String(List 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 String2List(String str,String splitor) {
+ public static List String2List(String str, String splitor) {
if (StringUtils.isEmpty(str)) {
return null;
}
diff --git a/common/src/main/java/org/apache/rocketmq/common/message/MessageClientIDSetter.java b/common/src/main/java/org/apache/rocketmq/common/message/MessageClientIDSetter.java
index d0b202ec64b120a07ecb555a594d8cd9b2d0efc8..df11556383de6c18d18751f43aa5cc7496197697 100644
--- a/common/src/main/java/org/apache/rocketmq/common/message/MessageClientIDSetter.java
+++ b/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;
}
diff --git a/common/src/main/java/org/apache/rocketmq/common/message/MessageDecoder.java b/common/src/main/java/org/apache/rocketmq/common/message/MessageDecoder.java
index af0b638d3d0ff62bed2718708255e52b26d8c85f..54d5e6b60f37cf683bc613176ae62edf0402e9c9 100644
--- a/common/src/main/java/org/apache/rocketmq/common/message/MessageDecoder.java
+++ b/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 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);
diff --git a/common/src/main/java/org/apache/rocketmq/common/message/MessageExt.java b/common/src/main/java/org/apache/rocketmq/common/message/MessageExt.java
index 3f77767eb8d00ff8322df2525b9225034f51f79d..20cb0572717ca138a26ca5d157bbe295dae98fc4 100644
--- a/common/src/main/java/org/apache/rocketmq/common/message/MessageExt.java
+++ b/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;
}
diff --git a/common/src/main/java/org/apache/rocketmq/common/sysflag/MessageSysFlag.java b/common/src/main/java/org/apache/rocketmq/common/sysflag/MessageSysFlag.java
index f1397706cc5e924ae4a8019045fd83a98643c20c..d534571eb55276f2452780a13bb6148ea8f5dd43 100644
--- a/common/src/main/java/org/apache/rocketmq/common/sysflag/MessageSysFlag.java
+++ b/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);
}
+
}
diff --git a/common/src/test/java/org/apache/rocketmq/common/UtilAllTest.java b/common/src/test/java/org/apache/rocketmq/common/UtilAllTest.java
index 265645213e4fe6f3ba4a6aefc73720013129d0e3..3991367330bedb522c46015b13b792e76a4fa278 100644
--- a/common/src/test/java/org/apache/rocketmq/common/UtilAllTest.java
+++ b/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;
diff --git a/common/src/test/java/org/apache/rocketmq/common/message/MessageClientIDSetterTest.java b/common/src/test/java/org/apache/rocketmq/common/message/MessageClientIDSetterTest.java
new file mode 100644
index 0000000000000000000000000000000000000000..987f1032fe7e6a1c3fa67bdf0119766a3097336f
--- /dev/null
+++ b/common/src/test/java/org/apache/rocketmq/common/message/MessageClientIDSetterTest.java
@@ -0,0 +1,47 @@
+/*
+ * 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);
+ }
+}
diff --git a/common/src/test/java/org/apache/rocketmq/common/message/MessageDecoderTest.java b/common/src/test/java/org/apache/rocketmq/common/message/MessageDecoderTest.java
index d14d6b060c89168cde510531da86740ec8b33ad5..82ad2714dd2c255eae0067a587aa3467554b0b7f 100644
--- a/common/src/test/java/org/apache/rocketmq/common/message/MessageDecoderTest.java
+++ b/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 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());
+ }
}
diff --git a/pom.xml b/pom.xml
index 46b88ceb04c88ca4f506b1a7b5a5972f4a79a26b..0161c7b924727521bb87ca3e78f5fef614a4236c 100644
--- a/pom.xml
+++ b/pom.xml
@@ -16,7 +16,8 @@
limitations under the License.
-->
-
+
org.apache
@@ -159,7 +160,7 @@
-
+
true
@@ -620,6 +621,13 @@
log4j-slf4j-impl
2.7
+
+ commons-validator
+ commons-validator
+ 1.6
+
+
+
diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/common/RemotingHelper.java b/remoting/src/main/java/org/apache/rocketmq/remoting/common/RemotingHelper.java
index 585b60b798cd0986748f9df035b6229690464f05..f244bf4c853551ad8b4810f95a78b039ae6d239b 100644
--- a/remoting/src/main/java/org/apache/rocketmq/remoting/common/RemotingHelper.java
+++ b/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;
}
diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/common/RemotingUtil.java b/remoting/src/main/java/org/apache/rocketmq/remoting/common/RemotingUtil.java
index 3da3a1839603396a1bd592971059b436a1082e08..a16940e561c5cec23c9df021c429e1c64d500b88 100644
--- a/remoting/src/main/java/org/apache/rocketmq/remoting/common/RemotingUtil.java
+++ b/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;
}
diff --git a/store/src/main/java/org/apache/rocketmq/store/CommitLog.java b/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
index 2ab66e84634f119b7dcb3b08e95c3399f26aff44..3d89fe4c5198cafc22fd33033716215a577fe6ef 100644
--- a/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
+++ b/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
diff --git a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
index 971b1e75b38da514fd7e8c5f12e5ff5e3dc9c2ab..d5ba5692a928e4b65b41ec94be0356b0c81dd1c9 100644
--- a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
+++ b/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++);
diff --git a/store/src/main/java/org/apache/rocketmq/store/dledger/DLedgerCommitLog.java b/store/src/main/java/org/apache/rocketmq/store/dledger/DLedgerCommitLog.java
index 5192e86defba1da4f0bb91080afec8fb3af5424c..13da48bfe9f08408c57212ea506b674939bbb3ac 100644
--- a/store/src/main/java/org/apache/rocketmq/store/dledger/DLedgerCommitLog.java
+++ b/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;
diff --git a/store/src/test/java/org/apache/rocketmq/store/AppendCallbackTest.java b/store/src/test/java/org/apache/rocketmq/store/AppendCallbackTest.java
index fbb2a746c4f96686b273adb4e0847f927a6cc1ff..f46b3befed63fc25f9c0901b60f97c54a68616f8 100644
--- a/store/src/test/java/org/apache/rocketmq/store/AppendCallbackTest.java
+++ b/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 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 messages = new ArrayList<>();
@@ -153,4 +190,64 @@ public class AppendCallbackTest {
}
+ @Test
+ public void testAppendIPv6HostMessageBatchSucc() throws Exception {
+ List 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 msgIds = new HashSet<>();
+ for (String msgId : allresult.getMsgId().split(",")) {
+ assertEquals(56, msgId.length());
+ msgIds.add(msgId);
+ }
+ assertEquals(messages.size(), msgIds.size());
+
+ List 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());
+ }
+
+ }
+
}
diff --git a/store/src/test/java/org/apache/rocketmq/store/BatchPutMessageTest.java b/store/src/test/java/org/apache/rocketmq/store/BatchPutMessageTest.java
index 0c8e5bb1472caf268cea6522de7ebf332a5db42c..8618dbb0e56ed2804303d1b08bbe94cc5e54f500 100644
--- a/store/src/test/java/org/apache/rocketmq/store/BatchPutMessageTest.java
+++ b/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 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 properties) {
StringBuilder sb = new StringBuilder();
if (properties != null) {
diff --git a/store/src/test/java/org/apache/rocketmq/store/ConsumeQueueTest.java b/store/src/test/java/org/apache/rocketmq/store/ConsumeQueueTest.java
index c9730306b5d6793811191697062b9efc3f8894a6..7c57813da7e8bdc5753886a461856db383b30e06 100644
--- a/store/src/test/java/org/apache/rocketmq/store/ConsumeQueueTest.java
+++ b/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());
+ }
}
}
diff --git a/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java b/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java
index 785977468e05138198b8da07ee28d10c17c404ef..fd13e71a427d07db60400392a8718bcc64de0298 100644
--- a/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java
+++ b/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 properties) {
+ byte[] filterBitMap, Map properties) {
}
}
}
diff --git a/store/src/test/java/org/apache/rocketmq/store/StoreTestBase.java b/store/src/test/java/org/apache/rocketmq/store/StoreTestBase.java
index 0dcb3b254c0c949ad4f4186b8148dccb70332abf..a736754de47b4ce6d699b63360dfebf1eb86ee0d 100644
--- a/store/src/test/java/org/apache/rocketmq/store/StoreTestBase.java
+++ b/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));
}
diff --git a/store/src/test/java/org/apache/rocketmq/store/dledger/DLedgerCommitlogTest.java b/store/src/test/java/org/apache/rocketmq/store/dledger/DLedgerCommitlogTest.java
index d0829d122a1ca1ad250053678f812eb1edac7c57..f0b9205302e965a832393d788d122e64bbb5d0aa 100644
--- a/store/src/test/java/org/apache/rocketmq/store/dledger/DLedgerCommitlogTest.java
+++ b/store/src/test/java/org/apache/rocketmq/store/dledger/DLedgerCommitlogTest.java
@@ -145,7 +145,8 @@ public class DLedgerCommitlogTest extends MessageStoreTestBase {
List 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();
+ }
+
}
diff --git a/store/src/test/java/org/apache/rocketmq/store/dledger/MessageStoreTestBase.java b/store/src/test/java/org/apache/rocketmq/store/dledger/MessageStoreTestBase.java
index 2da2fb7a2d5259a3fbbf5f4547607bee90e866f9..5b0ca347579dc7e391dab25645d5a87620a89269 100644
--- a/store/src/test/java/org/apache/rocketmq/store/dledger/MessageStoreTestBase.java
+++ b/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);