You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by du...@apache.org on 2019/08/22 07:39:00 UTC
[rocketmq] 01/01: Revert "[RIP-15]Add Ipv6 support for RocketMQ
(#1352)"
This is an automated email from the ASF dual-hosted git repository.
duhengforever pushed a commit to branch revert-1352-IPV6
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
commit eca166a5f7a5dd62998516c140f621a0cc8e312f
Author: Heng Du <du...@gmail.com>
AuthorDate: Thu Aug 22 15:38:54 2019 +0800
Revert "[RIP-15]Add Ipv6 support for RocketMQ (#1352)"
This reverts commit c7c9601d36d0179ae69592e541764b04130cf790.
---
CONTRIBUTING.md | 5 +-
README.md | 20 ---
acl/pom.xml | 6 +-
.../org/apache/rocketmq/acl/common/AclUtils.java | 164 ++------------------
.../rocketmq/acl/plain/PlainAccessValidator.java | 2 +-
.../acl/plain/RemoteAddressStrategyFactory.java | 133 ++++------------
.../apache/rocketmq/acl/common/AclUtilsTest.java | 99 ++----------
.../acl/plain/RemoteAddressStrategyTest.java | 147 +-----------------
broker/pom.xml | 2 +-
.../broker/processor/PullMessageProcessor.java | 21 +--
client/pom.xml | 2 +-
.../java/org/apache/rocketmq/client/MQAdmin.java | 2 +-
.../impl/consumer/DefaultMQPullConsumerImpl.java | 38 ++---
.../client/impl/consumer/PullAPIWrapper.java | 2 +-
common/pom.xml | 6 +-
.../java/org/apache/rocketmq/common/MQVersion.java | 2 +-
.../java/org/apache/rocketmq/common/MixAll.java | 6 +-
.../java/org/apache/rocketmq/common/UtilAll.java | 59 +------
.../common/message/MessageClientIDSetter.java | 32 ++--
.../rocketmq/common/message/MessageDecoder.java | 92 ++++-------
.../apache/rocketmq/common/message/MessageExt.java | 22 +--
.../rocketmq/common/sysflag/MessageSysFlag.java | 3 -
.../rocketmq/common/sysflag/PullSysFlag.java | 2 +-
.../org/apache/rocketmq/common/UtilAllTest.java | 9 --
.../common/message/MessageClientIDSetterTest.java | 47 ------
.../common/message/MessageDecoderTest.java | 169 ---------------------
.../common/protocol/route/TopicRouteDataTest.java | 2 +-
distribution/benchmark/runclass.sh | 34 +----
distribution/bin/mqadmin.xml | 43 ++++++
distribution/bin/mqbroker.xml | 45 ++++++
distribution/bin/mqnamesrv.xml | 45 ++++++
distribution/bin/runbroker.sh | 30 +---
distribution/bin/runserver.sh | 33 +---
distribution/bin/tools.cmd | 2 +-
distribution/bin/tools.sh | 2 +-
distribution/pom.xml | 2 +-
docs/cn/README.md | 4 +-
docs/cn/best_practice.md | 1 +
docs/cn/concept.md | 7 +-
docs/cn/features.md | 2 +-
example/pom.xml | 6 +-
filter/pom.xml | 2 +-
logappender/pom.xml | 2 +-
logging/pom.xml | 2 +-
namesrv/pom.xml | 2 +-
openmessaging/pom.xml | 2 +-
pom.xml | 14 +-
remoting/pom.xml | 2 +-
.../rocketmq/remoting/common/RemotingHelper.java | 19 ++-
.../rocketmq/remoting/common/RemotingUtil.java | 7 +-
srvutil/pom.xml | 2 +-
store/pom.xml | 2 +-
.../java/org/apache/rocketmq/store/CommitLog.java | 146 +++++-------------
.../apache/rocketmq/store/DefaultMessageStore.java | 10 +-
.../rocketmq/store/config/MessageStoreConfig.java | 2 +-
.../rocketmq/store/dledger/DLedgerCommitLog.java | 67 ++++----
.../org/apache/rocketmq/store/ha/HAService.java | 2 +-
.../apache/rocketmq/store/AppendCallbackTest.java | 97 ------------
.../apache/rocketmq/store/BatchPutMessageTest.java | 75 ---------
.../apache/rocketmq/store/ConsumeQueueTest.java | 29 +---
.../rocketmq/store/DefaultMessageStoreTest.java | 87 +++--------
.../org/apache/rocketmq/store/StoreTestBase.java | 30 +---
.../store/dledger/DLedgerCommitlogTest.java | 37 +----
.../store/dledger/MessageStoreTestBase.java | 3 +-
test/pom.xml | 2 +-
tools/pom.xml | 2 +-
66 files changed, 431 insertions(+), 1564 deletions(-)
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 952ad73..b7b1334 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -10,7 +10,7 @@ Nor is code the only way to contribute to the project. We strongly value documen
To submit a change for inclusion, please do the following:
#### If the change is non-trivial please include some unit tests that cover the new functionality.
-#### If you are introducing a completely new feature or API it is a good idea to start a [RIP](https://github.com/apache/rocketmq/wiki/RocketMQ-Improvement-Proposal) and get consensus on the basic design first.
+#### If you are introducing a completely new feature or API it is a good idea to start a wiki and get consensus on the basic design first.
#### It is our job to follow up on patches in a timely fashion. Nag us if we aren't doing our job (sometimes we drop things).
## Becoming a Committer
@@ -19,8 +19,9 @@ We are always interested in adding new contributors. What we look for are series
Nowadays,we have several important contribution points:
#### Wiki & JavaDoc
+#### RocketMQ Console
#### RocketMQ SDK(C++\.Net\Php\Python\Go\Node.js)
-#### RocketMQ Connectors
+#### RocketMQ MySQL(Oracle\PostgreSQL\Redis\MongoDB\HBase\MSSQL) Replicator
##### Prerequisite
If you want to contribute the above listing points, you must abide our some prerequisites:
diff --git a/README.md b/README.md
index 1c17c7e..98d97a0 100644
--- a/README.md
+++ b/README.md
@@ -51,23 +51,3 @@ We always welcome new contributions, whether for trivial cleanups, [big new feat
[Apache License, Version 2.0](http://www.apache.org/licenses/LICENSE-2.0.html) Copyright (C) Apache Software Foundation
-----------
-## Export Control Notice
-This distribution includes cryptographic software. The country in which you currently reside may have
-restrictions on the import, possession, use, and/or re-export to another country, of encryption software.
-BEFORE using any encryption software, please check your country's laws, regulations and policies concerning
-the import, possession, or use, and re-export of encryption software, to see if this is permitted. See
-<http://www.wassenaar.org/> for more information.
-
-The U.S. Government Department of Commerce, Bureau of Industry and Security (BIS), has classified this
-software as Export Commodity Control Number (ECCN) 5D002.C.1, which includes information security software
-using or performing cryptographic functions with asymmetric algorithms. The form and manner of this Apache
-Software Foundation distribution makes it eligible for export under the License Exception ENC Technology
-Software Unrestricted (TSU) exception (see the BIS Export Administration Regulations, Section 740.13) for
-both object code and source code.
-
-The following provides more details on the included cryptographic software:
-
-This software uses Apache Commons Crypto (https://commons.apache.org/proper/commons-crypto/) to
-support authentication, and encryption and decryption of data sent across the network between
-services.
diff --git a/acl/pom.xml b/acl/pom.xml
index ec5ab04..923acfe 100644
--- a/acl/pom.xml
+++ b/acl/pom.xml
@@ -13,7 +13,7 @@
<parent>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-all</artifactId>
- <version>4.6.0-SNAPSHOT</version>
+ <version>4.5.2-SNAPSHOT</version>
</parent>
<artifactId>rocketmq-acl</artifactId>
<name>rocketmq-acl ${project.version}</name>
@@ -67,10 +67,6 @@
<artifactId>logback-core</artifactId>
<scope>test</scope>
</dependency>
- <dependency>
- <groupId>commons-validator</groupId>
- <artifactId>commons-validator</artifactId>
- </dependency>
</dependencies>
</project>
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 8973320..20e1cfa 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,7 +23,6 @@ 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;
@@ -70,75 +69,24 @@ 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 partialAddress) {
- String[] parAddStrArray = StringUtils.split(partialAddress.substring(1, partialAddress.length() - 1), ",");
+ public static String[] getAddreeStrArray(String netaddress, String four) {
+ String[] fourStrArray = StringUtils.split(four.substring(1, four.length() - 1), ",");
String address = netaddress.substring(0, netaddress.indexOf("{"));
- String[] addreeStrArray = new String[parAddStrArray.length];
- for (int i = 0; i < parAddStrArray.length; i++) {
- addreeStrArray[i] = address + parAddStrArray[i];
+ String[] addreeStrArray = new String[fourStrArray.length];
+ for (int i = 0; i < fourStrArray.length; i++) {
+ addreeStrArray[i] = address + fourStrArray[i];
}
return addreeStrArray;
}
- 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, ".");
+ public static boolean isScope(String num, int index) {
+ String[] strArray = StringUtils.split(num, ".");
if (strArray.length != 4) {
return false;
}
@@ -159,10 +107,6 @@ 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()));
}
@@ -175,7 +119,7 @@ public class AclUtils {
return asterisk.indexOf('*') > -1;
}
- public static boolean isComma(String colon) {
+ public static boolean isColon(String colon) {
return colon.indexOf(',') > -1;
}
@@ -184,88 +128,6 @@ public class AclUtils {
}
- public static boolean isIPv6Scope(String[] num, int index) {
- for (int i = 0; i < index; i++) {
- int value;
- try {
- value = Integer.parseInt(num[i], 16);
- } catch (NumberFormatException e) {
- return false;
- }
- if (!isIPv6Scope(value)) {
- return false;
- }
- }
- return true;
- }
-
- public static boolean isIPv6Scope(int num) {
- int min = Integer.parseInt("0", 16);
- int max = Integer.parseInt("ffff", 16);
- return num >= min && num <= max;
- }
-
- public static String expandIP(String netaddress, int part) {
- boolean compress = false;
- int compressIndex = -1;
- String[] strArray = StringUtils.split(netaddress, ":");
- ArrayList<Integer> indexes = new ArrayList<>();
- for (int i = 0; i < netaddress.length(); i++) {
- if (netaddress.charAt(i) == ':') {
- if (indexes.size() > 0 && i - indexes.get(indexes.size() - 1) == 1) {
- compressIndex = i;
- compress = true;
- }
- indexes.add(i);
- }
- }
-
- for (int i = 0; i < strArray.length; i++) {
- if (strArray[i].length() < 4) {
- strArray[i] = "0000".substring(0, 4 - strArray[i].length()) + strArray[i];
- }
- }
-
- StringBuilder sb = new StringBuilder();
- if (compress) {
- int pos = indexes.indexOf(compressIndex);
- int index = 0;
- if (!netaddress.startsWith(":")) {
- for (int i = 0; i < pos; i++) {
- sb.append(strArray[index]).append(":");
- index += 1;
- }
- }
- int zeroNum = part - strArray.length;
- if (netaddress.endsWith(":")) {
- for (int i = 0; i < zeroNum; i++) {
- sb.append("0000");
- if (i != zeroNum - 1) {
- sb.append(":");
- }
- }
- } else {
- for (int i = 0; i < zeroNum; i++) {
- sb.append("0000").append(":");
- }
- for (int i = index; i < strArray.length; i++) {
- sb.append(strArray[i]);
- if (i != strArray.length - 1) {
- sb.append(":");
- }
- }
- }
- } else {
- for (int i = 0; i < strArray.length; i++) {
- sb.append(strArray[i]);
- if (i != strArray.length - 1) {
- sb.append(":");
- }
- }
- }
- return sb.toString().toUpperCase();
- }
-
public static <T> T getYamlDataObject(String path, Class<T> clazz) {
Yaml yaml = new Yaml();
FileInputStream fis = null;
@@ -286,7 +148,7 @@ public class AclUtils {
}
}
- public static boolean writeDataObject(String path, Map<String, Object> dataMap) {
+ public static boolean writeDataObject(String path, Map<String,Object> dataMap) {
Yaml yaml = new Yaml();
PrintWriter pw = null;
try {
@@ -310,15 +172,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);
@@ -327,7 +189,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 0d7ddb3..c8ce239 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
@@ -50,7 +50,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.substring(0, remoteAddr.lastIndexOf(':')));
+ accessResource.setWhiteRemoteAddress(remoteAddr.split(":")[0]);
} 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 6931eb7..cc2dcee 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,7 +19,6 @@ 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;
@@ -42,26 +41,17 @@ public class RemoteAddressStrategyFactory {
if (StringUtils.isBlank(remoteAddr)) {
return BLANK_NET_ADDRESS_STRATEGY;
}
- if ("*".equals(remoteAddr) || "*.*.*.*".equals(remoteAddr) || "*:*:*:*:*:*:*:*".equals(remoteAddr)) {
+ if ("*".equals(remoteAddr) || "*.*.*.*".equals(remoteAddr)) {
return NULL_NET_ADDRESS_STRATEGY;
}
if (remoteAddr.endsWith("}")) {
- 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));
+ String[] strArray = StringUtils.split(remoteAddr, ".");
+ String four = strArray[3];
+ if (!four.startsWith("{")) {
+ throw new AclException(String.format("MultipleRemoteAddressStrategy netaddress examine scope Exception netaddress", remoteAddr));
}
- } else if (AclUtils.isComma(remoteAddr)) {
+ return new MultipleRemoteAddressStrategy(AclUtils.getAddreeStrArray(remoteAddr, four));
+ } else if (AclUtils.isColon(remoteAddr)) {
return new MultipleRemoteAddressStrategy(StringUtils.split(remoteAddr, ","));
} else if (AclUtils.isAsterisk(remoteAddr) || AclUtils.isMinus(remoteAddr)) {
return new RangeRemoteAddressStrategy(remoteAddr);
@@ -91,26 +81,15 @@ public class RemoteAddressStrategyFactory {
private final Set<String> multipleSet = new HashSet<>();
public MultipleRemoteAddressStrategy(String[] strArray) {
- InetAddressValidator validator = InetAddressValidator.getInstance();
for (String netaddress : strArray) {
- 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));
- }
+ AclUtils.verify(netaddress, 4);
+ multipleSet.add(netaddress);
}
}
@Override
public boolean match(PlainAccessResource plainAccessResource) {
- InetAddressValidator validator = InetAddressValidator.getInstance();
- String whiteRemoteAddress = plainAccessResource.getWhiteRemoteAddress();
- if (validator.isValidInet6Address(whiteRemoteAddress)) {
- whiteRemoteAddress = AclUtils.expandIP(whiteRemoteAddress, 8);
- }
- return multipleSet.contains(whiteRemoteAddress);
+ return multipleSet.contains(plainAccessResource.getWhiteRemoteAddress());
}
}
@@ -121,16 +100,12 @@ public class RemoteAddressStrategyFactory {
public OneRemoteAddressStrategy(String netaddress) {
this.netaddress = netaddress;
- InetAddressValidator validator = InetAddressValidator.getInstance();
- if (!(validator.isValidInet4Address(netaddress) || validator.isValidInet6Address(netaddress))) {
- throw new AclException(String.format("Netaddress examine Exception netaddress is %s", netaddress));
- }
+ AclUtils.verify(netaddress, 4);
}
@Override
public boolean match(PlainAccessResource plainAccessResource) {
- String writeRemoteAddress = AclUtils.expandIP(plainAccessResource.getWhiteRemoteAddress(), 8).toUpperCase();
- return AclUtils.expandIP(netaddress, 8).toUpperCase().equals(writeRemoteAddress);
+ return netaddress.equals(plainAccessResource.getWhiteRemoteAddress());
}
}
@@ -146,29 +121,14 @@ public class RemoteAddressStrategyFactory {
private int index;
public RangeRemoteAddressStrategy(String remoteAddr) {
-// 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();
+ 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(".");
}
+ this.head = sb.toString();
}
}
@@ -192,27 +152,6 @@ 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;
@@ -221,33 +160,21 @@ public class RemoteAddressStrategyFactory {
@Override
public boolean match(PlainAccessResource plainAccessResource) {
String netAddress = plainAccessResource.getWhiteRemoteAddress();
- 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;
- }
+ 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('.'));
}
- } 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;
- }
+ Integer address = Integer.valueOf(value);
+ 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 5705b74..5b2627d 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,35 +46,20 @@ public class AclUtilsTest {
addressList.add("1.1.1.3");
addressList.add("1.1.1.4");
Assert.assertEquals(newAddressList, addressList);
-
-// IPv6 test
- String ipv6Address = "1:ac41:9987::bb22:666:{1,2,3,4}";
- String[] ipv6AddressArray = AclUtils.getAddreeStrArray(ipv6Address, "{1,2,3,4}");
- List<String> newIPv6AddressList = new ArrayList<>();
- for (String a : ipv6AddressArray) {
- newIPv6AddressList.add(a);
- }
-
- List<String> ipv6AddressList = new ArrayList<>();
- ipv6AddressList.add("1:ac41:9987::bb22:666:1");
- ipv6AddressList.add("1:ac41:9987::bb22:666:2");
- ipv6AddressList.add("1:ac41:9987::bb22:666:3");
- ipv6AddressList.add("1:ac41:9987::bb22:666:4");
- Assert.assertEquals(newIPv6AddressList, ipv6AddressList);
}
@Test
public void isScopeStringArray() {
- String address = "12";
+ String adderss = "12";
for (int i = 0; i < 6; i++) {
- boolean isScope = AclUtils.isScope(address, 4);
+ boolean isScope = AclUtils.isScope(adderss, 4);
if (i == 3) {
Assert.assertTrue(isScope);
} else {
Assert.assertFalse(isScope);
}
- address = address + ".12";
+ adderss = adderss + ".12";
}
}
@@ -92,25 +77,6 @@ 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
@@ -136,18 +102,6 @@ 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
@@ -161,10 +115,10 @@ public class AclUtilsTest {
@Test
public void isColonTest() {
- boolean isColon = AclUtils.isComma(",");
+ boolean isColon = AclUtils.isColon(",");
Assert.assertTrue(isColon);
- isColon = AclUtils.isComma("-");
+ isColon = AclUtils.isColon("-");
Assert.assertFalse(isColon);
}
@@ -177,36 +131,6 @@ 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() {
@@ -216,7 +140,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);
@@ -229,7 +153,7 @@ public class AclUtilsTest {
List<String> globalWhiteRemoteAddrs = new ArrayList<String>();
globalWhiteRemoteAddrs.add("10.10.103.*");
globalWhiteRemoteAddrs.add("192.168.0.*");
- aclYamlMap.put("globalWhiteRemoteAddrs", globalWhiteRemoteAddrs);
+ aclYamlMap.put("globalWhiteRemoteAddrs",globalWhiteRemoteAddrs);
// For accounts element in acl yaml config file
List<Map<String, Object>> accounts = new ArrayList<Map<String, Object>>();
@@ -242,14 +166,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);
@@ -262,7 +186,7 @@ public class AclUtilsTest {
List<String> globalWhiteRemoteAddrs = new ArrayList<String>();
globalWhiteRemoteAddrs.add("10.10.103.*");
globalWhiteRemoteAddrs.add("192.168.0.*");
- aclYamlMap.put("globalWhiteRemoteAddrs", globalWhiteRemoteAddrs);
+ aclYamlMap.put("globalWhiteRemoteAddrs",globalWhiteRemoteAddrs);
// Write file to yaml file
AclUtils.writeDataObject(targetFileName, aclYamlMap);
@@ -277,7 +201,7 @@ public class AclUtilsTest {
Map<String, Object> readableMap = AclUtils.getYamlDataObject(targetFileName, Map.class);
List<String> updatedGlobalWhiteRemoteAddrs = (List<String>) readableMap.get("globalWhiteRemoteAddrs");
- Assert.assertEquals("192.168.1.2", updatedGlobalWhiteRemoteAddrs.get(0));
+ Assert.assertEquals("192.168.1.2",updatedGlobalWhiteRemoteAddrs.get(0));
transport.delete();
}
@@ -311,4 +235,5 @@ 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 8998dd9..2c2e76b 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,35 +71,6 @@ 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)
@@ -109,8 +80,6 @@ public class RemoteAddressStrategyTest {
remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
plainAccessResource.setWhiteRemoteAddress("256.0.0.1");
remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
- plainAccessResource.setWhiteRemoteAddress("::1ggg");
- remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
}
@Test
@@ -125,7 +94,6 @@ public class RemoteAddressStrategyTest {
Assert.assertFalse(isMatch);
}
- @Test
public void oneNetaddressStrategyTest() {
PlainAccessResource plainAccessResource = new PlainAccessResource();
plainAccessResource.setWhiteRemoteAddress("127.0.0.1");
@@ -141,26 +109,6 @@ 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
@@ -174,21 +122,6 @@ 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)
@@ -196,8 +129,6 @@ 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) {
@@ -224,30 +155,6 @@ 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.";
@@ -255,7 +162,6 @@ 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);
@@ -263,40 +169,14 @@ 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);
- rangeNetaddressStrategyTest(remoteAddressStrategy, head, 0, 255, true);
-
+ rangeNetaddressStrategyThirdlyTest(remoteAddressStrategy, head, 1, 200);
+
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,
@@ -326,25 +206,6 @@ 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/pom.xml b/broker/pom.xml
index 3852baa..fae5517 100644
--- a/broker/pom.xml
+++ b/broker/pom.xml
@@ -13,7 +13,7 @@
<parent>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-all</artifactId>
- <version>4.6.0-SNAPSHOT</version>
+ <version>4.5.2-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
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 b4f6daa..10c0112 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,6 +41,8 @@ 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;
@@ -50,10 +52,7 @@ 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;
@@ -495,21 +494,7 @@ public class PullMessageProcessor implements NettyRequestProcessor {
for (ByteBuffer bb : messageBufferList) {
byteBuffer.put(bb);
- 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);
+ storeTimestamp = bb.getLong(MessageDecoder.MESSAGE_STORE_TIMESTAMP_POSTION);
}
} finally {
getMessageResult.release();
diff --git a/client/pom.xml b/client/pom.xml
index c2ec6fa..5604fad 100644
--- a/client/pom.xml
+++ b/client/pom.xml
@@ -19,7 +19,7 @@
<parent>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-all</artifactId>
- <version>4.6.0-SNAPSHOT</version>
+ <version>4.5.2-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/client/src/main/java/org/apache/rocketmq/client/MQAdmin.java b/client/src/main/java/org/apache/rocketmq/client/MQAdmin.java
index 63b2d14..019414b 100644
--- a/client/src/main/java/org/apache/rocketmq/client/MQAdmin.java
+++ b/client/src/main/java/org/apache/rocketmq/client/MQAdmin.java
@@ -82,7 +82,7 @@ public interface MQAdmin {
long earliestMsgStoreTime(final MessageQueue mq) throws MQClientException;
/**
- * Query message according to message id
+ * Query message according tto message id
*
* @param offsetMsgId message id
* @return message
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPullConsumerImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPullConsumerImpl.java
index 598dc94..8aff14b 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPullConsumerImpl.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPullConsumerImpl.java
@@ -94,13 +94,13 @@ public class DefaultMQPullConsumerImpl implements MQConsumerInner {
}
public void createTopic(String key, String newTopic, int queueNum, int topicSysFlag) throws MQClientException {
- this.isRunning();
+ this.makeSureStateOK();
this.mQClientFactory.getMQAdminImpl().createTopic(key, newTopic, queueNum, topicSysFlag);
}
- private void isRunning() throws MQClientException {
+ private void makeSureStateOK() throws MQClientException {
if (this.serviceState != ServiceState.RUNNING) {
- throw new MQClientException("The consumer is not in running status, "
+ throw new MQClientException("The consumer service state not OK, "
+ this.serviceState
+ FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
null);
@@ -108,12 +108,12 @@ public class DefaultMQPullConsumerImpl implements MQConsumerInner {
}
public long fetchConsumeOffset(MessageQueue mq, boolean fromStore) throws MQClientException {
- this.isRunning();
+ this.makeSureStateOK();
return this.offsetStore.readOffset(mq, fromStore ? ReadOffsetType.READ_FROM_STORE : ReadOffsetType.MEMORY_FIRST_THEN_STORE);
}
public Set<MessageQueue> fetchMessageQueuesInBalance(String topic) throws MQClientException {
- this.isRunning();
+ this.makeSureStateOK();
if (null == topic) {
throw new IllegalArgumentException("topic is null");
}
@@ -130,12 +130,12 @@ public class DefaultMQPullConsumerImpl implements MQConsumerInner {
}
public List<MessageQueue> fetchPublishMessageQueues(String topic) throws MQClientException {
- this.isRunning();
+ this.makeSureStateOK();
return this.mQClientFactory.getMQAdminImpl().fetchPublishMessageQueues(topic);
}
public Set<MessageQueue> fetchSubscribeMessageQueues(String topic) throws MQClientException {
- this.isRunning();
+ this.makeSureStateOK();
// check if has info in memory, otherwise invoke api.
Set<MessageQueue> result = this.rebalanceImpl.getTopicSubscribeInfoTable().get(topic);
if (null == result) {
@@ -156,17 +156,17 @@ public class DefaultMQPullConsumerImpl implements MQConsumerInner {
}
public long earliestMsgStoreTime(MessageQueue mq) throws MQClientException {
- this.isRunning();
+ this.makeSureStateOK();
return this.mQClientFactory.getMQAdminImpl().earliestMsgStoreTime(mq);
}
public long maxOffset(MessageQueue mq) throws MQClientException {
- this.isRunning();
+ this.makeSureStateOK();
return this.mQClientFactory.getMQAdminImpl().maxOffset(mq);
}
public long minOffset(MessageQueue mq) throws MQClientException {
- this.isRunning();
+ this.makeSureStateOK();
return this.mQClientFactory.getMQAdminImpl().minOffset(mq);
}
@@ -225,7 +225,7 @@ public class DefaultMQPullConsumerImpl implements MQConsumerInner {
private PullResult pullSyncImpl(MessageQueue mq, SubscriptionData subscriptionData, long offset, int maxNums, boolean block,
long timeout)
throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
- this.isRunning();
+ this.makeSureStateOK();
if (null == mq) {
throw new MQClientException("mq is null", null);
@@ -261,7 +261,7 @@ public class DefaultMQPullConsumerImpl implements MQConsumerInner {
null
);
this.pullAPIWrapper.processPullResult(mq, pullResult, subscriptionData);
- //If namespace is not null , reset Topic without namespace.
+ //If namespace not null , reset Topic without namespace.
this.resetTopic(pullResult.getMsgFoundList());
if (!this.consumeMessageHookList.isEmpty()) {
ConsumeMessageContext consumeMessageContext = null;
@@ -383,7 +383,7 @@ public class DefaultMQPullConsumerImpl implements MQConsumerInner {
@Override
public void persistConsumerOffset() {
try {
- this.isRunning();
+ this.makeSureStateOK();
Set<MessageQueue> mqs = new HashSet<MessageQueue>();
Set<MessageQueue> allocateMq = this.rebalanceImpl.getProcessQueueTable().keySet();
mqs.addAll(allocateMq);
@@ -466,7 +466,7 @@ public class DefaultMQPullConsumerImpl implements MQConsumerInner {
final PullCallback pullCallback,
final boolean block,
final long timeout) throws MQClientException, RemotingException, InterruptedException {
- this.isRunning();
+ this.makeSureStateOK();
if (null == mq) {
throw new MQClientException("mq is null", null);
@@ -543,18 +543,18 @@ public class DefaultMQPullConsumerImpl implements MQConsumerInner {
public QueryResult queryMessage(String topic, String key, int maxNum, long begin, long end)
throws MQClientException, InterruptedException {
- this.isRunning();
+ this.makeSureStateOK();
return this.mQClientFactory.getMQAdminImpl().queryMessage(topic, key, maxNum, begin, end);
}
public MessageExt queryMessageByUniqKey(String topic, String uniqKey)
throws MQClientException, InterruptedException {
- this.isRunning();
+ this.makeSureStateOK();
return this.mQClientFactory.getMQAdminImpl().queryMessageByUniqKey(topic, uniqKey);
}
public long searchOffset(MessageQueue mq, long timestamp) throws MQClientException {
- this.isRunning();
+ this.makeSureStateOK();
return this.mQClientFactory.getMQAdminImpl().searchOffset(mq, timestamp);
}
@@ -748,13 +748,13 @@ public class DefaultMQPullConsumerImpl implements MQConsumerInner {
}
public void updateConsumeOffset(MessageQueue mq, long offset) throws MQClientException {
- this.isRunning();
+ this.makeSureStateOK();
this.offsetStore.updateOffset(mq, offset, false);
}
public MessageExt viewMessage(String msgId)
throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
- this.isRunning();
+ this.makeSureStateOK();
return this.mQClientFactory.getMQAdminImpl().viewMessage(msgId);
}
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/PullAPIWrapper.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/PullAPIWrapper.java
index 1917d27..1d2d24f 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/PullAPIWrapper.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/PullAPIWrapper.java
@@ -97,7 +97,7 @@ public class PullAPIWrapper {
for (MessageExt msg : msgListFilterAgain) {
String traFlag = msg.getProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED);
- if (Boolean.parseBoolean(traFlag)) {
+ if (traFlag != null && Boolean.parseBoolean(traFlag)) {
msg.setTransactionId(msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX));
}
MessageAccessor.putProperty(msg, MessageConst.PROPERTY_MIN_OFFSET,
diff --git a/common/pom.xml b/common/pom.xml
index 11df833..b778831 100644
--- a/common/pom.xml
+++ b/common/pom.xml
@@ -19,7 +19,7 @@
<parent>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-all</artifactId>
- <version>4.6.0-SNAPSHOT</version>
+ <version>4.5.2-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
@@ -41,9 +41,5 @@
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
</dependency>
- <dependency>
- <groupId>commons-validator</groupId>
- <artifactId>commons-validator</artifactId>
- </dependency>
</dependencies>
</project>
diff --git a/common/src/main/java/org/apache/rocketmq/common/MQVersion.java b/common/src/main/java/org/apache/rocketmq/common/MQVersion.java
index 47cfb6e..71ca0d6 100644
--- a/common/src/main/java/org/apache/rocketmq/common/MQVersion.java
+++ b/common/src/main/java/org/apache/rocketmq/common/MQVersion.java
@@ -18,7 +18,7 @@ package org.apache.rocketmq.common;
public class MQVersion {
- public static final int CURRENT_VERSION = Version.V4_5_2.ordinal();
+ public static final int CURRENT_VERSION = Version.V4_5_1.ordinal();
public static String getVersionDesc(int value) {
int length = Version.values().length;
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 bba8b36..0af65df 100644
--- a/common/src/main/java/org/apache/rocketmq/common/MixAll.java
+++ b/common/src/main/java/org/apache/rocketmq/common/MixAll.java
@@ -124,10 +124,8 @@ public class MixAll {
public static String brokerVIPChannel(final boolean isChange, final String brokerAddr) {
if (isChange) {
- int split = brokerAddr.lastIndexOf(":");
- String ip = brokerAddr.substring(0, split);
- String port = brokerAddr.substring(split + 1);
- String brokerAddrNew = ip + ":" + (Integer.parseInt(port) - 2);
+ String[] ipAndPort = brokerAddr.split(":");
+ String brokerAddrNew = ipAndPort[0] + ":" + (Integer.parseInt(ipAndPort[1]) - 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 85a4cde..68d93f3 100644
--- a/common/src/main/java/org/apache/rocketmq/common/UtilAll.java
+++ b/common/src/main/java/org/apache/rocketmq/common/UtilAll.java
@@ -23,7 +23,6 @@ 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;
@@ -40,7 +39,6 @@ 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;
@@ -440,18 +438,6 @@ 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");
@@ -488,15 +474,6 @@ 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;
@@ -506,25 +483,6 @@ 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();
@@ -546,17 +504,6 @@ 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;
- }
- }
- }
}
}
}
@@ -585,12 +532,12 @@ public class UtilAll {
}
}
- public static String List2String(List<String> list, String splitor) {
+ public static String List2String(List<String> list,String splitor) {
if (list == null || list.size() == 0) {
return null;
}
StringBuffer str = new StringBuffer();
- for (int i = 0; i < list.size(); i++) {
+ for (int i = 0;i < list.size();i++) {
str.append(list.get(i));
if (i == list.size() - 1) {
continue;
@@ -600,7 +547,7 @@ public class UtilAll {
return str.toString();
}
- public static List<String> String2List(String str, String splitor) {
+ public static List<String> String2List(String str,String splitor) {
if (StringUtils.isEmpty(str)) {
return null;
}
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 df11556..d0b202e 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,19 +31,17 @@ public class MessageClientIDSetter {
private static long nextStartTime;
static {
- byte[] ip;
+ LEN = 4 + 2 + 4 + 4 + 2;
+ ByteBuffer tempBuffer = ByteBuffer.allocate(10);
+ tempBuffer.position(2);
+ tempBuffer.putInt(UtilAll.getPid());
+ tempBuffer.position(0);
try {
- ip = UtilAll.getIP();
+ tempBuffer.put(UtilAll.getIP());
} catch (Exception e) {
- ip = createFakeIP();
+ tempBuffer.put(createFakeIP());
}
- 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.position(6);
tempBuffer.putInt(MessageClientIDSetter.class.getClassLoader().hashCode());
FIX_STRING = UtilAll.bytes2string(tempBuffer.array());
setStartTime(System.currentTimeMillis());
@@ -66,12 +64,11 @@ 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, ipLength + 2 + 4, 4);
+ buf.put(bytes, 10, 4);
buf.position(0);
long spanMS = buf.getLong();
Calendar cal = Calendar.getInstance();
@@ -92,18 +89,13 @@ public class MessageClientIDSetter {
public static String getIPStrFromID(String msgID) {
byte[] ipBytes = getIPFromID(msgID);
- if (ipBytes.length == 16) {
- return UtilAll.ipToIPv6Str(ipBytes);
- } else {
- return UtilAll.ipToIPv4Str(ipBytes);
- }
+ return UtilAll.ipToIPv4Str(ipBytes);
}
public static byte[] getIPFromID(String msgID) {
+ byte[] result = new byte[4];
byte[] bytes = UtilAll.string2bytes(msgID);
- int ipLength = bytes.length == 28 ? 16 : 4;
- byte[] result = new byte[ipLength];
- System.arraycopy(bytes, 0, result, 0, ipLength);
+ System.arraycopy(bytes, 0, result, 0, 4);
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 54d5e6b..af0b638 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,7 +16,9 @@
*/
package org.apache.rocketmq.common.message;
-import java.net.Inet4Address;
+import org.apache.rocketmq.common.UtilAll;
+import org.apache.rocketmq.common.sysflag.MessageSysFlag;
+
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
@@ -27,41 +29,37 @@ 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 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 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 String createMessageId(final ByteBuffer input, final ByteBuffer addr, final long offset) {
input.flip();
- int msgIDLength = addr.limit() == 8 ? 16 : 28;
- input.limit(msgIDLength);
+ input.limit(MessageDecoder.MSG_ID_LENGTH);
input.put(addr);
input.putLong(offset);
@@ -70,9 +68,8 @@ 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);
@@ -83,16 +80,15 @@ 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, ipLength));
- byte[] port = UtilAll.string2bytes(msgId.substring(ipLength, ipLength + 8));
+ byte[] ip = UtilAll.string2bytes(msgId.substring(0, 8));
+ byte[] port = UtilAll.string2bytes(msgId.substring(8, 16));
ByteBuffer bb = ByteBuffer.wrap(port);
int portInt = bb.getInt(0);
address = new InetSocketAddress(InetAddress.getByAddress(ip), portInt);
// offset
- byte[] data = UtilAll.string2bytes(msgId.substring(ipLength + 8, ipLength + 8 + 16));
+ byte[] data = UtilAll.string2bytes(msgId.substring(16, 32));
bb = ByteBuffer.wrap(data);
offset = bb.getLong(0);
@@ -105,24 +101,7 @@ public class MessageDecoder {
* @param byteBuffer msg commit log buffer.
*/
public static Map<String, String> decodeProperties(java.nio.ByteBuffer byteBuffer) {
- 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);
+ int topicLengthPosition = BODY_SIZE_POSITION + 4 + byteBuffer.getInt(BODY_SIZE_POSITION);
byte topicLength = byteBuffer.get(topicLengthPosition);
@@ -160,8 +139,6 @@ 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);
@@ -181,9 +158,9 @@ public class MessageDecoder {
+ 8 // 7 PHYSICALOFFSET
+ 4 // 8 SYSFLAG
+ 8 // 9 BORNTIMESTAMP
- + bornhostLength // 10 BORNHOST
+ + 8 // 10 BORNHOST
+ 8 // 11 STORETIMESTAMP
- + storehostAddressLength // 12 STOREHOSTADDRESS
+ + 8 // 12 STOREHOSTADDRESS
+ 4 // 13 RECONSUMETIMES
+ 8 // 14 Prepared Transaction Offset
+ 4 + bodyLength // 14 BODY
@@ -314,9 +291,8 @@ public class MessageDecoder {
msgExt.setBornTimestamp(bornTimeStamp);
// 10 BORNHOST
- int bornhostIPLength = (sysFlag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 4 : 16;
- byte[] bornHost = new byte[bornhostIPLength];
- byteBuffer.get(bornHost, 0, bornhostIPLength);
+ byte[] bornHost = new byte[4];
+ byteBuffer.get(bornHost, 0, 4);
int port = byteBuffer.getInt();
msgExt.setBornHost(new InetSocketAddress(InetAddress.getByAddress(bornHost), port));
@@ -325,9 +301,8 @@ public class MessageDecoder {
msgExt.setStoreTimestamp(storeTimestamp);
// 12 STOREHOST
- int storehostIPLength = (sysFlag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 : 16;
- byte[] storeHost = new byte[storehostIPLength];
- byteBuffer.get(storeHost, 0, storehostIPLength);
+ byte[] storeHost = new byte[4];
+ byteBuffer.get(storeHost, 0, 4);
port = byteBuffer.getInt();
msgExt.setStoreHost(new InetSocketAddress(InetAddress.getByAddress(storeHost), port));
@@ -373,8 +348,7 @@ public class MessageDecoder {
msgExt.setProperties(map);
}
- int msgIDLength = storehostIPLength + 4 + 8;
- ByteBuffer byteBufferMsgId = ByteBuffer.allocate(msgIDLength);
+ ByteBuffer byteBufferMsgId = ByteBuffer.allocate(MSG_ID_LENGTH);
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 20cb057..3f77767 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,8 +16,6 @@
*/
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;
@@ -68,26 +66,14 @@ public class MessageExt extends Message {
public static ByteBuffer socketAddress2ByteBuffer(final SocketAddress socketAddress, final ByteBuffer byteBuffer) {
InetSocketAddress inetSocketAddress = (InetSocketAddress) socketAddress;
- InetAddress address = inetSocketAddress.getAddress();
- if (address instanceof Inet4Address) {
- byteBuffer.put(inetSocketAddress.getAddress().getAddress(), 0, 4);
- } else {
- byteBuffer.put(inetSocketAddress.getAddress().getAddress(), 0, 16);
- }
+ byteBuffer.put(inetSocketAddress.getAddress().getAddress(), 0, 4);
byteBuffer.putInt(inetSocketAddress.getPort());
byteBuffer.flip();
return byteBuffer;
}
public static ByteBuffer socketAddress2ByteBuffer(SocketAddress socketAddress) {
- 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);
- }
+ ByteBuffer byteBuffer = ByteBuffer.allocate(8);
return socketAddress2ByteBuffer(socketAddress, byteBuffer);
}
@@ -181,10 +167,6 @@ 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 d534571..f139770 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,8 +23,6 @@ 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;
@@ -37,5 +35,4 @@ public class MessageSysFlag {
public static int clearCompressedFlag(final int flag) {
return flag & (~COMPRESSED_FLAG);
}
-
}
diff --git a/common/src/main/java/org/apache/rocketmq/common/sysflag/PullSysFlag.java b/common/src/main/java/org/apache/rocketmq/common/sysflag/PullSysFlag.java
index d476a35..e2edca4 100644
--- a/common/src/main/java/org/apache/rocketmq/common/sysflag/PullSysFlag.java
+++ b/common/src/main/java/org/apache/rocketmq/common/sysflag/PullSysFlag.java
@@ -17,7 +17,7 @@
package org.apache.rocketmq.common.sysflag;
public class PullSysFlag {
- private final static int FLAG_COMMIT_OFFSET = 0x1;
+ private final static int FLAG_COMMIT_OFFSET = 0x1 << 0;
private final static int FLAG_SUSPEND = 0x1 << 1;
private final static int FLAG_SUBSCRIPTION = 0x1 << 2;
private final static int FLAG_CLASS_FILTER = 0x1 << 3;
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 3991367..2656452 100644
--- a/common/src/test/java/org/apache/rocketmq/common/UtilAllTest.java
+++ b/common/src/test/java/org/apache/rocketmq/common/UtilAllTest.java
@@ -98,15 +98,6 @@ 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
deleted file mode 100644
index 987f103..0000000
--- a/common/src/test/java/org/apache/rocketmq/common/message/MessageClientIDSetterTest.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-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 82ad271..d14d6b0 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,7 +25,6 @@ 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 {
@@ -78,172 +77,4 @@ public class MessageDecoderTest {
assertThat("hello").isEqualTo(properties.get("b"));
assertThat("3.14").isEqualTo(properties.get("c"));
}
-
- @Test
- public void testDecodePropertiesOnIPv6Host() {
- MessageExt messageExt = new MessageExt();
-
- messageExt.setMsgId("24084004018081003FAA1DDE2B3F898A00002A9F0000000000000CA0");
- messageExt.setBornHostV6Flag();
- messageExt.setStoreHostAddressV6Flag();
- messageExt.setTopic("abc");
- messageExt.setBody("hello!q!".getBytes());
- try {
- messageExt.setBornHost(new InetSocketAddress(InetAddress.getByName("1050:0000:0000:0000:0005:0600:300c:326b"), 0));
- } catch (UnknownHostException e) {
- e.printStackTrace();
- assertThat(Boolean.FALSE).isTrue();
- }
- messageExt.setBornTimestamp(System.currentTimeMillis());
- messageExt.setCommitLogOffset(123456);
- messageExt.setPreparedTransactionOffset(0);
- messageExt.setQueueId(0);
- messageExt.setQueueOffset(123);
- messageExt.setReconsumeTimes(0);
- try {
- messageExt.setStoreHost(new InetSocketAddress(InetAddress.getByName("::1"), 0));
- } catch (UnknownHostException e) {
- e.printStackTrace();
- assertThat(Boolean.FALSE).isTrue();
- }
-
- messageExt.putUserProperty("a", "123");
- messageExt.putUserProperty("b", "hello");
- messageExt.putUserProperty("c", "3.14");
-
- byte[] msgBytes = new byte[0];
- try {
- msgBytes = MessageDecoder.encode(messageExt, false);
- } catch (Exception e) {
- e.printStackTrace();
- assertThat(Boolean.FALSE).isTrue();
- }
-
- ByteBuffer byteBuffer = ByteBuffer.allocate(msgBytes.length);
- byteBuffer.put(msgBytes);
-
- Map<String, String> properties = MessageDecoder.decodeProperties(byteBuffer);
-
- assertThat(properties).isNotNull();
- assertThat("123").isEqualTo(properties.get("a"));
- assertThat("hello").isEqualTo(properties.get("b"));
- assertThat("3.14").isEqualTo(properties.get("c"));
- }
-
- @Test
- public void testEncodeAndDecode() {
- MessageExt messageExt = new MessageExt();
-
- messageExt.setMsgId("645100FA00002A9F000000489A3AA09E");
- messageExt.setTopic("abc");
- messageExt.setBody("hello!q!".getBytes());
- try {
- messageExt.setBornHost(new InetSocketAddress(InetAddress.getByName("127.0.0.1"), 0));
- } catch (UnknownHostException e) {
- e.printStackTrace();
- assertThat(Boolean.FALSE).isTrue();
- }
- messageExt.setBornTimestamp(System.currentTimeMillis());
- messageExt.setCommitLogOffset(123456);
- messageExt.setPreparedTransactionOffset(0);
- messageExt.setQueueId(1);
- messageExt.setQueueOffset(123);
- messageExt.setReconsumeTimes(0);
- try {
- messageExt.setStoreHost(new InetSocketAddress(InetAddress.getLocalHost(), 0));
- } catch (UnknownHostException e) {
- e.printStackTrace();
- assertThat(Boolean.FALSE).isTrue();
- }
-
- messageExt.putUserProperty("a", "123");
- messageExt.putUserProperty("b", "hello");
- messageExt.putUserProperty("c", "3.14");
-
- byte[] msgBytes = new byte[0];
- try {
- msgBytes = MessageDecoder.encode(messageExt, false);
- } catch (Exception e) {
- e.printStackTrace();
- assertThat(Boolean.FALSE).isTrue();
- }
-
- ByteBuffer byteBuffer = ByteBuffer.allocate(msgBytes.length);
- byteBuffer.put(msgBytes);
-
- byteBuffer.clear();
- MessageExt decodedMsg = MessageDecoder.decode(byteBuffer);
-
- assertThat(decodedMsg).isNotNull();
- assertThat(1).isEqualTo(decodedMsg.getQueueId());
- assertThat(123456L).isEqualTo(decodedMsg.getCommitLogOffset());
- assertThat("hello!q!".getBytes()).isEqualTo(decodedMsg.getBody());
-
- int msgIDLength = 4 + 4 + 8;
- ByteBuffer byteBufferMsgId = ByteBuffer.allocate(msgIDLength);
- String msgId = createMessageId(byteBufferMsgId, messageExt.getStoreHostBytes(), messageExt.getCommitLogOffset());
- assertThat(msgId).isEqualTo(decodedMsg.getMsgId());
-
- assertThat("abc").isEqualTo(decodedMsg.getTopic());
- }
-
- @Test
- public void testEncodeAndDecodeOnIPv6Host() {
- MessageExt messageExt = new MessageExt();
-
- messageExt.setMsgId("24084004018081003FAA1DDE2B3F898A00002A9F0000000000000CA0");
- messageExt.setBornHostV6Flag();
- messageExt.setStoreHostAddressV6Flag();
- messageExt.setTopic("abc");
- messageExt.setBody("hello!q!".getBytes());
- try {
- messageExt.setBornHost(new InetSocketAddress(InetAddress.getByName("1050:0000:0000:0000:0005:0600:300c:326b"), 0));
- } catch (UnknownHostException e) {
- e.printStackTrace();
- assertThat(Boolean.FALSE).isTrue();
- }
- messageExt.setBornTimestamp(System.currentTimeMillis());
- messageExt.setCommitLogOffset(123456);
- messageExt.setPreparedTransactionOffset(0);
- messageExt.setQueueId(1);
- messageExt.setQueueOffset(123);
- messageExt.setReconsumeTimes(0);
- try {
- messageExt.setStoreHost(new InetSocketAddress(InetAddress.getByName("::1"), 0));
- } catch (UnknownHostException e) {
- e.printStackTrace();
- assertThat(Boolean.FALSE).isTrue();
- }
-
- messageExt.putUserProperty("a", "123");
- messageExt.putUserProperty("b", "hello");
- messageExt.putUserProperty("c", "3.14");
-
- byte[] msgBytes = new byte[0];
- try {
- msgBytes = MessageDecoder.encode(messageExt, false);
- } catch (Exception e) {
- e.printStackTrace();
- assertThat(Boolean.FALSE).isTrue();
- }
-
- ByteBuffer byteBuffer = ByteBuffer.allocate(msgBytes.length);
- byteBuffer.put(msgBytes);
-
- byteBuffer.clear();
- MessageExt decodedMsg = MessageDecoder.decode(byteBuffer);
-
- assertThat(decodedMsg).isNotNull();
- assertThat(1).isEqualTo(decodedMsg.getQueueId());
- assertThat(123456L).isEqualTo(decodedMsg.getCommitLogOffset());
- assertThat("hello!q!".getBytes()).isEqualTo(decodedMsg.getBody());
- assertThat(48).isEqualTo(decodedMsg.getSysFlag());
-
- int msgIDLength = 16 + 4 + 8;
- ByteBuffer byteBufferMsgId = ByteBuffer.allocate(msgIDLength);
- String msgId = createMessageId(byteBufferMsgId, messageExt.getStoreHostBytes(), messageExt.getCommitLogOffset());
- assertThat(msgId).isEqualTo(decodedMsg.getMsgId());
-
- assertThat("abc").isEqualTo(decodedMsg.getTopic());
- }
}
diff --git a/common/src/test/java/org/apache/rocketmq/common/protocol/route/TopicRouteDataTest.java b/common/src/test/java/org/apache/rocketmq/common/protocol/route/TopicRouteDataTest.java
index 764225d..2057b74 100644
--- a/common/src/test/java/org/apache/rocketmq/common/protocol/route/TopicRouteDataTest.java
+++ b/common/src/test/java/org/apache/rocketmq/common/protocol/route/TopicRouteDataTest.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.rocketmq.common.protocol.route;
+package org.apache.rocketmq.common.protocol;
import org.apache.rocketmq.common.protocol.route.BrokerData;
diff --git a/distribution/benchmark/runclass.sh b/distribution/benchmark/runclass.sh
index 339e11a..13c58d1 100644
--- a/distribution/benchmark/runclass.sh
+++ b/distribution/benchmark/runclass.sh
@@ -24,41 +24,13 @@ fi
BASE_DIR=$(dirname $0)/..
CLASSPATH=.:${BASE_DIR}/conf:${CLASSPATH}
-# The RAMDisk initializing size in MB on Darwin OS for gc-log
-DIR_SIZE_IN_MB=600
-
-choose_gc_log_directory()
-{
- case "`uname`" in
- Darwin)
- if [ ! -d "/Volumes/RAMDisk" ]; then
- # create ram disk on Darwin systems as gc-log directory
- DEV=`hdiutil attach -nomount ram://$((2 * 1024 * DIR_SIZE_IN_MB))` > /dev/null
- diskutil eraseVolume HFS+ RAMDisk ${DEV} > /dev/null
- echo "Create RAMDisk /Volumes/RAMDisk for gc logging on Darwin OS."
- fi
- GC_LOG_DIR="/Volumes/RAMDisk"
- ;;
- *)
- # check if /dev/shm exists on other systems
- if [ -d "/dev/shm" ]; then
- GC_LOG_DIR="/dev/shm"
- else
- GC_LOG_DIR=${BASE_DIR}
- fi
- ;;
- esac
-}
-
-choose_gc_log_directory
-
JAVA_OPT="${JAVA_OPT} -server -Xms1g -Xmx1g -Xmn256m -XX:PermSize=128m -XX:MaxPermSize=320m"
JAVA_OPT="${JAVA_OPT} -XX:+UseConcMarkSweepGC -XX:+UseCMSCompactAtFullCollection -XX:CMSInitiatingOccupancyFraction=70 -XX:+CMSParallelRemarkEnabled -XX:SoftRefLRUPolicyMSPerMB=0 -XX:+CMSClassUnloadingEnabled -XX:SurvivorRatio=8 -XX:+DisableExplicitGC"
-JAVA_OPT="${JAVA_OPT} -verbose:gc -Xloggc:${GC_LOG_DIR}/rmq_run_class_gc_%p_%t.log -XX:+PrintGCDetails"
+JAVA_OPT="${JAVA_OPT} -verbose:gc -Xloggc:/dev/shm/rmq_srv_gc.log -XX:+PrintGCDetails"
JAVA_OPT="${JAVA_OPT} -XX:-OmitStackTraceInFastThrow"
JAVA_OPT="${JAVA_OPT} -Djava.ext.dirs=${BASE_DIR}/lib"
-JAVA_OPT="${JAVA_OPT} -XX:-UseLargePages"
-JAVA_OPT="${JAVA_OPT} -XX:+PerfDisableSharedMem"
+JAVA_OPT="${JAVA_OPT} -XX:-UseLargePages"
+JAVA_OPT="${JAVA_OPT} -XX:+PerfDisableSharedMem"
#JAVA_OPT="${JAVA_OPT} -Xdebug -Xrunjdwp:transport=dt_socket,address=9555,server=y,suspend=n"
JAVA_OPT="${JAVA_OPT} -cp ${CLASSPATH}"
diff --git a/distribution/bin/mqadmin.xml b/distribution/bin/mqadmin.xml
new file mode 100644
index 0000000..0f07da4
--- /dev/null
+++ b/distribution/bin/mqadmin.xml
@@ -0,0 +1,43 @@
+<!--
+ 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.
+ -->
+
+<java>
+ <debug>false</debug>
+
+ <javahome>${JAVA_HOME}</javahome>
+
+ <jvmtype>server</jvmtype>
+
+ <mainclass>org.apache.rocketmq.tools.command.MQAdminStartup</mainclass>
+
+ <properties>
+ <java.ext.dirs>${cpd}/../lib</java.ext.dirs>
+ <rocketmq.home.dir>${cpd}/..</rocketmq.home.dir>
+ </properties>
+
+ <classpaths>
+ </classpaths>
+
+ <options>
+ <-Xms512m></-Xms512m>
+ <-Xmx1g></-Xmx1g>
+ <-XX:NewSize>256M</-XX:NewSize>
+ <-XX:MaxNewSize>512M</-XX:MaxNewSize>
+ <-XX:PermSize>128M</-XX:PermSize>
+ <-XX:MaxPermSize>128M</-XX:MaxPermSize>
+ </options>
+</java>
diff --git a/distribution/bin/mqbroker.xml b/distribution/bin/mqbroker.xml
new file mode 100644
index 0000000..3043cc0
--- /dev/null
+++ b/distribution/bin/mqbroker.xml
@@ -0,0 +1,45 @@
+<!--
+ 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.
+ -->
+
+<java>
+ <debug>false</debug>
+
+ <javahome>${JAVA_HOME}</javahome>
+
+ <jvmtype>server</jvmtype>
+
+ <mainclass>org.apache.rocketmq.broker.BrokerStartup</mainclass>
+
+ <properties>
+ <java.ext.dirs>${cpd}/../lib</java.ext.dirs>
+ <rocketmq.home.dir>${cpd}/..</rocketmq.home.dir>
+ </properties>
+
+ <classpaths>
+ </classpaths>
+
+ <options>
+ <-Xms512m>
+ </-Xms512m>
+ <-Xmx1g>
+</-Xmx1g>
+<-XX:NewSize>256M</-XX:NewSize>
+<-XX:MaxNewSize>512M</-XX:MaxNewSize>
+<-XX:PermSize>128M</-XX:PermSize>
+<-XX:MaxPermSize>128M</-XX:MaxPermSize>
+ </options>
+ </java>
diff --git a/distribution/bin/mqnamesrv.xml b/distribution/bin/mqnamesrv.xml
new file mode 100644
index 0000000..1f050d1
--- /dev/null
+++ b/distribution/bin/mqnamesrv.xml
@@ -0,0 +1,45 @@
+<!--
+ 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.
+ -->
+
+<java>
+ <debug>false</debug>
+
+ <javahome>${JAVA_HOME}</javahome>
+
+ <jvmtype>server</jvmtype>
+
+ <mainclass>org.apache.rocketmq.namesrv.NamesrvStartup</mainclass>
+
+ <properties>
+ <java.ext.dirs>${cpd}/../lib</java.ext.dirs>
+ <rocketmq.home.dir>${cpd}/..</rocketmq.home.dir>
+ </properties>
+
+ <classpaths>
+ </classpaths>
+
+ <options>
+ <-Xms512m>
+ </-Xms512m>
+ <-Xmx1g>
+</-Xmx1g>
+<-XX:NewSize>256M</-XX:NewSize>
+<-XX:MaxNewSize>512M</-XX:MaxNewSize>
+<-XX:PermSize>128M</-XX:PermSize>
+<-XX:MaxPermSize>128M</-XX:MaxPermSize>
+ </options>
+ </java>
diff --git a/distribution/bin/runbroker.sh b/distribution/bin/runbroker.sh
index 0ccbf77..df962b4 100644
--- a/distribution/bin/runbroker.sh
+++ b/distribution/bin/runbroker.sh
@@ -36,37 +36,9 @@ export CLASSPATH=.:${BASE_DIR}/conf:${CLASSPATH}
#===========================================================================================
# JVM Configuration
#===========================================================================================
-# The RAMDisk initializing size in MB on Darwin OS for gc-log
-DIR_SIZE_IN_MB=600
-
-choose_gc_log_directory()
-{
- case "`uname`" in
- Darwin)
- if [ ! -d "/Volumes/RAMDisk" ]; then
- # create ram disk on Darwin systems as gc-log directory
- DEV=`hdiutil attach -nomount ram://$((2 * 1024 * DIR_SIZE_IN_MB))` > /dev/null
- diskutil eraseVolume HFS+ RAMDisk ${DEV} > /dev/null
- echo "Create RAMDisk /Volumes/RAMDisk for gc logging on Darwin OS."
- fi
- GC_LOG_DIR="/Volumes/RAMDisk"
- ;;
- *)
- # check if /dev/shm exists on other systems
- if [ -d "/dev/shm" ]; then
- GC_LOG_DIR="/dev/shm"
- else
- GC_LOG_DIR=${BASE_DIR}
- fi
- ;;
- esac
-}
-
-choose_gc_log_directory
-
JAVA_OPT="${JAVA_OPT} -server -Xms8g -Xmx8g -Xmn4g"
JAVA_OPT="${JAVA_OPT} -XX:+UseG1GC -XX:G1HeapRegionSize=16m -XX:G1ReservePercent=25 -XX:InitiatingHeapOccupancyPercent=30 -XX:SoftRefLRUPolicyMSPerMB=0"
-JAVA_OPT="${JAVA_OPT} -verbose:gc -Xloggc:${GC_LOG_DIR}/rmq_broker_gc_%p_%t.log -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCApplicationStoppedTime -XX:+PrintAdaptiveSizePolicy"
+JAVA_OPT="${JAVA_OPT} -verbose:gc -Xloggc:/dev/shm/mq_gc_%p.log -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCApplicationStoppedTime -XX:+PrintAdaptiveSizePolicy"
JAVA_OPT="${JAVA_OPT} -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=5 -XX:GCLogFileSize=30m"
JAVA_OPT="${JAVA_OPT} -XX:-OmitStackTraceInFastThrow"
JAVA_OPT="${JAVA_OPT} -XX:+AlwaysPreTouch"
diff --git a/distribution/bin/runserver.sh b/distribution/bin/runserver.sh
index d2bbded..908c79d 100644
--- a/distribution/bin/runserver.sh
+++ b/distribution/bin/runserver.sh
@@ -36,40 +36,11 @@ export CLASSPATH=.:${BASE_DIR}/conf:${CLASSPATH}
#===========================================================================================
# JVM Configuration
#===========================================================================================
-# The RAMDisk initializing size in MB on Darwin OS for gc-log
-DIR_SIZE_IN_MB=600
-
-choose_gc_log_directory()
-{
- case "`uname`" in
- Darwin)
- if [ ! -d "/Volumes/RAMDisk" ]; then
- # create ram disk on Darwin systems as gc-log directory
- DEV=`hdiutil attach -nomount ram://$((2 * 1024 * DIR_SIZE_IN_MB))` > /dev/null
- diskutil eraseVolume HFS+ RAMDisk ${DEV} > /dev/null
- echo "Create RAMDisk /Volumes/RAMDisk for gc logging on Darwin OS."
- fi
- GC_LOG_DIR="/Volumes/RAMDisk"
- ;;
- *)
- # check if /dev/shm exists on other systems
- if [ -d "/dev/shm" ]; then
- GC_LOG_DIR="/dev/shm"
- else
- GC_LOG_DIR=${BASE_DIR}
- fi
- ;;
- esac
-}
-
-choose_gc_log_directory
-
JAVA_OPT="${JAVA_OPT} -server -Xms4g -Xmx4g -Xmn2g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
JAVA_OPT="${JAVA_OPT} -XX:+UseConcMarkSweepGC -XX:+UseCMSCompactAtFullCollection -XX:CMSInitiatingOccupancyFraction=70 -XX:+CMSParallelRemarkEnabled -XX:SoftRefLRUPolicyMSPerMB=0 -XX:+CMSClassUnloadingEnabled -XX:SurvivorRatio=8 -XX:-UseParNewGC"
-JAVA_OPT="${JAVA_OPT} -verbose:gc -Xloggc:${GC_LOG_DIR}/rmq_srv_gc_%p_%t.log -XX:+PrintGCDetails"
-JAVA_OPT="${JAVA_OPT} -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=5 -XX:GCLogFileSize=30m"
+JAVA_OPT="${JAVA_OPT} -verbose:gc -Xloggc:/dev/shm/rmq_srv_gc.log -XX:+PrintGCDetails"
JAVA_OPT="${JAVA_OPT} -XX:-OmitStackTraceInFastThrow"
-JAVA_OPT="${JAVA_OPT} -XX:-UseLargePages"
+JAVA_OPT="${JAVA_OPT} -XX:-UseLargePages"
JAVA_OPT="${JAVA_OPT} -Djava.ext.dirs=${JAVA_HOME}/jre/lib/ext:${BASE_DIR}/lib"
#JAVA_OPT="${JAVA_OPT} -Xdebug -Xrunjdwp:transport=dt_socket,address=9555,server=y,suspend=n"
JAVA_OPT="${JAVA_OPT} ${JAVA_OPT_EXT}"
diff --git a/distribution/bin/tools.cmd b/distribution/bin/tools.cmd
index c61fbcf..0e25c39 100644
--- a/distribution/bin/tools.cmd
+++ b/distribution/bin/tools.cmd
@@ -28,7 +28,7 @@ set CLASSPATH=.;%BASE_DIR%conf;%CLASSPATH%
rem ===========================================================================================
rem JVM Configuration
rem ===========================================================================================
-set "JAVA_OPT=%JAVA_OPT% -server -Xms1g -Xmx1g -Xmn256m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=128m"
+set "JAVA_OPT=%JAVA_OPT% -server -Xms1g -Xmx1g -Xmn256m -XX:PermSize=128m -XX:MaxPermSize=128m"
set "JAVA_OPT=%JAVA_OPT% -Djava.ext.dirs="%BASE_DIR%\lib";"%JAVA_HOME%\jre\lib\ext";"%JAVA_HOME%\lib\ext""
set "JAVA_OPT=%JAVA_OPT% -cp "%CLASSPATH%""
diff --git a/distribution/bin/tools.sh b/distribution/bin/tools.sh
index 071794c..645a595 100644
--- a/distribution/bin/tools.sh
+++ b/distribution/bin/tools.sh
@@ -36,7 +36,7 @@ export CLASSPATH=.:${BASE_DIR}/conf:${CLASSPATH}
#===========================================================================================
# JVM Configuration
#===========================================================================================
-JAVA_OPT="${JAVA_OPT} -server -Xms1g -Xmx1g -Xmn256m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=128m"
+JAVA_OPT="${JAVA_OPT} -server -Xms1g -Xmx1g -Xmn256m -XX:PermSize=128m -XX:MaxPermSize=128m"
JAVA_OPT="${JAVA_OPT} -Djava.ext.dirs=${BASE_DIR}/lib:${JAVA_HOME}/jre/lib/ext:${JAVA_HOME}/lib/ext"
JAVA_OPT="${JAVA_OPT} -cp ${CLASSPATH}"
diff --git a/distribution/pom.xml b/distribution/pom.xml
index 4e3e78f..1e3a4dd 100644
--- a/distribution/pom.xml
+++ b/distribution/pom.xml
@@ -20,7 +20,7 @@
<parent>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-all</artifactId>
- <version>4.6.0-SNAPSHOT</version>
+ <version>4.5.2-SNAPSHOT</version>
</parent>
<artifactId>rocketmq-distribution</artifactId>
<name>rocketmq-distribution ${project.version}</name>
diff --git a/docs/cn/README.md b/docs/cn/README.md
index c8bdc8d..a47ec64 100644
--- a/docs/cn/README.md
+++ b/docs/cn/README.md
@@ -1,4 +1,4 @@
-Apache RocketMQ开发者指南
+Apache RocketMQ开发者指南
--------
##### 这个开发者指南是帮忙您快速了解,并使用 Apache RocketMQ
@@ -14,7 +14,7 @@
- [架构(Architecture)](architecture.md)：介绍RocketMQ部署架构和技术架构。
-- [设计(Design)](design.md)：介绍RocketMQ关键机制的设计原理，主要包括消息存储、通信机制、消息过滤、负载均衡、事务消息等。
+- [设计(Design)](design.md)：介绍RocketMQ关键机制的设计原理，主要包括消息存储、通信机制、消息过滤、负载均衡、事物消息等。
### 3. 样例
diff --git a/docs/cn/best_practice.md b/docs/cn/best_practice.md
index e4b04e0..39febbc 100755
--- a/docs/cn/best_practice.md
+++ b/docs/cn/best_practice.md
@@ -289,6 +289,7 @@ DefaultMQProducer、TransactionMQProducer、DefaultMQPushConsumer、DefaultMQPul
| offsetStore | | 消费进度存储 |
| consumeThreadMin | 10 | 消费线程池最小线程数 |
| consumeThreadMax | 20 | 消费线程池最大线程数 |
+| | | |
| consumeConcurrentlyMaxSpan | 2000 | 单队列并行消费允许的最大跨度 |
| pullThresholdForQueue | 1000 | 拉消息本地队列缓存消息最大数 |
| pullInterval | 0 | 拉消息间隔，由于是长轮询，所以为0，但是如果应用为了流控，也可以设置大于0的值，单位毫秒 |
diff --git a/docs/cn/concept.md b/docs/cn/concept.md
index b486b38..87d691b 100644
--- a/docs/cn/concept.md
+++ b/docs/cn/concept.md
@@ -43,8 +43,11 @@ RocketMQ主要由 Producer、Broker、Consumer 三部分组成，其中Producer
## 14 严格顺序消息（Strictly Ordered Message）
严格顺序消息模式下，消费者收到的所有消息均是有顺序的。
-## 15 消息（Message）
+## 15 代理服务器（Broker Server）
+消息中转角色，负责存储消息、转发消息。代理服务器在RocketMQ系统中负责接收从生产者发送来的消息并存储、同时为消费者的拉取请求作准备。代理服务器也存储消息相关的元数据，包括消费者组、消费进度偏移和主题和队列消息等。
+
+## 16 消息（Message）
消息系统所传输信息的物理载体，生产和消费数据的最小单位，每条消息必须属于一个主题。RocketMQ中每个消息拥有唯一的Message ID，且可以携带具有业务标识的Key。系统提供了通过Message ID和Key查询消息的功能。
-## 16 标签（Tag）
+## 17 标签（Tag）
为消息设置的标志，用于同一主题下区分不同类型的消息。来自同一业务单元的消息，可以根据不同业务目的在同一主题下设置不同标签。标签能够有效地保持代码的清晰度和连贯性，并优化RocketMQ提供的查询系统。消费者可以根据Tag实现对不同子主题的不同消费逻辑，实现更好的扩展性。
diff --git a/docs/cn/features.md b/docs/cn/features.md
index 859e0f8..1c0456b 100644
--- a/docs/cn/features.md
+++ b/docs/cn/features.md
@@ -12,7 +12,7 @@
- 分区顺序
对于指定的一个 Topic，所有消息根据 sharding key 进行区块分区。 同一个分区内的消息按照严格的 FIFO 顺序进行发布和消费。 Sharding key 是顺序消息中用来区分不同分区的关键字段，和普通消息的 Key 是完全不同的概念。
适用场景：性能要求高，以 sharding key 作为分区字段，在同一个区块中严格的按照 FIFO 原则进行消息发布和消费的场景。
-## 3 消息过滤
+##3 消息过滤
RocketMQ的消费者可以根据Tag进行消息过滤，也支持自定义属性过滤。消息过滤目前是在Broker端实现的，优点是减少了对于Consumer无用消息的网络传输，缺点是增加了Broker的负担、而且实现相对复杂。
## 4 消息可靠性
RocketMQ支持消息的高可靠，影响消息可靠性的几种情况：
diff --git a/example/pom.xml b/example/pom.xml
index 5e16027..ac83bea 100644
--- a/example/pom.xml
+++ b/example/pom.xml
@@ -19,7 +19,7 @@
<parent>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-all</artifactId>
- <version>4.6.0-SNAPSHOT</version>
+ <version>4.5.2-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
@@ -51,12 +51,12 @@
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-openmessaging</artifactId>
- <version>4.6.0-SNAPSHOT</version>
+ <version>4.5.2-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-acl</artifactId>
- <version>4.6.0-SNAPSHOT</version>
+ <version>4.5.2-SNAPSHOT</version>
</dependency>
</dependencies>
</project>
diff --git a/filter/pom.xml b/filter/pom.xml
index 9b2128d..7f33e99 100644
--- a/filter/pom.xml
+++ b/filter/pom.xml
@@ -20,7 +20,7 @@
<parent>
<artifactId>rocketmq-all</artifactId>
<groupId>org.apache.rocketmq</groupId>
- <version>4.6.0-SNAPSHOT</version>
+ <version>4.5.2-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/logappender/pom.xml b/logappender/pom.xml
index 667f213..2fd6da2 100644
--- a/logappender/pom.xml
+++ b/logappender/pom.xml
@@ -19,7 +19,7 @@
<parent>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-all</artifactId>
- <version>4.6.0-SNAPSHOT</version>
+ <version>4.5.2-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>rocketmq-logappender</artifactId>
diff --git a/logging/pom.xml b/logging/pom.xml
index 95c70b5..fb10452 100644
--- a/logging/pom.xml
+++ b/logging/pom.xml
@@ -19,7 +19,7 @@
<parent>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-all</artifactId>
- <version>4.6.0-SNAPSHOT</version>
+ <version>4.5.2-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/namesrv/pom.xml b/namesrv/pom.xml
index bb858b9..d98403e 100644
--- a/namesrv/pom.xml
+++ b/namesrv/pom.xml
@@ -19,7 +19,7 @@
<parent>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-all</artifactId>
- <version>4.6.0-SNAPSHOT</version>
+ <version>4.5.2-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/openmessaging/pom.xml b/openmessaging/pom.xml
index 8215a3f..0f6aa2e 100644
--- a/openmessaging/pom.xml
+++ b/openmessaging/pom.xml
@@ -20,7 +20,7 @@
<parent>
<artifactId>rocketmq-all</artifactId>
<groupId>org.apache.rocketmq</groupId>
- <version>4.6.0-SNAPSHOT</version>
+ <version>4.5.2-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pom.xml b/pom.xml
index f622df7..7d183e7 100644
--- a/pom.xml
+++ b/pom.xml
@@ -16,8 +16,7 @@
limitations under the License.
-->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<parent>
<groupId>org.apache</groupId>
@@ -30,7 +29,7 @@
<inceptionYear>2012</inceptionYear>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-all</artifactId>
- <version>4.6.0-SNAPSHOT</version>
+ <version>4.5.2-SNAPSHOT</version>
<packaging>pom</packaging>
<name>Apache RocketMQ ${project.version}</name>
<url>http://rocketmq.apache.org/</url>
@@ -159,7 +158,7 @@
</executions>
<configuration>
<rules>
- <banCircularDependencies/>
+ <banCircularDependencies />
</rules>
<fail>true</fail>
</configuration>
@@ -608,13 +607,6 @@
<artifactId>log4j-slf4j-impl</artifactId>
<version>2.7</version>
</dependency>
- <dependency>
- <groupId>commons-validator</groupId>
- <artifactId>commons-validator</artifactId>
- <version>1.6</version>
- </dependency>
-
-
</dependencies>
</dependencyManagement>
</project>
diff --git a/remoting/pom.xml b/remoting/pom.xml
index 68aebd9..132e21b 100644
--- a/remoting/pom.xml
+++ b/remoting/pom.xml
@@ -19,7 +19,7 @@
<parent>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-all</artifactId>
- <version>4.6.0-SNAPSHOT</version>
+ <version>4.5.2-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
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 f244bf4..585b60b 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,17 +17,18 @@
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";
@@ -52,10 +53,8 @@ public class RemotingHelper {
}
public static SocketAddress string2SocketAddress(final String addr) {
- int split = addr.lastIndexOf(":");
- String host = addr.substring(0, split);
- String port = addr.substring(split + 1);
- InetSocketAddress isa = new InetSocketAddress(host, Integer.parseInt(port));
+ String[] s = addr.split(":");
+ InetSocketAddress isa = new InetSocketAddress(s[0], Integer.parseInt(s[1]));
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 a16940e..3da3a18 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,6 +31,7 @@ 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;
@@ -144,10 +145,8 @@ public class RemotingUtil {
}
public static SocketAddress string2SocketAddress(final String addr) {
- int split = addr.lastIndexOf(":");
- String host = addr.substring(0, split);
- String port = addr.substring(split + 1);
- InetSocketAddress isa = new InetSocketAddress(host, Integer.parseInt(port));
+ String[] s = addr.split(":");
+ InetSocketAddress isa = new InetSocketAddress(s[0], Integer.parseInt(s[1]));
return isa;
}
diff --git a/srvutil/pom.xml b/srvutil/pom.xml
index 6aa9fa3..d616edd 100644
--- a/srvutil/pom.xml
+++ b/srvutil/pom.xml
@@ -19,7 +19,7 @@
<parent>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-all</artifactId>
- <version>4.6.0-SNAPSHOT</version>
+ <version>4.5.2-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/store/pom.xml b/store/pom.xml
index 79e337b..b96f879 100644
--- a/store/pom.xml
+++ b/store/pom.xml
@@ -19,7 +19,7 @@
<parent>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-all</artifactId>
- <version>4.6.0-SNAPSHOT</version>
+ <version>4.5.2-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
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 3d89fe4..2ab66e8 100644
--- a/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
+++ b/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
@@ -16,8 +16,6 @@
*/
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;
@@ -28,14 +26,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;
@@ -272,21 +270,11 @@ public class CommitLog {
long bornTimeStamp = byteBuffer.getLong();
- ByteBuffer byteBuffer1;
- if ((sysFlag & MessageSysFlag.BORNHOST_V6_FLAG) == 0) {
- byteBuffer1 = byteBuffer.get(bytesContent, 0, 4 + 4);
- } else {
- byteBuffer1 = byteBuffer.get(bytesContent, 0, 16 + 4);
- }
+ ByteBuffer byteBuffer1 = byteBuffer.get(bytesContent, 0, 8);
long storeTimestamp = byteBuffer.getLong();
- ByteBuffer byteBuffer2;
- if ((sysFlag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0) {
- byteBuffer2 = byteBuffer.get(bytesContent, 0, 4 + 4);
- } else {
- byteBuffer2 = byteBuffer.get(bytesContent, 0, 16 + 4);
- }
+ ByteBuffer byteBuffer2 = byteBuffer.get(bytesContent, 0, 8);
int reconsumeTimes = byteBuffer.getInt();
@@ -351,7 +339,7 @@ public class CommitLog {
}
}
- int readLength = calMsgLength(sysFlag, bodyLen, topicLen, propertiesLength);
+ int readLength = calMsgLength(bodyLen, topicLen, propertiesLength);
if (totalSize != readLength) {
doNothingForDeadCode(reconsumeTimes);
doNothingForDeadCode(flag);
@@ -384,9 +372,7 @@ public class CommitLog {
return new DispatchRequest(-1, false /* success */);
}
- 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;
+ protected static int calMsgLength(int bodyLength, int topicLength, int propertiesLength) {
final int msgLen = 4 //TOTALSIZE
+ 4 //MAGICCODE
+ 4 //BODYCRC
@@ -396,9 +382,9 @@ public class CommitLog {
+ 8 //PHYSICALOFFSET
+ 4 //SYSFLAG
+ 8 //BORNTIMESTAMP
- + bornhostLength //BORNHOST
+ + 8 //BORNHOST
+ 8 //STORETIMESTAMP
- + storehostAddressLength //STOREHOSTADDRESS
+ + 8 //STOREHOSTADDRESS
+ 4 //RECONSUMETIMES
+ 8 //Prepared Transaction Offset
+ 4 + (bodyLength > 0 ? bodyLength : 0) //BODY
@@ -510,10 +496,7 @@ public class CommitLog {
return false;
}
- 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);
+ long storeTimestamp = byteBuffer.getLong(MessageDecoder.MESSAGE_STORE_TIMESTAMP_POSTION);
if (0 == storeTimestamp) {
return false;
}
@@ -586,18 +569,7 @@ public class CommitLog {
}
}
- 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;
-
+ long elapsedTimeInLock = 0;
MappedFile unlockMappedFile = null;
MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
@@ -647,14 +619,14 @@ public class CommitLog {
return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result);
}
- eclipsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
+ elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
beginTimeInLock = 0;
} finally {
putMessageLock.unlock();
}
- if (eclipsedTimeInLock > 500) {
- log.warn("[NOTIFYME]putMessage in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", eclipsedTimeInLock, msg.getBody().length, result);
+ if (elapsedTimeInLock > 500) {
+ log.warn("[NOTIFYME]putMessage in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", elapsedTimeInLock, msg.getBody().length, result);
}
if (null != unlockMappedFile && this.defaultMessageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {
@@ -742,17 +714,7 @@ public class CommitLog {
return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, null);
}
- 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;
+ long elapsedTimeInLock = 0;
MappedFile unlockMappedFile = null;
MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
@@ -807,14 +769,14 @@ public class CommitLog {
return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result);
}
- eclipsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
+ elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
beginTimeInLock = 0;
} finally {
putMessageLock.unlock();
}
- if (eclipsedTimeInLock > 500) {
- log.warn("[NOTIFYME]putMessages in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", eclipsedTimeInLock, messageExtBatch.getBody().length, result);
+ if (elapsedTimeInLock > 500) {
+ log.warn("[NOTIFYME]putMessages in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", elapsedTimeInLock, messageExtBatch.getBody().length, result);
}
if (null != unlockMappedFile && this.defaultMessageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {
@@ -842,10 +804,7 @@ public class CommitLog {
SelectMappedBufferResult result = this.getMessage(offset, size);
if (null != result) {
try {
- 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);
+ return result.getByteBuffer().getLong(MessageDecoder.MESSAGE_STORE_TIMESTAMP_POSTION);
} finally {
result.release();
}
@@ -1211,7 +1170,6 @@ 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
@@ -1221,9 +1179,10 @@ public class CommitLog {
private final StringBuilder msgIdBuilder = new StringBuilder();
+ private final ByteBuffer hostHolder = ByteBuffer.allocate(8);
+
DefaultAppendMessageCallback(final int size) {
- this.msgIdMemory = ByteBuffer.allocate(4 + 4 + 8);
- this.msgIdV6Memory = ByteBuffer.allocate(16 + 4 + 8);
+ this.msgIdMemory = ByteBuffer.allocate(MessageDecoder.MSG_ID_LENGTH);
this.msgStoreItemMemory = ByteBuffer.allocate(size + END_FILE_MIN_BLANK_LENGTH);
this.maxMessageSize = size;
}
@@ -1239,20 +1198,8 @@ public class CommitLog {
// PHY OFFSET
long wroteOffset = fileFromOffset + byteBuffer.position();
- 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);
- }
+ this.resetByteBuffer(hostHolder, 8);
+ String msgId = MessageDecoder.createMessageId(this.msgIdMemory, msgInner.getStoreHostBytes(hostHolder), wroteOffset);
// Record ConsumeQueue information
keyBuilder.setLength(0);
@@ -1299,7 +1246,7 @@ public class CommitLog {
final int bodyLength = msgInner.getBody() == null ? 0 : msgInner.getBody().length;
- final int msgLen = calMsgLength(msgInner.getSysFlag(), bodyLength, topicLength, propertiesLength);
+ final int msgLen = calMsgLength(bodyLength, topicLength, propertiesLength);
// Exceeds the maximum message
if (msgLen > this.maxMessageSize) {
@@ -1344,13 +1291,14 @@ public class CommitLog {
// 9 BORNTIMESTAMP
this.msgStoreItemMemory.putLong(msgInner.getBornTimestamp());
// 10 BORNHOST
- this.resetByteBuffer(bornHostHolder, bornHostLength);
- this.msgStoreItemMemory.put(msgInner.getBornHostBytes(bornHostHolder));
+ this.resetByteBuffer(hostHolder, 8);
+ this.msgStoreItemMemory.put(msgInner.getBornHostBytes(hostHolder));
// 11 STORETIMESTAMP
this.msgStoreItemMemory.putLong(msgInner.getStoreTimestamp());
// 12 STOREHOSTADDRESS
- this.resetByteBuffer(storeHostHolder, storeHostLength);
- this.msgStoreItemMemory.put(msgInner.getStoreHostBytes(storeHostHolder));
+ this.resetByteBuffer(hostHolder, 8);
+ this.msgStoreItemMemory.put(msgInner.getStoreHostBytes(hostHolder));
+ //this.msgBatchMemory.put(msgInner.getStoreHostBytes());
// 13 RECONSUMETIMES
this.msgStoreItemMemory.putInt(msgInner.getReconsumeTimes());
// 14 Prepared Transaction Offset
@@ -1411,13 +1359,8 @@ public class CommitLog {
msgIdBuilder.setLength(0);
final long beginTimeMills = CommitLog.this.defaultMessageStore.now();
ByteBuffer messagesByteBuff = messageExtBatch.getEncodedBuff();
-
- 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);
+ this.resetByteBuffer(hostHolder, 8);
+ ByteBuffer storeHostBytes = messageExtBatch.getStoreHostBytes(hostHolder);
messagesByteBuff.mark();
while (messagesByteBuff.hasRemaining()) {
// 1 TOTALSIZE
@@ -1453,13 +1396,7 @@ public class CommitLog {
messagesByteBuff.putLong(wroteOffset + totalMsgLen - msgLen);
storeHostBytes.rewind();
- 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);
- }
-
+ String msgId = MessageDecoder.createMessageId(this.msgIdMemory, storeHostBytes, wroteOffset + totalMsgLen - msgLen);
if (msgIdBuilder.length() > 0) {
msgIdBuilder.append(',').append(msgId);
} else {
@@ -1495,6 +1432,8 @@ 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;
@@ -1504,13 +1443,6 @@ 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();
@@ -1534,7 +1466,7 @@ public class CommitLog {
final int topicLength = topicData.length;
- final int msgLen = calMsgLength(messageExtBatch.getSysFlag(), bodyLen, topicLength, propertiesLen);
+ final int msgLen = calMsgLength(bodyLen, topicLength, propertiesLen);
// Exceeds the maximum message
if (msgLen > this.maxMessageSize) {
@@ -1568,13 +1500,13 @@ public class CommitLog {
// 9 BORNTIMESTAMP
this.msgBatchMemory.putLong(messageExtBatch.getBornTimestamp());
// 10 BORNHOST
- this.resetByteBuffer(bornHostHolder, bornHostLength);
- this.msgBatchMemory.put(messageExtBatch.getBornHostBytes(bornHostHolder));
+ this.resetByteBuffer(hostHolder, 8);
+ this.msgBatchMemory.put(messageExtBatch.getBornHostBytes(hostHolder));
// 11 STORETIMESTAMP
this.msgBatchMemory.putLong(messageExtBatch.getStoreTimestamp());
// 12 STOREHOSTADDRESS
- this.resetByteBuffer(storeHostHolder, storeHostLength);
- this.msgBatchMemory.put(messageExtBatch.getStoreHostBytes(storeHostHolder));
+ this.resetByteBuffer(hostHolder, 8);
+ this.msgBatchMemory.put(messageExtBatch.getStoreHostBytes(hostHolder));
// 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 d5ba569..971b1e7 100644
--- a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
+++ b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
@@ -19,8 +19,6 @@ 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;
@@ -287,6 +285,8 @@ 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,9 +1042,7 @@ public class DefaultMessageStore implements MessageStore {
int i = 0;
for (; i < bufferConsumeQueue.getSize(); i += ConsumeQueue.CQ_STORE_UNIT_SIZE) {
long offsetPy = bufferConsumeQueue.getByteBuffer().getLong();
- InetSocketAddress inetSocketAddress = (InetSocketAddress) storeHost;
- int msgIdLength = (inetSocketAddress.getAddress() instanceof Inet6Address) ? 16 + 4 + 8 : 4 + 4 + 8;
- final ByteBuffer msgIdMemory = ByteBuffer.allocate(msgIdLength);
+ final ByteBuffer msgIdMemory = ByteBuffer.allocate(MessageDecoder.MSG_ID_LENGTH);
String msgId =
MessageDecoder.createMessageId(msgIdMemory, MessageExt.socketAddress2ByteBuffer(storeHost), offsetPy);
messageIds.put(msgId, nextOffset++);
diff --git a/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java b/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java
index 7891f71..2543e96 100644
--- a/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java
+++ b/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java
@@ -80,7 +80,7 @@ public class MessageStoreConfig {
private int fileReservedTime = 72;
// Flow control for ConsumeQueue
private int putMsgIndexHightWater = 600000;
- // The maximum size of message,default is 4M
+ // The maximum size of a single log file,default is 512K
private int maxMessageSize = 1024 * 1024 * 4;
// Whether check the CRC32 of the records consumed.
// This ensures no on-the-wire or on-disk corruption to the messages occurred.
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 13da48b..17be11b 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,11 +68,12 @@ 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());
@@ -98,7 +99,12 @@ public class DLedgerCommitLog extends CommitLog {
@Override
public boolean load() {
- return super.load();
+ boolean result = super.load();
+ if (!result) {
+ return false;
+ }
+
+ return true;
}
private void refreshConfig() {
@@ -157,6 +163,8 @@ public class DLedgerCommitLog extends CommitLog {
log.warn("Should not set confirm offset {} for dleger commitlog", phyOffset);
}
+
+
@Override
public long remainHowManyDataToCommit() {
return dLedgerFileList.remainHowManyDataToCommit();
@@ -177,7 +185,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();
}
@@ -198,6 +206,7 @@ public class DLedgerCommitLog extends CommitLog {
return 1;
}
+
public SelectMappedBufferResult convertSbr(SelectMmapBufferResult sbr) {
if (sbr == null) {
return null;
@@ -228,6 +237,7 @@ public class DLedgerCommitLog extends CommitLog {
return this.getData(offset, offset == 0);
}
+
@Override
public SelectMappedBufferResult getData(final long offset, final boolean returnFirstOnNotFound) {
if (offset < dividedCommitlogOffset) {
@@ -241,7 +251,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;
@@ -273,7 +283,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
@@ -306,7 +316,7 @@ public class DLedgerCommitLog extends CommitLog {
}
@Override
- public void recoverAbnormally(long maxPhyOffsetOfConsumeQueue) {
+ public void recoverAbnormally(long maxPhyOffsetOfConsumeQueue) {
recover(maxPhyOffsetOfConsumeQueue);
}
@@ -324,9 +334,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);
@@ -404,10 +414,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();
@@ -418,11 +428,8 @@ 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;
-
- int msgIdLength = (msg.getSysFlag() & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 + 4 + 8 : 16 + 4 + 8;
- ByteBuffer buffer = ByteBuffer.allocate(msgIdLength);
-
+ long wroteOffset = dledgerFuture.getPos() + DLedgerEntry.BODY_OFFSET;
+ ByteBuffer buffer = ByteBuffer.allocate(MessageDecoder.MSG_ID_LENGTH);
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);
@@ -489,6 +496,8 @@ 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) {
@@ -498,7 +507,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;
}
@@ -555,7 +564,6 @@ 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;
@@ -567,7 +575,6 @@ 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
@@ -577,11 +584,10 @@ 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(4 + 4 + 8);
- this.msgIdV6Memory = ByteBuffer.allocate(16 + 4 + 8);
+ this.msgIdMemory = ByteBuffer.allocate(MessageDecoder.MSG_ID_LENGTH);
this.msgStoreItemMemory = ByteBuffer.allocate(size + END_FILE_MIN_BLANK_LENGTH);
this.maxMessageSize = size;
}
@@ -596,13 +602,7 @@ public class DLedgerCommitLog extends CommitLog {
// PHY OFFSET
long wroteOffset = 0;
- 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(hostHolder, 8);
// Record ConsumeQueue information
keyBuilder.setLength(0);
keyBuilder.append(msgInner.getTopic());
@@ -649,7 +649,7 @@ public class DLedgerCommitLog extends CommitLog {
final int bodyLength = msgInner.getBody() == null ? 0 : msgInner.getBody().length;
- final int msgLen = calMsgLength(msgInner.getSysFlag(), bodyLength, topicLength, propertiesLength);
+ final int msgLen = calMsgLength(bodyLength, topicLength, propertiesLength);
// Exceeds the maximum message
if (msgLen > this.maxMessageSize) {
@@ -678,13 +678,13 @@ public class DLedgerCommitLog extends CommitLog {
// 9 BORNTIMESTAMP
this.msgStoreItemMemory.putLong(msgInner.getBornTimestamp());
// 10 BORNHOST
- this.resetByteBuffer(bornHostHolder, bornHostLength);
- this.msgStoreItemMemory.put(msgInner.getBornHostBytes(bornHostHolder));
+ this.resetByteBuffer(hostHolder, 8);
+ this.msgStoreItemMemory.put(msgInner.getBornHostBytes(hostHolder));
// 11 STORETIMESTAMP
this.msgStoreItemMemory.putLong(msgInner.getStoreTimestamp());
// 12 STOREHOSTADDRESS
- this.resetByteBuffer(storeHostHolder, storeHostLength);
- this.msgStoreItemMemory.put(msgInner.getStoreHostBytes(storeHostHolder));
+ this.resetByteBuffer(hostHolder, 8);
+ this.msgStoreItemMemory.put(msgInner.getStoreHostBytes(hostHolder));
//this.msgBatchMemory.put(msgInner.getStoreHostBytes());
// 13 RECONSUMETIMES
this.msgStoreItemMemory.putInt(msgInner.getReconsumeTimes());
@@ -719,7 +719,6 @@ 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/main/java/org/apache/rocketmq/store/ha/HAService.java b/store/src/main/java/org/apache/rocketmq/store/ha/HAService.java
index 84fb421..bbadd1b 100644
--- a/store/src/main/java/org/apache/rocketmq/store/ha/HAService.java
+++ b/store/src/main/java/org/apache/rocketmq/store/ha/HAService.java
@@ -374,7 +374,6 @@ public class HAService {
}
}
- lastWriteTimestamp = HAService.this.defaultMessageStore.getSystemClock().now();
return !this.reportOffset.hasRemaining();
}
@@ -407,6 +406,7 @@ public class HAService {
try {
int readSize = this.socketChannel.read(this.byteBufferRead);
if (readSize > 0) {
+ lastWriteTimestamp = HAService.this.defaultMessageStore.getSystemClock().now();
readSizeZeroTimes = 0;
boolean result = this.dispatchReadRequest();
if (!result) {
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 f46b3be..fbb2a74 100644
--- a/store/src/test/java/org/apache/rocketmq/store/AppendCallbackTest.java
+++ b/store/src/test/java/org/apache/rocketmq/store/AppendCallbackTest.java
@@ -98,43 +98,6 @@ public class AppendCallbackTest {
}
@Test
- public void testAppendIPv6HostMessageBatchEndOfFile() throws Exception {
- List<Message> messages = new ArrayList<>();
- String topic = "test-topic";
- int queue = 0;
- for (int i = 0; i < 10; i++) {
- Message msg = new Message();
- msg.setBody("body".getBytes());
- msg.setTopic(topic);
- msg.setTags("abc");
- messages.add(msg);
- }
- MessageExtBatch messageExtBatch = new MessageExtBatch();
- messageExtBatch.setTopic(topic);
- messageExtBatch.setQueueId(queue);
- messageExtBatch.setBornTimestamp(System.currentTimeMillis());
- messageExtBatch.setMsgId("24084004018081003FAA1DDE2B3F898A00002A9F0000000000000CA0");
- messageExtBatch.setSysFlag(0);
- messageExtBatch.setBornHostV6Flag();
- messageExtBatch.setStoreHostAddressV6Flag();
- messageExtBatch.setBornHost(new InetSocketAddress("1050:0000:0000:0000:0005:0600:300c:326b", 123));
- messageExtBatch.setStoreHost(new InetSocketAddress("::1", 124));
- messageExtBatch.setBody(MessageDecoder.encodeMessages(messages));
-
- messageExtBatch.setEncodedBuff(batchEncoder.encode(messageExtBatch));
- ByteBuffer buff = ByteBuffer.allocate(1024 * 10);
- //encounter end of file when append half of the data
- AppendMessageResult result = callback.doAppend(0, buff, 1000, messageExtBatch);
- assertEquals(AppendMessageStatus.END_OF_FILE, result.getStatus());
- assertEquals(0, result.getWroteOffset());
- assertEquals(0, result.getLogicsOffset());
- assertEquals(1000, result.getWroteBytes());
- assertEquals(8, buff.position()); //write blank size and magic value
-
- assertTrue(result.getMsgId().length() > 0); //should have already constructed some message ids
- }
-
- @Test
public void testAppendMessageBatchSucc() throws Exception {
List<Message> messages = new ArrayList<>();
String topic = "test-topic";
@@ -190,64 +153,4 @@ public class AppendCallbackTest {
}
- @Test
- public void testAppendIPv6HostMessageBatchSucc() throws Exception {
- List<Message> messages = new ArrayList<>();
- String topic = "test-topic";
- int queue = 0;
- for (int i = 0; i < 10; i++) {
- Message msg = new Message();
- msg.setBody("body".getBytes());
- msg.setTopic(topic);
- msg.setTags("abc");
- messages.add(msg);
- }
- MessageExtBatch messageExtBatch = new MessageExtBatch();
- messageExtBatch.setTopic(topic);
- messageExtBatch.setQueueId(queue);
- messageExtBatch.setBornTimestamp(System.currentTimeMillis());
- messageExtBatch.setMsgId("24084004018081003FAA1DDE2B3F898A00002A9F0000000000000CA0");
- messageExtBatch.setSysFlag(0);
- messageExtBatch.setBornHostV6Flag();
- messageExtBatch.setStoreHostAddressV6Flag();
- messageExtBatch.setBornHost(new InetSocketAddress("1050:0000:0000:0000:0005:0600:300c:326b", 123));
- messageExtBatch.setStoreHost(new InetSocketAddress("::1", 124));
- messageExtBatch.setBody(MessageDecoder.encodeMessages(messages));
-
- messageExtBatch.setEncodedBuff(batchEncoder.encode(messageExtBatch));
- ByteBuffer buff = ByteBuffer.allocate(1024 * 10);
- AppendMessageResult allresult = callback.doAppend(0, buff, 1024 * 10, messageExtBatch);
-
- assertEquals(AppendMessageStatus.PUT_OK, allresult.getStatus());
- assertEquals(0, allresult.getWroteOffset());
- assertEquals(0, allresult.getLogicsOffset());
- assertEquals(buff.position(), allresult.getWroteBytes());
-
- assertEquals(messages.size(), allresult.getMsgNum());
-
- Set<String> msgIds = new HashSet<>();
- for (String msgId : allresult.getMsgId().split(",")) {
- assertEquals(56, msgId.length());
- msgIds.add(msgId);
- }
- assertEquals(messages.size(), msgIds.size());
-
- List<MessageExt> decodeMsgs = MessageDecoder.decodes((ByteBuffer) buff.flip());
- assertEquals(decodeMsgs.size(), decodeMsgs.size());
- long queueOffset = decodeMsgs.get(0).getQueueOffset();
- long storeTimeStamp = decodeMsgs.get(0).getStoreTimestamp();
- for (int i = 0; i < messages.size(); i++) {
- assertEquals(messages.get(i).getTopic(), decodeMsgs.get(i).getTopic());
- assertEquals(new String(messages.get(i).getBody()), new String(decodeMsgs.get(i).getBody()));
- assertEquals(messages.get(i).getTags(), decodeMsgs.get(i).getTags());
-
- assertEquals(messageExtBatch.getBornHostNameString(), decodeMsgs.get(i).getBornHostNameString());
-
- assertEquals(messageExtBatch.getBornTimestamp(), decodeMsgs.get(i).getBornTimestamp());
- assertEquals(storeTimeStamp, decodeMsgs.get(i).getStoreTimestamp());
- assertEquals(queueOffset++, decodeMsgs.get(i).getQueueOffset());
- }
-
- }
-
}
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 8618dbb..0c8e5bb 100644
--- a/store/src/test/java/org/apache/rocketmq/store/BatchPutMessageTest.java
+++ b/store/src/test/java/org/apache/rocketmq/store/BatchPutMessageTest.java
@@ -23,7 +23,6 @@ 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;
@@ -126,58 +125,6 @@ public class BatchPutMessageTest {
}
- @Test
- public void testPutIPv6HostMessages() throws Exception {
- List<Message> messages = new ArrayList<>();
- String topic = "batch-write-topic";
- int queue = 0;
- int[] msgLengthArr = new int[11];
- msgLengthArr[0] = 0;
- int j = 1;
- for (int i = 0; i < 10; i++) {
- Message msg = new Message();
- msg.setBody(("body" + i).getBytes());
- msg.setTopic(topic);
- msg.setTags("TAG1");
- msg.setKeys(String.valueOf(System.currentTimeMillis()));
- messages.add(msg);
- String properties = messageProperties2String(msg.getProperties());
- byte[] propertiesBytes = properties.getBytes(CHARSET_UTF8);
- short propertiesLength = (short) propertiesBytes.length;
- final byte[] topicData = msg.getTopic().getBytes(MessageDecoder.CHARSET_UTF8);
- final int topicLength = topicData.length;
- msgLengthArr[j] = calIPv6HostMsgLength(msg.getBody().length, topicLength, propertiesLength) + msgLengthArr[j - 1];
- j++;
- }
- byte[] batchMessageBody = MessageDecoder.encodeMessages(messages);
- MessageExtBatch messageExtBatch = new MessageExtBatch();
- messageExtBatch.setTopic(topic);
- messageExtBatch.setQueueId(queue);
- messageExtBatch.setBody(batchMessageBody);
- messageExtBatch.setMsgId("24084004018081003FAA1DDE2B3F898A00002A9F0000000000000CA0");
- messageExtBatch.setBornTimestamp(System.currentTimeMillis());
- messageExtBatch.setSysFlag(0);
- messageExtBatch.setBornHostV6Flag();
- messageExtBatch.setStoreHostAddressV6Flag();
- messageExtBatch.setStoreHost(new InetSocketAddress("1050:0000:0000:0000:0005:0600:300c:326b", 125));
- messageExtBatch.setBornHost(new InetSocketAddress("::1", 126));
-
- PutMessageResult putMessageResult = messageStore.putMessages(messageExtBatch);
- assertThat(putMessageResult.isOk()).isTrue();
-
- Thread.sleep(3 * 1000);
-
- for (long i = 0; i < 10; i++) {
- MessageExt messageExt = messageStore.lookMessageByOffset(msgLengthArr[(int) i]);
- assertThat(messageExt).isNotNull();
- GetMessageResult result = messageStore.getMessage("batch_write_group", topic, queue, i, 1024 * 1024, null);
- assertThat(result).isNotNull();
- assertThat(result.getStatus()).isEqualTo(GetMessageStatus.FOUND);
- result.release();
- }
-
- }
-
private int calMsgLength(int bodyLength, int topicLength, int propertiesLength) {
final int msgLen = 4 //TOTALSIZE
+ 4 //MAGICCODE
@@ -200,28 +147,6 @@ public class BatchPutMessageTest {
return msgLen;
}
- private int calIPv6HostMsgLength(int bodyLength, int topicLength, int propertiesLength) {
- final int msgLen = 4 //TOTALSIZE
- + 4 //MAGICCODE
- + 4 //BODYCRC
- + 4 //QUEUEID
- + 4 //FLAG
- + 8 //QUEUEOFFSET
- + 8 //PHYSICALOFFSET
- + 4 //SYSFLAG
- + 8 //BORNTIMESTAMP
- + 20 //BORNHOST
- + 8 //STORETIMESTAMP
- + 20 //STOREHOSTADDRESS
- + 4 //RECONSUMETIMES
- + 8 //Prepared Transaction Offset
- + 4 + (bodyLength > 0 ? bodyLength : 0) //BODY
- + 1 + topicLength //TOPIC
- + 2 + (propertiesLength > 0 ? propertiesLength : 0) //propertiesLength
- + 0;
- return msgLen;
- }
-
public String messageProperties2String(Map<String, String> properties) {
StringBuilder sb = new StringBuilder();
if (properties != null) {
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 7c57813..c973030 100644
--- a/store/src/test/java/org/apache/rocketmq/store/ConsumeQueueTest.java
+++ b/store/src/test/java/org/apache/rocketmq/store/ConsumeQueueTest.java
@@ -83,29 +83,6 @@ 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();
@@ -150,11 +127,7 @@ public class ConsumeQueueTest {
long totalMsgs = 200;
for (long i = 0; i < totalMsgs; i++) {
- if (i < totalMsgs / 2) {
- master.putMessage(buildMessage());
- } else {
- master.putMessage(buildIPv6HostMessage());
- }
+ master.putMessage(buildMessage());
}
}
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 fd13e71..7859774 100644
--- a/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java
+++ b/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java
@@ -24,7 +24,6 @@ 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;
@@ -117,19 +116,13 @@ public class DefaultMessageStoreTest {
@Test
public void testWriteAndRead() {
- long ipv4HostMsgs = 10;
- long ipv6HostMsgs = 10;
- long totalMsgs = ipv4HostMsgs + ipv6HostMsgs;
+ long totalMsgs = 10;
QUEUE_TOTAL = 1;
MessageBody = StoreMessage.getBytes();
- for (long i = 0; i < ipv4HostMsgs; i++) {
+ for (long i = 0; i < totalMsgs; 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++) {
@@ -141,7 +134,7 @@ public class DefaultMessageStoreTest {
}
@Test
- public void testLookMessageByOffset_OffsetIsFirst() {
+ public void should_look_message_successfully_when_offset_is_first() {
final int totalCount = 10;
int queueId = new Random().nextInt(10);
String topic = "FooBar";
@@ -157,7 +150,7 @@ public class DefaultMessageStoreTest {
}
@Test
- public void testLookMessageByOffset_OffsetIsLast() {
+ public void should_look_message_successfully_when_offset_is_last() {
final int totalCount = 10;
int queueId = new Random().nextInt(10);
String topic = "FooBar";
@@ -171,7 +164,7 @@ public class DefaultMessageStoreTest {
}
@Test
- public void testLookMessageByOffset_OffsetIsOutOfBound() {
+ public void should_look_message_failed_and_return_null_when_offset_is_out_of_bound() {
final int totalCount = 10;
int queueId = new Random().nextInt(10);
String topic = "FooBar";
@@ -184,7 +177,7 @@ public class DefaultMessageStoreTest {
}
@Test
- public void testGetOffsetInQueueByTime() {
+ public void should_get_consume_queue_offset_successfully_when_incomming_by_timestamp() throws InterruptedException {
final int totalCount = 10;
int queueId = 0;
String topic = "FooBar";
@@ -203,7 +196,7 @@ public class DefaultMessageStoreTest {
}
@Test
- public void testGetOffsetInQueueByTime_TimestampIsSkewing() {
+ public void should_get_consume_queue_offset_successfully_when_timestamp_is_skewing() throws InterruptedException {
final int totalCount = 10;
int queueId = 0;
String topic = "FooBar";
@@ -228,7 +221,7 @@ public class DefaultMessageStoreTest {
}
@Test
- public void testGetOffsetInQueueByTime_TimestampSkewingIsLarge() {
+ public void should_get_min_of_max_consume_queue_offset_when_timestamp_s_skewing_is_large() throws InterruptedException {
final int totalCount = 10;
int queueId = 0;
String topic = "FooBar";
@@ -254,7 +247,7 @@ public class DefaultMessageStoreTest {
}
@Test
- public void testGetOffsetInQueueByTime_ConsumeQueueNotFound1() {
+ public void should_return_zero_when_consume_queue_not_found() throws InterruptedException {
final int totalCount = 10;
int queueId = 0;
int wrongQueueId = 1;
@@ -270,7 +263,7 @@ public class DefaultMessageStoreTest {
}
@Test
- public void testGetOffsetInQueueByTime_ConsumeQueueNotFound2() {
+ public void should_return_negative_one_when_invoke_getMessageStoreTimeStamp_if_consume_queue_not_found() throws InterruptedException {
final int totalCount = 10;
int queueId = 0;
int wrongQueueId = 1;
@@ -285,7 +278,7 @@ public class DefaultMessageStoreTest {
}
@Test
- public void testGetOffsetInQueueByTime_ConsumeQueueOffsetNotExist() {
+ public void should_return_negative_one_when_invoke_getMessageStoreTimeStamp_if_consumeQueueOffset_not_exist() throws InterruptedException {
final int totalCount = 10;
int queueId = 0;
int wrongQueueId = 1;
@@ -300,8 +293,9 @@ public class DefaultMessageStoreTest {
assertThat(messageStoreTimeStamp).isEqualTo(-1);
}
+
@Test
- public void testGetMessageStoreTimeStamp() {
+ public void should_get_message_store_timestamp_successfully_when_incomming_by_topic_queueId_and_consumeQueueOffset() throws InterruptedException {
final int totalCount = 10;
int queueId = 0;
String topic = "FooBar";
@@ -310,7 +304,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());
@@ -318,14 +312,14 @@ public class DefaultMessageStoreTest {
}
@Test
- public void testGetStoreTime_ParamIsNull() {
+ public void should_return_negative_one_when_invoke_getStoreTime_if_incomming_param_is_null() {
long storeTime = getStoreTime(null);
assertThat(storeTime).isEqualTo(-1);
}
@Test
- public void testGetStoreTime_EverythingIsOk() {
+ public void should_get_store_time_successfully_when_invoke_getStoreTime_if_everything_is_ok() throws InterruptedException {
final int totalCount = 10;
int queueId = 0;
String topic = "FooBar";
@@ -343,7 +337,7 @@ public class DefaultMessageStoreTest {
}
@Test
- public void testGetStoreTime_PhyOffsetIsLessThanCommitLogMinOffset() {
+ public void should_return_negative_one_when_invoke_getStoreTime_if_phyOffset_is_less_than_commitLog_s_minOffset() {
long phyOffset = -10;
int size = 138;
ByteBuffer byteBuffer = ByteBuffer.allocate(100);
@@ -360,7 +354,7 @@ public class DefaultMessageStoreTest {
}
private DefaultMessageStore getDefaultMessageStore() {
- return (DefaultMessageStore) this.messageStore;
+ return (DefaultMessageStore)this.messageStore;
}
private AppendMessageResult[] putMessages(int totalCount, String topic, int queueId) {
@@ -371,9 +365,7 @@ public class DefaultMessageStoreTest {
AppendMessageResult[] appendMessageResultArray = new AppendMessageResult[totalCount];
for (int i = 0; i < totalCount; i++) {
String messageBody = buildMessageBodyByOffset(StoreMessage, i);
-
- MessageExtBrokerInner msgInner =
- i < totalCount / 2 ? buildMessage(messageBody.getBytes(), topic) : buildIPv6HostMessage(messageBody.getBytes(), topic);
+ MessageExtBrokerInner msgInner = buildMessage(messageBody.getBytes(), topic);
msgInner.setQueueId(queueId);
PutMessageResult result = messageStore.putMessage(msgInner);
appendMessageResultArray[i] = result.getAppendMessageResult();
@@ -382,7 +374,7 @@ public class DefaultMessageStoreTest {
try {
Thread.sleep(10);
} catch (InterruptedException e) {
- throw new RuntimeException("Thread sleep ERROR");
+ throw new RuntimeException("Thread sleep ERROR");
}
}
}
@@ -405,7 +397,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);
}
@@ -426,43 +418,10 @@ 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());
@@ -518,7 +477,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();
@@ -628,7 +587,7 @@ public class DefaultMessageStoreTest {
private class MyMessageArrivingListener implements MessageArrivingListener {
@Override
public void arriving(String topic, int queueId, long logicOffset, long tagsCode, long msgStoreTime,
- byte[] filterBitMap, Map<String, String> properties) {
+ byte[] filterBitMap, Map<String, String> properties) {
}
}
}
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 a736754..0dcb3b2 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,33 +59,6 @@ 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);
@@ -101,6 +74,7 @@ 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 f0b9205..d0829d1 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,8 +145,7 @@ public class DLedgerCommitlogTest extends MessageStoreTestBase {
List<PutMessageResult> results = new ArrayList<>();
for (int i = 0; i < 10; i++) {
- MessageExtBrokerInner msgInner =
- i < 5 ? buildMessage() : buildIPv6HostMessage();
+ MessageExtBrokerInner msgInner = buildMessage();
msgInner.setTopic(topic);
msgInner.setQueueId(0);
PutMessageResult putMessageResult = messageStore.putMessage(msgInner);
@@ -210,39 +209,5 @@ 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 5b0ca34..2da2fb7 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,7 +19,6 @@ 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;
@@ -119,7 +118,7 @@ public class MessageStoreTestBase extends StoreTestBase {
return defaultMessageStore;
}
- protected void doPutMessages(MessageStore messageStore, String topic, int queueId, int num, long beginLogicsOffset) throws UnknownHostException {
+ protected void doPutMessages(MessageStore messageStore, String topic, int queueId, int num, long beginLogicsOffset) {
for (int i = 0; i < num; i++) {
MessageExtBrokerInner msgInner = buildMessage();
msgInner.setTopic(topic);
diff --git a/test/pom.xml b/test/pom.xml
index c3d1318..87bb776 100644
--- a/test/pom.xml
+++ b/test/pom.xml
@@ -20,7 +20,7 @@
<parent>
<artifactId>rocketmq-all</artifactId>
<groupId>org.apache.rocketmq</groupId>
- <version>4.6.0-SNAPSHOT</version>
+ <version>4.5.2-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/tools/pom.xml b/tools/pom.xml
index 061f3c7..ac6f6f4 100644
--- a/tools/pom.xml
+++ b/tools/pom.xml
@@ -19,7 +19,7 @@
<parent>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-all</artifactId>
- <version>4.6.0-SNAPSHOT</version>
+ <version>4.5.2-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>