You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@rocketmq.apache.org by GitBox <gi...@apache.org> on 2019/01/10 12:54:47 UTC
[GitHub] wangjuneng closed pull request #679: Develop
wangjuneng closed pull request #679: Develop
URL: https://github.com/apache/rocketmq/pull/679
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/.gitignore b/.gitignore
index 80c6f5698..8abdfd8fd 100644
--- a/.gitignore
+++ b/.gitignore
@@ -10,4 +10,6 @@ devenv
*.versionsBackup
!NOTICE-BIN
!LICENSE-BIN
-.DS_Store
\ No newline at end of file
+.DS_Store
+localbin
+nohup.out
diff --git a/README.md b/README.md
index 7518f06a8..686ba4629 100644
--- a/README.md
+++ b/README.md
@@ -38,7 +38,6 @@ It offers a variety of features:
## Apache RocketMQ Community
* [RocketMQ Community Projects](https://github.com/apache/rocketmq-externals)
-
----------
## Contributing
diff --git a/acl/pom.xml b/acl/pom.xml
new file mode 100644
index 000000000..4978ad5e9
--- /dev/null
+++ b/acl/pom.xml
@@ -0,0 +1,74 @@
+<!-- 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. -->
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
+ xmlns="http://maven.apache.org/POM/4.0.0">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-all</artifactId>
+ <version>4.4.0-SNAPSHOT</version>
+ </parent>
+ <artifactId>rocketmq-acl</artifactId>
+ <name>rocketmq-acl ${project.version}</name>
+
+ <url>http://maven.apache.org</url>
+ <properties>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ </properties>
+ <dependencies>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>rocketmq-remoting</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>rocketmq-logging</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>rocketmq-common</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>rocketmq-srvutil</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.yaml</groupId>
+ <artifactId>snakeyaml</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>commons-codec</groupId>
+ <artifactId>commons-codec</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-lang3</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>logback-classic</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>logback-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ </dependencies>
+</project>
diff --git a/acl/src/main/java/org/apache/rocketmq/acl/AccessResource.java b/acl/src/main/java/org/apache/rocketmq/acl/AccessResource.java
new file mode 100644
index 000000000..e30febc57
--- /dev/null
+++ b/acl/src/main/java/org/apache/rocketmq/acl/AccessResource.java
@@ -0,0 +1,21 @@
+/*
+ * 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.acl;
+
+public interface AccessResource {
+}
diff --git a/acl/src/main/java/org/apache/rocketmq/acl/AccessValidator.java b/acl/src/main/java/org/apache/rocketmq/acl/AccessValidator.java
new file mode 100644
index 000000000..c915cf35d
--- /dev/null
+++ b/acl/src/main/java/org/apache/rocketmq/acl/AccessValidator.java
@@ -0,0 +1,38 @@
+/*
+ * 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.acl;
+
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+
+public interface AccessValidator {
+ /**
+ * Parse to get the AccessResource(user, resource, needed permission)
+ *
+ * @param request
+ * @param remoteAddr
+ * @return Plain access resource result,include access key,signature and some other access attributes.
+ */
+ AccessResource parse(RemotingCommand request, String remoteAddr);
+
+ /**
+ * Validate the access resource.
+ *
+ * @param accessResource
+ */
+ void validate(AccessResource accessResource);
+}
diff --git a/acl/src/main/java/org/apache/rocketmq/acl/common/AclClientRPCHook.java b/acl/src/main/java/org/apache/rocketmq/acl/common/AclClientRPCHook.java
new file mode 100644
index 000000000..9e5bf1fb5
--- /dev/null
+++ b/acl/src/main/java/org/apache/rocketmq/acl/common/AclClientRPCHook.java
@@ -0,0 +1,98 @@
+/*
+ * 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.acl.common;
+
+import java.lang.reflect.Field;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.rocketmq.remoting.CommandCustomHeader;
+import org.apache.rocketmq.remoting.RPCHook;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+
+import static org.apache.rocketmq.acl.common.SessionCredentials.ACCESS_KEY;
+import static org.apache.rocketmq.acl.common.SessionCredentials.SECURITY_TOKEN;
+import static org.apache.rocketmq.acl.common.SessionCredentials.SIGNATURE;
+
+public class AclClientRPCHook implements RPCHook {
+ private final SessionCredentials sessionCredentials;
+ protected ConcurrentHashMap<Class<? extends CommandCustomHeader>, Field[]> fieldCache =
+ new ConcurrentHashMap<Class<? extends CommandCustomHeader>, Field[]>();
+
+ public AclClientRPCHook(SessionCredentials sessionCredentials) {
+ this.sessionCredentials = sessionCredentials;
+ }
+
+ @Override
+ public void doBeforeRequest(String remoteAddr, RemotingCommand request) {
+ byte[] total = AclUtils.combineRequestContent(request,
+ parseRequestContent(request, sessionCredentials.getAccessKey(), sessionCredentials.getSecurityToken()));
+ String signature = AclUtils.calSignature(total, sessionCredentials.getSecretKey());
+ request.addExtField(SIGNATURE, signature);
+ request.addExtField(ACCESS_KEY, sessionCredentials.getAccessKey());
+
+ // The SecurityToken value is unneccessary,user can choose this one.
+ if (sessionCredentials.getSecurityToken() != null) {
+ request.addExtField(SECURITY_TOKEN, sessionCredentials.getSecurityToken());
+ }
+ }
+
+ @Override
+ public void doAfterResponse(String remoteAddr, RemotingCommand request, RemotingCommand response) {
+
+ }
+
+ protected SortedMap<String, String> parseRequestContent(RemotingCommand request, String ak, String securityToken) {
+ CommandCustomHeader header = request.readCustomHeader();
+ // Sort property
+ SortedMap<String, String> map = new TreeMap<String, String>();
+ map.put(ACCESS_KEY, ak);
+ if (securityToken != null) {
+ map.put(SECURITY_TOKEN, securityToken);
+ }
+ try {
+ // Add header properties
+ if (null != header) {
+ Field[] fields = fieldCache.get(header.getClass());
+ if (null == fields) {
+ fields = header.getClass().getDeclaredFields();
+ for (Field field : fields) {
+ field.setAccessible(true);
+ }
+ Field[] tmp = fieldCache.putIfAbsent(header.getClass(), fields);
+ if (null != tmp) {
+ fields = tmp;
+ }
+ }
+
+ for (Field field : fields) {
+ Object value = field.get(header);
+ if (null != value && !field.isSynthetic()) {
+ map.put(field.getName(), value.toString());
+ }
+ }
+ }
+ return map;
+ } catch (Exception e) {
+ throw new RuntimeException("incompatible exception.", e);
+ }
+ }
+
+ public SessionCredentials getSessionCredentials() {
+ return sessionCredentials;
+ }
+}
diff --git a/acl/src/main/java/org/apache/rocketmq/acl/common/AclException.java b/acl/src/main/java/org/apache/rocketmq/acl/common/AclException.java
new file mode 100644
index 000000000..54579d48a
--- /dev/null
+++ b/acl/src/main/java/org/apache/rocketmq/acl/common/AclException.java
@@ -0,0 +1,66 @@
+/*
+ * 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.acl.common;
+
+public class AclException extends RuntimeException {
+ private static final long serialVersionUID = -7256002576788700354L;
+
+ private String status;
+ private int code;
+
+ public AclException(String status, int code) {
+ super();
+ this.status = status;
+ this.code = code;
+ }
+
+ public AclException(String status, int code, String message) {
+ super(message);
+ this.status = status;
+ this.code = code;
+ }
+
+ public AclException(String message) {
+ super(message);
+ }
+
+ public AclException(String message, Throwable throwable) {
+ super(message, throwable);
+ }
+
+ public AclException(String status, int code, String message, Throwable throwable) {
+ super(message, throwable);
+ this.status = status;
+ this.code = code;
+ }
+
+ public String getStatus() {
+ return status;
+ }
+
+ public void setStatus(String status) {
+ this.status = status;
+ }
+
+ public int getCode() {
+ return code;
+ }
+
+ public void setCode(int code) {
+ this.code = code;
+ }
+}
diff --git a/acl/src/main/java/org/apache/rocketmq/acl/common/AclSigner.java b/acl/src/main/java/org/apache/rocketmq/acl/common/AclSigner.java
new file mode 100644
index 000000000..61e935066
--- /dev/null
+++ b/acl/src/main/java/org/apache/rocketmq/acl/common/AclSigner.java
@@ -0,0 +1,88 @@
+/*
+ * 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.acl.common;
+
+import java.nio.charset.Charset;
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+import org.apache.commons.codec.binary.Base64;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
+
+public class AclSigner {
+ public static final Charset DEFAULT_CHARSET = Charset.forName("UTF-8");
+ public static final SigningAlgorithm DEFAULT_ALGORITHM = SigningAlgorithm.HmacSHA1;
+ private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.ROCKETMQ_AUTHORIZE_LOGGER_NAME);
+ private static final int CAL_SIGNATURE_FAILED = 10015;
+ private static final String CAL_SIGNATURE_FAILED_MSG = "[%s:signature-failed] unable to calculate a request signature. error=%s";
+
+ public static String calSignature(String data, String key) throws AclException {
+ return calSignature(data, key, DEFAULT_ALGORITHM, DEFAULT_CHARSET);
+ }
+
+ public static String calSignature(String data, String key, SigningAlgorithm algorithm,
+ Charset charset) throws AclException {
+ return signAndBase64Encode(data, key, algorithm, charset);
+ }
+
+ private static String signAndBase64Encode(String data, String key, SigningAlgorithm algorithm, Charset charset)
+ throws AclException {
+ try {
+ byte[] signature = sign(data.getBytes(charset), key.getBytes(charset), algorithm);
+ return new String(Base64.encodeBase64(signature), DEFAULT_CHARSET);
+ } catch (Exception e) {
+ String message = String.format(CAL_SIGNATURE_FAILED_MSG, CAL_SIGNATURE_FAILED, e.getMessage());
+ log.error(message, e);
+ throw new AclException("CAL_SIGNATURE_FAILED", CAL_SIGNATURE_FAILED, message, e);
+ }
+ }
+
+ private static byte[] sign(byte[] data, byte[] key, SigningAlgorithm algorithm) throws AclException {
+ try {
+ Mac mac = Mac.getInstance(algorithm.toString());
+ mac.init(new SecretKeySpec(key, algorithm.toString()));
+ return mac.doFinal(data);
+ } catch (Exception e) {
+ String message = String.format(CAL_SIGNATURE_FAILED_MSG, CAL_SIGNATURE_FAILED, e.getMessage());
+ log.error(message, e);
+ throw new AclException("CAL_SIGNATURE_FAILED", CAL_SIGNATURE_FAILED, message, e);
+ }
+ }
+
+ public static String calSignature(byte[] data, String key) throws AclException {
+ return calSignature(data, key, DEFAULT_ALGORITHM, DEFAULT_CHARSET);
+ }
+
+ public static String calSignature(byte[] data, String key, SigningAlgorithm algorithm,
+ Charset charset) throws AclException {
+ return signAndBase64Encode(data, key, algorithm, charset);
+ }
+
+ private static String signAndBase64Encode(byte[] data, String key, SigningAlgorithm algorithm, Charset charset)
+ throws AclException {
+ try {
+ byte[] signature = sign(data, key.getBytes(charset), algorithm);
+ return new String(Base64.encodeBase64(signature), DEFAULT_CHARSET);
+ } catch (Exception e) {
+ String message = String.format(CAL_SIGNATURE_FAILED_MSG, CAL_SIGNATURE_FAILED, e.getMessage());
+ log.error(message, e);
+ throw new AclException("CAL_SIGNATURE_FAILED", CAL_SIGNATURE_FAILED, message, e);
+ }
+ }
+
+}
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
new file mode 100644
index 000000000..1a618456f
--- /dev/null
+++ b/acl/src/main/java/org/apache/rocketmq/acl/common/AclUtils.java
@@ -0,0 +1,140 @@
+/*
+ * 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.acl.common;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.util.Map;
+import java.util.SortedMap;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.yaml.snakeyaml.Yaml;
+
+import static org.apache.rocketmq.acl.common.SessionCredentials.CHARSET;
+
+public class AclUtils {
+
+ public static byte[] combineRequestContent(RemotingCommand request, SortedMap<String, String> fieldsMap) {
+ try {
+ StringBuilder sb = new StringBuilder("");
+ for (Map.Entry<String, String> entry : fieldsMap.entrySet()) {
+ if (!SessionCredentials.SIGNATURE.equals(entry.getKey())) {
+ sb.append(entry.getValue());
+ }
+ }
+
+ return AclUtils.combineBytes(sb.toString().getBytes(CHARSET), request.getBody());
+ } catch (Exception e) {
+ throw new RuntimeException("incompatible exception.", e);
+ }
+ }
+
+ public static byte[] combineBytes(byte[] b1, byte[] b2) {
+ int size = (null != b1 ? b1.length : 0) + (null != b2 ? b2.length : 0);
+ byte[] total = new byte[size];
+ if (null != b1)
+ System.arraycopy(b1, 0, total, 0, b1.length);
+ if (null != b2)
+ System.arraycopy(b2, 0, total, b1.length, b2.length);
+ return total;
+ }
+
+ public static String calSignature(byte[] data, String secretKey) {
+ String signature = AclSigner.calSignature(data, secretKey);
+ return signature;
+ }
+
+ public static void verify(String netaddress, int index) {
+ if (!AclUtils.isScope(netaddress, index)) {
+ throw new AclException(String.format("netaddress examine scope Exception netaddress is %s", netaddress));
+ }
+ }
+
+ public static String[] getAddreeStrArray(String netaddress, String four) {
+ String[] fourStrArray = StringUtils.split(four.substring(1, four.length() - 1), ",");
+ String address = netaddress.substring(0, netaddress.indexOf("{"));
+ String[] addreeStrArray = new String[fourStrArray.length];
+ for (int i = 0; i < fourStrArray.length; i++) {
+ addreeStrArray[i] = address + fourStrArray[i];
+ }
+ return addreeStrArray;
+ }
+
+ public static boolean isScope(String num, int index) {
+ String[] strArray = StringUtils.split(num, ".");
+ if (strArray.length != 4) {
+ return false;
+ }
+ return isScope(strArray, index);
+
+ }
+
+ public static boolean isScope(String[] num, int index) {
+ if (num.length <= index) {
+
+ }
+ for (int i = 0; i < index; i++) {
+ if (!isScope(num[i])) {
+ return false;
+ }
+ }
+ return true;
+
+ }
+
+ public static boolean isScope(String num) {
+ return isScope(Integer.valueOf(num.trim()));
+ }
+
+ public static boolean isScope(int num) {
+ return num >= 0 && num <= 255;
+ }
+
+ public static boolean isAsterisk(String asterisk) {
+ return asterisk.indexOf('*') > -1;
+ }
+
+ public static boolean isColon(String colon) {
+ return colon.indexOf(',') > -1;
+ }
+
+ public static boolean isMinus(String minus) {
+ return minus.indexOf('-') > -1;
+
+ }
+
+ public static <T> T getYamlDataObject(String path, Class<T> clazz) {
+ Yaml ymal = new Yaml();
+ FileInputStream fis = null;
+ try {
+ fis = new FileInputStream(new File(path));
+ return ymal.loadAs(fis, clazz);
+ } catch (Exception e) {
+ throw new AclException(String.format("The file for Plain mode was not found , paths %s", path), e);
+ } finally {
+ if (fis != null) {
+ try {
+ fis.close();
+ } catch (IOException e) {
+ throw new AclException("close transport fileInputStream Exception", e);
+ }
+ }
+ }
+ }
+
+}
diff --git a/acl/src/main/java/org/apache/rocketmq/acl/common/Permission.java b/acl/src/main/java/org/apache/rocketmq/acl/common/Permission.java
new file mode 100644
index 000000000..0acc8e950
--- /dev/null
+++ b/acl/src/main/java/org/apache/rocketmq/acl/common/Permission.java
@@ -0,0 +1,96 @@
+/*
+ * 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.acl.common;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.acl.plain.PlainAccessResource;
+import org.apache.rocketmq.common.protocol.RequestCode;
+
+public class Permission {
+
+ public static final byte DENY = 1;
+ public static final byte ANY = 1 << 1;
+ public static final byte PUB = 1 << 2;
+ public static final byte SUB = 1 << 3;
+
+ public static final Set<Integer> ADMIN_CODE = new HashSet<Integer>();
+
+ static {
+ // UPDATE_AND_CREATE_TOPIC
+ ADMIN_CODE.add(RequestCode.UPDATE_AND_CREATE_TOPIC);
+ // UPDATE_BROKER_CONFIG
+ ADMIN_CODE.add(RequestCode.UPDATE_BROKER_CONFIG);
+ // DELETE_TOPIC_IN_BROKER
+ ADMIN_CODE.add(RequestCode.DELETE_TOPIC_IN_BROKER);
+ // UPDATE_AND_CREATE_SUBSCRIPTIONGROUP
+ ADMIN_CODE.add(RequestCode.UPDATE_AND_CREATE_SUBSCRIPTIONGROUP);
+ // DELETE_SUBSCRIPTIONGROUP
+ ADMIN_CODE.add(RequestCode.DELETE_SUBSCRIPTIONGROUP);
+ }
+
+ public static boolean checkPermission(byte neededPerm, byte ownedPerm) {
+ if ((ownedPerm & DENY) > 0) {
+ return false;
+ }
+ if ((neededPerm & ANY) > 0) {
+ return ((ownedPerm & PUB) > 0) || ((ownedPerm & SUB) > 0);
+ }
+ return (neededPerm & ownedPerm) > 0;
+ }
+
+ public static byte parsePermFromString(String permString) {
+ if (permString == null) {
+ return Permission.DENY;
+ }
+ switch (permString.trim()) {
+ case "PUB":
+ return Permission.PUB;
+ case "SUB":
+ return Permission.SUB;
+ case "PUB|SUB":
+ return Permission.PUB | Permission.SUB;
+ case "SUB|PUB":
+ return Permission.PUB | Permission.SUB;
+ case "DENY":
+ return Permission.DENY;
+ default:
+ return Permission.DENY;
+ }
+ }
+
+ public static void parseResourcePerms(PlainAccessResource plainAccessResource, Boolean isTopic,
+ List<String> resources) {
+ if (resources == null || resources.isEmpty()) {
+ return;
+ }
+ for (String resource : resources) {
+ String[] items = StringUtils.split(resource, "=");
+ if (items.length == 2) {
+ plainAccessResource.addResourceAndPerm(isTopic ? items[0].trim() : PlainAccessResource.getRetryTopic(items[0].trim()), parsePermFromString(items[1].trim()));
+ } else {
+ throw new AclException(String.format("Parse resource permission failed for %s:%s", isTopic ? "topic" : "group", resource));
+ }
+ }
+ }
+
+ public static boolean needAdminPerm(Integer code) {
+ return ADMIN_CODE.contains(code);
+ }
+}
diff --git a/acl/src/main/java/org/apache/rocketmq/acl/common/SessionCredentials.java b/acl/src/main/java/org/apache/rocketmq/acl/common/SessionCredentials.java
new file mode 100644
index 000000000..33a8a3435
--- /dev/null
+++ b/acl/src/main/java/org/apache/rocketmq/acl/common/SessionCredentials.java
@@ -0,0 +1,163 @@
+/*
+ * 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.acl.common;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.util.Properties;
+import org.apache.rocketmq.common.MixAll;
+
+public class SessionCredentials {
+ public static final Charset CHARSET = Charset.forName("UTF-8");
+ public static final String ACCESS_KEY = "AccessKey";
+ public static final String SECRET_KEY = "SecretKey";
+ public static final String SIGNATURE = "Signature";
+ public static final String SECURITY_TOKEN = "SecurityToken";
+
+ public static final String KEY_FILE = System.getProperty("rocketmq.client.keyFile",
+ System.getProperty("user.home") + File.separator + "key");
+
+ private String accessKey;
+ private String secretKey;
+ private String securityToken;
+ private String signature;
+
+ public SessionCredentials() {
+ String keyContent = null;
+ try {
+ keyContent = MixAll.file2String(KEY_FILE);
+ } catch (IOException ignore) {
+ }
+ if (keyContent != null) {
+ Properties prop = MixAll.string2Properties(keyContent);
+ if (prop != null) {
+ this.updateContent(prop);
+ }
+ }
+ }
+
+ public SessionCredentials(String accessKey, String secretKey) {
+ this.accessKey = accessKey;
+ this.secretKey = secretKey;
+ }
+
+ public SessionCredentials(String accessKey, String secretKey, String securityToken) {
+ this(accessKey, secretKey);
+ this.securityToken = securityToken;
+ }
+
+ public void updateContent(Properties prop) {
+ {
+ String value = prop.getProperty(ACCESS_KEY);
+ if (value != null) {
+ this.accessKey = value.trim();
+ }
+ }
+ {
+ String value = prop.getProperty(SECRET_KEY);
+ if (value != null) {
+ this.secretKey = value.trim();
+ }
+ }
+ {
+ String value = prop.getProperty(SECURITY_TOKEN);
+ if (value != null) {
+ this.securityToken = value.trim();
+ }
+ }
+ }
+
+ public String getAccessKey() {
+ return accessKey;
+ }
+
+ public void setAccessKey(String accessKey) {
+ this.accessKey = accessKey;
+ }
+
+ public String getSecretKey() {
+ return secretKey;
+ }
+
+ public void setSecretKey(String secretKey) {
+ this.secretKey = secretKey;
+ }
+
+ public String getSignature() {
+ return signature;
+ }
+
+ public void setSignature(String signature) {
+ this.signature = signature;
+ }
+
+ public String getSecurityToken() {
+ return securityToken;
+ }
+
+ public void setSecurityToken(final String securityToken) {
+ this.securityToken = securityToken;
+ }
+
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = 1;
+ result = prime * result + ((accessKey == null) ? 0 : accessKey.hashCode());
+ result = prime * result + ((secretKey == null) ? 0 : secretKey.hashCode());
+ result = prime * result + ((signature == null) ? 0 : signature.hashCode());
+ return result;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj)
+ return true;
+ if (obj == null)
+ return false;
+ if (getClass() != obj.getClass())
+ return false;
+
+ SessionCredentials other = (SessionCredentials) obj;
+ if (accessKey == null) {
+ if (other.accessKey != null)
+ return false;
+ } else if (!accessKey.equals(other.accessKey))
+ return false;
+
+ if (secretKey == null) {
+ if (other.secretKey != null)
+ return false;
+ } else if (!secretKey.equals(other.secretKey))
+ return false;
+
+ if (signature == null) {
+ if (other.signature != null)
+ return false;
+ } else if (!signature.equals(other.signature))
+ return false;
+
+ return true;
+ }
+
+ @Override
+ public String toString() {
+ return "SessionCredentials [accessKey=" + accessKey + ", secretKey=" + secretKey + ", signature="
+ + signature + ", SecurityToken=" + securityToken + "]";
+ }
+}
\ No newline at end of file
diff --git a/acl/src/main/java/org/apache/rocketmq/acl/common/SigningAlgorithm.java b/acl/src/main/java/org/apache/rocketmq/acl/common/SigningAlgorithm.java
new file mode 100644
index 000000000..bfed7b2f2
--- /dev/null
+++ b/acl/src/main/java/org/apache/rocketmq/acl/common/SigningAlgorithm.java
@@ -0,0 +1,24 @@
+/*
+ * 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.acl.common;
+
+public enum SigningAlgorithm {
+ HmacSHA1,
+ HmacSHA256,
+ HmacMD5;
+
+}
diff --git a/acl/src/main/java/org/apache/rocketmq/acl/plain/PlainAccessResource.java b/acl/src/main/java/org/apache/rocketmq/acl/plain/PlainAccessResource.java
new file mode 100644
index 000000000..00072e8e2
--- /dev/null
+++ b/acl/src/main/java/org/apache/rocketmq/acl/plain/PlainAccessResource.java
@@ -0,0 +1,201 @@
+/*
+ * 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.acl.plain;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.commons.lang3.builder.ToStringBuilder;
+import org.apache.rocketmq.acl.AccessResource;
+import org.apache.rocketmq.common.MixAll;
+
+public class PlainAccessResource implements AccessResource {
+
+ // Identify the user
+ private String accessKey;
+
+ private String secretKey;
+
+ private String whiteRemoteAddress;
+
+ private boolean admin;
+
+ private byte defaultTopicPerm = 1;
+
+ private byte defaultGroupPerm = 1;
+
+ private Map<String, Byte> resourcePermMap;
+
+ private RemoteAddressStrategy remoteAddressStrategy;
+
+ private int requestCode;
+
+ //the content to calculate the content
+ private byte[] content;
+
+ private String signature;
+
+ private String secretToken;
+
+ private String recognition;
+
+ public PlainAccessResource() {
+ }
+
+ public static boolean isRetryTopic(String topic) {
+ return null != topic && topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX);
+ }
+
+ public static String printStr(String resource, boolean isGroup) {
+ if (resource == null) {
+ return null;
+ }
+ if (isGroup) {
+ return String.format("%s:%s", "group", getGroupFromRetryTopic(resource));
+ } else {
+ return String.format("%s:%s", "topic", resource);
+ }
+ }
+
+ public static String getGroupFromRetryTopic(String retryTopic) {
+ if (retryTopic == null) {
+ return null;
+ }
+ return retryTopic.substring(MixAll.RETRY_GROUP_TOPIC_PREFIX.length());
+ }
+
+ public static String getRetryTopic(String group) {
+ if (group == null) {
+ return null;
+ }
+ return MixAll.getRetryTopic(group);
+ }
+
+ public void addResourceAndPerm(String resource, byte perm) {
+ if (resource == null) {
+ return;
+ }
+ if (resourcePermMap == null) {
+ resourcePermMap = new HashMap<>();
+ }
+ resourcePermMap.put(resource, perm);
+ }
+
+ public String getAccessKey() {
+ return accessKey;
+ }
+
+ public void setAccessKey(String accessKey) {
+ this.accessKey = accessKey;
+ }
+
+ public String getSecretKey() {
+ return secretKey;
+ }
+
+ public void setSecretKey(String secretKey) {
+ this.secretKey = secretKey;
+ }
+
+ public String getWhiteRemoteAddress() {
+ return whiteRemoteAddress;
+ }
+
+ public void setWhiteRemoteAddress(String whiteRemoteAddress) {
+ this.whiteRemoteAddress = whiteRemoteAddress;
+ }
+
+ public boolean isAdmin() {
+ return admin;
+ }
+
+ public void setAdmin(boolean admin) {
+ this.admin = admin;
+ }
+
+ public byte getDefaultTopicPerm() {
+ return defaultTopicPerm;
+ }
+
+ public void setDefaultTopicPerm(byte defaultTopicPerm) {
+ this.defaultTopicPerm = defaultTopicPerm;
+ }
+
+ public byte getDefaultGroupPerm() {
+ return defaultGroupPerm;
+ }
+
+ public void setDefaultGroupPerm(byte defaultGroupPerm) {
+ this.defaultGroupPerm = defaultGroupPerm;
+ }
+
+ public Map<String, Byte> getResourcePermMap() {
+ return resourcePermMap;
+ }
+
+ public String getRecognition() {
+ return recognition;
+ }
+
+ public void setRecognition(String recognition) {
+ this.recognition = recognition;
+ }
+
+ public int getRequestCode() {
+ return requestCode;
+ }
+
+ public void setRequestCode(int requestCode) {
+ this.requestCode = requestCode;
+ }
+
+ public String getSecretToken() {
+ return secretToken;
+ }
+
+ public void setSecretToken(String secretToken) {
+ this.secretToken = secretToken;
+ }
+
+ public RemoteAddressStrategy getRemoteAddressStrategy() {
+ return remoteAddressStrategy;
+ }
+
+ public void setRemoteAddressStrategy(RemoteAddressStrategy remoteAddressStrategy) {
+ this.remoteAddressStrategy = remoteAddressStrategy;
+ }
+
+ public String getSignature() {
+ return signature;
+ }
+
+ public void setSignature(String signature) {
+ this.signature = signature;
+ }
+
+ @Override
+ public String toString() {
+ return ToStringBuilder.reflectionToString(this);
+ }
+
+ public byte[] getContent() {
+ return content;
+ }
+
+ public void setContent(byte[] content) {
+ this.content = content;
+ }
+}
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
new file mode 100644
index 000000000..34bb1b439
--- /dev/null
+++ b/acl/src/main/java/org/apache/rocketmq/acl/plain/PlainAccessValidator.java
@@ -0,0 +1,130 @@
+/*
+ * 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.acl.plain;
+
+import java.util.Map;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import org.apache.rocketmq.acl.AccessResource;
+import org.apache.rocketmq.acl.AccessValidator;
+import org.apache.rocketmq.acl.common.AclException;
+import org.apache.rocketmq.acl.common.AclUtils;
+import org.apache.rocketmq.acl.common.Permission;
+import org.apache.rocketmq.acl.common.SessionCredentials;
+import org.apache.rocketmq.common.protocol.RequestCode;
+import org.apache.rocketmq.common.protocol.header.GetConsumerListByGroupRequestHeader;
+import org.apache.rocketmq.common.protocol.header.UnregisterClientRequestHeader;
+import org.apache.rocketmq.common.protocol.header.UpdateConsumerOffsetRequestHeader;
+import org.apache.rocketmq.common.protocol.heartbeat.ConsumerData;
+import org.apache.rocketmq.common.protocol.heartbeat.HeartbeatData;
+import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+
+import static org.apache.rocketmq.acl.plain.PlainAccessResource.getRetryTopic;
+
+public class PlainAccessValidator implements AccessValidator {
+
+ private PlainPermissionLoader aclPlugEngine;
+
+ public PlainAccessValidator() {
+ aclPlugEngine = new PlainPermissionLoader();
+ }
+
+ @Override
+ public AccessResource parse(RemotingCommand request, String remoteAddr) {
+ PlainAccessResource accessResource = new PlainAccessResource();
+ if (remoteAddr != null && remoteAddr.contains(":")) {
+ accessResource.setWhiteRemoteAddress(remoteAddr.split(":")[0]);
+ } else {
+ accessResource.setWhiteRemoteAddress(remoteAddr);
+ }
+ accessResource.setRequestCode(request.getCode());
+ accessResource.setAccessKey(request.getExtFields().get(SessionCredentials.ACCESS_KEY));
+ accessResource.setSignature(request.getExtFields().get(SessionCredentials.SIGNATURE));
+ accessResource.setSecretToken(request.getExtFields().get(SessionCredentials.SECURITY_TOKEN));
+
+ try {
+ switch (request.getCode()) {
+ case RequestCode.SEND_MESSAGE:
+ accessResource.addResourceAndPerm(request.getExtFields().get("topic"), Permission.PUB);
+ break;
+ case RequestCode.SEND_MESSAGE_V2:
+ accessResource.addResourceAndPerm(request.getExtFields().get("b"), Permission.PUB);
+ break;
+ case RequestCode.CONSUMER_SEND_MSG_BACK:
+ accessResource.addResourceAndPerm(request.getExtFields().get("originTopic"), Permission.PUB);
+ accessResource.addResourceAndPerm(getRetryTopic(request.getExtFields().get("group")), Permission.SUB);
+ break;
+ case RequestCode.PULL_MESSAGE:
+ accessResource.addResourceAndPerm(request.getExtFields().get("topic"), Permission.SUB);
+ accessResource.addResourceAndPerm(getRetryTopic(request.getExtFields().get("consumerGroup")), Permission.SUB);
+ break;
+ case RequestCode.QUERY_MESSAGE:
+ accessResource.addResourceAndPerm(request.getExtFields().get("topic"), Permission.SUB);
+ break;
+ case RequestCode.HEART_BEAT:
+ HeartbeatData heartbeatData = HeartbeatData.decode(request.getBody(), HeartbeatData.class);
+ for (ConsumerData data : heartbeatData.getConsumerDataSet()) {
+ accessResource.addResourceAndPerm(getRetryTopic(data.getGroupName()), Permission.SUB);
+ for (SubscriptionData subscriptionData : data.getSubscriptionDataSet()) {
+ accessResource.addResourceAndPerm(subscriptionData.getTopic(), Permission.SUB);
+ }
+ }
+ break;
+ case RequestCode.UNREGISTER_CLIENT:
+ final UnregisterClientRequestHeader unregisterClientRequestHeader =
+ (UnregisterClientRequestHeader) request
+ .decodeCommandCustomHeader(UnregisterClientRequestHeader.class);
+ accessResource.addResourceAndPerm(getRetryTopic(unregisterClientRequestHeader.getConsumerGroup()), Permission.SUB);
+ break;
+ case RequestCode.GET_CONSUMER_LIST_BY_GROUP:
+ final GetConsumerListByGroupRequestHeader getConsumerListByGroupRequestHeader =
+ (GetConsumerListByGroupRequestHeader) request
+ .decodeCommandCustomHeader(GetConsumerListByGroupRequestHeader.class);
+ accessResource.addResourceAndPerm(getRetryTopic(getConsumerListByGroupRequestHeader.getConsumerGroup()), Permission.SUB);
+ break;
+ case RequestCode.UPDATE_CONSUMER_OFFSET:
+ final UpdateConsumerOffsetRequestHeader updateConsumerOffsetRequestHeader =
+ (UpdateConsumerOffsetRequestHeader) request
+ .decodeCommandCustomHeader(UpdateConsumerOffsetRequestHeader.class);
+ accessResource.addResourceAndPerm(getRetryTopic(updateConsumerOffsetRequestHeader.getConsumerGroup()), Permission.SUB);
+ accessResource.addResourceAndPerm(updateConsumerOffsetRequestHeader.getTopic(), Permission.SUB);
+ break;
+ default:
+ break;
+
+ }
+ } catch (Throwable t) {
+ throw new AclException(t.getMessage(), t);
+ }
+ // Content
+ SortedMap<String, String> map = new TreeMap<String, String>();
+ for (Map.Entry<String, String> entry : request.getExtFields().entrySet()) {
+ if (!SessionCredentials.SIGNATURE.equals(entry.getKey())) {
+ map.put(entry.getKey(), entry.getValue());
+ }
+ }
+ accessResource.setContent(AclUtils.combineRequestContent(request, map));
+ return accessResource;
+ }
+
+ @Override
+ public void validate(AccessResource accessResource) {
+ aclPlugEngine.validate((PlainAccessResource) accessResource);
+ }
+
+}
diff --git a/acl/src/main/java/org/apache/rocketmq/acl/plain/PlainPermissionLoader.java b/acl/src/main/java/org/apache/rocketmq/acl/plain/PlainPermissionLoader.java
new file mode 100644
index 000000000..1da7380b6
--- /dev/null
+++ b/acl/src/main/java/org/apache/rocketmq/acl/plain/PlainPermissionLoader.java
@@ -0,0 +1,300 @@
+/*
+ * 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.acl.plain;
+
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import java.io.File;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import org.apache.rocketmq.acl.common.AclException;
+import org.apache.rocketmq.acl.common.AclUtils;
+import org.apache.rocketmq.acl.common.Permission;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
+import org.apache.rocketmq.srvutil.FileWatchService;
+
+public class PlainPermissionLoader {
+
+ private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME);
+
+ private static final String DEFAULT_PLAIN_ACL_FILE = "/conf/plain_acl.yml";
+
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ private String fileHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY,
+ System.getenv(MixAll.ROCKETMQ_HOME_ENV));
+
+ private String fileName = System.getProperty("rocketmq.acl.plain.file", DEFAULT_PLAIN_ACL_FILE);
+
+ private Map<String/** AccessKey **/, PlainAccessResource> plainAccessResourceMap = new HashMap<>();
+
+ private List<RemoteAddressStrategy> globalWhiteRemoteAddressStrategy = new ArrayList<>();
+
+ private RemoteAddressStrategyFactory remoteAddressStrategyFactory = new RemoteAddressStrategyFactory();
+
+ private boolean isWatchStart;
+
+ public PlainPermissionLoader() {
+ load();
+ watch();
+ }
+
+ public void load() {
+
+ Map<String, PlainAccessResource> plainAccessResourceMap = new HashMap<>();
+ List<RemoteAddressStrategy> globalWhiteRemoteAddressStrategy = new ArrayList<>();
+
+ JSONObject plainAclConfData = AclUtils.getYamlDataObject(fileHome + File.separator + fileName,
+ JSONObject.class);
+
+ if (plainAclConfData == null || plainAclConfData.isEmpty()) {
+ throw new AclException(String.format("%s file is not data", fileHome + File.separator + fileName));
+ }
+ log.info("Broker plain acl conf data is : ", plainAclConfData.toString());
+ JSONArray globalWhiteRemoteAddressesList = plainAclConfData.getJSONArray("globalWhiteRemoteAddresses");
+ if (globalWhiteRemoteAddressesList != null && !globalWhiteRemoteAddressesList.isEmpty()) {
+ for (int i = 0; i < globalWhiteRemoteAddressesList.size(); i++) {
+ globalWhiteRemoteAddressStrategy.add(remoteAddressStrategyFactory.
+ getRemoteAddressStrategy(globalWhiteRemoteAddressesList.getString(i)));
+ }
+ }
+
+ JSONArray accounts = plainAclConfData.getJSONArray("accounts");
+ if (accounts != null && !accounts.isEmpty()) {
+ List<PlainAccessConfig> plainAccessConfigList = accounts.toJavaList(PlainAccessConfig.class);
+ for (PlainAccessConfig plainAccessConfig : plainAccessConfigList) {
+ PlainAccessResource plainAccessResource = buildPlainAccessResource(plainAccessConfig);
+ plainAccessResourceMap.put(plainAccessResource.getAccessKey(),plainAccessResource);
+ }
+ }
+
+ this.globalWhiteRemoteAddressStrategy = globalWhiteRemoteAddressStrategy;
+ this.plainAccessResourceMap = plainAccessResourceMap;
+ }
+
+ private void watch() {
+ try {
+ String watchFilePath = fileHome + fileName;
+ FileWatchService fileWatchService = new FileWatchService(new String[] {watchFilePath}, new FileWatchService.Listener() {
+ @Override
+ public void onChanged(String path) {
+ log.info("The plain acl yml changed, reload the context");
+ load();
+ }
+ });
+ fileWatchService.start();
+ log.info("Succeed to start AclWatcherService");
+ this.isWatchStart = true;
+ } catch (Exception e) {
+ log.error("Failed to start AclWatcherService", e);
+ }
+ }
+
+ void checkPerm(PlainAccessResource needCheckedAccess, PlainAccessResource ownedAccess) {
+ if (Permission.needAdminPerm(needCheckedAccess.getRequestCode()) && !ownedAccess.isAdmin()) {
+ throw new AclException(String.format("Need admin permission for request code=%d, but accessKey=%s is not", needCheckedAccess.getRequestCode(), ownedAccess.getAccessKey()));
+ }
+ Map<String, Byte> needCheckedPermMap = needCheckedAccess.getResourcePermMap();
+ Map<String, Byte> ownedPermMap = ownedAccess.getResourcePermMap();
+
+ if (needCheckedPermMap == null) {
+ // If the needCheckedPermMap is null,then return
+ return;
+ }
+
+ for (Map.Entry<String, Byte> needCheckedEntry : needCheckedPermMap.entrySet()) {
+ String resource = needCheckedEntry.getKey();
+ Byte neededPerm = needCheckedEntry.getValue();
+ boolean isGroup = PlainAccessResource.isRetryTopic(resource);
+
+ if (!ownedPermMap.containsKey(resource)) {
+ // Check the default perm
+ byte ownedPerm = isGroup ? needCheckedAccess.getDefaultGroupPerm() :
+ needCheckedAccess.getDefaultTopicPerm();
+ if (!Permission.checkPermission(neededPerm, ownedPerm)) {
+ throw new AclException(String.format("No default permission for %s", PlainAccessResource.printStr(resource, isGroup)));
+ }
+ continue;
+ }
+ if (!Permission.checkPermission(neededPerm, ownedPermMap.get(resource))) {
+ throw new AclException(String.format("No default permission for %s", PlainAccessResource.printStr(resource, isGroup)));
+ }
+ }
+ }
+
+ void clearPermissionInfo() {
+ this.plainAccessResourceMap.clear();
+ this.globalWhiteRemoteAddressStrategy.clear();
+ }
+
+ public PlainAccessResource buildPlainAccessResource(PlainAccessConfig plainAccessConfig) throws AclException {
+ if (plainAccessConfig.getAccessKey() == null
+ || plainAccessConfig.getSecretKey() == null
+ || plainAccessConfig.getAccessKey().length() <= 6
+ || plainAccessConfig.getSecretKey().length() <= 6) {
+ throw new AclException(String.format(
+ "The accessKey=%s and secretKey=%s cannot be null and length should longer than 6",
+ plainAccessConfig.getAccessKey(), plainAccessConfig.getSecretKey()));
+ }
+ PlainAccessResource plainAccessResource = new PlainAccessResource();
+ plainAccessResource.setAccessKey(plainAccessConfig.getAccessKey());
+ plainAccessResource.setSecretKey(plainAccessConfig.getSecretKey());
+ plainAccessResource.setWhiteRemoteAddress(plainAccessConfig.getWhiteRemoteAddress());
+
+ plainAccessResource.setAdmin(plainAccessConfig.isAdmin());
+
+ plainAccessResource.setDefaultGroupPerm(Permission.parsePermFromString(plainAccessConfig.getDefaultGroupPerm()));
+ plainAccessResource.setDefaultTopicPerm(Permission.parsePermFromString(plainAccessConfig.getDefaultTopicPerm()));
+
+ Permission.parseResourcePerms(plainAccessResource, false, plainAccessConfig.getGroupPerms());
+ Permission.parseResourcePerms(plainAccessResource, true, plainAccessConfig.getTopicPerms());
+
+ plainAccessResource.setRemoteAddressStrategy(remoteAddressStrategyFactory.
+ getRemoteAddressStrategy(plainAccessResource.getWhiteRemoteAddress()));
+
+ return plainAccessResource;
+ }
+
+ public void validate(PlainAccessResource plainAccessResource) {
+
+ // Check the global white remote addr
+ for (RemoteAddressStrategy remoteAddressStrategy : globalWhiteRemoteAddressStrategy) {
+ if (remoteAddressStrategy.match(plainAccessResource)) {
+ return;
+ }
+ }
+
+ if (plainAccessResource.getAccessKey() == null) {
+ throw new AclException(String.format("No accessKey is configured"));
+ }
+
+ if (!plainAccessResourceMap.containsKey(plainAccessResource.getAccessKey())) {
+ throw new AclException(String.format("No acl config for %s", plainAccessResource.getAccessKey()));
+ }
+
+ // Check the white addr for accesskey
+ PlainAccessResource ownedAccess = plainAccessResourceMap.get(plainAccessResource.getAccessKey());
+ if (ownedAccess.getRemoteAddressStrategy().match(plainAccessResource)) {
+ return;
+ }
+
+ // Check the signature
+ String signature = AclUtils.calSignature(plainAccessResource.getContent(), ownedAccess.getSecretKey());
+ if (!signature.equals(plainAccessResource.getSignature())) {
+ throw new AclException(String.format("Check signature failed for accessKey=%s", plainAccessResource.getAccessKey()));
+ }
+ // Check perm of each resource
+
+ checkPerm(plainAccessResource, ownedAccess);
+ }
+
+ public boolean isWatchStart() {
+ return isWatchStart;
+ }
+
+ static class PlainAccessConfig {
+
+ private String accessKey;
+
+ private String secretKey;
+
+ private String whiteRemoteAddress;
+
+ private boolean admin;
+
+ private String defaultTopicPerm;
+
+ private String defaultGroupPerm;
+
+ private List<String> topicPerms;
+
+ private List<String> groupPerms;
+
+ public String getAccessKey() {
+ return accessKey;
+ }
+
+ public void setAccessKey(String accessKey) {
+ this.accessKey = accessKey;
+ }
+
+ public String getSecretKey() {
+ return secretKey;
+ }
+
+ public void setSecretKey(String secretKey) {
+ this.secretKey = secretKey;
+ }
+
+ public String getWhiteRemoteAddress() {
+ return whiteRemoteAddress;
+ }
+
+ public void setWhiteRemoteAddress(String whiteRemoteAddress) {
+ this.whiteRemoteAddress = whiteRemoteAddress;
+ }
+
+ public boolean isAdmin() {
+ return admin;
+ }
+
+ public void setAdmin(boolean admin) {
+ this.admin = admin;
+ }
+
+ public String getDefaultTopicPerm() {
+ return defaultTopicPerm;
+ }
+
+ public void setDefaultTopicPerm(String defaultTopicPerm) {
+ this.defaultTopicPerm = defaultTopicPerm;
+ }
+
+ public String getDefaultGroupPerm() {
+ return defaultGroupPerm;
+ }
+
+ public void setDefaultGroupPerm(String defaultGroupPerm) {
+ this.defaultGroupPerm = defaultGroupPerm;
+ }
+
+ public List<String> getTopicPerms() {
+ return topicPerms;
+ }
+
+ public void setTopicPerms(List<String> topicPerms) {
+ this.topicPerms = topicPerms;
+ }
+
+ public List<String> getGroupPerms() {
+ return groupPerms;
+ }
+
+ public void setGroupPerms(List<String> groupPerms) {
+ this.groupPerms = groupPerms;
+ }
+
+ }
+
+}
diff --git a/acl/src/main/java/org/apache/rocketmq/acl/plain/RemoteAddressStrategy.java b/acl/src/main/java/org/apache/rocketmq/acl/plain/RemoteAddressStrategy.java
new file mode 100644
index 000000000..8eab40c95
--- /dev/null
+++ b/acl/src/main/java/org/apache/rocketmq/acl/plain/RemoteAddressStrategy.java
@@ -0,0 +1,22 @@
+/*
+ * 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.acl.plain;
+
+public interface RemoteAddressStrategy {
+
+ boolean match(PlainAccessResource plainAccessResource);
+}
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
new file mode 100644
index 000000000..de29e92fd
--- /dev/null
+++ b/acl/src/main/java/org/apache/rocketmq/acl/plain/RemoteAddressStrategyFactory.java
@@ -0,0 +1,180 @@
+/*
+ * 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.acl.plain;
+
+import java.util.HashSet;
+import java.util.Set;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.acl.common.AclException;
+import org.apache.rocketmq.acl.common.AclUtils;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
+
+public class RemoteAddressStrategyFactory {
+
+ private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME);
+
+ public static final NullRemoteAddressStrategy NULL_NET_ADDRESS_STRATEGY = new NullRemoteAddressStrategy();
+
+ public static final BlankRemoteAddressStrategy BLANK_NET_ADDRESS_STRATEGY = new BlankRemoteAddressStrategy();
+
+ public RemoteAddressStrategy getRemoteAddressStrategy(PlainAccessResource plainAccessResource) {
+ return getRemoteAddressStrategy(plainAccessResource.getWhiteRemoteAddress());
+ }
+
+ public RemoteAddressStrategy getRemoteAddressStrategy(String remoteAddr) {
+ if (StringUtils.isBlank(remoteAddr)) {
+ return BLANK_NET_ADDRESS_STRATEGY;
+ }
+ if ("*".equals(remoteAddr)) {
+ return NULL_NET_ADDRESS_STRATEGY;
+ }
+ if (remoteAddr.endsWith("}")) {
+ String[] strArray = StringUtils.split(remoteAddr, ".");
+ String four = strArray[3];
+ if (!four.startsWith("{")) {
+ throw new AclException(String.format("MultipleRemoteAddressStrategy netaddress examine scope Exception netaddress", remoteAddr));
+ }
+ 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);
+ }
+ return new OneRemoteAddressStrategy(remoteAddr);
+
+ }
+
+ public static class NullRemoteAddressStrategy implements RemoteAddressStrategy {
+ @Override
+ public boolean match(PlainAccessResource plainAccessResource) {
+ return true;
+ }
+
+ }
+
+ public static class BlankRemoteAddressStrategy implements RemoteAddressStrategy {
+ @Override
+ public boolean match(PlainAccessResource plainAccessResource) {
+ return false;
+ }
+
+ }
+
+ public static class MultipleRemoteAddressStrategy implements RemoteAddressStrategy {
+
+ private final Set<String> multipleSet = new HashSet<>();
+
+ public MultipleRemoteAddressStrategy(String[] strArray) {
+ for (String netaddress : strArray) {
+ AclUtils.verify(netaddress, 4);
+ multipleSet.add(netaddress);
+ }
+ }
+
+ @Override
+ public boolean match(PlainAccessResource plainAccessResource) {
+ return multipleSet.contains(plainAccessResource.getWhiteRemoteAddress());
+ }
+
+ }
+
+ public static class OneRemoteAddressStrategy implements RemoteAddressStrategy {
+
+ private String netaddress;
+
+ public OneRemoteAddressStrategy(String netaddress) {
+ this.netaddress = netaddress;
+ AclUtils.verify(netaddress, 4);
+ }
+
+ @Override
+ public boolean match(PlainAccessResource plainAccessResource) {
+ return netaddress.equals(plainAccessResource.getWhiteRemoteAddress());
+ }
+
+ }
+
+ public static class RangeRemoteAddressStrategy implements RemoteAddressStrategy {
+
+ private String head;
+
+ private int start;
+
+ private int end;
+
+ private int index;
+
+ public RangeRemoteAddressStrategy(String remoteAddr) {
+ String[] strArray = StringUtils.split(remoteAddr, ".");
+ if (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();
+ }
+ }
+
+ private boolean analysis(String[] strArray, int index) {
+ String value = strArray[index].trim();
+ this.index = index;
+ if ("*".equals(value)) {
+ setValue(0, 255);
+ } 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.valueOf(valueArray[0]);
+ this.end = Integer.valueOf(valueArray[1]);
+ if (!(AclUtils.isScope(end) && AclUtils.isScope(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;
+ }
+
+ @Override
+ public boolean match(PlainAccessResource plainAccessResource) {
+ String netAddress = plainAccessResource.getWhiteRemoteAddress();
+ if (netAddress.startsWith(this.head)) {
+ String value;
+ if (index == 3) {
+ value = netAddress.substring(this.head.length());
+ } else {
+ value = netAddress.substring(this.head.length(), netAddress.lastIndexOf('.'));
+ }
+ 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/AclSignerTest.java b/acl/src/test/java/org/apache/rocketmq/acl/common/AclSignerTest.java
new file mode 100644
index 000000000..eec626357
--- /dev/null
+++ b/acl/src/test/java/org/apache/rocketmq/acl/common/AclSignerTest.java
@@ -0,0 +1,34 @@
+/*
+ * 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.acl.common;
+
+import org.junit.Test;
+
+public class AclSignerTest {
+
+ @Test(expected = Exception.class)
+ public void calSignatureExceptionTest(){
+ AclSigner.calSignature(new byte[]{},"");
+ }
+
+ @Test
+ public void calSignatureTest(){
+ AclSigner.calSignature("RocketMQ","12345678");
+ AclSigner.calSignature("RocketMQ".getBytes(),"12345678");
+ }
+
+}
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
new file mode 100644
index 000000000..72bcda6bb
--- /dev/null
+++ b/acl/src/test/java/org/apache/rocketmq/acl/common/AclUtilsTest.java
@@ -0,0 +1,141 @@
+/*
+ * 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.acl.common;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import org.apache.commons.lang3.StringUtils;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class AclUtilsTest {
+
+ @Test
+ public void getAddreeStrArray() {
+ String address = "1.1.1.{1,2,3,4}";
+ String[] addressArray = AclUtils.getAddreeStrArray(address, "{1,2,3,4}");
+ List<String> newAddressList = new ArrayList<>();
+ for (String a : addressArray) {
+ newAddressList.add(a);
+ }
+
+ List<String> addressList = new ArrayList<>();
+ addressList.add("1.1.1.1");
+ addressList.add("1.1.1.2");
+ addressList.add("1.1.1.3");
+ addressList.add("1.1.1.4");
+ Assert.assertEquals(newAddressList, addressList);
+ }
+
+ @Test
+ public void isScopeStringArray() {
+ String adderss = "12";
+
+ for (int i = 0; i < 6; i++) {
+ boolean isScope = AclUtils.isScope(adderss, 4);
+ if (i == 3) {
+ Assert.assertTrue(isScope);
+ } else {
+ Assert.assertFalse(isScope);
+ }
+ adderss = adderss + ".12";
+ }
+ }
+
+ @Test
+ public void isScopeArray() {
+ String[] adderss = StringUtils.split("12.12.12.12", ".");
+ boolean isScope = AclUtils.isScope(adderss, 4);
+ Assert.assertTrue(isScope);
+ isScope = AclUtils.isScope(adderss, 3);
+ Assert.assertTrue(isScope);
+
+ adderss = StringUtils.split("12.12.1222.1222", ".");
+ isScope = AclUtils.isScope(adderss, 4);
+ Assert.assertFalse(isScope);
+ isScope = AclUtils.isScope(adderss, 3);
+ Assert.assertFalse(isScope);
+
+ }
+
+ @Test
+ public void isScopeStringTest() {
+ for (int i = 0; i < 256; i++) {
+ boolean isScope = AclUtils.isScope(i + "");
+ Assert.assertTrue(isScope);
+ }
+ boolean isScope = AclUtils.isScope("-1");
+ Assert.assertFalse(isScope);
+ isScope = AclUtils.isScope("256");
+ Assert.assertFalse(isScope);
+ }
+
+ @Test
+ public void isScopeTest() {
+ for (int i = 0; i < 256; i++) {
+ boolean isScope = AclUtils.isScope(i);
+ Assert.assertTrue(isScope);
+ }
+ boolean isScope = AclUtils.isScope(-1);
+ Assert.assertFalse(isScope);
+ isScope = AclUtils.isScope(256);
+ Assert.assertFalse(isScope);
+
+ }
+
+ @Test
+ public void isAsteriskTest() {
+ boolean isAsterisk = AclUtils.isAsterisk("*");
+ Assert.assertTrue(isAsterisk);
+
+ isAsterisk = AclUtils.isAsterisk(",");
+ Assert.assertFalse(isAsterisk);
+ }
+
+ @Test
+ public void isColonTest() {
+ boolean isColon = AclUtils.isColon(",");
+ Assert.assertTrue(isColon);
+
+ isColon = AclUtils.isColon("-");
+ Assert.assertFalse(isColon);
+ }
+
+ @Test
+ public void isMinusTest() {
+ boolean isMinus = AclUtils.isMinus("-");
+ Assert.assertTrue(isMinus);
+
+ isMinus = AclUtils.isMinus("*");
+ Assert.assertFalse(isMinus);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test
+ public void getYamlDataObjectTest() {
+
+ Map<String, Object> map = AclUtils.getYamlDataObject("src/test/resources/conf/plain_acl.yml", Map.class);
+ Assert.assertFalse(map.isEmpty());
+ }
+
+ @Test(expected = Exception.class)
+ public void getYamlDataObjectExceptionTest() {
+
+ AclUtils.getYamlDataObject("plain_acl.yml", Map.class);
+ }
+}
diff --git a/acl/src/test/java/org/apache/rocketmq/acl/common/PermissionTest.java b/acl/src/test/java/org/apache/rocketmq/acl/common/PermissionTest.java
new file mode 100644
index 000000000..253b5b241
--- /dev/null
+++ b/acl/src/test/java/org/apache/rocketmq/acl/common/PermissionTest.java
@@ -0,0 +1,168 @@
+/*
+ * 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.acl.common;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.rocketmq.acl.plain.PlainAccessResource;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class PermissionTest {
+
+ @Test
+ public void fromStringGetPermissionTest() {
+ byte perm = Permission.parsePermFromString("PUB");
+ Assert.assertEquals(perm, Permission.PUB);
+
+ perm = Permission.parsePermFromString("SUB");
+ Assert.assertEquals(perm, Permission.SUB);
+
+ perm = Permission.parsePermFromString("PUB|SUB");
+ Assert.assertEquals(perm, Permission.PUB|Permission.SUB);
+
+ perm = Permission.parsePermFromString("SUB|PUB");
+ Assert.assertEquals(perm, Permission.PUB|Permission.SUB);
+
+ perm = Permission.parsePermFromString("DENY");
+ Assert.assertEquals(perm, Permission.DENY);
+
+ perm = Permission.parsePermFromString("1");
+ Assert.assertEquals(perm, Permission.DENY);
+
+ perm = Permission.parsePermFromString(null);
+ Assert.assertEquals(perm, Permission.DENY);
+
+ }
+
+ @Test
+ public void checkPermissionTest() {
+ boolean boo = Permission.checkPermission(Permission.DENY, Permission.DENY);
+ Assert.assertFalse(boo);
+
+ boo = Permission.checkPermission(Permission.PUB, Permission.PUB);
+ Assert.assertTrue(boo);
+
+ boo = Permission.checkPermission(Permission.SUB, Permission.SUB);
+ Assert.assertTrue(boo);
+
+ boo = Permission.checkPermission(Permission.PUB, (byte) (Permission.PUB|Permission.SUB));
+ Assert.assertTrue(boo);
+
+ boo = Permission.checkPermission(Permission.SUB, (byte) (Permission.PUB|Permission.SUB));
+ Assert.assertTrue(boo);
+
+ boo = Permission.checkPermission(Permission.ANY, (byte) (Permission.PUB|Permission.SUB));
+ Assert.assertTrue(boo);
+
+ boo = Permission.checkPermission(Permission.ANY, Permission.SUB);
+ Assert.assertTrue(boo);
+
+ boo = Permission.checkPermission(Permission.ANY, Permission.PUB);
+ Assert.assertTrue(boo);
+
+ boo = Permission.checkPermission(Permission.DENY, Permission.ANY);
+ Assert.assertFalse(boo);
+
+ boo = Permission.checkPermission(Permission.DENY, Permission.PUB);
+ Assert.assertFalse(boo);
+
+ boo = Permission.checkPermission(Permission.DENY, Permission.SUB);
+ Assert.assertFalse(boo);
+
+ }
+
+ @Test(expected = AclException.class)
+ public void setTopicPermTest() {
+ PlainAccessResource plainAccessResource = new PlainAccessResource();
+ Map<String, Byte> resourcePermMap = plainAccessResource.getResourcePermMap();
+
+ Permission.parseResourcePerms(plainAccessResource, false, null);
+ Assert.assertNull(resourcePermMap);
+
+ List<String> groups = new ArrayList<>();
+ Permission.parseResourcePerms(plainAccessResource, false, groups);
+ Assert.assertNull(resourcePermMap);
+
+ groups.add("groupA=DENY");
+ groups.add("groupB=PUB|SUB");
+ groups.add("groupC=PUB");
+ Permission.parseResourcePerms(plainAccessResource, false, groups);
+ resourcePermMap = plainAccessResource.getResourcePermMap();
+
+ byte perm = resourcePermMap.get(PlainAccessResource.getRetryTopic("groupA"));
+ Assert.assertEquals(perm, Permission.DENY);
+
+ perm = resourcePermMap.get(PlainAccessResource.getRetryTopic("groupB"));
+ Assert.assertEquals(perm,Permission.PUB|Permission.SUB);
+
+ perm = resourcePermMap.get(PlainAccessResource.getRetryTopic("groupC"));
+ Assert.assertEquals(perm, Permission.PUB);
+
+ List<String> topics = new ArrayList<>();
+ topics.add("topicA=DENY");
+ topics.add("topicB=PUB|SUB");
+ topics.add("topicC=PUB");
+
+ Permission.parseResourcePerms(plainAccessResource, true, topics);
+
+ perm = resourcePermMap.get("topicA");
+ Assert.assertEquals(perm, Permission.DENY);
+
+ perm = resourcePermMap.get("topicB");
+ Assert.assertEquals(perm, Permission.PUB|Permission.SUB);
+
+ perm = resourcePermMap.get("topicC");
+ Assert.assertEquals(perm, Permission.PUB);
+
+ List<String> erron = new ArrayList<>();
+ erron.add("");
+ Permission.parseResourcePerms(plainAccessResource, false, erron);
+ }
+
+ @Test
+ public void checkAdminCodeTest() {
+ Set<Integer> code = new HashSet<>();
+ code.add(17);
+ code.add(25);
+ code.add(215);
+ code.add(200);
+ code.add(207);
+
+ for (int i = 0; i < 400; i++) {
+ boolean boo = Permission.needAdminPerm(i);
+ if (boo) {
+ Assert.assertTrue(code.contains(i));
+ }
+ }
+ }
+
+ @Test
+ public void AclExceptionTest(){
+ AclException aclException = new AclException("CAL_SIGNATURE_FAILED",10015);
+ AclException aclExceptionWithMessage = new AclException("CAL_SIGNATURE_FAILED",10015,"CAL_SIGNATURE_FAILED Exception");
+ Assert.assertEquals(aclException.getCode(),10015);
+ Assert.assertEquals(aclExceptionWithMessage.getStatus(),"CAL_SIGNATURE_FAILED");
+ aclException.setCode(10016);
+ Assert.assertEquals(aclException.getCode(),10016);
+ aclException.setStatus("netaddress examine scope Exception netaddress");
+ Assert.assertEquals(aclException.getStatus(),"netaddress examine scope Exception netaddress");
+ }
+}
diff --git a/acl/src/test/java/org/apache/rocketmq/acl/common/SessionCredentialsTest.java b/acl/src/test/java/org/apache/rocketmq/acl/common/SessionCredentialsTest.java
new file mode 100644
index 000000000..a1a4bde4f
--- /dev/null
+++ b/acl/src/test/java/org/apache/rocketmq/acl/common/SessionCredentialsTest.java
@@ -0,0 +1,91 @@
+/*
+ * 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.acl.common;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Properties;
+
+public class SessionCredentialsTest {
+
+ @Test
+ public void equalsTest(){
+ SessionCredentials sessionCredentials=new SessionCredentials("RocketMQ","12345678");
+ sessionCredentials.setSecurityToken("abcd");
+ SessionCredentials other=new SessionCredentials("RocketMQ","12345678","abcd");
+ Assert.assertTrue(sessionCredentials.equals(other));
+ }
+
+ @Test
+ public void updateContentTest(){
+ SessionCredentials sessionCredentials=new SessionCredentials();
+ Properties properties=new Properties();
+ properties.setProperty(SessionCredentials.ACCESS_KEY,"RocketMQ");
+ properties.setProperty(SessionCredentials.SECRET_KEY,"12345678");
+ properties.setProperty(SessionCredentials.SECURITY_TOKEN,"abcd");
+ sessionCredentials.updateContent(properties);
+ }
+
+ @Test
+ public void SessionCredentialHashCodeTest(){
+ SessionCredentials sessionCredentials=new SessionCredentials();
+ Properties properties=new Properties();
+ properties.setProperty(SessionCredentials.ACCESS_KEY,"RocketMQ");
+ properties.setProperty(SessionCredentials.SECRET_KEY,"12345678");
+ properties.setProperty(SessionCredentials.SECURITY_TOKEN,"abcd");
+ sessionCredentials.updateContent(properties);
+ Assert.assertEquals(sessionCredentials.hashCode(),353652211);
+ }
+
+ @Test
+ public void SessionCredentialEqualsTest(){
+ SessionCredentials sessionCredential1 =new SessionCredentials();
+ Properties properties1=new Properties();
+ properties1.setProperty(SessionCredentials.ACCESS_KEY,"RocketMQ");
+ properties1.setProperty(SessionCredentials.SECRET_KEY,"12345678");
+ properties1.setProperty(SessionCredentials.SECURITY_TOKEN,"abcd");
+ sessionCredential1.updateContent(properties1);
+
+ SessionCredentials sessionCredential2 =new SessionCredentials();
+ Properties properties2=new Properties();
+ properties2.setProperty(SessionCredentials.ACCESS_KEY,"RocketMQ");
+ properties2.setProperty(SessionCredentials.SECRET_KEY,"12345678");
+ properties2.setProperty(SessionCredentials.SECURITY_TOKEN,"abcd");
+ sessionCredential2.updateContent(properties2);
+
+ Assert.assertTrue(sessionCredential2.equals(sessionCredential1));
+ sessionCredential2.setSecretKey("1234567899");
+ sessionCredential2.setSignature("1234567899");
+ Assert.assertFalse(sessionCredential2.equals(sessionCredential1));
+ }
+
+ @Test
+ public void SessionCredentialToStringTest(){
+ SessionCredentials sessionCredential1 =new SessionCredentials();
+ Properties properties1=new Properties();
+ properties1.setProperty(SessionCredentials.ACCESS_KEY,"RocketMQ");
+ properties1.setProperty(SessionCredentials.SECRET_KEY,"12345678");
+ properties1.setProperty(SessionCredentials.SECURITY_TOKEN,"abcd");
+ sessionCredential1.updateContent(properties1);
+
+ Assert.assertEquals(sessionCredential1.toString(),
+ "SessionCredentials [accessKey=RocketMQ, secretKey=12345678, signature=null, SecurityToken=abcd]");
+ }
+
+
+}
diff --git a/acl/src/test/java/org/apache/rocketmq/acl/plain/PlainAccessValidatorTest.java b/acl/src/test/java/org/apache/rocketmq/acl/plain/PlainAccessValidatorTest.java
new file mode 100644
index 000000000..16e770206
--- /dev/null
+++ b/acl/src/test/java/org/apache/rocketmq/acl/plain/PlainAccessValidatorTest.java
@@ -0,0 +1,270 @@
+/*
+ * 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.acl.plain;
+
+import java.nio.ByteBuffer;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.rocketmq.acl.common.AclClientRPCHook;
+import org.apache.rocketmq.acl.common.AclException;
+import org.apache.rocketmq.acl.common.AclUtils;
+import org.apache.rocketmq.acl.common.SessionCredentials;
+import org.apache.rocketmq.common.protocol.RequestCode;
+import org.apache.rocketmq.common.protocol.header.*;
+import org.apache.rocketmq.common.protocol.heartbeat.ConsumerData;
+import org.apache.rocketmq.common.protocol.heartbeat.HeartbeatData;
+import org.apache.rocketmq.common.protocol.heartbeat.ProducerData;
+import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class PlainAccessValidatorTest {
+
+ private PlainAccessValidator plainAccessValidator;
+ private AclClientRPCHook aclClient;
+ private SessionCredentials sessionCredentials;
+ @Before
+ public void init() {
+ System.setProperty("rocketmq.home.dir", "src/test/resources");
+ plainAccessValidator = new PlainAccessValidator();
+ sessionCredentials = new SessionCredentials();
+ sessionCredentials.setAccessKey("RocketMQ");
+ sessionCredentials.setSecretKey("12345678");
+ sessionCredentials.setSecurityToken("87654321");
+ aclClient = new AclClientRPCHook(sessionCredentials);
+ }
+
+ @Test
+ public void contentTest() {
+ SendMessageRequestHeader messageRequestHeader = new SendMessageRequestHeader();
+ messageRequestHeader.setTopic("topicA");
+ RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, messageRequestHeader);
+ aclClient.doBeforeRequest("", remotingCommand);
+
+ ByteBuffer buf = remotingCommand.encodeHeader();
+ buf.getInt();
+ buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
+ buf.position(0);
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "127.0.0.1");
+ String signature = AclUtils.calSignature(accessResource.getContent(), sessionCredentials.getSecretKey());
+
+ Assert.assertEquals(accessResource.getSignature(), signature);
+
+ }
+
+ @Test
+ public void validateTest() {
+ SendMessageRequestHeader messageRequestHeader = new SendMessageRequestHeader();
+ messageRequestHeader.setTopic("topicB");
+ RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, messageRequestHeader);
+ aclClient.doBeforeRequest("", remotingCommand);
+
+ ByteBuffer buf = remotingCommand.encodeHeader();
+ buf.getInt();
+ buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
+ buf.position(0);
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "192.168.0.1");
+ plainAccessValidator.validate(accessResource);
+
+ }
+
+ @Test
+ public void validateSendMessageTest() {
+ SendMessageRequestHeader messageRequestHeader = new SendMessageRequestHeader();
+ messageRequestHeader.setTopic("topicB");
+ RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, messageRequestHeader);
+ aclClient.doBeforeRequest("", remotingCommand);
+
+ ByteBuffer buf = remotingCommand.encodeHeader();
+ buf.getInt();
+ buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
+ buf.position(0);
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "192.168.0.1");
+ plainAccessValidator.validate(accessResource);
+ }
+
+ @Test
+ public void validateSendMessageV2Test() {
+ SendMessageRequestHeader messageRequestHeader = new SendMessageRequestHeader();
+ messageRequestHeader.setTopic("topicC");
+ RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE_V2, SendMessageRequestHeaderV2.createSendMessageRequestHeaderV2(messageRequestHeader));
+ aclClient.doBeforeRequest("", remotingCommand);
+
+ ByteBuffer buf = remotingCommand.encodeHeader();
+ buf.getInt();
+ buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
+ buf.position(0);
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "192.168.0.1:9876");
+ plainAccessValidator.validate(accessResource);
+ }
+
+ @Test
+ public void validatePullMessageTest() {
+ PullMessageRequestHeader pullMessageRequestHeader=new PullMessageRequestHeader();
+ pullMessageRequestHeader.setTopic("topicC");
+ pullMessageRequestHeader.setConsumerGroup("consumerGroupA");
+ RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE,pullMessageRequestHeader);
+ aclClient.doBeforeRequest("", remotingCommand);
+ ByteBuffer buf = remotingCommand.encodeHeader();
+ buf.getInt();
+ buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
+ buf.position(0);
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "192.168.0.1:9876");
+ plainAccessValidator.validate(accessResource);
+ }
+
+ @Test
+ public void validateConsumeMessageBackTest() {
+ ConsumerSendMsgBackRequestHeader consumerSendMsgBackRequestHeader=new ConsumerSendMsgBackRequestHeader();
+ consumerSendMsgBackRequestHeader.setOriginTopic("topicC");
+ consumerSendMsgBackRequestHeader.setGroup("consumerGroupA");
+ RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.CONSUMER_SEND_MSG_BACK,consumerSendMsgBackRequestHeader);
+ aclClient.doBeforeRequest("", remotingCommand);
+ ByteBuffer buf = remotingCommand.encodeHeader();
+ buf.getInt();
+ buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
+ buf.position(0);
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "192.168.0.1:9876");
+ plainAccessValidator.validate(accessResource);
+ }
+
+ @Test
+ public void validateQueryMessageTest() {
+ QueryMessageRequestHeader queryMessageRequestHeader=new QueryMessageRequestHeader();
+ queryMessageRequestHeader.setTopic("topicC");
+ RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.QUERY_MESSAGE,queryMessageRequestHeader);
+ aclClient.doBeforeRequest("", remotingCommand);
+ ByteBuffer buf = remotingCommand.encodeHeader();
+ buf.getInt();
+ buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
+ buf.position(0);
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "192.168.0.1:9876");
+ plainAccessValidator.validate(accessResource);
+ }
+
+ @Test
+ public void validateHeartBeatTest() {
+ HeartbeatData heartbeatData=new HeartbeatData();
+ Set<ProducerData> producerDataSet=new HashSet<>();
+ Set<ConsumerData> consumerDataSet=new HashSet<>();
+ Set<SubscriptionData> subscriptionDataSet=new HashSet<>();
+ ProducerData producerData=new ProducerData();
+ producerData.setGroupName("producerGroupA");
+ ConsumerData consumerData=new ConsumerData();
+ consumerData.setGroupName("consumerGroupA");
+ SubscriptionData subscriptionData=new SubscriptionData();
+ subscriptionData.setTopic("topicC");
+ producerDataSet.add(producerData);
+ consumerDataSet.add(consumerData);
+ subscriptionDataSet.add(subscriptionData);
+ consumerData.setSubscriptionDataSet(subscriptionDataSet);
+ heartbeatData.setProducerDataSet(producerDataSet);
+ heartbeatData.setConsumerDataSet(consumerDataSet);
+ RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.HEART_BEAT,null);
+ remotingCommand.setBody(heartbeatData.encode());
+ aclClient.doBeforeRequest("", remotingCommand);
+ ByteBuffer buf = remotingCommand.encode();
+ buf.getInt();
+ buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
+ buf.position(0);
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "192.168.0.1:9876");
+ plainAccessValidator.validate(accessResource);
+ }
+
+ @Test
+ public void validateUnRegisterClientTest() {
+ UnregisterClientRequestHeader unregisterClientRequestHeader=new UnregisterClientRequestHeader();
+ unregisterClientRequestHeader.setConsumerGroup("consumerGroupA");
+ RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.UNREGISTER_CLIENT,unregisterClientRequestHeader);
+ aclClient.doBeforeRequest("", remotingCommand);
+ ByteBuffer buf = remotingCommand.encodeHeader();
+ buf.getInt();
+ buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
+ buf.position(0);
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "192.168.0.1:9876");
+ plainAccessValidator.validate(accessResource);
+ }
+
+ @Test
+ public void validateGetConsumerListByGroupTest() {
+ GetConsumerListByGroupRequestHeader getConsumerListByGroupRequestHeader=new GetConsumerListByGroupRequestHeader();
+ getConsumerListByGroupRequestHeader.setConsumerGroup("consumerGroupA");
+ RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.GET_CONSUMER_LIST_BY_GROUP,getConsumerListByGroupRequestHeader);
+ aclClient.doBeforeRequest("", remotingCommand);
+ ByteBuffer buf = remotingCommand.encodeHeader();
+ buf.getInt();
+ buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
+ buf.position(0);
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "192.168.0.1:9876");
+ plainAccessValidator.validate(accessResource);
+ }
+
+ @Test
+ public void validateUpdateConsumerOffSetTest() {
+ UpdateConsumerOffsetRequestHeader updateConsumerOffsetRequestHeader=new UpdateConsumerOffsetRequestHeader();
+ updateConsumerOffsetRequestHeader.setConsumerGroup("consumerGroupA");
+ RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.UPDATE_CONSUMER_OFFSET,updateConsumerOffsetRequestHeader);
+ aclClient.doBeforeRequest("", remotingCommand);
+ ByteBuffer buf = remotingCommand.encodeHeader();
+ buf.getInt();
+ buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
+ buf.position(0);
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "192.168.0.1:9876");
+ plainAccessValidator.validate(accessResource);
+ }
+
+ @Test(expected = AclException.class)
+ public void validateNullAccessKeyTest() {
+ SessionCredentials sessionCredentials=new SessionCredentials();
+ sessionCredentials.setAccessKey("RocketMQ1");
+ sessionCredentials.setSecretKey("1234");
+ AclClientRPCHook aclClientRPCHook=new AclClientRPCHook(sessionCredentials);
+ SendMessageRequestHeader messageRequestHeader = new SendMessageRequestHeader();
+ messageRequestHeader.setTopic("topicB");
+ RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, messageRequestHeader);
+ aclClientRPCHook.doBeforeRequest("", remotingCommand);
+
+ ByteBuffer buf = remotingCommand.encodeHeader();
+ buf.getInt();
+ buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
+ buf.position(0);
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "192.168.1.1");
+ plainAccessValidator.validate(accessResource);
+ }
+
+ @Test(expected = AclException.class)
+ public void validateErrorSecretKeyTest() {
+ SessionCredentials sessionCredentials=new SessionCredentials();
+ sessionCredentials.setAccessKey("RocketMQ");
+ sessionCredentials.setSecretKey("1234");
+ AclClientRPCHook aclClientRPCHook=new AclClientRPCHook(sessionCredentials);
+ SendMessageRequestHeader messageRequestHeader = new SendMessageRequestHeader();
+ messageRequestHeader.setTopic("topicB");
+ RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, messageRequestHeader);
+ aclClientRPCHook.doBeforeRequest("", remotingCommand);
+
+ ByteBuffer buf = remotingCommand.encodeHeader();
+ buf.getInt();
+ buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
+ buf.position(0);
+ PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "192.168.1.1");
+ plainAccessValidator.validate(accessResource);
+ }
+}
diff --git a/acl/src/test/java/org/apache/rocketmq/acl/plain/PlainPermissionLoaderTest.java b/acl/src/test/java/org/apache/rocketmq/acl/plain/PlainPermissionLoaderTest.java
new file mode 100644
index 000000000..ebbc4fd26
--- /dev/null
+++ b/acl/src/test/java/org/apache/rocketmq/acl/plain/PlainPermissionLoaderTest.java
@@ -0,0 +1,275 @@
+/*
+ * 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.acl.plain;
+
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.rocketmq.acl.common.AclException;
+import org.apache.rocketmq.acl.common.Permission;
+import org.apache.rocketmq.acl.plain.PlainPermissionLoader.PlainAccessConfig;
+import org.apache.rocketmq.common.UtilAll;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class PlainPermissionLoaderTest {
+
+ PlainPermissionLoader plainPermissionLoader;
+ PlainAccessResource PUBPlainAccessResource;
+ PlainAccessResource SUBPlainAccessResource;
+ PlainAccessResource ANYPlainAccessResource;
+ PlainAccessResource DENYPlainAccessResource;
+ PlainAccessResource plainAccessResource = new PlainAccessResource();
+ PlainAccessConfig plainAccessConfig = new PlainAccessConfig();
+ PlainAccessResource plainAccessResourceTwo = new PlainAccessResource();
+ Set<Integer> adminCode = new HashSet<>();
+
+ @Before
+ public void init() throws NoSuchFieldException, SecurityException, IOException {
+ // UPDATE_AND_CREATE_TOPIC
+ adminCode.add(17);
+ // UPDATE_BROKER_CONFIG
+ adminCode.add(25);
+ // DELETE_TOPIC_IN_BROKER
+ adminCode.add(215);
+ // UPDATE_AND_CREATE_SUBSCRIPTIONGROUP
+ adminCode.add(200);
+ // DELETE_SUBSCRIPTIONGROUP
+ adminCode.add(207);
+
+ PUBPlainAccessResource = clonePlainAccessResource(Permission.PUB);
+ SUBPlainAccessResource = clonePlainAccessResource(Permission.SUB);
+ ANYPlainAccessResource = clonePlainAccessResource(Permission.ANY);
+ DENYPlainAccessResource = clonePlainAccessResource(Permission.DENY);
+
+ System.setProperty("rocketmq.home.dir", "src/test/resources");
+ System.setProperty("rocketmq.acl.plain.file", "/conf/plain_acl.yml");
+ plainPermissionLoader = new PlainPermissionLoader();
+
+ }
+
+ public PlainAccessResource clonePlainAccessResource(byte perm) {
+ PlainAccessResource painAccessResource = new PlainAccessResource();
+ painAccessResource.setAccessKey("RocketMQ");
+ painAccessResource.setSecretKey("12345678");
+ painAccessResource.setWhiteRemoteAddress("127.0." + perm + ".*");
+ painAccessResource.setDefaultGroupPerm(perm);
+ painAccessResource.setDefaultTopicPerm(perm);
+ painAccessResource.addResourceAndPerm(PlainAccessResource.getRetryTopic("groupA"), Permission.PUB);
+ painAccessResource.addResourceAndPerm(PlainAccessResource.getRetryTopic("groupB"), Permission.SUB);
+ painAccessResource.addResourceAndPerm(PlainAccessResource.getRetryTopic("groupC"), Permission.ANY);
+ painAccessResource.addResourceAndPerm(PlainAccessResource.getRetryTopic("groupD"), Permission.DENY);
+
+ painAccessResource.addResourceAndPerm("topicA", Permission.PUB);
+ painAccessResource.addResourceAndPerm("topicB", Permission.SUB);
+ painAccessResource.addResourceAndPerm("topicC", Permission.ANY);
+ painAccessResource.addResourceAndPerm("topicD", Permission.DENY);
+ return painAccessResource;
+ }
+
+ @Test
+ public void buildPlainAccessResourceTest() {
+ PlainAccessResource plainAccessResource = null;
+ PlainAccessConfig plainAccess = new PlainAccessConfig();
+
+ plainAccess.setAccessKey("RocketMQ");
+ plainAccess.setSecretKey("12345678");
+ plainAccessResource = plainPermissionLoader.buildPlainAccessResource(plainAccess);
+ Assert.assertEquals(plainAccessResource.getAccessKey(), "RocketMQ");
+ Assert.assertEquals(plainAccessResource.getSecretKey(), "12345678");
+
+ plainAccess.setWhiteRemoteAddress("127.0.0.1");
+ plainAccessResource = plainPermissionLoader.buildPlainAccessResource(plainAccess);
+ Assert.assertEquals(plainAccessResource.getWhiteRemoteAddress(), "127.0.0.1");
+
+ plainAccess.setAdmin(true);
+ plainAccessResource = plainPermissionLoader.buildPlainAccessResource(plainAccess);
+ Assert.assertEquals(plainAccessResource.isAdmin(), true);
+
+ List<String> groups = new ArrayList<String>();
+ groups.add("groupA=DENY");
+ groups.add("groupB=PUB|SUB");
+ groups.add("groupC=PUB");
+ plainAccess.setGroupPerms(groups);
+ plainAccessResource = plainPermissionLoader.buildPlainAccessResource(plainAccess);
+ Map<String, Byte> resourcePermMap = plainAccessResource.getResourcePermMap();
+ Assert.assertEquals(resourcePermMap.size(), 3);
+
+ Assert.assertEquals(resourcePermMap.get(PlainAccessResource.getRetryTopic("groupA")).byteValue(), Permission.DENY);
+ Assert.assertEquals(resourcePermMap.get(PlainAccessResource.getRetryTopic("groupB")).byteValue(), Permission.PUB|Permission.SUB);
+ Assert.assertEquals(resourcePermMap.get(PlainAccessResource.getRetryTopic("groupC")).byteValue(), Permission.PUB);
+
+ List<String> topics = new ArrayList<String>();
+ topics.add("topicA=DENY");
+ topics.add("topicB=PUB|SUB");
+ topics.add("topicC=PUB");
+ plainAccess.setTopicPerms(topics);
+ plainAccessResource = plainPermissionLoader.buildPlainAccessResource(plainAccess);
+ resourcePermMap = plainAccessResource.getResourcePermMap();
+ Assert.assertEquals(resourcePermMap.size(), 6);
+
+ Assert.assertEquals(resourcePermMap.get("topicA").byteValue(), Permission.DENY);
+ Assert.assertEquals(resourcePermMap.get("topicB").byteValue(), Permission.PUB|Permission.SUB);
+ Assert.assertEquals(resourcePermMap.get("topicC").byteValue(), Permission.PUB);
+ }
+
+ @Test(expected = AclException.class)
+ public void checkPermAdmin() {
+ PlainAccessResource plainAccessResource = new PlainAccessResource();
+ plainAccessResource.setRequestCode(17);
+ plainPermissionLoader.checkPerm(plainAccessResource, PUBPlainAccessResource);
+ }
+
+ @Test
+ public void checkPerm() {
+
+ PlainAccessResource plainAccessResource = new PlainAccessResource();
+ plainAccessResource.addResourceAndPerm("topicA", Permission.PUB);
+ plainPermissionLoader.checkPerm(plainAccessResource, PUBPlainAccessResource);
+ plainAccessResource.addResourceAndPerm("topicB", Permission.SUB);
+ plainPermissionLoader.checkPerm(plainAccessResource, ANYPlainAccessResource);
+
+ plainAccessResource = new PlainAccessResource();
+ plainAccessResource.addResourceAndPerm("topicB", Permission.SUB);
+ plainPermissionLoader.checkPerm(plainAccessResource, SUBPlainAccessResource);
+ plainAccessResource.addResourceAndPerm("topicA", Permission.PUB);
+ plainPermissionLoader.checkPerm(plainAccessResource, ANYPlainAccessResource);
+
+ }
+ @Test(expected = AclException.class)
+ public void checkErrorPerm() {
+
+ plainAccessResource = new PlainAccessResource();
+ plainAccessResource.addResourceAndPerm("topicF", Permission.SUB);
+ plainPermissionLoader.checkPerm(plainAccessResource, SUBPlainAccessResource);
+ }
+ @Test(expected = AclException.class)
+ public void accountNullTest() {
+ plainAccessConfig.setAccessKey(null);
+ plainPermissionLoader.buildPlainAccessResource(plainAccessConfig);
+ }
+
+ @Test(expected = AclException.class)
+ public void accountThanTest() {
+ plainAccessConfig.setAccessKey("123");
+ plainPermissionLoader.buildPlainAccessResource(plainAccessConfig);
+ }
+
+ @Test(expected = AclException.class)
+ public void passWordtNullTest() {
+ plainAccessConfig.setAccessKey(null);
+ plainPermissionLoader.buildPlainAccessResource(plainAccessConfig);
+ }
+
+ @Test(expected = AclException.class)
+ public void passWordThanTest() {
+ plainAccessConfig.setAccessKey("123");
+ plainPermissionLoader.buildPlainAccessResource(plainAccessConfig);
+ }
+
+ @Test(expected = AclException.class)
+ public void testPlainAclPlugEngineInit() {
+ System.setProperty("rocketmq.home.dir", "");
+ new PlainPermissionLoader().load();
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test
+ public void cleanAuthenticationInfoTest() throws IllegalAccessException {
+ //plainPermissionLoader.addPlainAccessResource(plainAccessResource);
+ Map<String, List<PlainAccessResource>> plainAccessResourceMap = (Map<String, List<PlainAccessResource>>) FieldUtils.readDeclaredField(plainPermissionLoader, "plainAccessResourceMap", true);
+ Assert.assertFalse(plainAccessResourceMap.isEmpty());
+
+ plainPermissionLoader.clearPermissionInfo();
+ plainAccessResourceMap = (Map<String, List<PlainAccessResource>>) FieldUtils.readDeclaredField(plainPermissionLoader, "plainAccessResourceMap", true);
+ Assert.assertTrue(plainAccessResourceMap.isEmpty());
+ }
+
+ @Test
+ public void isWatchStartTest() {
+
+ PlainPermissionLoader plainPermissionLoader = new PlainPermissionLoader();
+ Assert.assertTrue(plainPermissionLoader.isWatchStart());
+ }
+
+
+ @Test
+ public void testWatch() throws IOException, IllegalAccessException ,InterruptedException{
+ System.setProperty("rocketmq.home.dir", "src/test/resources");
+ System.setProperty("rocketmq.acl.plain.file", "/conf/plain_acl-test.yml");
+ String fileName =System.getProperty("rocketmq.home.dir", "src/test/resources")+System.getProperty("rocketmq.acl.plain.file", "/conf/plain_acl.yml");
+ File transport = new File(fileName);
+ transport.delete();
+ transport.createNewFile();
+ FileWriter writer = new FileWriter(transport);
+ writer.write("accounts:\r\n");
+ writer.write("- accessKey: watchrocketmq\r\n");
+ writer.write(" secretKey: 12345678\r\n");
+ writer.write(" whiteRemoteAddress: 127.0.0.1\r\n");
+ writer.write(" admin: true\r\n");
+ writer.flush();
+ writer.close();
+
+ PlainPermissionLoader plainPermissionLoader = new PlainPermissionLoader();
+ Assert.assertTrue(plainPermissionLoader.isWatchStart());
+
+ {
+ Map<String, PlainAccessResource> plainAccessResourceMap = (Map<String, PlainAccessResource>) FieldUtils.readDeclaredField(plainPermissionLoader, "plainAccessResourceMap", true);
+ PlainAccessResource accessResource = plainAccessResourceMap.get("watchrocketmq");
+ Assert.assertNotNull(accessResource);
+ Assert.assertEquals(accessResource.getSecretKey(), "12345678");
+ Assert.assertTrue(accessResource.isAdmin());
+
+ }
+
+ writer = new FileWriter(new File(fileName), true);
+ writer.write("- accessKey: watchrocketmq1\r\n");
+ writer.write(" secretKey: 88888888\r\n");
+ writer.write(" whiteRemoteAddress: 127.0.0.1\r\n");
+ writer.write(" admin: false\r\n");
+ writer.flush();
+ writer.close();
+
+ Thread.sleep(1000);
+ {
+ Map<String, PlainAccessResource> plainAccessResourceMap = (Map<String, PlainAccessResource>) FieldUtils.readDeclaredField(plainPermissionLoader, "plainAccessResourceMap", true);
+ PlainAccessResource accessResource = plainAccessResourceMap.get("watchrocketmq1");
+ Assert.assertNotNull(accessResource);
+ Assert.assertEquals(accessResource.getSecretKey(), "88888888");
+ Assert.assertFalse(accessResource.isAdmin());
+
+ }
+ transport.delete();
+ System.setProperty("rocketmq.home.dir", "src/test/resources");
+ System.setProperty("rocketmq.acl.plain.file", "/conf/plain_acl.yml");
+ }
+
+ @Test(expected = AclException.class)
+ public void initializeTest() {
+ System.setProperty("rocketmq.acl.plain.file", "/conf/plain_acl_null.yml");
+ new PlainPermissionLoader();
+
+ }
+
+}
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
new file mode 100644
index 000000000..53391f411
--- /dev/null
+++ b/acl/src/test/java/org/apache/rocketmq/acl/plain/RemoteAddressStrategyTest.java
@@ -0,0 +1,218 @@
+/*
+ * 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.acl.plain;
+
+import org.apache.rocketmq.acl.common.AclException;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class RemoteAddressStrategyTest {
+
+ RemoteAddressStrategyFactory remoteAddressStrategyFactory = new RemoteAddressStrategyFactory();
+
+ @Test
+ public void netaddressStrategyFactoryExceptionTest() {
+ PlainAccessResource plainAccessResource = new PlainAccessResource();
+ remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+ Assert.assertEquals(remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource).getClass(),
+ RemoteAddressStrategyFactory.BlankRemoteAddressStrategy.class);
+ }
+
+ @Test
+ public void netaddressStrategyFactoryTest() {
+ PlainAccessResource plainAccessResource = new PlainAccessResource();
+
+ plainAccessResource.setWhiteRemoteAddress("*");
+ RemoteAddressStrategy remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+ Assert.assertEquals(remoteAddressStrategy, RemoteAddressStrategyFactory.NULL_NET_ADDRESS_STRATEGY);
+
+ plainAccessResource.setWhiteRemoteAddress("127.0.0.1");
+ remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+ Assert.assertEquals(remoteAddressStrategy.getClass(), RemoteAddressStrategyFactory.OneRemoteAddressStrategy.class);
+
+ plainAccessResource.setWhiteRemoteAddress("127.0.0.1,127.0.0.2,127.0.0.3");
+ remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+ Assert.assertEquals(remoteAddressStrategy.getClass(), RemoteAddressStrategyFactory.MultipleRemoteAddressStrategy.class);
+
+ plainAccessResource.setWhiteRemoteAddress("127.0.0.{1,2,3}");
+ remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+ Assert.assertEquals(remoteAddressStrategy.getClass(), RemoteAddressStrategyFactory.MultipleRemoteAddressStrategy.class);
+
+ plainAccessResource.setWhiteRemoteAddress("127.0.0.1-200");
+ remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+ Assert.assertEquals(remoteAddressStrategy.getClass(), RemoteAddressStrategyFactory.RangeRemoteAddressStrategy.class);
+
+ plainAccessResource.setWhiteRemoteAddress("127.0.0.*");
+ remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+ Assert.assertEquals(remoteAddressStrategy.getClass(), RemoteAddressStrategyFactory.RangeRemoteAddressStrategy.class);
+
+ plainAccessResource.setWhiteRemoteAddress("127.0.1-20.*");
+ remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+ Assert.assertEquals(remoteAddressStrategy.getClass(), RemoteAddressStrategyFactory.RangeRemoteAddressStrategy.class);
+
+ plainAccessResource.setWhiteRemoteAddress("");
+ remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+ Assert.assertEquals(remoteAddressStrategy.getClass(), RemoteAddressStrategyFactory.BlankRemoteAddressStrategy.class);
+ }
+
+ @Test(expected = AclException.class)
+ public void verifyTest() {
+ PlainAccessResource plainAccessResource = new PlainAccessResource();
+ plainAccessResource.setWhiteRemoteAddress("127.0.0.1");
+ remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+ plainAccessResource.setWhiteRemoteAddress("256.0.0.1");
+ remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+ }
+
+ @Test
+ public void nullNetaddressStrategyTest() {
+ boolean isMatch = RemoteAddressStrategyFactory.NULL_NET_ADDRESS_STRATEGY.match(new PlainAccessResource());
+ Assert.assertTrue(isMatch);
+ }
+
+ @Test
+ public void blankNetaddressStrategyTest() {
+ boolean isMatch = RemoteAddressStrategyFactory.BLANK_NET_ADDRESS_STRATEGY.match(new PlainAccessResource());
+ Assert.assertFalse(isMatch);
+ }
+
+ public void oneNetaddressStrategyTest() {
+ PlainAccessResource plainAccessResource = new PlainAccessResource();
+ plainAccessResource.setWhiteRemoteAddress("127.0.0.1");
+ RemoteAddressStrategy remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+ plainAccessResource.setWhiteRemoteAddress("");
+ boolean match = remoteAddressStrategy.match(plainAccessResource);
+ Assert.assertFalse(match);
+
+ plainAccessResource.setWhiteRemoteAddress("127.0.0.2");
+ match = remoteAddressStrategy.match(plainAccessResource);
+ Assert.assertFalse(match);
+
+ plainAccessResource.setWhiteRemoteAddress("127.0.0.1");
+ match = remoteAddressStrategy.match(plainAccessResource);
+ Assert.assertTrue(match);
+ }
+
+ @Test
+ public void multipleNetaddressStrategyTest() {
+ PlainAccessResource plainAccessResource = new PlainAccessResource();
+ plainAccessResource.setWhiteRemoteAddress("127.0.0.1,127.0.0.2,127.0.0.3");
+ RemoteAddressStrategy remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+ multipleNetaddressStrategyTest(remoteAddressStrategy);
+
+ plainAccessResource.setWhiteRemoteAddress("127.0.0.{1,2,3}");
+ remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+ multipleNetaddressStrategyTest(remoteAddressStrategy);
+
+ }
+
+ @Test(expected = AclException.class)
+ public void multipleNetaddressStrategyExceptionTest() {
+ PlainAccessResource plainAccessResource = new PlainAccessResource();
+ plainAccessResource.setWhiteRemoteAddress("127.0.0.1,2,3}");
+ remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+ }
+
+ private void multipleNetaddressStrategyTest(RemoteAddressStrategy remoteAddressStrategy) {
+ PlainAccessResource plainAccessResource = new PlainAccessResource();
+ plainAccessResource.setWhiteRemoteAddress("127.0.0.1");
+ boolean match = remoteAddressStrategy.match(plainAccessResource);
+ Assert.assertTrue(match);
+
+ plainAccessResource.setWhiteRemoteAddress("127.0.0.2");
+ match = remoteAddressStrategy.match(plainAccessResource);
+ Assert.assertTrue(match);
+
+ plainAccessResource.setWhiteRemoteAddress("127.0.0.3");
+ match = remoteAddressStrategy.match(plainAccessResource);
+ Assert.assertTrue(match);
+
+ plainAccessResource.setWhiteRemoteAddress("127.0.0.4");
+ match = remoteAddressStrategy.match(plainAccessResource);
+ Assert.assertFalse(match);
+
+ plainAccessResource.setWhiteRemoteAddress("127.0.0.0");
+ match = remoteAddressStrategy.match(plainAccessResource);
+ Assert.assertFalse(match);
+
+ }
+
+ @Test
+ public void rangeNetaddressStrategyTest() {
+ String head = "127.0.0.";
+ PlainAccessResource plainAccessResource = new PlainAccessResource();
+ 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);
+
+ plainAccessResource.setWhiteRemoteAddress("127.0.1-200.*");
+ remoteAddressStrategy = remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+ rangeNetaddressStrategyThirdlyTest(remoteAddressStrategy, head, 1, 200);
+ }
+
+ private void rangeNetaddressStrategyTest(RemoteAddressStrategy remoteAddressStrategy, String head, int start,
+ int end,
+ boolean isFalse) {
+ PlainAccessResource plainAccessResource = new PlainAccessResource();
+ for (int i = -10; i < 300; i++) {
+ plainAccessResource.setWhiteRemoteAddress(head + i);
+ boolean match = remoteAddressStrategy.match(plainAccessResource);
+ if (isFalse && i >= start && i <= end) {
+ Assert.assertTrue(match);
+ continue;
+ }
+ Assert.assertFalse(match);
+
+ }
+ }
+
+ private void rangeNetaddressStrategyThirdlyTest(RemoteAddressStrategy remoteAddressStrategy, String head, int start,
+ int end) {
+ String newHead;
+ for (int i = -10; i < 300; i++) {
+ newHead = head + i;
+ if (i >= start && i <= end) {
+ rangeNetaddressStrategyTest(remoteAddressStrategy, newHead, 0, 255, false);
+ }
+ }
+ }
+
+ @Test(expected = AclException.class)
+ public void rangeNetaddressStrategyExceptionStartGreaterEndTest() {
+ rangeNetaddressStrategyExceptionTest("127.0.0.2-1");
+ }
+
+ @Test(expected = AclException.class)
+ public void rangeNetaddressStrategyExceptionScopeTest() {
+ rangeNetaddressStrategyExceptionTest("127.0.0.-1-200");
+ }
+
+ @Test(expected = AclException.class)
+ public void rangeNetaddressStrategyExceptionScopeTwoTest() {
+ rangeNetaddressStrategyExceptionTest("127.0.0.0-256");
+ }
+
+ private void rangeNetaddressStrategyExceptionTest(String netaddress) {
+ PlainAccessResource plainAccessResource = new PlainAccessResource();
+ plainAccessResource.setWhiteRemoteAddress(netaddress);
+ remoteAddressStrategyFactory.getRemoteAddressStrategy(plainAccessResource);
+ }
+
+}
diff --git a/acl/src/test/resources/conf/plain_acl.yml b/acl/src/test/resources/conf/plain_acl.yml
new file mode 100644
index 000000000..2c24795ff
--- /dev/null
+++ b/acl/src/test/resources/conf/plain_acl.yml
@@ -0,0 +1,44 @@
+# 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.
+
+## suggested format
+
+globalWhiteRemoteAddresses:
+- 10.10.103.*
+- 192.168.0.*
+
+accounts:
+- accessKey: RocketMQ
+ secretKey: 12345678
+ whiteRemoteAddress: 192.168.0.*
+ admin: false
+ defaultTopicPerm: DENY
+ defaultGroupPerm: SUB
+ topicPerms:
+ - topicA=DENY
+ - topicB=PUB|SUB
+ - topicC=SUB
+ groupPerms:
+ # the group should convert to retry topic
+ - groupA=DENY
+ - groupB=SUB
+ - groupC=SUB
+
+- accessKey: rocketmq2
+ secretKey: 12345678
+ whiteRemoteAddress: 192.168.1.*
+ # if it is admin, it could access all resources
+ admin: true
+
diff --git a/acl/src/test/resources/conf/plain_acl_null.yml b/acl/src/test/resources/conf/plain_acl_null.yml
new file mode 100644
index 000000000..bc30380c8
--- /dev/null
+++ b/acl/src/test/resources/conf/plain_acl_null.yml
@@ -0,0 +1,18 @@
+# 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.
+
+## suggested format
+
+
diff --git a/acl/src/test/resources/conf/watch/plain_acl_watch.yml b/acl/src/test/resources/conf/watch/plain_acl_watch.yml
new file mode 100644
index 000000000..9d2c39549
--- /dev/null
+++ b/acl/src/test/resources/conf/watch/plain_acl_watch.yml
@@ -0,0 +1,25 @@
+# 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.
+
+## suggested format
+accounts:
+- accessKey: watchrocketmq
+ secretKey: 12345678
+ whiteRemoteAddress: 127.0.0.1
+ admin: true
+- accessKey: watchrocketmq1
+ secretKey: 88888888
+ whiteRemoteAddress: 127.0.0.1
+ admin: false
diff --git a/acl/src/test/resources/logback-test.xml b/acl/src/test/resources/logback-test.xml
new file mode 100644
index 000000000..e556c649e
--- /dev/null
+++ b/acl/src/test/resources/logback-test.xml
@@ -0,0 +1,34 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+ -->
+<configuration>
+
+ <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+ <encoder>
+ <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{5} - %msg%n</pattern>
+ <charset class="java.nio.charset.Charset">UTF-8</charset>
+ </encoder>
+ </appender>
+
+ <logger name="RocketmqCommon" level="INFO" additivity="false">
+ <appender-ref ref="STDOUT"/>
+ </logger>
+ <root level="INFO">
+ <appender-ref ref="STDOUT"/>
+ </root>
+
+</configuration>
diff --git a/broker/pom.xml b/broker/pom.xml
index f10ae5373..f617d2492 100644
--- a/broker/pom.xml
+++ b/broker/pom.xml
@@ -1,21 +1,17 @@
-<!--
- 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
+<!-- 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. -->
- 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.
- -->
-
-<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/xsd/maven-4.0.0.xsd">
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xmlns="http://maven.apache.org/POM/4.0.0"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-all</artifactId>
@@ -52,6 +48,10 @@
<groupId>${project.groupId}</groupId>
<artifactId>rocketmq-filter</artifactId>
</dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>rocketmq-acl</artifactId>
+ </dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java b/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
index e7ef46d0c..c3431ca41 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
@@ -31,6 +31,7 @@
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import org.apache.rocketmq.acl.AccessValidator;
import org.apache.rocketmq.broker.client.ClientHousekeepingService;
import org.apache.rocketmq.broker.client.ConsumerIdsChangeListener;
import org.apache.rocketmq.broker.client.ConsumerManager;
@@ -91,6 +92,7 @@
import org.apache.rocketmq.remoting.netty.NettyServerConfig;
import org.apache.rocketmq.remoting.netty.RequestTask;
import org.apache.rocketmq.remoting.netty.TlsSystemConfig;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.srvutil.FileWatchService;
import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.MessageArrivingListener;
@@ -157,6 +159,7 @@
private TransactionalMessageService transactionalMessageService;
private AbstractTransactionalMessageCheckListener transactionalMessageCheckListener;
+
public BrokerController(
final BrokerConfig brokerConfig,
final NettyServerConfig nettyServerConfig,
@@ -467,6 +470,8 @@ private void reloadServerSslContext() {
}
}
initialTransaction();
+ initialAcl();
+ initialRpcHooks();
}
return result;
}
@@ -486,6 +491,47 @@ private void initialTransaction() {
this.transactionalMessageCheckService = new TransactionalMessageCheckService(this);
}
+ private void initialAcl() {
+ if (!this.brokerConfig.isAclEnable()) {
+ log.info("The broker dose not enable acl");
+ return;
+ }
+
+ List<AccessValidator> accessValidators = ServiceProvider.load(ServiceProvider.ACL_VALIDATOR_ID, AccessValidator.class);
+ if (accessValidators == null || accessValidators.isEmpty()) {
+ log.info("The broker dose not load the AccessValidator");
+ return;
+ }
+
+ for (AccessValidator accessValidator: accessValidators) {
+ final AccessValidator validator = accessValidator;
+ this.registerServerRPCHook(new RPCHook() {
+
+ @Override
+ public void doBeforeRequest(String remoteAddr, RemotingCommand request) {
+ //Do not catch the exception
+ validator.validate(validator.parse(request, remoteAddr));
+ }
+
+ @Override
+ public void doAfterResponse(String remoteAddr, RemotingCommand request, RemotingCommand response) {
+ }
+ });
+ }
+ }
+
+
+ private void initialRpcHooks() {
+
+ List<RPCHook> rpcHooks = ServiceProvider.load(ServiceProvider.RPC_HOOK_ID, RPCHook.class);
+ if (rpcHooks == null || rpcHooks.isEmpty()) {
+ return;
+ }
+ for (RPCHook rpcHook: rpcHooks) {
+ this.registerServerRPCHook(rpcHook);
+ }
+ }
+
public void registerProcessor() {
/**
* SendMessageProcessor
@@ -989,6 +1035,7 @@ public void registerConsumeMessageHook(final ConsumeMessageHook hook) {
public void registerServerRPCHook(RPCHook rpcHook) {
getRemotingServer().registerRPCHook(rpcHook);
+ this.fastRemotingServer.registerRPCHook(rpcHook);
}
public RemotingServer getRemotingServer() {
@@ -1049,7 +1096,9 @@ public void setTransactionalMessageCheckListener(
this.transactionalMessageCheckListener = transactionalMessageCheckListener;
}
+
public BlockingQueue<Runnable> getEndTransactionThreadPoolQueue() {
return endTransactionThreadPoolQueue;
+
}
}
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicConfigManager.java b/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicConfigManager.java
index 59c7895eb..ed353da4b 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicConfigManager.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicConfigManager.java
@@ -124,6 +124,16 @@ public TopicConfigManager(BrokerController brokerController) {
topicConfig.setWriteQueueNums(1);
this.topicConfigTable.put(topicConfig.getTopicName(), topicConfig);
}
+ {
+ if (this.brokerController.getBrokerConfig().isTraceTopicEnable()) {
+ String topic = this.brokerController.getBrokerConfig().getMsgTraceTopicName();
+ TopicConfig topicConfig = new TopicConfig(topic);
+ this.systemTopicList.add(topic);
+ topicConfig.setReadQueueNums(1);
+ topicConfig.setWriteQueueNums(1);
+ this.topicConfigTable.put(topicConfig.getTopicName(), topicConfig);
+ }
+ }
}
public boolean isSystemTopic(final String topic) {
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/transaction/queue/TransactionalMessageServiceImpl.java b/broker/src/main/java/org/apache/rocketmq/broker/transaction/queue/TransactionalMessageServiceImpl.java
index 15e5c84ff..1c227af15 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/transaction/queue/TransactionalMessageServiceImpl.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/transaction/queue/TransactionalMessageServiceImpl.java
@@ -198,7 +198,7 @@ public void check(long transactionTimeout, int transactionCheckMax,
if (null != checkImmunityTimeStr) {
checkImmunityTime = getImmunityTime(checkImmunityTimeStr, transactionTimeout);
if (valueOfCurrentMinusBorn < checkImmunityTime) {
- if (checkPrepareQueueOffset(removeMap, doneOpOffset, msgExt, checkImmunityTime)) {
+ if (checkPrepareQueueOffset(removeMap, doneOpOffset, msgExt)) {
newOffset = i + 1;
i++;
continue;
@@ -315,33 +315,26 @@ private PullResult fillOpRemoveMap(HashMap<Long, Long> removeMap,
* @param removeMap Op message map to determine whether a half message was responded by producer.
* @param doneOpOffset Op Message which has been checked.
* @param msgExt Half message
- * @param checkImmunityTime User defined time to avoid being detected early.
* @return Return true if put success, otherwise return false.
*/
- private boolean checkPrepareQueueOffset(HashMap<Long, Long> removeMap, List<Long> doneOpOffset, MessageExt msgExt,
- long checkImmunityTime) {
- if (System.currentTimeMillis() - msgExt.getBornTimestamp() < checkImmunityTime) {
- String prepareQueueOffsetStr = msgExt.getUserProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET);
- if (null == prepareQueueOffsetStr) {
- return putImmunityMsgBackToHalfQueue(msgExt);
+ private boolean checkPrepareQueueOffset(HashMap<Long, Long> removeMap, List<Long> doneOpOffset,
+ MessageExt msgExt) {
+ String prepareQueueOffsetStr = msgExt.getUserProperty(MessageConst.PROPERTY_TRANSACTION_PREPARED_QUEUE_OFFSET);
+ if (null == prepareQueueOffsetStr) {
+ return putImmunityMsgBackToHalfQueue(msgExt);
+ } else {
+ long prepareQueueOffset = getLong(prepareQueueOffsetStr);
+ if (-1 == prepareQueueOffset) {
+ return false;
} else {
- long prepareQueueOffset = getLong(prepareQueueOffsetStr);
- if (-1 == prepareQueueOffset) {
- return false;
+ if (removeMap.containsKey(prepareQueueOffset)) {
+ long tmpOpOffset = removeMap.remove(prepareQueueOffset);
+ doneOpOffset.add(tmpOpOffset);
+ return true;
} else {
- if (removeMap.containsKey(prepareQueueOffset)) {
- long tmpOpOffset = removeMap.remove(prepareQueueOffset);
- doneOpOffset.add(tmpOpOffset);
- return true;
- } else {
- return putImmunityMsgBackToHalfQueue(msgExt);
- }
+ return putImmunityMsgBackToHalfQueue(msgExt);
}
-
}
-
- } else {
- return true;
}
}
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/util/ServiceProvider.java b/broker/src/main/java/org/apache/rocketmq/broker/util/ServiceProvider.java
index 8b9b63e4d..e67966010 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/util/ServiceProvider.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/util/ServiceProvider.java
@@ -34,6 +34,14 @@
public static final String TRANSACTION_LISTENER_ID = "META-INF/service/org.apache.rocketmq.broker.transaction.AbstractTransactionalMessageCheckListener";
+
+ public static final String RPC_HOOK_ID = "META-INF/service/org.apache.rocketmq.remoting.RPCHook";
+
+
+ public static final String ACL_VALIDATOR_ID = "META-INF/service/org.apache.rocketmq.acl.AccessValidator";
+
+
+
static {
thisClassLoader = getClassLoader(ServiceProvider.class);
}
diff --git a/broker/src/main/resources/META-INF/service/org.apache.rocketmq.acl.AccessValidator b/broker/src/main/resources/META-INF/service/org.apache.rocketmq.acl.AccessValidator
new file mode 100644
index 000000000..1abc92e01
--- /dev/null
+++ b/broker/src/main/resources/META-INF/service/org.apache.rocketmq.acl.AccessValidator
@@ -0,0 +1 @@
+org.apache.rocketmq.acl.plain.PlainAccessValidator
\ No newline at end of file
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/BrokerControllerTest.java b/broker/src/test/java/org/apache/rocketmq/broker/BrokerControllerTest.java
index 56abf084a..dae133554 100644
--- a/broker/src/test/java/org/apache/rocketmq/broker/BrokerControllerTest.java
+++ b/broker/src/test/java/org/apache/rocketmq/broker/BrokerControllerTest.java
@@ -24,6 +24,7 @@
import org.apache.rocketmq.remoting.netty.NettyServerConfig;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.junit.After;
+import org.junit.Ignore;
import org.junit.Test;
import static org.assertj.core.api.Assertions.assertThat;
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/pagecache/ManyMessageTransferTest.java b/broker/src/test/java/org/apache/rocketmq/broker/pagecache/ManyMessageTransferTest.java
new file mode 100644
index 000000000..508635c04
--- /dev/null
+++ b/broker/src/test/java/org/apache/rocketmq/broker/pagecache/ManyMessageTransferTest.java
@@ -0,0 +1,64 @@
+/*
+ * 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.broker.pagecache;
+
+import java.nio.ByteBuffer;
+import org.apache.rocketmq.store.GetMessageResult;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class ManyMessageTransferTest {
+
+ @Test
+ public void ManyMessageTransferBuilderTest(){
+ ByteBuffer byteBuffer = ByteBuffer.allocate(20);
+ byteBuffer.putInt(20);
+ GetMessageResult getMessageResult = new GetMessageResult();
+ ManyMessageTransfer manyMessageTransfer = new ManyMessageTransfer(byteBuffer,getMessageResult);
+ }
+
+ @Test
+ public void ManyMessageTransferPosTest(){
+ ByteBuffer byteBuffer = ByteBuffer.allocate(20);
+ byteBuffer.putInt(20);
+ GetMessageResult getMessageResult = new GetMessageResult();
+ ManyMessageTransfer manyMessageTransfer = new ManyMessageTransfer(byteBuffer,getMessageResult);
+ Assert.assertEquals(manyMessageTransfer.position(),4);
+ }
+
+ @Test
+ public void ManyMessageTransferCountTest(){
+ ByteBuffer byteBuffer = ByteBuffer.allocate(20);
+ byteBuffer.putInt(20);
+ GetMessageResult getMessageResult = new GetMessageResult();
+ ManyMessageTransfer manyMessageTransfer = new ManyMessageTransfer(byteBuffer,getMessageResult);
+
+ Assert.assertEquals(manyMessageTransfer.count(),20);
+
+ }
+
+ @Test
+ public void ManyMessageTransferCloseTest(){
+ ByteBuffer byteBuffer = ByteBuffer.allocate(20);
+ byteBuffer.putInt(20);
+ GetMessageResult getMessageResult = new GetMessageResult();
+ ManyMessageTransfer manyMessageTransfer = new ManyMessageTransfer(byteBuffer,getMessageResult);
+ manyMessageTransfer.close();
+ manyMessageTransfer.deallocate();
+ }
+}
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/pagecache/OneMessageTransferTest.java b/broker/src/test/java/org/apache/rocketmq/broker/pagecache/OneMessageTransferTest.java
new file mode 100644
index 000000000..2cd4bdc16
--- /dev/null
+++ b/broker/src/test/java/org/apache/rocketmq/broker/pagecache/OneMessageTransferTest.java
@@ -0,0 +1,53 @@
+/*
+ * 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.broker.pagecache;
+
+import java.nio.ByteBuffer;
+import org.apache.rocketmq.store.MappedFile;
+import org.apache.rocketmq.store.SelectMappedBufferResult;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class OneMessageTransferTest {
+
+ @Test
+ public void OneMessageTransferTest(){
+ ByteBuffer byteBuffer = ByteBuffer.allocate(20);
+ byteBuffer.putInt(20);
+ SelectMappedBufferResult selectMappedBufferResult = new SelectMappedBufferResult(0,byteBuffer,20,new MappedFile());
+ OneMessageTransfer manyMessageTransfer = new OneMessageTransfer(byteBuffer,selectMappedBufferResult);
+ }
+
+ @Test
+ public void OneMessageTransferCountTest(){
+ ByteBuffer byteBuffer = ByteBuffer.allocate(20);
+ byteBuffer.putInt(20);
+ SelectMappedBufferResult selectMappedBufferResult = new SelectMappedBufferResult(0,byteBuffer,20,new MappedFile());
+ OneMessageTransfer manyMessageTransfer = new OneMessageTransfer(byteBuffer,selectMappedBufferResult);
+ Assert.assertEquals(manyMessageTransfer.count(),40);
+ }
+
+ @Test
+ public void OneMessageTransferPosTest(){
+ ByteBuffer byteBuffer = ByteBuffer.allocate(20);
+ byteBuffer.putInt(20);
+ SelectMappedBufferResult selectMappedBufferResult = new SelectMappedBufferResult(0,byteBuffer,20,new MappedFile());
+ OneMessageTransfer manyMessageTransfer = new OneMessageTransfer(byteBuffer,selectMappedBufferResult);
+ Assert.assertEquals(manyMessageTransfer.position(),8);
+ }
+}
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/util/ServiceProviderTest.java b/broker/src/test/java/org/apache/rocketmq/broker/util/ServiceProviderTest.java
index 22228a6e0..a3a35c883 100644
--- a/broker/src/test/java/org/apache/rocketmq/broker/util/ServiceProviderTest.java
+++ b/broker/src/test/java/org/apache/rocketmq/broker/util/ServiceProviderTest.java
@@ -17,12 +17,15 @@
package org.apache.rocketmq.broker.util;
+import org.apache.rocketmq.acl.AccessValidator;
import org.apache.rocketmq.broker.transaction.AbstractTransactionalMessageCheckListener;
import org.apache.rocketmq.broker.transaction.TransactionalMessageService;
import org.junit.Test;
import static org.assertj.core.api.Assertions.assertThat;
+import java.util.List;
+
public class ServiceProviderTest {
@Test
@@ -38,4 +41,10 @@ public void loadAbstractTransactionListenerTest() {
AbstractTransactionalMessageCheckListener.class);
assertThat(listener).isNotNull();
}
+
+ @Test
+ public void loadAccessValidatorTest() {
+ List<AccessValidator> accessValidators = ServiceProvider.load(ServiceProvider.ACL_VALIDATOR_ID, AccessValidator.class);
+ assertThat(accessValidators).isNotNull();
+ }
}
diff --git a/broker/src/test/resources/META-INF/service/org.apache.rocketmq.acl.AccessValidator b/broker/src/test/resources/META-INF/service/org.apache.rocketmq.acl.AccessValidator
new file mode 100644
index 000000000..1abc92e01
--- /dev/null
+++ b/broker/src/test/resources/META-INF/service/org.apache.rocketmq.acl.AccessValidator
@@ -0,0 +1 @@
+org.apache.rocketmq.acl.plain.PlainAccessValidator
\ No newline at end of file
diff --git a/client/src/main/java/org/apache/rocketmq/client/consumer/DefaultMQPullConsumer.java b/client/src/main/java/org/apache/rocketmq/client/consumer/DefaultMQPullConsumer.java
index cd7067030..6befbf3b5 100644
--- a/client/src/main/java/org/apache/rocketmq/client/consumer/DefaultMQPullConsumer.java
+++ b/client/src/main/java/org/apache/rocketmq/client/consumer/DefaultMQPullConsumer.java
@@ -257,6 +257,18 @@ public PullResult pull(MessageQueue mq, String subExpression, long offset, int m
return this.defaultMQPullConsumerImpl.pull(mq, subExpression, offset, maxNums, timeout);
}
+ @Override
+ public PullResult pull(MessageQueue mq, MessageSelector messageSelector, long offset, int maxNums)
+ throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
+ return this.defaultMQPullConsumerImpl.pull(mq, messageSelector, offset, maxNums);
+ }
+
+ @Override
+ public PullResult pull(MessageQueue mq, MessageSelector messageSelector, long offset, int maxNums, long timeout)
+ throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
+ return this.defaultMQPullConsumerImpl.pull(mq, messageSelector, offset, maxNums, timeout);
+ }
+
@Override
public void pull(MessageQueue mq, String subExpression, long offset, int maxNums, PullCallback pullCallback)
throws MQClientException, RemotingException, InterruptedException {
@@ -270,6 +282,20 @@ public void pull(MessageQueue mq, String subExpression, long offset, int maxNums
this.defaultMQPullConsumerImpl.pull(mq, subExpression, offset, maxNums, pullCallback, timeout);
}
+ @Override
+ public void pull(MessageQueue mq, MessageSelector messageSelector, long offset, int maxNums,
+ PullCallback pullCallback)
+ throws MQClientException, RemotingException, InterruptedException {
+ this.defaultMQPullConsumerImpl.pull(mq, messageSelector, offset, maxNums, pullCallback);
+ }
+
+ @Override
+ public void pull(MessageQueue mq, MessageSelector messageSelector, long offset, int maxNums,
+ PullCallback pullCallback, long timeout)
+ throws MQClientException, RemotingException, InterruptedException {
+ this.defaultMQPullConsumerImpl.pull(mq, messageSelector, offset, maxNums, pullCallback, timeout);
+ }
+
@Override
public PullResult pullBlockIfNotFound(MessageQueue mq, String subExpression, long offset, int maxNums)
throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
diff --git a/client/src/main/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumer.java b/client/src/main/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumer.java
index d51030a15..6cd2ad18a 100644
--- a/client/src/main/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumer.java
+++ b/client/src/main/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumer.java
@@ -29,6 +29,10 @@
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl;
+import org.apache.rocketmq.client.log.ClientLogger;
+import org.apache.rocketmq.client.trace.AsyncTraceDispatcher;
+import org.apache.rocketmq.client.trace.TraceDispatcher;
+import org.apache.rocketmq.client.trace.hook.ConsumeMessageTraceHookImpl;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
@@ -36,6 +40,7 @@
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
+import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.exception.RemotingException;
@@ -56,6 +61,8 @@
*/
public class DefaultMQPushConsumer extends ClientConfig implements MQPushConsumer {
+ private final InternalLogger log = ClientLogger.getLog();
+
/**
* Internal implementation. Most of the functions herein are delegated to it.
*/
@@ -246,6 +253,11 @@
*/
private long consumeTimeout = 15;
+ /**
+ * Interface of asynchronous transfer data
+ */
+ private TraceDispatcher traceDispatcher = null;
+
/**
* Default constructor.
*/
@@ -258,7 +270,7 @@ public DefaultMQPushConsumer() {
*
* @param consumerGroup Consume queue.
* @param rpcHook RPC hook to execute before each remoting command.
- * @param allocateMessageQueueStrategy message queue allocating algorithm.
+ * @param allocateMessageQueueStrategy Message queue allocating algorithm.
*/
public DefaultMQPushConsumer(final String consumerGroup, RPCHook rpcHook,
AllocateMessageQueueStrategy allocateMessageQueueStrategy) {
@@ -267,6 +279,33 @@ public DefaultMQPushConsumer(final String consumerGroup, RPCHook rpcHook,
defaultMQPushConsumerImpl = new DefaultMQPushConsumerImpl(this, rpcHook);
}
+ /**
+ * Constructor specifying consumer group, RPC hook, message queue allocating algorithm, enabled msg trace flag and customized trace topic name.
+ *
+ * @param consumerGroup Consume queue.
+ * @param rpcHook RPC hook to execute before each remoting command.
+ * @param allocateMessageQueueStrategy message queue allocating algorithm.
+ * @param enableMsgTrace Switch flag instance for message trace.
+ * @param customizedTraceTopic The name value of message trace topic.If you don't config,you can use the default trace topic name.
+ */
+ public DefaultMQPushConsumer(final String consumerGroup, RPCHook rpcHook,
+ AllocateMessageQueueStrategy allocateMessageQueueStrategy, boolean enableMsgTrace, final String customizedTraceTopic) {
+ this.consumerGroup = consumerGroup;
+ this.allocateMessageQueueStrategy = allocateMessageQueueStrategy;
+ defaultMQPushConsumerImpl = new DefaultMQPushConsumerImpl(this, rpcHook);
+ if (enableMsgTrace) {
+ try {
+ AsyncTraceDispatcher dispatcher = new AsyncTraceDispatcher(customizedTraceTopic, rpcHook);
+ dispatcher.setHostConsumer(this.getDefaultMQPushConsumerImpl());
+ traceDispatcher = dispatcher;
+ this.getDefaultMQPushConsumerImpl().registerConsumeMessageHook(
+ new ConsumeMessageTraceHookImpl(traceDispatcher));
+ } catch (Throwable e) {
+ log.error("system mqtrace hook init failed ,maybe can't send msg trace data");
+ }
+ }
+ }
+
/**
* Constructor specifying RPC hook.
*
@@ -276,6 +315,28 @@ public DefaultMQPushConsumer(RPCHook rpcHook) {
this(MixAll.DEFAULT_CONSUMER_GROUP, rpcHook, new AllocateMessageQueueAveragely());
}
+
+ /**
+ * Constructor specifying consumer group and enabled msg trace flag.
+ *
+ * @param consumerGroup Consumer group.
+ * @param enableMsgTrace Switch flag instance for message trace.
+ */
+ public DefaultMQPushConsumer(final String consumerGroup, boolean enableMsgTrace) {
+ this(consumerGroup, null, new AllocateMessageQueueAveragely(), enableMsgTrace, null);
+ }
+
+ /**
+ * Constructor specifying consumer group, enabled msg trace flag and customized trace topic name.
+ *
+ * @param consumerGroup Consumer group.
+ * @param enableMsgTrace Switch flag instance for message trace.
+ * @param customizedTraceTopic The name value of message trace topic.If you don't config,you can use the default trace topic name.
+ */
+ public DefaultMQPushConsumer(final String consumerGroup, boolean enableMsgTrace, final String customizedTraceTopic) {
+ this(consumerGroup, null, new AllocateMessageQueueAveragely(), enableMsgTrace, customizedTraceTopic);
+ }
+
/**
* Constructor specifying consumer group.
*
@@ -518,6 +579,13 @@ public void sendMessageBack(MessageExt msg, int delayLevel, String brokerName)
@Override
public void start() throws MQClientException {
this.defaultMQPushConsumerImpl.start();
+ if (null != traceDispatcher) {
+ try {
+ traceDispatcher.start(this.getNamesrvAddr());
+ } catch (MQClientException e) {
+ log.warn("trace dispatcher start failed ", e);
+ }
+ }
}
/**
@@ -526,6 +594,9 @@ public void start() throws MQClientException {
@Override
public void shutdown() {
this.defaultMQPushConsumerImpl.shutdown();
+ if (null != traceDispatcher) {
+ traceDispatcher.shutdown();
+ }
}
@Override
@@ -694,4 +765,8 @@ public long getConsumeTimeout() {
public void setConsumeTimeout(final long consumeTimeout) {
this.consumeTimeout = consumeTimeout;
}
+
+ public TraceDispatcher getTraceDispatcher() {
+ return traceDispatcher;
+ }
}
diff --git a/client/src/main/java/org/apache/rocketmq/client/consumer/MQPullConsumer.java b/client/src/main/java/org/apache/rocketmq/client/consumer/MQPullConsumer.java
index 33002c983..28b807c2e 100644
--- a/client/src/main/java/org/apache/rocketmq/client/consumer/MQPullConsumer.java
+++ b/client/src/main/java/org/apache/rocketmq/client/consumer/MQPullConsumer.java
@@ -66,6 +66,39 @@ PullResult pull(final MessageQueue mq, final String subExpression, final long of
final int maxNums, final long timeout) throws MQClientException, RemotingException,
MQBrokerException, InterruptedException;
+ /**
+ * Pulling the messages, not blocking
+ * <p>
+ * support other message selection, such as {@link org.apache.rocketmq.common.filter.ExpressionType#SQL92}
+ * </p>
+ *
+ * @param mq from which message queue
+ * @param selector message selector({@link MessageSelector}), can be null.
+ * @param offset from where to pull
+ * @param maxNums max pulling numbers
+ * @return The resulting {@code PullRequest}
+ */
+ PullResult pull(final MessageQueue mq, final MessageSelector selector, final long offset,
+ final int maxNums) throws MQClientException, RemotingException, MQBrokerException,
+ InterruptedException;
+
+ /**
+ * Pulling the messages in the specified timeout
+ * <p>
+ * support other message selection, such as {@link org.apache.rocketmq.common.filter.ExpressionType#SQL92}
+ * </p>
+ *
+ * @param mq from which message queue
+ * @param selector message selector({@link MessageSelector}), can be null.
+ * @param offset from where to pull
+ * @param maxNums max pulling numbers
+ * @param timeout Pulling the messages in the specified timeout
+ * @return The resulting {@code PullRequest}
+ */
+ PullResult pull(final MessageQueue mq, final MessageSelector selector, final long offset,
+ final int maxNums, final long timeout) throws MQClientException, RemotingException, MQBrokerException,
+ InterruptedException;
+
/**
* Pulling the messages in a async. way
*/
@@ -80,6 +113,20 @@ void pull(final MessageQueue mq, final String subExpression, final long offset,
final PullCallback pullCallback, long timeout) throws MQClientException, RemotingException,
InterruptedException;
+ /**
+ * Pulling the messages in a async. way. Support message selection
+ */
+ void pull(final MessageQueue mq, final MessageSelector selector, final long offset, final int maxNums,
+ final PullCallback pullCallback) throws MQClientException, RemotingException,
+ InterruptedException;
+
+ /**
+ * Pulling the messages in a async. way. Support message selection
+ */
+ void pull(final MessageQueue mq, final MessageSelector selector, final long offset, final int maxNums,
+ final PullCallback pullCallback, long timeout) throws MQClientException, RemotingException,
+ InterruptedException;
+
/**
* Pulling the messages,if no message arrival,blocking some time
*
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 420d89b2f..39c43d592 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
@@ -26,6 +26,7 @@
import org.apache.rocketmq.client.QueryResult;
import org.apache.rocketmq.client.Validators;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
+import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.PullCallback;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
@@ -46,6 +47,7 @@
import org.apache.rocketmq.common.ServiceState;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
+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;
@@ -158,17 +160,58 @@ public PullResult pull(MessageQueue mq, String subExpression, long offset, int m
public PullResult pull(MessageQueue mq, String subExpression, long offset, int maxNums, long timeout)
throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
- return this.pullSyncImpl(mq, subExpression, offset, maxNums, false, timeout);
+ SubscriptionData subscriptionData = getSubscriptionData(mq, subExpression);
+ return this.pullSyncImpl(mq, subscriptionData, offset, maxNums, false, timeout);
}
- private PullResult pullSyncImpl(MessageQueue mq, String subExpression, long offset, int maxNums, boolean block,
+ public PullResult pull(MessageQueue mq, MessageSelector messageSelector, long offset, int maxNums)
+ throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
+ return pull(mq, messageSelector, offset, maxNums, this.defaultMQPullConsumer.getConsumerPullTimeoutMillis());
+ }
+
+ public PullResult pull(MessageQueue mq, MessageSelector messageSelector, long offset, int maxNums, long timeout)
+ throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
+ SubscriptionData subscriptionData = getSubscriptionData(mq, messageSelector);
+ return this.pullSyncImpl(mq, subscriptionData, offset, maxNums, false, timeout);
+ }
+
+ private SubscriptionData getSubscriptionData(MessageQueue mq, String subExpression)
+ throws MQClientException {
+
+ if (null == mq) {
+ throw new MQClientException("mq is null", null);
+ }
+
+ try {
+ return FilterAPI.buildSubscriptionData(this.defaultMQPullConsumer.getConsumerGroup(),
+ mq.getTopic(), subExpression);
+ } catch (Exception e) {
+ throw new MQClientException("parse subscription error", e);
+ }
+ }
+
+ private SubscriptionData getSubscriptionData(MessageQueue mq, MessageSelector messageSelector)
+ throws MQClientException {
+
+ if (null == mq) {
+ throw new MQClientException("mq is null", null);
+ }
+
+ try {
+ return FilterAPI.build(mq.getTopic(),
+ messageSelector.getExpression(), messageSelector.getExpressionType());
+ } catch (Exception e) {
+ throw new MQClientException("parse subscription error", e);
+ }
+ }
+
+ private PullResult pullSyncImpl(MessageQueue mq, SubscriptionData subscriptionData, long offset, int maxNums, boolean block,
long timeout)
throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
this.makeSureStateOK();
if (null == mq) {
throw new MQClientException("mq is null", null);
-
}
if (offset < 0) {
@@ -183,20 +226,14 @@ private PullResult pullSyncImpl(MessageQueue mq, String subExpression, long offs
int sysFlag = PullSysFlag.buildSysFlag(false, block, true, false);
- SubscriptionData subscriptionData;
- try {
- subscriptionData = FilterAPI.buildSubscriptionData(this.defaultMQPullConsumer.getConsumerGroup(),
- mq.getTopic(), subExpression);
- } catch (Exception e) {
- throw new MQClientException("parse subscription error", e);
- }
-
long timeoutMillis = block ? this.defaultMQPullConsumer.getConsumerTimeoutMillisWhenSuspend() : timeout;
+ boolean isTagType = ExpressionType.isTagType(subscriptionData.getExpressionType());
PullResult pullResult = this.pullAPIWrapper.pullKernelImpl(
mq,
subscriptionData.getSubString(),
- 0L,
+ subscriptionData.getExpressionType(),
+ isTagType ? 0L : subscriptionData.getSubVersion(),
offset,
maxNums,
sysFlag,
@@ -369,12 +406,27 @@ public void pull(MessageQueue mq, String subExpression, long offset, int maxNums
public void pull(MessageQueue mq, String subExpression, long offset, int maxNums, PullCallback pullCallback,
long timeout)
throws MQClientException, RemotingException, InterruptedException {
- this.pullAsyncImpl(mq, subExpression, offset, maxNums, pullCallback, false, timeout);
+ SubscriptionData subscriptionData = getSubscriptionData(mq, subExpression);
+ this.pullAsyncImpl(mq, subscriptionData, offset, maxNums, pullCallback, false, timeout);
+ }
+
+ public void pull(MessageQueue mq, MessageSelector messageSelector, long offset, int maxNums,
+ PullCallback pullCallback)
+ throws MQClientException, RemotingException, InterruptedException {
+ pull(mq, messageSelector, offset, maxNums, pullCallback, this.defaultMQPullConsumer.getConsumerPullTimeoutMillis());
+ }
+
+ public void pull(MessageQueue mq, MessageSelector messageSelector, long offset, int maxNums,
+ PullCallback pullCallback,
+ long timeout)
+ throws MQClientException, RemotingException, InterruptedException {
+ SubscriptionData subscriptionData = getSubscriptionData(mq, messageSelector);
+ this.pullAsyncImpl(mq, subscriptionData, offset, maxNums, pullCallback, false, timeout);
}
private void pullAsyncImpl(
final MessageQueue mq,
- final String subExpression,
+ final SubscriptionData subscriptionData,
final long offset,
final int maxNums,
final PullCallback pullCallback,
@@ -403,20 +455,14 @@ private void pullAsyncImpl(
try {
int sysFlag = PullSysFlag.buildSysFlag(false, block, true, false);
- final SubscriptionData subscriptionData;
- try {
- subscriptionData = FilterAPI.buildSubscriptionData(this.defaultMQPullConsumer.getConsumerGroup(),
- mq.getTopic(), subExpression);
- } catch (Exception e) {
- throw new MQClientException("parse subscription error", e);
- }
-
long timeoutMillis = block ? this.defaultMQPullConsumer.getConsumerTimeoutMillisWhenSuspend() : timeout;
+ boolean isTagType = ExpressionType.isTagType(subscriptionData.getExpressionType());
this.pullAPIWrapper.pullKernelImpl(
mq,
subscriptionData.getSubString(),
- 0L,
+ subscriptionData.getExpressionType(),
+ isTagType ? 0L : subscriptionData.getSubVersion(),
offset,
maxNums,
sysFlag,
@@ -444,7 +490,8 @@ public void onException(Throwable e) {
public PullResult pullBlockIfNotFound(MessageQueue mq, String subExpression, long offset, int maxNums)
throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
- return this.pullSyncImpl(mq, subExpression, offset, maxNums, true, this.getDefaultMQPullConsumer().getConsumerPullTimeoutMillis());
+ SubscriptionData subscriptionData = getSubscriptionData(mq, subExpression);
+ return this.pullSyncImpl(mq, subscriptionData, offset, maxNums, true, this.getDefaultMQPullConsumer().getConsumerPullTimeoutMillis());
}
public DefaultMQPullConsumer getDefaultMQPullConsumer() {
@@ -454,7 +501,8 @@ public DefaultMQPullConsumer getDefaultMQPullConsumer() {
public void pullBlockIfNotFound(MessageQueue mq, String subExpression, long offset, int maxNums,
PullCallback pullCallback)
throws MQClientException, RemotingException, InterruptedException {
- this.pullAsyncImpl(mq, subExpression, offset, maxNums, pullCallback, true,
+ SubscriptionData subscriptionData = getSubscriptionData(mq, subExpression);
+ this.pullAsyncImpl(mq, subscriptionData, offset, maxNums, pullCallback, true,
this.getDefaultMQPullConsumer().getConsumerPullTimeoutMillis());
}
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 b650e35e0..1d2d24fa3 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
@@ -209,34 +209,6 @@ public PullResult pullKernelImpl(
throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);
}
- public PullResult pullKernelImpl(
- final MessageQueue mq,
- final String subExpression,
- final long subVersion,
- final long offset,
- final int maxNums,
- final int sysFlag,
- final long commitOffset,
- final long brokerSuspendMaxTimeMillis,
- final long timeoutMillis,
- final CommunicationMode communicationMode,
- final PullCallback pullCallback
- ) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
- return pullKernelImpl(
- mq,
- subExpression,
- ExpressionType.TAG,
- subVersion, offset,
- maxNums,
- sysFlag,
- commitOffset,
- brokerSuspendMaxTimeMillis,
- timeoutMillis,
- communicationMode,
- pullCallback
- );
- }
-
public long recalculatePullFromWhichNode(final MessageQueue mq) {
if (this.isConnectBrokerByUser()) {
return this.defaultBrokerId;
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java b/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java
index 9ffaed0a4..80347d105 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java
@@ -1046,6 +1046,19 @@ public int findBrokerVersion(String brokerName, String brokerAddr) {
if (this.brokerVersionTable.get(brokerName).containsKey(brokerAddr)) {
return this.brokerVersionTable.get(brokerName).get(brokerAddr);
}
+ } else {
+ HeartbeatData heartbeatData = prepareHeartbeatData();
+ try {
+ int version = this.mQClientAPIImpl.sendHearbeat(brokerAddr, heartbeatData, 3000);
+ return version;
+ } catch (Exception e) {
+ if (this.isBrokerInNameServer(brokerAddr)) {
+ log.info("send heart beat to broker[{} {}] failed", brokerName, brokerAddr);
+ } else {
+ log.info("send heart beat to broker[{} {}] exception, because the broker not up, forget it", brokerName,
+ brokerAddr);
+ }
+ }
}
return 0;
}
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java
index 7ace9d5b0..90f4f7876 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java
@@ -30,8 +30,10 @@
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.rocketmq.client.QueryResult;
import org.apache.rocketmq.client.Validators;
import org.apache.rocketmq.client.common.ClientErrorCode;
@@ -101,6 +103,10 @@
private MQFaultStrategy mqFaultStrategy = new MQFaultStrategy();
+ private final BlockingQueue<Runnable> asyncSenderThreadPoolQueue;
+ private final ExecutorService defaultAsyncSenderExecutor;
+ private ExecutorService asyncSenderExecutor;
+
public DefaultMQProducerImpl(final DefaultMQProducer defaultMQProducer) {
this(defaultMQProducer, null);
}
@@ -108,6 +114,22 @@ public DefaultMQProducerImpl(final DefaultMQProducer defaultMQProducer) {
public DefaultMQProducerImpl(final DefaultMQProducer defaultMQProducer, RPCHook rpcHook) {
this.defaultMQProducer = defaultMQProducer;
this.rpcHook = rpcHook;
+
+ this.asyncSenderThreadPoolQueue = new LinkedBlockingQueue<Runnable>(50000);
+ this.defaultAsyncSenderExecutor = new ThreadPoolExecutor(
+ Runtime.getRuntime().availableProcessors(),
+ Runtime.getRuntime().availableProcessors(),
+ 1000 * 60,
+ TimeUnit.MILLISECONDS,
+ this.asyncSenderThreadPoolQueue,
+ new ThreadFactory() {
+ private AtomicInteger threadIndex = new AtomicInteger(0);
+
+ @Override
+ public Thread newThread(Runnable r) {
+ return new Thread(r, "AsyncSenderExecutor_" + this.threadIndex.incrementAndGet());
+ }
+ });
}
public void registerCheckForbiddenHook(CheckForbiddenHook checkForbiddenHook) {
@@ -456,7 +478,7 @@ public void send(Message msg,
public void send(final Message msg, final SendCallback sendCallback, final long timeout)
throws MQClientException, RemotingException, InterruptedException {
final long beginStartTime = System.currentTimeMillis();
- ExecutorService executor = this.getCallbackExecutor();
+ ExecutorService executor = this.getAsyncSenderExecutor();
try {
executor.submit(new Runnable() {
@Override
@@ -957,7 +979,7 @@ public void send(Message msg, MessageQueue mq, SendCallback sendCallback)
public void send(final Message msg, final MessageQueue mq, final SendCallback sendCallback, final long timeout)
throws MQClientException, RemotingException, InterruptedException {
final long beginStartTime = System.currentTimeMillis();
- ExecutorService executor = this.getCallbackExecutor();
+ ExecutorService executor = this.getAsyncSenderExecutor();
try {
executor.submit(new Runnable() {
@Override
@@ -1079,7 +1101,7 @@ public void send(Message msg, MessageQueueSelector selector, Object arg, SendCal
public void send(final Message msg, final MessageQueueSelector selector, final Object arg, final SendCallback sendCallback, final long timeout)
throws MQClientException, RemotingException, InterruptedException {
final long beginStartTime = System.currentTimeMillis();
- ExecutorService executor = this.getCallbackExecutor();
+ ExecutorService executor = this.getAsyncSenderExecutor();
try {
executor.submit(new Runnable() {
@Override
@@ -1243,9 +1265,13 @@ public void endTransaction(
public void setCallbackExecutor(final ExecutorService callbackExecutor) {
this.mQClientFactory.getMQClientAPIImpl().getRemotingClient().setCallbackExecutor(callbackExecutor);
}
- public ExecutorService getCallbackExecutor() {
- return this.mQClientFactory.getMQClientAPIImpl().getRemotingClient().getCallbackExecutor();
+ public ExecutorService getAsyncSenderExecutor() {
+ return null == asyncSenderExecutor ? defaultAsyncSenderExecutor : asyncSenderExecutor;
+ }
+
+ public void setAsyncSenderExecutor(ExecutorService asyncSenderExecutor) {
+ this.asyncSenderExecutor = asyncSenderExecutor;
}
public SendResult send(Message msg,
diff --git a/client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java b/client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java
index 9732d0eb8..233914261 100644
--- a/client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java
+++ b/client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java
@@ -25,6 +25,10 @@
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl;
+import org.apache.rocketmq.client.log.ClientLogger;
+import org.apache.rocketmq.client.trace.AsyncTraceDispatcher;
+import org.apache.rocketmq.client.trace.TraceDispatcher;
+import org.apache.rocketmq.client.trace.hook.SendMessageTraceHookImpl;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageBatch;
@@ -33,6 +37,7 @@
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageId;
import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.remoting.netty.NettyRemotingClient;
@@ -56,6 +61,8 @@
*/
public class DefaultMQProducer extends ClientConfig implements MQProducer {
+ private final InternalLogger log = ClientLogger.getLog();
+
/**
* Wrapping internal implementations for virtually all methods presented in this class.
*/
@@ -119,6 +126,11 @@
*/
private int maxMessageSize = 1024 * 1024 * 4; // 4M
+ /**
+ * Interface of asynchronous transfer data
+ */
+ private TraceDispatcher traceDispatcher = null;
+
/**
* Default constructor.
*/
@@ -137,6 +149,31 @@ public DefaultMQProducer(final String producerGroup, RPCHook rpcHook) {
defaultMQProducerImpl = new DefaultMQProducerImpl(this, rpcHook);
}
+ /**
+ * Constructor specifying producer group, RPC hook, enabled msgTrace flag and customized trace topic name.
+ *
+ * @param producerGroup Producer group, see the name-sake field.
+ * @param rpcHook RPC hook to execute per each remoting command execution.
+ * @param enableMsgTrace Switch flag instance for message trace.
+ * @param customizedTraceTopic The name value of message trace topic.If you don't config,you can use the default trace topic name.
+ */
+ public DefaultMQProducer(final String producerGroup, RPCHook rpcHook, boolean enableMsgTrace,final String customizedTraceTopic) {
+ this.producerGroup = producerGroup;
+ defaultMQProducerImpl = new DefaultMQProducerImpl(this, rpcHook);
+ //if client open the message trace feature
+ if (enableMsgTrace) {
+ try {
+ AsyncTraceDispatcher dispatcher = new AsyncTraceDispatcher(customizedTraceTopic, rpcHook);
+ dispatcher.setHostProducer(this.getDefaultMQProducerImpl());
+ traceDispatcher = dispatcher;
+ this.getDefaultMQProducerImpl().registerSendMessageHook(
+ new SendMessageTraceHookImpl(traceDispatcher));
+ } catch (Throwable e) {
+ log.error("system mqtrace hook init failed ,maybe can't send msg trace data");
+ }
+ }
+ }
+
/**
* Constructor specifying producer group.
*
@@ -147,8 +184,30 @@ public DefaultMQProducer(final String producerGroup) {
}
/**
- * Constructor specifying the RPC hook.
+ * Constructor specifying producer group and enabled msg trace flag.
+ *
+ * @param producerGroup Producer group, see the name-sake field.
+ * @param enableMsgTrace Switch flag instance for message trace.
+ */
+ public DefaultMQProducer(final String producerGroup, boolean enableMsgTrace) {
+ this(producerGroup, null, enableMsgTrace, null);
+ }
+
+
+ /**
+ * Constructor specifying producer group, enabled msgTrace flag and customized trace topic name.
*
+ * @param producerGroup Producer group, see the name-sake field.
+ * @param enableMsgTrace Switch flag instance for message trace.
+ * @param customizedTraceTopic The name value of message trace topic.If you don't config,you can use the default trace topic name.
+ */
+ public DefaultMQProducer(final String producerGroup, boolean enableMsgTrace, final String customizedTraceTopic) {
+ this(producerGroup, null, enableMsgTrace, customizedTraceTopic);
+ }
+
+ /**
+ * Constructor specifying the RPC hook.
+ *
* @param rpcHook RPC hook to execute per each remoting command execution.
*/
public DefaultMQProducer(RPCHook rpcHook) {
@@ -170,6 +229,13 @@ public DefaultMQProducer(RPCHook rpcHook) {
@Override
public void start() throws MQClientException {
this.defaultMQProducerImpl.start();
+ if (null != traceDispatcher) {
+ try {
+ traceDispatcher.start(this.getNamesrvAddr());
+ } catch (MQClientException e) {
+ log.warn("trace dispatcher start failed ", e);
+ }
+ }
}
/**
@@ -178,6 +244,9 @@ public void start() throws MQClientException {
@Override
public void shutdown() {
this.defaultMQProducerImpl.shutdown();
+ if (null != traceDispatcher) {
+ traceDispatcher.shutdown();
+ }
}
/**
@@ -655,6 +724,16 @@ public void setCallbackExecutor(final ExecutorService callbackExecutor) {
this.defaultMQProducerImpl.setCallbackExecutor(callbackExecutor);
}
+ /**
+ * Sets an Executor to be used for executing asynchronous send. If the Executor is not set, {@link
+ * DefaultMQProducerImpl#defaultAsyncSenderExecutor} will be used.
+ *
+ * @param asyncSenderExecutor the instance of Executor
+ */
+ public void setAsyncSenderExecutor(final ExecutorService asyncSenderExecutor) {
+ this.defaultMQProducerImpl.setAsyncSenderExecutor(asyncSenderExecutor);
+ }
+
private MessageBatch batch(Collection<Message> msgs) throws MQClientException {
MessageBatch msgBatch;
try {
@@ -777,4 +856,9 @@ public int getRetryTimesWhenSendAsyncFailed() {
public void setRetryTimesWhenSendAsyncFailed(final int retryTimesWhenSendAsyncFailed) {
this.retryTimesWhenSendAsyncFailed = retryTimesWhenSendAsyncFailed;
}
+
+ public TraceDispatcher getTraceDispatcher() {
+ return traceDispatcher;
+ }
+
}
diff --git a/client/src/main/java/org/apache/rocketmq/client/trace/AsyncTraceDispatcher.java b/client/src/main/java/org/apache/rocketmq/client/trace/AsyncTraceDispatcher.java
new file mode 100644
index 000000000..554232441
--- /dev/null
+++ b/client/src/main/java/org/apache/rocketmq/client/trace/AsyncTraceDispatcher.java
@@ -0,0 +1,385 @@
+/*
+ * 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.client.trace;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.rocketmq.client.common.ThreadLocalIndex;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl;
+import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl;
+import org.apache.rocketmq.client.impl.producer.TopicPublishInfo;
+import org.apache.rocketmq.client.log.ClientLogger;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.MessageQueueSelector;
+import org.apache.rocketmq.client.producer.SendCallback;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.ThreadFactoryImpl;
+import org.apache.rocketmq.common.UtilAll;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.logging.InternalLogger;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.HashMap;
+import java.util.UUID;
+import java.util.Set;
+import java.util.HashSet;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.rocketmq.remoting.RPCHook;
+
+import static org.apache.rocketmq.client.trace.TraceConstants.TRACE_INSTANCE_NAME;
+
+public class AsyncTraceDispatcher implements TraceDispatcher {
+
+ private final static InternalLogger log = ClientLogger.getLog();
+ private final int queueSize;
+ private final int batchSize;
+ private final int maxMsgSize;
+ private final DefaultMQProducer traceProducer;
+ private final ThreadPoolExecutor traceExecuter;
+ // The last discard number of log
+ private AtomicLong discardCount;
+ private Thread worker;
+ private ArrayBlockingQueue<TraceContext> traceContextQueue;
+ private ArrayBlockingQueue<Runnable> appenderQueue;
+ private volatile Thread shutDownHook;
+ private volatile boolean stopped = false;
+ private DefaultMQProducerImpl hostProducer;
+ private DefaultMQPushConsumerImpl hostConsumer;
+ private volatile ThreadLocalIndex sendWhichQueue = new ThreadLocalIndex();
+ private String dispatcherId = UUID.randomUUID().toString();
+ private String traceTopicName;
+ private static AtomicBoolean isStarted = new AtomicBoolean(false);
+
+
+ public AsyncTraceDispatcher(String traceTopicName, RPCHook rpcHook) throws MQClientException {
+ // queueSize is greater than or equal to the n power of 2 of value
+ this.queueSize = 2048;
+ this.batchSize = 100;
+ this.maxMsgSize = 128000;
+ this.discardCount = new AtomicLong(0L);
+ this.traceContextQueue = new ArrayBlockingQueue<TraceContext>(1024);
+ this.appenderQueue = new ArrayBlockingQueue<Runnable>(queueSize);
+ if (!UtilAll.isBlank(traceTopicName)) {
+ this.traceTopicName = traceTopicName;
+ } else {
+ this.traceTopicName = MixAll.RMQ_SYS_TRACE_TOPIC;
+ }
+ this.traceExecuter = new ThreadPoolExecutor(//
+ 10, //
+ 20, //
+ 1000 * 60, //
+ TimeUnit.MILLISECONDS, //
+ this.appenderQueue, //
+ new ThreadFactoryImpl("MQTraceSendThread_"));
+ traceProducer = getAndCreateTraceProducer(rpcHook);
+ }
+
+ public String getTraceTopicName() {
+ return traceTopicName;
+ }
+
+ public void setTraceTopicName(String traceTopicName) {
+ this.traceTopicName = traceTopicName;
+ }
+
+ public DefaultMQProducer getTraceProducer() {
+ return traceProducer;
+ }
+
+ public DefaultMQProducerImpl getHostProducer() {
+ return hostProducer;
+ }
+
+ public void setHostProducer(DefaultMQProducerImpl hostProducer) {
+ this.hostProducer = hostProducer;
+ }
+
+ public DefaultMQPushConsumerImpl getHostConsumer() {
+ return hostConsumer;
+ }
+
+ public void setHostConsumer(DefaultMQPushConsumerImpl hostConsumer) {
+ this.hostConsumer = hostConsumer;
+ }
+
+ public void start(String nameSrvAddr) throws MQClientException {
+ if (isStarted.compareAndSet(false, true)) {
+ traceProducer.setNamesrvAddr(nameSrvAddr);
+ traceProducer.start();
+ }
+ this.worker = new Thread(new AsyncRunnable(), "MQ-AsyncTraceDispatcher-Thread-" + dispatcherId);
+ this.worker.setDaemon(true);
+ this.worker.start();
+ this.registerShutDownHook();
+ }
+
+ private DefaultMQProducer getAndCreateTraceProducer(RPCHook rpcHook) {
+ DefaultMQProducer traceProducerInstance = this.traceProducer;
+ if (traceProducerInstance == null) {
+ traceProducerInstance = new DefaultMQProducer(rpcHook);
+ traceProducerInstance.setProducerGroup(TraceConstants.GROUP_NAME);
+ traceProducerInstance.setSendMsgTimeout(5000);
+ traceProducerInstance.setInstanceName(TRACE_INSTANCE_NAME);
+ traceProducerInstance.setVipChannelEnabled(false);
+ // The max size of message is 128K
+ traceProducerInstance.setMaxMessageSize(maxMsgSize - 10 * 1000);
+ }
+ return traceProducerInstance;
+ }
+
+ @Override
+ public boolean append(final Object ctx) {
+ boolean result = traceContextQueue.offer((TraceContext) ctx);
+ if (!result) {
+ log.info("buffer full" + discardCount.incrementAndGet() + " ,context is " + ctx);
+ }
+ return result;
+ }
+
+ @Override
+ public void flush() throws IOException {
+ // The maximum waiting time for refresh,avoid being written all the time, resulting in failure to return.
+ long end = System.currentTimeMillis() + 500;
+ while (traceContextQueue.size() > 0 || appenderQueue.size() > 0 && System.currentTimeMillis() <= end) {
+ try {
+ Thread.sleep(1);
+ } catch (InterruptedException e) {
+ break;
+ }
+ }
+ log.info("------end trace send " + traceContextQueue.size() + " " + appenderQueue.size());
+ }
+
+ @Override
+ public void shutdown() {
+ this.stopped = true;
+ this.traceExecuter.shutdown();
+ if (isStarted.get()) {
+ traceProducer.shutdown();
+ }
+ this.removeShutdownHook();
+ }
+
+ public void registerShutDownHook() {
+ if (shutDownHook == null) {
+ shutDownHook = new Thread(new Runnable() {
+ private volatile boolean hasShutdown = false;
+
+ @Override
+ public void run() {
+ synchronized (this) {
+ if (!this.hasShutdown) {
+ try {
+ flush();
+ } catch (IOException e) {
+ log.error("system MQTrace hook shutdown failed ,maybe loss some trace data");
+ }
+ }
+ }
+ }
+ }, "ShutdownHookMQTrace");
+ Runtime.getRuntime().addShutdownHook(shutDownHook);
+ }
+ }
+
+ public void removeShutdownHook() {
+ if (shutDownHook != null) {
+ Runtime.getRuntime().removeShutdownHook(shutDownHook);
+ }
+ }
+
+ class AsyncRunnable implements Runnable {
+ private boolean stopped;
+
+ @Override
+ public void run() {
+ while (!stopped) {
+ List<TraceContext> contexts = new ArrayList<TraceContext>(batchSize);
+ for (int i = 0; i < batchSize; i++) {
+ TraceContext context = null;
+ try {
+ //get trace data element from blocking Queue — traceContextQueue
+ context = traceContextQueue.poll(5, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ }
+ if (context != null) {
+ contexts.add(context);
+ } else {
+ break;
+ }
+ }
+ if (contexts.size() > 0) {
+ AsyncAppenderRequest request = new AsyncAppenderRequest(contexts);
+ traceExecuter.submit(request);
+ } else if (AsyncTraceDispatcher.this.stopped) {
+ this.stopped = true;
+ }
+ }
+
+ }
+ }
+
+ class AsyncAppenderRequest implements Runnable {
+ List<TraceContext> contextList;
+
+ public AsyncAppenderRequest(final List<TraceContext> contextList) {
+ if (contextList != null) {
+ this.contextList = contextList;
+ } else {
+ this.contextList = new ArrayList<TraceContext>(1);
+ }
+ }
+
+ @Override
+ public void run() {
+ sendTraceData(contextList);
+ }
+
+ public void sendTraceData(List<TraceContext> contextList) {
+ Map<String, List<TraceTransferBean>> transBeanMap = new HashMap<String, List<TraceTransferBean>>();
+ for (TraceContext context : contextList) {
+ if (context.getTraceBeans().isEmpty()) {
+ continue;
+ }
+ // Topic value corresponding to original message entity content
+ String topic = context.getTraceBeans().get(0).getTopic();
+ // Use original message entity's topic as key
+ String key = topic;
+ List<TraceTransferBean> transBeanList = transBeanMap.get(key);
+ if (transBeanList == null) {
+ transBeanList = new ArrayList<TraceTransferBean>();
+ transBeanMap.put(key, transBeanList);
+ }
+ TraceTransferBean traceData = TraceDataEncoder.encoderFromContextBean(context);
+ transBeanList.add(traceData);
+ }
+ for (Map.Entry<String, List<TraceTransferBean>> entry : transBeanMap.entrySet()) {
+ flushData(entry.getValue());
+ }
+ }
+
+ /**
+ * Batch sending data actually
+ */
+ private void flushData(List<TraceTransferBean> transBeanList) {
+ if (transBeanList.size() == 0) {
+ return;
+ }
+ // Temporary buffer
+ StringBuilder buffer = new StringBuilder(1024);
+ int count = 0;
+ Set<String> keySet = new HashSet<String>();
+
+ for (TraceTransferBean bean : transBeanList) {
+ // Keyset of message trace includes msgId of or original message
+ keySet.addAll(bean.getTransKey());
+ buffer.append(bean.getTransData());
+ count++;
+ // Ensure that the size of the package should not exceed the upper limit.
+ if (buffer.length() >= traceProducer.getMaxMessageSize()) {
+ sendTraceDataByMQ(keySet, buffer.toString());
+ // Clear temporary buffer after finishing
+ buffer.delete(0, buffer.length());
+ keySet.clear();
+ count = 0;
+ }
+ }
+ if (count > 0) {
+ sendTraceDataByMQ(keySet, buffer.toString());
+ }
+ transBeanList.clear();
+ }
+
+ /**
+ * Send message trace data
+ *
+ * @param keySet the keyset in this batch(including msgId in original message not offsetMsgId)
+ * @param data the message trace data in this batch
+ */
+ private void sendTraceDataByMQ(Set<String> keySet, final String data) {
+ String topic = traceTopicName;
+ final Message message = new Message(topic, data.getBytes());
+
+ // Keyset of message trace includes msgId of or original message
+ message.setKeys(keySet);
+ try {
+ Set<String> traceBrokerSet = tryGetMessageQueueBrokerSet(traceProducer.getDefaultMQProducerImpl(), topic);
+ SendCallback callback = new SendCallback() {
+ @Override
+ public void onSuccess(SendResult sendResult) {
+ }
+
+ @Override
+ public void onException(Throwable e) {
+ log.info("send trace data ,the traceData is " + data);
+ }
+ };
+ if (traceBrokerSet.isEmpty()) {
+ // No cross set
+ traceProducer.send(message, callback, 5000);
+ } else {
+ traceProducer.send(message, new MessageQueueSelector() {
+ @Override
+ public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
+ Set<String> brokerSet = (Set<String>) arg;
+ List<MessageQueue> filterMqs = new ArrayList<MessageQueue>();
+ for (MessageQueue queue : mqs) {
+ if (brokerSet.contains(queue.getBrokerName())) {
+ filterMqs.add(queue);
+ }
+ }
+ int index = sendWhichQueue.getAndIncrement();
+ int pos = Math.abs(index) % filterMqs.size();
+ if (pos < 0) {
+ pos = 0;
+ }
+ return filterMqs.get(pos);
+ }
+ }, traceBrokerSet, callback);
+ }
+
+ } catch (Exception e) {
+ log.info("send trace data,the traceData is" + data);
+ }
+ }
+
+ private Set<String> tryGetMessageQueueBrokerSet(DefaultMQProducerImpl producer, String topic) {
+ Set<String> brokerSet = new HashSet<String>();
+ TopicPublishInfo topicPublishInfo = producer.getTopicPublishInfoTable().get(topic);
+ if (null == topicPublishInfo || !topicPublishInfo.ok()) {
+ producer.getTopicPublishInfoTable().putIfAbsent(topic, new TopicPublishInfo());
+ producer.getmQClientFactory().updateTopicRouteInfoFromNameServer(topic);
+ topicPublishInfo = producer.getTopicPublishInfoTable().get(topic);
+ }
+ if (topicPublishInfo.isHaveTopicRouterInfo() || topicPublishInfo.ok()) {
+ for (MessageQueue queue : topicPublishInfo.getMessageQueueList()) {
+ brokerSet.add(queue.getBrokerName());
+ }
+ }
+ return brokerSet;
+ }
+ }
+
+}
diff --git a/client/src/main/java/org/apache/rocketmq/client/trace/TraceBean.java b/client/src/main/java/org/apache/rocketmq/client/trace/TraceBean.java
new file mode 100644
index 000000000..f93aa38b8
--- /dev/null
+++ b/client/src/main/java/org/apache/rocketmq/client/trace/TraceBean.java
@@ -0,0 +1,144 @@
+/*
+ * 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.client.trace;
+
+import org.apache.rocketmq.common.UtilAll;
+import org.apache.rocketmq.common.message.MessageType;
+
+public class TraceBean {
+ private static final String LOCAL_ADDRESS = UtilAll.ipToIPv4Str(UtilAll.getIP());
+ private String topic = "";
+ private String msgId = "";
+ private String offsetMsgId = "";
+ private String tags = "";
+ private String keys = "";
+ private String storeHost = LOCAL_ADDRESS;
+ private String clientHost = LOCAL_ADDRESS;
+ private long storeTime;
+ private int retryTimes;
+ private int bodyLength;
+ private MessageType msgType;
+
+
+ public MessageType getMsgType() {
+ return msgType;
+ }
+
+
+ public void setMsgType(final MessageType msgType) {
+ this.msgType = msgType;
+ }
+
+
+ public String getOffsetMsgId() {
+ return offsetMsgId;
+ }
+
+
+ public void setOffsetMsgId(final String offsetMsgId) {
+ this.offsetMsgId = offsetMsgId;
+ }
+
+ public String getTopic() {
+ return topic;
+ }
+
+
+ public void setTopic(String topic) {
+ this.topic = topic;
+ }
+
+
+ public String getMsgId() {
+ return msgId;
+ }
+
+
+ public void setMsgId(String msgId) {
+ this.msgId = msgId;
+ }
+
+
+ public String getTags() {
+ return tags;
+ }
+
+
+ public void setTags(String tags) {
+ this.tags = tags;
+ }
+
+
+ public String getKeys() {
+ return keys;
+ }
+
+
+ public void setKeys(String keys) {
+ this.keys = keys;
+ }
+
+
+ public String getStoreHost() {
+ return storeHost;
+ }
+
+
+ public void setStoreHost(String storeHost) {
+ this.storeHost = storeHost;
+ }
+
+
+ public String getClientHost() {
+ return clientHost;
+ }
+
+
+ public void setClientHost(String clientHost) {
+ this.clientHost = clientHost;
+ }
+
+
+ public long getStoreTime() {
+ return storeTime;
+ }
+
+
+ public void setStoreTime(long storeTime) {
+ this.storeTime = storeTime;
+ }
+
+
+ public int getRetryTimes() {
+ return retryTimes;
+ }
+
+
+ public void setRetryTimes(int retryTimes) {
+ this.retryTimes = retryTimes;
+ }
+
+
+ public int getBodyLength() {
+ return bodyLength;
+ }
+
+
+ public void setBodyLength(int bodyLength) {
+ this.bodyLength = bodyLength;
+ }
+}
diff --git a/client/src/main/java/org/apache/rocketmq/client/trace/TraceConstants.java b/client/src/main/java/org/apache/rocketmq/client/trace/TraceConstants.java
new file mode 100644
index 000000000..b9fd8778e
--- /dev/null
+++ b/client/src/main/java/org/apache/rocketmq/client/trace/TraceConstants.java
@@ -0,0 +1,25 @@
+/*
+ * 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.client.trace;
+
+public class TraceConstants {
+
+ public static final String GROUP_NAME = "_INNER_TRACE_PRODUCER";
+ public static final char CONTENT_SPLITOR = (char) 1;
+ public static final char FIELD_SPLITOR = (char) 2;
+ public static final String TRACE_INSTANCE_NAME = "PID_CLIENT_INNER_TRACE_PRODUCER";
+}
diff --git a/client/src/main/java/org/apache/rocketmq/client/trace/TraceContext.java b/client/src/main/java/org/apache/rocketmq/client/trace/TraceContext.java
new file mode 100644
index 000000000..f61ba888c
--- /dev/null
+++ b/client/src/main/java/org/apache/rocketmq/client/trace/TraceContext.java
@@ -0,0 +1,136 @@
+/*
+ * 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.client.trace;
+
+import org.apache.rocketmq.common.message.MessageClientIDSetter;
+
+import java.util.List;
+
+/**
+ * The context of Trace
+ */
+public class TraceContext implements Comparable<TraceContext> {
+
+ private TraceType traceType;
+ private long timeStamp = System.currentTimeMillis();
+ private String regionId = "";
+ private String regionName = "";
+ private String groupName = "";
+ private int costTime = 0;
+ private boolean isSuccess = true;
+ private String requestId = MessageClientIDSetter.createUniqID();
+ private int contextCode = 0;
+ private List<TraceBean> traceBeans;
+
+ public int getContextCode() {
+ return contextCode;
+ }
+
+ public void setContextCode(final int contextCode) {
+ this.contextCode = contextCode;
+ }
+
+ public List<TraceBean> getTraceBeans() {
+ return traceBeans;
+ }
+
+ public void setTraceBeans(List<TraceBean> traceBeans) {
+ this.traceBeans = traceBeans;
+ }
+
+ public String getRegionId() {
+ return regionId;
+ }
+
+ public void setRegionId(String regionId) {
+ this.regionId = regionId;
+ }
+
+ public TraceType getTraceType() {
+ return traceType;
+ }
+
+ public void setTraceType(TraceType traceType) {
+ this.traceType = traceType;
+ }
+
+ public long getTimeStamp() {
+ return timeStamp;
+ }
+
+ public void setTimeStamp(long timeStamp) {
+ this.timeStamp = timeStamp;
+ }
+
+ public String getGroupName() {
+ return groupName;
+ }
+
+ public void setGroupName(String groupName) {
+ this.groupName = groupName;
+ }
+
+ public int getCostTime() {
+ return costTime;
+ }
+
+ public void setCostTime(int costTime) {
+ this.costTime = costTime;
+ }
+
+ public boolean isSuccess() {
+ return isSuccess;
+ }
+
+ public void setSuccess(boolean success) {
+ isSuccess = success;
+ }
+
+ public String getRequestId() {
+ return requestId;
+ }
+
+ public void setRequestId(String requestId) {
+ this.requestId = requestId;
+ }
+
+ public String getRegionName() {
+ return regionName;
+ }
+
+ public void setRegionName(String regionName) {
+ this.regionName = regionName;
+ }
+
+ @Override
+ public int compareTo(TraceContext o) {
+ return (int) (this.timeStamp - o.getTimeStamp());
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder(1024);
+ sb.append(traceType).append("_").append(groupName)
+ .append("_").append(regionId).append("_").append(isSuccess).append("_");
+ if (traceBeans != null && traceBeans.size() > 0) {
+ for (TraceBean bean : traceBeans) {
+ sb.append(bean.getMsgId() + "_" + bean.getTopic() + "_");
+ }
+ }
+ return "TraceContext{" + sb.toString() + '}';
+ }
+}
diff --git a/client/src/main/java/org/apache/rocketmq/client/trace/TraceDataEncoder.java b/client/src/main/java/org/apache/rocketmq/client/trace/TraceDataEncoder.java
new file mode 100644
index 000000000..5a1afaf36
--- /dev/null
+++ b/client/src/main/java/org/apache/rocketmq/client/trace/TraceDataEncoder.java
@@ -0,0 +1,173 @@
+/*
+ * 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.client.trace;
+
+import org.apache.rocketmq.common.message.MessageType;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Encode/decode for Trace Data
+ */
+public class TraceDataEncoder {
+
+ /**
+ * Resolving traceContext list From trace data String
+ *
+ * @param traceData
+ * @return
+ */
+ public static List<TraceContext> decoderFromTraceDataString(String traceData) {
+ List<TraceContext> resList = new ArrayList<TraceContext>();
+ if (traceData == null || traceData.length() <= 0) {
+ return resList;
+ }
+ String[] contextList = traceData.split(String.valueOf(TraceConstants.FIELD_SPLITOR));
+ for (String context : contextList) {
+ String[] line = context.split(String.valueOf(TraceConstants.CONTENT_SPLITOR));
+ if (line[0].equals(TraceType.Pub.name())) {
+ TraceContext pubContext = new TraceContext();
+ pubContext.setTraceType(TraceType.Pub);
+ pubContext.setTimeStamp(Long.parseLong(line[1]));
+ pubContext.setRegionId(line[2]);
+ pubContext.setGroupName(line[3]);
+ TraceBean bean = new TraceBean();
+ bean.setTopic(line[4]);
+ bean.setMsgId(line[5]);
+ bean.setTags(line[6]);
+ bean.setKeys(line[7]);
+ bean.setStoreHost(line[8]);
+ bean.setBodyLength(Integer.parseInt(line[9]));
+ pubContext.setCostTime(Integer.parseInt(line[10]));
+ bean.setMsgType(MessageType.values()[Integer.parseInt(line[11])]);
+
+ if (line.length == 13) {
+ pubContext.setSuccess(Boolean.parseBoolean(line[12]));
+ } else if (line.length == 14) {
+ bean.setOffsetMsgId(line[12]);
+ pubContext.setSuccess(Boolean.parseBoolean(line[13]));
+ }
+ pubContext.setTraceBeans(new ArrayList<TraceBean>(1));
+ pubContext.getTraceBeans().add(bean);
+ resList.add(pubContext);
+ } else if (line[0].equals(TraceType.SubBefore.name())) {
+ TraceContext subBeforeContext = new TraceContext();
+ subBeforeContext.setTraceType(TraceType.SubBefore);
+ subBeforeContext.setTimeStamp(Long.parseLong(line[1]));
+ subBeforeContext.setRegionId(line[2]);
+ subBeforeContext.setGroupName(line[3]);
+ subBeforeContext.setRequestId(line[4]);
+ TraceBean bean = new TraceBean();
+ bean.setMsgId(line[5]);
+ bean.setRetryTimes(Integer.parseInt(line[6]));
+ bean.setKeys(line[7]);
+ subBeforeContext.setTraceBeans(new ArrayList<TraceBean>(1));
+ subBeforeContext.getTraceBeans().add(bean);
+ resList.add(subBeforeContext);
+ } else if (line[0].equals(TraceType.SubAfter.name())) {
+ TraceContext subAfterContext = new TraceContext();
+ subAfterContext.setTraceType(TraceType.SubAfter);
+ subAfterContext.setRequestId(line[1]);
+ TraceBean bean = new TraceBean();
+ bean.setMsgId(line[2]);
+ bean.setKeys(line[5]);
+ subAfterContext.setTraceBeans(new ArrayList<TraceBean>(1));
+ subAfterContext.getTraceBeans().add(bean);
+ subAfterContext.setCostTime(Integer.parseInt(line[3]));
+ subAfterContext.setSuccess(Boolean.parseBoolean(line[4]));
+ if (line.length >= 7) {
+ // add the context type
+ subAfterContext.setContextCode(Integer.parseInt(line[6]));
+ }
+ resList.add(subAfterContext);
+ }
+ }
+ return resList;
+ }
+
+ /**
+ * Encoding the trace context into data strings and keyset sets
+ *
+ * @param ctx
+ * @return
+ */
+ public static TraceTransferBean encoderFromContextBean(TraceContext ctx) {
+ if (ctx == null) {
+ return null;
+ }
+ //build message trace of the transfering entity content bean
+ TraceTransferBean transferBean = new TraceTransferBean();
+ StringBuilder sb = new StringBuilder(256);
+ switch (ctx.getTraceType()) {
+ case Pub: {
+ TraceBean bean = ctx.getTraceBeans().get(0);
+ //append the content of context and traceBean to transferBean's TransData
+ sb.append(ctx.getTraceType()).append(TraceConstants.CONTENT_SPLITOR)//
+ .append(ctx.getTimeStamp()).append(TraceConstants.CONTENT_SPLITOR)//
+ .append(ctx.getRegionId()).append(TraceConstants.CONTENT_SPLITOR)//
+ .append(ctx.getGroupName()).append(TraceConstants.CONTENT_SPLITOR)//
+ .append(bean.getTopic()).append(TraceConstants.CONTENT_SPLITOR)//
+ .append(bean.getMsgId()).append(TraceConstants.CONTENT_SPLITOR)//
+ .append(bean.getTags()).append(TraceConstants.CONTENT_SPLITOR)//
+ .append(bean.getKeys()).append(TraceConstants.CONTENT_SPLITOR)//
+ .append(bean.getStoreHost()).append(TraceConstants.CONTENT_SPLITOR)//
+ .append(bean.getBodyLength()).append(TraceConstants.CONTENT_SPLITOR)//
+ .append(ctx.getCostTime()).append(TraceConstants.CONTENT_SPLITOR)//
+ .append(bean.getMsgType().ordinal()).append(TraceConstants.CONTENT_SPLITOR)//
+ .append(bean.getOffsetMsgId()).append(TraceConstants.CONTENT_SPLITOR)//
+ .append(ctx.isSuccess()).append(TraceConstants.FIELD_SPLITOR);
+ }
+ break;
+ case SubBefore: {
+ for (TraceBean bean : ctx.getTraceBeans()) {
+ sb.append(ctx.getTraceType()).append(TraceConstants.CONTENT_SPLITOR)//
+ .append(ctx.getTimeStamp()).append(TraceConstants.CONTENT_SPLITOR)//
+ .append(ctx.getRegionId()).append(TraceConstants.CONTENT_SPLITOR)//
+ .append(ctx.getGroupName()).append(TraceConstants.CONTENT_SPLITOR)//
+ .append(ctx.getRequestId()).append(TraceConstants.CONTENT_SPLITOR)//
+ .append(bean.getMsgId()).append(TraceConstants.CONTENT_SPLITOR)//
+ .append(bean.getRetryTimes()).append(TraceConstants.CONTENT_SPLITOR)//
+ .append(bean.getKeys()).append(TraceConstants.FIELD_SPLITOR);//
+ }
+ }
+ break;
+ case SubAfter: {
+ for (TraceBean bean : ctx.getTraceBeans()) {
+ sb.append(ctx.getTraceType()).append(TraceConstants.CONTENT_SPLITOR)//
+ .append(ctx.getRequestId()).append(TraceConstants.CONTENT_SPLITOR)//
+ .append(bean.getMsgId()).append(TraceConstants.CONTENT_SPLITOR)//
+ .append(ctx.getCostTime()).append(TraceConstants.CONTENT_SPLITOR)//
+ .append(ctx.isSuccess()).append(TraceConstants.CONTENT_SPLITOR)//
+ .append(bean.getKeys()).append(TraceConstants.CONTENT_SPLITOR)//
+ .append(ctx.getContextCode()).append(TraceConstants.FIELD_SPLITOR);
+ }
+ }
+ break;
+ default:
+ }
+ transferBean.setTransData(sb.toString());
+ for (TraceBean bean : ctx.getTraceBeans()) {
+
+ transferBean.getTransKey().add(bean.getMsgId());
+ if (bean.getKeys() != null && bean.getKeys().length() > 0) {
+ transferBean.getTransKey().add(bean.getKeys());
+ }
+ }
+ return transferBean;
+ }
+}
diff --git a/client/src/main/java/org/apache/rocketmq/client/trace/TraceDispatcher.java b/client/src/main/java/org/apache/rocketmq/client/trace/TraceDispatcher.java
new file mode 100644
index 000000000..275e6a322
--- /dev/null
+++ b/client/src/main/java/org/apache/rocketmq/client/trace/TraceDispatcher.java
@@ -0,0 +1,50 @@
+/*
+ * 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.client.trace;
+
+import org.apache.rocketmq.client.exception.MQClientException;
+import java.io.IOException;
+
+/**
+ * Interface of asynchronous transfer data
+ */
+public interface TraceDispatcher {
+
+ /**
+ * Initialize asynchronous transfer data module
+ */
+ void start(String nameSrvAddr) throws MQClientException;
+
+ /**
+ * Append the transfering data
+ * @param ctx data infomation
+ * @return
+ */
+ boolean append(Object ctx);
+
+ /**
+ * Write flush action
+ *
+ * @throws IOException
+ */
+ void flush() throws IOException;
+
+ /**
+ * Close the trace Hook
+ */
+ void shutdown();
+}
diff --git a/client/src/main/java/org/apache/rocketmq/client/trace/TraceDispatcherType.java b/client/src/main/java/org/apache/rocketmq/client/trace/TraceDispatcherType.java
new file mode 100644
index 000000000..f09c9b8db
--- /dev/null
+++ b/client/src/main/java/org/apache/rocketmq/client/trace/TraceDispatcherType.java
@@ -0,0 +1,22 @@
+/*
+ * 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.client.trace;
+
+public enum TraceDispatcherType {
+ PRODUCER,
+ CONSUMER
+}
diff --git a/client/src/main/java/org/apache/rocketmq/client/trace/TraceTransferBean.java b/client/src/main/java/org/apache/rocketmq/client/trace/TraceTransferBean.java
new file mode 100644
index 000000000..052ca3652
--- /dev/null
+++ b/client/src/main/java/org/apache/rocketmq/client/trace/TraceTransferBean.java
@@ -0,0 +1,44 @@
+/*
+ * 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.client.trace;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Trace transfering bean
+ */
+public class TraceTransferBean {
+ private String transData;
+ private Set<String> transKey = new HashSet<String>();
+
+ public String getTransData() {
+ return transData;
+ }
+
+ public void setTransData(String transData) {
+ this.transData = transData;
+ }
+
+ public Set<String> getTransKey() {
+ return transKey;
+ }
+
+ public void setTransKey(Set<String> transKey) {
+ this.transKey = transKey;
+ }
+}
diff --git a/client/src/main/java/org/apache/rocketmq/client/trace/TraceType.java b/client/src/main/java/org/apache/rocketmq/client/trace/TraceType.java
new file mode 100644
index 000000000..79b19c17e
--- /dev/null
+++ b/client/src/main/java/org/apache/rocketmq/client/trace/TraceType.java
@@ -0,0 +1,23 @@
+/*
+ * 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.client.trace;
+
+public enum TraceType {
+ Pub,
+ SubBefore,
+ SubAfter,
+}
diff --git a/client/src/main/java/org/apache/rocketmq/client/trace/hook/ConsumeMessageTraceHookImpl.java b/client/src/main/java/org/apache/rocketmq/client/trace/hook/ConsumeMessageTraceHookImpl.java
new file mode 100644
index 000000000..38ec8b97a
--- /dev/null
+++ b/client/src/main/java/org/apache/rocketmq/client/trace/hook/ConsumeMessageTraceHookImpl.java
@@ -0,0 +1,113 @@
+/*
+ * 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.client.trace.hook;
+
+import org.apache.rocketmq.client.consumer.listener.ConsumeReturnType;
+import org.apache.rocketmq.client.hook.ConsumeMessageContext;
+import org.apache.rocketmq.client.hook.ConsumeMessageHook;
+import org.apache.rocketmq.client.trace.TraceContext;
+import org.apache.rocketmq.client.trace.TraceDispatcher;
+import org.apache.rocketmq.client.trace.TraceBean;
+import org.apache.rocketmq.client.trace.TraceType;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.message.MessageConst;
+import org.apache.rocketmq.common.message.MessageExt;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class ConsumeMessageTraceHookImpl implements ConsumeMessageHook {
+
+ private TraceDispatcher localDispatcher;
+
+ public ConsumeMessageTraceHookImpl(TraceDispatcher localDispatcher) {
+ this.localDispatcher = localDispatcher;
+ }
+
+ @Override
+ public String hookName() {
+ return "ConsumeMessageTraceHook";
+ }
+
+ @Override
+ public void consumeMessageBefore(ConsumeMessageContext context) {
+ if (context == null || context.getMsgList() == null || context.getMsgList().isEmpty()) {
+ return;
+ }
+ TraceContext traceContext = new TraceContext();
+ context.setMqTraceContext(traceContext);
+ traceContext.setTraceType(TraceType.SubBefore);//
+ traceContext.setGroupName(context.getConsumerGroup());//
+ List<TraceBean> beans = new ArrayList<TraceBean>();
+ for (MessageExt msg : context.getMsgList()) {
+ if (msg == null) {
+ continue;
+ }
+ String regionId = msg.getProperty(MessageConst.PROPERTY_MSG_REGION);
+ String traceOn = msg.getProperty(MessageConst.PROPERTY_TRACE_SWITCH);
+
+ if (traceOn != null && traceOn.equals("false")) {
+ // If trace switch is false ,skip it
+ continue;
+ }
+ TraceBean traceBean = new TraceBean();
+ traceBean.setTopic(msg.getTopic());//
+ traceBean.setMsgId(msg.getMsgId());//
+ traceBean.setTags(msg.getTags());//
+ traceBean.setKeys(msg.getKeys());//
+ traceBean.setStoreTime(msg.getStoreTimestamp());//
+ traceBean.setBodyLength(msg.getStoreSize());//
+ traceBean.setRetryTimes(msg.getReconsumeTimes());//
+ traceContext.setRegionId(regionId);//
+ beans.add(traceBean);
+ }
+ if (beans.size() > 0) {
+ traceContext.setTraceBeans(beans);
+ traceContext.setTimeStamp(System.currentTimeMillis());
+ localDispatcher.append(traceContext);
+ }
+ }
+
+ @Override
+ public void consumeMessageAfter(ConsumeMessageContext context) {
+ if (context == null || context.getMsgList() == null || context.getMsgList().isEmpty()) {
+ return;
+ }
+ TraceContext subBeforeContext = (TraceContext) context.getMqTraceContext();
+
+ if (subBeforeContext.getTraceBeans() == null || subBeforeContext.getTraceBeans().size() < 1) {
+ // If subbefore bean is null ,skip it
+ return;
+ }
+ TraceContext subAfterContext = new TraceContext();
+ subAfterContext.setTraceType(TraceType.SubAfter);//
+ subAfterContext.setRegionId(subBeforeContext.getRegionId());//
+ subAfterContext.setGroupName(subBeforeContext.getGroupName());//
+ subAfterContext.setRequestId(subBeforeContext.getRequestId());//
+ subAfterContext.setSuccess(context.isSuccess());//
+
+ // Caculate the cost time for processing messages
+ int costTime = (int) ((System.currentTimeMillis() - subBeforeContext.getTimeStamp()) / context.getMsgList().size());
+ subAfterContext.setCostTime(costTime);//
+ subAfterContext.setTraceBeans(subBeforeContext.getTraceBeans());
+ String contextType = context.getProps().get(MixAll.CONSUME_CONTEXT_TYPE);
+ if (contextType != null) {
+ subAfterContext.setContextCode(ConsumeReturnType.valueOf(contextType).ordinal());
+ }
+ localDispatcher.append(subAfterContext);
+ }
+}
diff --git a/client/src/main/java/org/apache/rocketmq/client/trace/hook/SendMessageTraceHookImpl.java b/client/src/main/java/org/apache/rocketmq/client/trace/hook/SendMessageTraceHookImpl.java
new file mode 100644
index 000000000..20396c6dd
--- /dev/null
+++ b/client/src/main/java/org/apache/rocketmq/client/trace/hook/SendMessageTraceHookImpl.java
@@ -0,0 +1,97 @@
+/*
+ * 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.client.trace.hook;
+
+import org.apache.rocketmq.client.hook.SendMessageContext;
+import org.apache.rocketmq.client.hook.SendMessageHook;
+import org.apache.rocketmq.client.producer.SendStatus;
+import org.apache.rocketmq.client.trace.AsyncTraceDispatcher;
+import org.apache.rocketmq.client.trace.TraceContext;
+import org.apache.rocketmq.client.trace.TraceDispatcher;
+import org.apache.rocketmq.client.trace.TraceBean;
+import org.apache.rocketmq.client.trace.TraceType;
+import java.util.ArrayList;
+
+public class SendMessageTraceHookImpl implements SendMessageHook {
+
+ private TraceDispatcher localDispatcher;
+
+ public SendMessageTraceHookImpl(TraceDispatcher localDispatcher) {
+ this.localDispatcher = localDispatcher;
+ }
+
+ @Override
+ public String hookName() {
+ return "SendMessageTraceHook";
+ }
+
+ @Override
+ public void sendMessageBefore(SendMessageContext context) {
+ //if it is message trace data,then it doesn't recorded
+ if (context == null || context.getMessage().getTopic().startsWith(((AsyncTraceDispatcher) localDispatcher).getTraceTopicName())) {
+ return;
+ }
+ //build the context content of TuxeTraceContext
+ TraceContext tuxeContext = new TraceContext();
+ tuxeContext.setTraceBeans(new ArrayList<TraceBean>(1));
+ context.setMqTraceContext(tuxeContext);
+ tuxeContext.setTraceType(TraceType.Pub);
+ tuxeContext.setGroupName(context.getProducerGroup());
+ //build the data bean object of message trace
+ TraceBean traceBean = new TraceBean();
+ traceBean.setTopic(context.getMessage().getTopic());
+ traceBean.setTags(context.getMessage().getTags());
+ traceBean.setKeys(context.getMessage().getKeys());
+ traceBean.setStoreHost(context.getBrokerAddr());
+ traceBean.setBodyLength(context.getMessage().getBody().length);
+ traceBean.setMsgType(context.getMsgType());
+ tuxeContext.getTraceBeans().add(traceBean);
+ }
+
+ @Override
+ public void sendMessageAfter(SendMessageContext context) {
+ //if it is message trace data,then it doesn't recorded
+ if (context == null || context.getMessage().getTopic().startsWith(((AsyncTraceDispatcher) localDispatcher).getTraceTopicName())
+ || context.getMqTraceContext() == null) {
+ return;
+ }
+ if (context.getSendResult() == null) {
+ return;
+ }
+
+ if (context.getSendResult().getRegionId() == null
+ || !context.getSendResult().isTraceOn()) {
+ // if switch is false,skip it
+ return;
+ }
+
+ TraceContext tuxeContext = (TraceContext) context.getMqTraceContext();
+ TraceBean traceBean = tuxeContext.getTraceBeans().get(0);
+ int costTime = (int) ((System.currentTimeMillis() - tuxeContext.getTimeStamp()) / tuxeContext.getTraceBeans().size());
+ tuxeContext.setCostTime(costTime);
+ if (context.getSendResult().getSendStatus().equals(SendStatus.SEND_OK)) {
+ tuxeContext.setSuccess(true);
+ } else {
+ tuxeContext.setSuccess(false);
+ }
+ tuxeContext.setRegionId(context.getSendResult().getRegionId());
+ traceBean.setMsgId(context.getSendResult().getMsgId());
+ traceBean.setOffsetMsgId(context.getSendResult().getOffsetMsgId());
+ traceBean.setStoreTime(tuxeContext.getTimeStamp() + costTime / 2);
+ localDispatcher.append(tuxeContext);
+ }
+}
diff --git a/client/src/test/java/org/apache/rocketmq/client/producer/DefaultMQProducerTest.java b/client/src/test/java/org/apache/rocketmq/client/producer/DefaultMQProducerTest.java
index c225afd68..9540755fe 100644
--- a/client/src/test/java/org/apache/rocketmq/client/producer/DefaultMQProducerTest.java
+++ b/client/src/test/java/org/apache/rocketmq/client/producer/DefaultMQProducerTest.java
@@ -167,10 +167,7 @@ public void testSendMessageSync_WithBodyCompressed() throws RemotingException, I
@Test
public void testSendMessageAsync_Success() throws RemotingException, InterruptedException, MQBrokerException, MQClientException {
- ExecutorService callbackExecutor = Executors.newSingleThreadExecutor();
final CountDownLatch countDownLatch = new CountDownLatch(1);
- when(mQClientAPIImpl.getRemotingClient()).thenReturn((nettyRemotingClient));
- when(nettyRemotingClient.getCallbackExecutor()).thenReturn(callbackExecutor);
producer.send(message, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
@@ -186,15 +183,11 @@ public void onException(Throwable e) {
}
});
countDownLatch.await(3000L, TimeUnit.MILLISECONDS);
- callbackExecutor.shutdown();
}
@Test
public void testSendMessageAsync() throws RemotingException, MQClientException, InterruptedException {
final AtomicInteger cc = new AtomicInteger(0);
final CountDownLatch countDownLatch = new CountDownLatch(6);
- ExecutorService callbackExecutor = Executors.newSingleThreadExecutor();
- when(mQClientAPIImpl.getRemotingClient()).thenReturn((nettyRemotingClient));
- when(nettyRemotingClient.getCallbackExecutor()).thenReturn(callbackExecutor);
SendCallback sendCallback = new SendCallback() {
@Override
@@ -226,16 +219,13 @@ public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
producer.send(message,messageQueueSelector,null,sendCallback,1000);
countDownLatch.await(3000L, TimeUnit.MILLISECONDS);
- callbackExecutor.shutdown();
assertThat(cc.get()).isEqualTo(6);
}
@Test
public void testSendMessageAsync_BodyCompressed() throws RemotingException, InterruptedException, MQBrokerException, MQClientException {
- ExecutorService callbackExecutor = Executors.newSingleThreadExecutor();
+
final CountDownLatch countDownLatch = new CountDownLatch(1);
- when(mQClientAPIImpl.getRemotingClient()).thenReturn((nettyRemotingClient));
- when(nettyRemotingClient.getCallbackExecutor()).thenReturn(callbackExecutor);
producer.send(bigMessage, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
@@ -251,7 +241,6 @@ public void onException(Throwable e) {
}
});
countDownLatch.await(3000L, TimeUnit.MILLISECONDS);
- callbackExecutor.shutdown();
}
@Test
diff --git a/client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQConsumerWithTraceTest.java b/client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQConsumerWithTraceTest.java
new file mode 100644
index 000000000..b45ad0281
--- /dev/null
+++ b/client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQConsumerWithTraceTest.java
@@ -0,0 +1,314 @@
+/*
+ * 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.client.trace;
+
+import java.io.ByteArrayOutputStream;
+import java.lang.reflect.Field;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.PullCallback;
+import org.apache.rocketmq.client.consumer.PullResult;
+import org.apache.rocketmq.client.consumer.PullStatus;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.impl.CommunicationMode;
+import org.apache.rocketmq.client.impl.FindBrokerResult;
+import org.apache.rocketmq.client.impl.MQClientAPIImpl;
+import org.apache.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService;
+import org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl;
+import org.apache.rocketmq.client.impl.consumer.ProcessQueue;
+import org.apache.rocketmq.client.impl.consumer.PullAPIWrapper;
+import org.apache.rocketmq.client.impl.consumer.PullMessageService;
+import org.apache.rocketmq.client.impl.consumer.PullRequest;
+import org.apache.rocketmq.client.impl.consumer.PullResultExt;
+import org.apache.rocketmq.client.impl.consumer.RebalancePushImpl;
+import org.apache.rocketmq.client.impl.factory.MQClientInstance;
+import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.client.producer.SendStatus;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.message.MessageClientExt;
+import org.apache.rocketmq.common.message.MessageDecoder;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.common.protocol.header.PullMessageRequestHeader;
+
+import org.apache.rocketmq.common.protocol.route.BrokerData;
+import org.apache.rocketmq.common.protocol.route.QueueData;
+import org.apache.rocketmq.common.protocol.route.TopicRouteData;
+import org.apache.rocketmq.remoting.exception.RemotingException;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.junit.MockitoJUnitRunner;
+import org.mockito.stubbing.Answer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.nullable;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
+public class DefaultMQConsumerWithTraceTest {
+ private String consumerGroup;
+ private String consumerGroupNormal;
+ private String producerGroupTraceTemp = MixAll.RMQ_SYS_TRACE_TOPIC + System.currentTimeMillis();
+
+ private String topic = "FooBar";
+ private String brokerName = "BrokerA";
+ private MQClientInstance mQClientFactory;
+
+ @Mock
+ private MQClientAPIImpl mQClientAPIImpl;
+ private PullAPIWrapper pullAPIWrapper;
+ private RebalancePushImpl rebalancePushImpl;
+ private DefaultMQPushConsumer pushConsumer;
+ private DefaultMQPushConsumer normalPushConsumer;
+ private DefaultMQPushConsumer customTraceTopicpushConsumer;
+
+
+ private AsyncTraceDispatcher asyncTraceDispatcher;
+ private MQClientInstance mQClientTraceFactory;
+ @Mock
+ private MQClientAPIImpl mQClientTraceAPIImpl;
+ private DefaultMQProducer traceProducer;
+ private String customerTraceTopic = "rmq_trace_topic_12345";
+
+ @Before
+ public void init() throws Exception {
+ consumerGroup = "FooBarGroup" + System.currentTimeMillis();
+ pushConsumer = new DefaultMQPushConsumer(consumerGroup,true,"");
+ consumerGroupNormal = "FooBarGroup" + System.currentTimeMillis();
+ normalPushConsumer = new DefaultMQPushConsumer(consumerGroupNormal,false,"");
+ customTraceTopicpushConsumer = new DefaultMQPushConsumer(consumerGroup,true,customerTraceTopic);
+ pushConsumer.setNamesrvAddr("127.0.0.1:9876");
+ pushConsumer.setPullInterval(60 * 1000);
+
+ asyncTraceDispatcher = (AsyncTraceDispatcher)pushConsumer.getTraceDispatcher();
+ traceProducer = asyncTraceDispatcher.getTraceProducer();
+
+
+ pushConsumer.registerMessageListener(new MessageListenerConcurrently() {
+ @Override
+ public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
+ ConsumeConcurrentlyContext context) {
+ return null;
+ }
+ });
+
+ DefaultMQPushConsumerImpl pushConsumerImpl = pushConsumer.getDefaultMQPushConsumerImpl();
+ rebalancePushImpl = spy(new RebalancePushImpl(pushConsumer.getDefaultMQPushConsumerImpl()));
+ Field field = DefaultMQPushConsumerImpl.class.getDeclaredField("rebalanceImpl");
+ field.setAccessible(true);
+ field.set(pushConsumerImpl, rebalancePushImpl);
+ pushConsumer.subscribe(topic, "*");
+ pushConsumer.start();
+
+ mQClientFactory = spy(pushConsumerImpl.getmQClientFactory());
+ mQClientTraceFactory = spy(pushConsumerImpl.getmQClientFactory());
+
+ field = DefaultMQPushConsumerImpl.class.getDeclaredField("mQClientFactory");
+ field.setAccessible(true);
+ field.set(pushConsumerImpl, mQClientFactory);
+
+ field = MQClientInstance.class.getDeclaredField("mQClientAPIImpl");
+ field.setAccessible(true);
+ field.set(mQClientFactory, mQClientAPIImpl);
+
+ Field fieldTrace = DefaultMQProducerImpl.class.getDeclaredField("mQClientFactory");
+ fieldTrace.setAccessible(true);
+ fieldTrace.set(traceProducer.getDefaultMQProducerImpl(), mQClientTraceFactory);
+
+ fieldTrace = MQClientInstance.class.getDeclaredField("mQClientAPIImpl");
+ fieldTrace.setAccessible(true);
+ fieldTrace.set(mQClientTraceFactory, mQClientTraceAPIImpl);
+
+ pullAPIWrapper = spy(new PullAPIWrapper(mQClientFactory, consumerGroup, false));
+ field = DefaultMQPushConsumerImpl.class.getDeclaredField("pullAPIWrapper");
+ field.setAccessible(true);
+ field.set(pushConsumerImpl, pullAPIWrapper);
+
+ pushConsumer.getDefaultMQPushConsumerImpl().getRebalanceImpl().setmQClientFactory(mQClientFactory);
+ mQClientFactory.registerConsumer(consumerGroup, pushConsumerImpl);
+
+ when(mQClientFactory.getMQClientAPIImpl().pullMessage(anyString(), any(PullMessageRequestHeader.class),
+ anyLong(), any(CommunicationMode.class), nullable(PullCallback.class)))
+ .thenAnswer(new Answer<Object>() {
+ @Override
+ public Object answer(InvocationOnMock mock) throws Throwable {
+ PullMessageRequestHeader requestHeader = mock.getArgument(1);
+ MessageClientExt messageClientExt = new MessageClientExt();
+ messageClientExt.setTopic(topic);
+ messageClientExt.setQueueId(0);
+ messageClientExt.setMsgId("123");
+ messageClientExt.setBody(new byte[] {'a'});
+ messageClientExt.setOffsetMsgId("234");
+ messageClientExt.setBornHost(new InetSocketAddress(8080));
+ messageClientExt.setStoreHost(new InetSocketAddress(8080));
+ PullResult pullResult = createPullResult(requestHeader, PullStatus.FOUND, Collections.<MessageExt>singletonList(messageClientExt));
+ ((PullCallback) mock.getArgument(4)).onSuccess(pullResult);
+ return pullResult;
+ }
+ });
+
+ doReturn(new FindBrokerResult("127.0.0.1:10911", false)).when(mQClientFactory).findBrokerAddressInSubscribe(anyString(), anyLong(), anyBoolean());
+ Set<MessageQueue> messageQueueSet = new HashSet<MessageQueue>();
+ messageQueueSet.add(createPullRequest().getMessageQueue());
+ pushConsumer.getDefaultMQPushConsumerImpl().updateTopicSubscribeInfo(topic, messageQueueSet);
+ }
+
+ @After
+ public void terminate() {
+ pushConsumer.shutdown();
+ }
+
+ @Test
+ public void testPullMessage_WithTrace_Success() throws InterruptedException, RemotingException, MQBrokerException, MQClientException {
+ traceProducer.getDefaultMQProducerImpl().getmQClientFactory().registerProducer(producerGroupTraceTemp, traceProducer.getDefaultMQProducerImpl());
+ //when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(createTopicRoute());
+ //when(mQClientTraceAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(createTraceTopicRoute());
+
+
+ final CountDownLatch countDownLatch = new CountDownLatch(1);
+ final MessageExt[] messageExts = new MessageExt[1];
+ pushConsumer.getDefaultMQPushConsumerImpl().setConsumeMessageService(new ConsumeMessageConcurrentlyService(pushConsumer.getDefaultMQPushConsumerImpl(), new MessageListenerConcurrently() {
+ @Override
+ public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
+ ConsumeConcurrentlyContext context) {
+ messageExts[0] = msgs.get(0);
+ countDownLatch.countDown();
+ return null;
+ }
+ }));
+
+ PullMessageService pullMessageService = mQClientFactory.getPullMessageService();
+ pullMessageService.executePullRequestImmediately(createPullRequest());
+ countDownLatch.await(3000L, TimeUnit.MILLISECONDS);
+ assertThat(messageExts[0].getTopic()).isEqualTo(topic);
+ assertThat(messageExts[0].getBody()).isEqualTo(new byte[] {'a'});
+ }
+
+ private PullRequest createPullRequest() {
+ PullRequest pullRequest = new PullRequest();
+ pullRequest.setConsumerGroup(consumerGroup);
+ pullRequest.setNextOffset(1024);
+
+ MessageQueue messageQueue = new MessageQueue();
+ messageQueue.setBrokerName(brokerName);
+ messageQueue.setQueueId(0);
+ messageQueue.setTopic(topic);
+ pullRequest.setMessageQueue(messageQueue);
+ ProcessQueue processQueue = new ProcessQueue();
+ processQueue.setLocked(true);
+ processQueue.setLastLockTimestamp(System.currentTimeMillis());
+ pullRequest.setProcessQueue(processQueue);
+
+ return pullRequest;
+ }
+
+ private PullResultExt createPullResult(PullMessageRequestHeader requestHeader, PullStatus pullStatus,
+ List<MessageExt> messageExtList) throws Exception {
+ ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+ for (MessageExt messageExt : messageExtList) {
+ outputStream.write(MessageDecoder.encode(messageExt, false));
+ }
+ return new PullResultExt(pullStatus, requestHeader.getQueueOffset() + messageExtList.size(), 123, 2048, messageExtList, 0, outputStream.toByteArray());
+ }
+
+ public static TopicRouteData createTopicRoute() {
+ TopicRouteData topicRouteData = new TopicRouteData();
+
+ topicRouteData.setFilterServerTable(new HashMap<String, List<String>>());
+ List<BrokerData> brokerDataList = new ArrayList<BrokerData>();
+ BrokerData brokerData = new BrokerData();
+ brokerData.setBrokerName("BrokerA");
+ brokerData.setCluster("DefaultCluster");
+ HashMap<Long, String> brokerAddrs = new HashMap<Long, String>();
+ brokerAddrs.put(0L, "127.0.0.1:10911");
+ brokerData.setBrokerAddrs(brokerAddrs);
+ brokerDataList.add(brokerData);
+ topicRouteData.setBrokerDatas(brokerDataList);
+
+ List<QueueData> queueDataList = new ArrayList<QueueData>();
+ QueueData queueData = new QueueData();
+ queueData.setBrokerName("BrokerA");
+ queueData.setPerm(6);
+ queueData.setReadQueueNums(3);
+ queueData.setWriteQueueNums(4);
+ queueData.setTopicSynFlag(0);
+ queueDataList.add(queueData);
+ topicRouteData.setQueueDatas(queueDataList);
+ return topicRouteData;
+ }
+
+ private SendResult createSendResult(SendStatus sendStatus) {
+ SendResult sendResult = new SendResult();
+ sendResult.setMsgId("123");
+ sendResult.setOffsetMsgId("123");
+ sendResult.setQueueOffset(456);
+ sendResult.setSendStatus(sendStatus);
+ sendResult.setRegionId("HZ");
+ return sendResult;
+ }
+
+ public static TopicRouteData createTraceTopicRoute() {
+ TopicRouteData topicRouteData = new TopicRouteData();
+
+ topicRouteData.setFilterServerTable(new HashMap<String, List<String>>());
+ List<BrokerData> brokerDataList = new ArrayList<BrokerData>();
+ BrokerData brokerData = new BrokerData();
+ brokerData.setBrokerName("broker-trace");
+ brokerData.setCluster("DefaultCluster");
+ HashMap<Long, String> brokerAddrs = new HashMap<Long, String>();
+ brokerAddrs.put(0L, "127.0.0.1:10912");
+ brokerData.setBrokerAddrs(brokerAddrs);
+ brokerDataList.add(brokerData);
+ topicRouteData.setBrokerDatas(brokerDataList);
+
+ List<QueueData> queueDataList = new ArrayList<QueueData>();
+ QueueData queueData = new QueueData();
+ queueData.setBrokerName("broker-trace");
+ queueData.setPerm(6);
+ queueData.setReadQueueNums(1);
+ queueData.setWriteQueueNums(1);
+ queueData.setTopicSynFlag(1);
+ queueDataList.add(queueData);
+ topicRouteData.setQueueDatas(queueDataList);
+ return topicRouteData;
+ }
+}
diff --git a/client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQProducerWithTraceTest.java b/client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQProducerWithTraceTest.java
new file mode 100644
index 000000000..6dcceeb5c
--- /dev/null
+++ b/client/src/test/java/org/apache/rocketmq/client/trace/DefaultMQProducerWithTraceTest.java
@@ -0,0 +1,226 @@
+/*
+ * 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.client.trace;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.apache.rocketmq.client.ClientConfig;
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.hook.SendMessageContext;
+import org.apache.rocketmq.client.impl.CommunicationMode;
+import org.apache.rocketmq.client.impl.MQClientAPIImpl;
+import org.apache.rocketmq.client.impl.MQClientManager;
+import org.apache.rocketmq.client.impl.factory.MQClientInstance;
+import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl;
+import org.apache.rocketmq.client.impl.producer.TopicPublishInfo;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.SendCallback;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.client.producer.SendStatus;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeader;
+import org.apache.rocketmq.common.protocol.route.BrokerData;
+import org.apache.rocketmq.common.protocol.route.QueueData;
+import org.apache.rocketmq.common.protocol.route.TopicRouteData;
+import org.apache.rocketmq.remoting.exception.RemotingException;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.Spy;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.nullable;
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
+public class DefaultMQProducerWithTraceTest {
+
+ @Spy
+ private MQClientInstance mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(new ClientConfig());
+ @Mock
+ private MQClientAPIImpl mQClientAPIImpl;
+
+ @Spy
+ private MQClientInstance mQClientTraceFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(new ClientConfig());
+ @Mock
+ private MQClientAPIImpl mQClientTraceAPIImpl;
+
+ private AsyncTraceDispatcher asyncTraceDispatcher;
+
+ private DefaultMQProducer producer;
+ private DefaultMQProducer customTraceTopicproducer;
+ private DefaultMQProducer traceProducer;
+ private DefaultMQProducer normalProducer;
+
+ private Message message;
+ private String topic = "FooBar";
+ private String producerGroupPrefix = "FooBar_PID";
+ private String producerGroupTemp = producerGroupPrefix + System.currentTimeMillis();
+ private String producerGroupTraceTemp = MixAll.RMQ_SYS_TRACE_TOPIC + System.currentTimeMillis();
+ private String customerTraceTopic = "rmq_trace_topic_12345";
+
+ @Before
+ public void init() throws Exception {
+
+ customTraceTopicproducer = new DefaultMQProducer(producerGroupTemp,false, customerTraceTopic);
+ normalProducer = new DefaultMQProducer(producerGroupTemp,false,"");
+ producer = new DefaultMQProducer(producerGroupTemp,true,"");
+ producer.setNamesrvAddr("127.0.0.1:9876");
+ normalProducer.setNamesrvAddr("127.0.0.1:9877");
+ customTraceTopicproducer.setNamesrvAddr("127.0.0.1:9878");
+ message = new Message(topic, new byte[] {'a', 'b' ,'c'});
+ asyncTraceDispatcher = (AsyncTraceDispatcher)producer.getTraceDispatcher();
+ asyncTraceDispatcher.setTraceTopicName(customerTraceTopic);
+ asyncTraceDispatcher.getHostProducer();
+ asyncTraceDispatcher.getHostConsumer();
+ traceProducer = asyncTraceDispatcher.getTraceProducer();
+
+ producer.start();
+
+ Field field = DefaultMQProducerImpl.class.getDeclaredField("mQClientFactory");
+ field.setAccessible(true);
+ field.set(producer.getDefaultMQProducerImpl(), mQClientFactory);
+
+ Field fieldTrace = DefaultMQProducerImpl.class.getDeclaredField("mQClientFactory");
+ fieldTrace.setAccessible(true);
+ fieldTrace.set(traceProducer.getDefaultMQProducerImpl(), mQClientTraceFactory);
+
+ field = MQClientInstance.class.getDeclaredField("mQClientAPIImpl");
+ field.setAccessible(true);
+ field.set(mQClientFactory, mQClientAPIImpl);
+
+ field = MQClientInstance.class.getDeclaredField("mQClientAPIImpl");
+ field.setAccessible(true);
+ field.set(mQClientTraceFactory, mQClientTraceAPIImpl);
+
+ producer.getDefaultMQProducerImpl().getmQClientFactory().registerProducer(producerGroupTemp, producer.getDefaultMQProducerImpl());
+
+ when(mQClientAPIImpl.sendMessage(anyString(), anyString(), any(Message.class), any(SendMessageRequestHeader.class), anyLong(), any(CommunicationMode.class),
+ nullable(SendMessageContext.class), any(DefaultMQProducerImpl.class))).thenCallRealMethod();
+ when(mQClientAPIImpl.sendMessage(anyString(), anyString(), any(Message.class), any(SendMessageRequestHeader.class), anyLong(), any(CommunicationMode.class),
+ nullable(SendCallback.class), nullable(TopicPublishInfo.class), nullable(MQClientInstance.class), anyInt(), nullable(SendMessageContext.class), any(DefaultMQProducerImpl.class)))
+ .thenReturn(createSendResult(SendStatus.SEND_OK));
+
+ }
+
+ @Test
+ public void testSendMessageSync_WithTrace_Success() throws RemotingException, InterruptedException, MQBrokerException, MQClientException {
+ traceProducer.getDefaultMQProducerImpl().getmQClientFactory().registerProducer(producerGroupTraceTemp, traceProducer.getDefaultMQProducerImpl());
+ when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(createTopicRoute());
+ when(mQClientTraceAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(createTraceTopicRoute());
+ final CountDownLatch countDownLatch = new CountDownLatch(1);
+ try {
+ producer.send(message);
+ }catch (MQClientException e){
+ }
+ countDownLatch.await(3000L, TimeUnit.MILLISECONDS);
+
+ }
+
+ @Test
+ public void testSendMessageSync_WithTrace_NoBrokerSet_Exception() throws RemotingException, InterruptedException, MQBrokerException, MQClientException {
+ when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(createTopicRoute());
+ when(mQClientTraceAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(createTraceTopicRoute());
+ final CountDownLatch countDownLatch = new CountDownLatch(1);
+ try {
+ producer.send(message);
+ }catch (MQClientException e){
+ }
+ countDownLatch.await(3000L, TimeUnit.MILLISECONDS);
+
+ }
+
+ @After
+ public void terminate() {
+ producer.shutdown();
+ }
+
+ public static TopicRouteData createTopicRoute() {
+ TopicRouteData topicRouteData = new TopicRouteData();
+
+ topicRouteData.setFilterServerTable(new HashMap<String, List<String>>());
+ List<BrokerData> brokerDataList = new ArrayList<BrokerData>();
+ BrokerData brokerData = new BrokerData();
+ brokerData.setBrokerName("BrokerA");
+ brokerData.setCluster("DefaultCluster");
+ HashMap<Long, String> brokerAddrs = new HashMap<Long, String>();
+ brokerAddrs.put(0L, "127.0.0.1:10911");
+ brokerData.setBrokerAddrs(brokerAddrs);
+ brokerDataList.add(brokerData);
+ topicRouteData.setBrokerDatas(brokerDataList);
+
+ List<QueueData> queueDataList = new ArrayList<QueueData>();
+ QueueData queueData = new QueueData();
+ queueData.setBrokerName("BrokerA");
+ queueData.setPerm(6);
+ queueData.setReadQueueNums(3);
+ queueData.setWriteQueueNums(4);
+ queueData.setTopicSynFlag(0);
+ queueDataList.add(queueData);
+ topicRouteData.setQueueDatas(queueDataList);
+ return topicRouteData;
+ }
+
+ private SendResult createSendResult(SendStatus sendStatus) {
+ SendResult sendResult = new SendResult();
+ sendResult.setMsgId("123");
+ sendResult.setOffsetMsgId("123");
+ sendResult.setQueueOffset(456);
+ sendResult.setSendStatus(sendStatus);
+ sendResult.setRegionId("HZ");
+ return sendResult;
+ }
+
+ public static TopicRouteData createTraceTopicRoute() {
+ TopicRouteData topicRouteData = new TopicRouteData();
+
+ topicRouteData.setFilterServerTable(new HashMap<String, List<String>>());
+ List<BrokerData> brokerDataList = new ArrayList<BrokerData>();
+ BrokerData brokerData = new BrokerData();
+ brokerData.setBrokerName("broker-trace");
+ brokerData.setCluster("DefaultCluster");
+ HashMap<Long, String> brokerAddrs = new HashMap<Long, String>();
+ brokerAddrs.put(0L, "127.0.0.1:10912");
+ brokerData.setBrokerAddrs(brokerAddrs);
+ brokerDataList.add(brokerData);
+ topicRouteData.setBrokerDatas(brokerDataList);
+
+ List<QueueData> queueDataList = new ArrayList<QueueData>();
+ QueueData queueData = new QueueData();
+ queueData.setBrokerName("broker-trace");
+ queueData.setPerm(6);
+ queueData.setReadQueueNums(1);
+ queueData.setWriteQueueNums(1);
+ queueData.setTopicSynFlag(1);
+ queueDataList.add(queueData);
+ topicRouteData.setQueueDatas(queueDataList);
+ return topicRouteData;
+ }
+}
diff --git a/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java b/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
index f81af2165..1c3f37d00 100644
--- a/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
+++ b/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
@@ -51,7 +51,10 @@
@ImportantField
private boolean autoCreateSubscriptionGroup = true;
private String messageStorePlugIn = "";
-
+ @ImportantField
+ private String msgTraceTopicName = MixAll.RMQ_SYS_TRACE_TOPIC;
+ @ImportantField
+ private boolean traceTopicEnable = false;
/**
* thread numbers for send message thread pool, since spin lock will be used by default since 4.0.x, the default
* value is 1.
@@ -171,6 +174,22 @@
@ImportantField
private long transactionCheckInterval = 60 * 1000;
+ /**
+ * Acl feature switch
+ */
+ @ImportantField
+ private boolean aclEnable = false;
+
+ public static String localHostName() {
+ try {
+ return InetAddress.getLocalHost().getHostName();
+ } catch (UnknownHostException e) {
+ log.error("Failed to obtain the host name", e);
+ }
+
+ return "DEFAULT_BROKER";
+ }
+
public boolean isTraceOn() {
return traceOn;
}
@@ -235,16 +254,6 @@ public void setSlaveReadEnable(final boolean slaveReadEnable) {
this.slaveReadEnable = slaveReadEnable;
}
- public static String localHostName() {
- try {
- return InetAddress.getLocalHost().getHostName();
- } catch (UnknownHostException e) {
- log.error("Failed to obtain the host name", e);
- }
-
- return "DEFAULT_BROKER";
- }
-
public int getRegisterBrokerTimeoutMills() {
return registerBrokerTimeoutMills;
}
@@ -732,4 +741,28 @@ public long getWaitTimeMillsInTransactionQueue() {
public void setWaitTimeMillsInTransactionQueue(long waitTimeMillsInTransactionQueue) {
this.waitTimeMillsInTransactionQueue = waitTimeMillsInTransactionQueue;
}
+
+ public String getMsgTraceTopicName() {
+ return msgTraceTopicName;
+ }
+
+ public void setMsgTraceTopicName(String msgTraceTopicName) {
+ this.msgTraceTopicName = msgTraceTopicName;
+ }
+
+ public boolean isTraceTopicEnable() {
+ return traceTopicEnable;
+ }
+
+ public void setTraceTopicEnable(boolean traceTopicEnable) {
+ this.traceTopicEnable = traceTopicEnable;
+ }
+
+ public boolean isAclEnable() {
+ return aclEnable;
+ }
+
+ public void setAclEnable(boolean aclEnable) {
+ this.aclEnable = aclEnable;
+ }
}
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 20d186764..d39e63386 100644
--- a/common/src/main/java/org/apache/rocketmq/common/MixAll.java
+++ b/common/src/main/java/org/apache/rocketmq/common/MixAll.java
@@ -90,6 +90,7 @@
public static final String CONSUME_CONTEXT_TYPE = "ConsumeContextType";
public static final String RMQ_SYS_TRANS_HALF_TOPIC = "RMQ_SYS_TRANS_HALF_TOPIC";
+ public static final String RMQ_SYS_TRACE_TOPIC = "RMQ_SYS_TRACE_TOPIC";
public static final String RMQ_SYS_TRANS_OP_HALF_TOPIC = "RMQ_SYS_TRANS_OP_HALF_TOPIC";
public static final String CID_SYS_RMQ_TRANS = "CID_RMQ_SYS_TRANS";
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 a846755d8..dee6ca291 100644
--- a/common/src/main/java/org/apache/rocketmq/common/UtilAll.java
+++ b/common/src/main/java/org/apache/rocketmq/common/UtilAll.java
@@ -60,6 +60,18 @@ public static int getPid() {
}
}
+ public static void sleep(long sleepMs) {
+ if (sleepMs < 0) {
+ return;
+ }
+ try {
+ Thread.sleep(sleepMs);
+ } catch (Throwable ignored) {
+
+ }
+
+ }
+
public static String currentStackTrace() {
StringBuilder sb = new StringBuilder();
StackTraceElement[] stackTrace = Thread.currentThread().getStackTrace();
diff --git a/common/src/main/java/org/apache/rocketmq/common/constant/DBMsgConstants.java b/common/src/main/java/org/apache/rocketmq/common/constant/DBMsgConstants.java
index 8273aaa78..8d2b34497 100644
--- a/common/src/main/java/org/apache/rocketmq/common/constant/DBMsgConstants.java
+++ b/common/src/main/java/org/apache/rocketmq/common/constant/DBMsgConstants.java
@@ -18,5 +18,5 @@
package org.apache.rocketmq.common.constant;
public class DBMsgConstants {
- public static final int MAX_BODY_SIZE = 64 * 1024 * 1204; //64KB
+ public static final int MAX_BODY_SIZE = 64 * 1024 * 1024; //64KB
}
diff --git a/common/src/test/java/org/apache/rocketmq/common/BrokerConfigTest.java b/common/src/test/java/org/apache/rocketmq/common/BrokerConfigTest.java
index 0fb9b3afb..aa8bcfa72 100644
--- a/common/src/test/java/org/apache/rocketmq/common/BrokerConfigTest.java
+++ b/common/src/test/java/org/apache/rocketmq/common/BrokerConfigTest.java
@@ -27,4 +27,21 @@ public void testConsumerFallBehindThresholdOverflow() {
long expect = 1024L * 1024 * 1024 * 16;
assertThat(new BrokerConfig().getConsumerFallbehindThreshold()).isEqualTo(expect);
}
+
+ @Test
+ public void testBrokerConfigAttribute() {
+ BrokerConfig brokerConfig = new BrokerConfig();
+ brokerConfig.setNamesrvAddr("127.0.0.1:9876");
+ brokerConfig.setAutoCreateTopicEnable(false);
+ brokerConfig.setBrokerName("broker-a");
+ brokerConfig.setBrokerId(0);
+ brokerConfig.setBrokerClusterName("DefaultCluster");
+ brokerConfig.setMsgTraceTopicName("RMQ_SYS_TRACE_TOPIC4");
+ assertThat(brokerConfig.getBrokerClusterName()).isEqualTo("DefaultCluster");
+ assertThat(brokerConfig.getNamesrvAddr()).isEqualTo("127.0.0.1:9876");
+ assertThat(brokerConfig.getMsgTraceTopicName()).isEqualTo("RMQ_SYS_TRACE_TOPIC4");
+ assertThat(brokerConfig.getBrokerId()).isEqualTo(0);
+ assertThat(brokerConfig.getBrokerName()).isEqualTo("broker-a");
+ assertThat(brokerConfig.isAutoCreateTopicEnable()).isEqualTo(false);
+ }
}
\ No newline at end of file
diff --git a/common/src/test/java/org/apache/rocketmq/common/MixAllTest.java b/common/src/test/java/org/apache/rocketmq/common/MixAllTest.java
index 0d2dec6fa..8d86544be 100644
--- a/common/src/test/java/org/apache/rocketmq/common/MixAllTest.java
+++ b/common/src/test/java/org/apache/rocketmq/common/MixAllTest.java
@@ -17,14 +17,13 @@
package org.apache.rocketmq.common;
-import org.junit.Test;
-
import java.io.File;
import java.io.IOException;
import java.io.PrintWriter;
import java.net.InetAddress;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
+import org.junit.Test;
import static org.assertj.core.api.Assertions.assertThat;
diff --git a/distribution/conf/2m-noslave/broker-trace.properties b/distribution/conf/2m-noslave/broker-trace.properties
new file mode 100644
index 000000000..9dd57a73d
--- /dev/null
+++ b/distribution/conf/2m-noslave/broker-trace.properties
@@ -0,0 +1,23 @@
+#
+# 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.
+#
+brokerClusterName=DefaultCluster
+brokerName=broker-trace
+brokerId=0
+deleteWhen=04
+fileReservedTime=48
+brokerRole=ASYNC_MASTER
+flushDiskType=ASYNC_FLUSH
diff --git a/distribution/conf/plain_acl.yml b/distribution/conf/plain_acl.yml
new file mode 100644
index 000000000..413a7120f
--- /dev/null
+++ b/distribution/conf/plain_acl.yml
@@ -0,0 +1,40 @@
+# 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.
+
+globalWhiteRemoteAddresses:
+
+accounts:
+- accessKey: RocketMQ
+ secretKey: 12345678
+ whiteRemoteAddress:
+ admin: false
+ defaultTopicPerm: DENY
+ defaultGroupPerm: SUB
+ topicPerms:
+ - topicA=DENY
+ - topicB=PUB|SUB
+ - topicC=SUB
+ groupPerms:
+ # the group should convert to retry topic
+ - groupA=DENY
+ - groupB=PUB|SUB
+ - groupC=SUB
+
+- accessKey: rocketmq2
+ secretKey: 12345678
+ whiteRemoteAddress: 192.168.1.*
+ # if it is admin, it could access all resources
+ admin: true
+
diff --git a/distribution/conf/tools.yml b/distribution/conf/tools.yml
new file mode 100644
index 000000000..a4a9ad1b5
--- /dev/null
+++ b/distribution/conf/tools.yml
@@ -0,0 +1,19 @@
+# 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.
+
+
+accessKey: rocketmq
+secretKey: 12345678
+
diff --git a/docs/cn/architecture.md b/docs/cn/architecture.md
new file mode 100644
index 000000000..a26bcbdd0
--- /dev/null
+++ b/docs/cn/architecture.md
@@ -0,0 +1,46 @@
+# 架构设计
+
+## 技术架构
+
+
+RocketMQ架构上主要分为四部分，如上图所示:
+
+
+- Producer：消息发布的角色，支持分布式集群方式部署。producer通过MQ的负载均衡模块选择相应的Broker集群队列进行消息投递。投递的过程支持快速失败并且低延迟
+
+- Consumer：消息消费者的角色，支持分布式集群方式部署。支持以push推，pull拉两种模式对消息进行消费。同时也支持集群方式和广播形式的消费，它提供实时消息订阅机制，可以满足大多数用户的需求
+
+- NameServer：NameServer是一个非常简单的Topic路由注册中心，其角色类似dubbo中的zookeeper，支持Broker的动态注册与发现。主要包括两个功能：Broker管理，NameServer接受Broker集群的注册信息并且保存下来作为路由信息的基本数据。然后提供心跳检测机制，检查Broker是否还存活。路由信息管理。每个NameServer将保存关于Broker集群的整个路由信息和用于客户端查询的队列信息。然后Producer和Conumser通过NameServer就可以知道整个Broker集群的路由信息，从而进行消息的投递和消费。NameServer通常也是集群的方式部署，各实例间相互不进行信息通讯。Broker是向每一台NameServer注册自己的路由信息，所以每一个NameServer实例上面都保存一份完整的路由信息。当某个NameServer因某种原因下线了，Broker仍然可以向其它NameServer同步其路由信息，Produce,Consumer仍然可以动态感知Broker的路由的信息。
+
+- BrokerServer：Broker主要负责消息的存储、投递和查询以及服务高可用保证，为了实现这些功能broker包含了以下几个重要子模块。
+1. Remoting Module：整个broker的实体，负责处理来自clients端的请求。
+2. Client Manager：负责管理客户端(Producer/Consumer)和维护Consumer的topic订阅信息
+3. Store Service：提供方便简单的API接口处理消息存储到物理硬盘和查询功能。
+4. HA Service：高可用服务，提供master broker 和 slave broker之间的数据同步功能。
+5. Index Service：根据特定的Message key对投递到broker的消息进行索引服务，以提供消息的快速查询。
+
+
+
+## 部署架构
+
+
+
+
+
+### RocketMQ 网络部署特点
+
+- NameServer 是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。
+
+- Broker 部署相对复杂，Broker 分为Master 与Slave，一个Master 可以对应多个Slave，但是一个Slave 只能对应一个Master，Master 与Slave 的对应关系通过指定相同的BrokerName，不同的BrokerId 来定义，BrokerId 为0 表示Master，非0 表示Slave。Master 也可以部署多个。每个Broker 与Name Server 集群中的所有节点建立长连接，定时注册Topic 信息到所有Name Server。 注意：当前RocketMQ版本在部署架构上支持一master多slave，但只有brokerId=1的从服务器才会参与消息的读负载。
+
+- Producer与NameServer 集群中的其中一个节点（随机选择）建立长连接，定期从NameServer 获取Topic 路由信息，并向提供Topic 服务的Master 建立长连接，且定时向Master 发送心跳。Producer 完全无状态，可集群部署。
+
+- Consumer与NameServer 集群中的其中一个节点（随机选择）建立长连接，定期从NameServer取Topic路由信息，并向提供Topic服务的Master、Slave建立长连接，且定时向Master、Slave发送心跳。Consumer既可以从Master订阅消息，也可以从Slave订阅消息，消费者在向master拉取消息时Master服务器会根据拉取偏移量与最大偏移量的距离（判断是否读老消息，产生读IO），以及从服务器是否可读等因素建议下一次是从Master还是Slave拉取。
+
+结合部署结构图，描述集群工作流程：
+
+1. 启动Namesrv，Namesrv起来后监听端口，等待Broker、Produer、Consumer连上来，相当于一个路由控制中心。
+2. Broker启动，跟所有的Namesrv保持长连接，定时发送心跳包。心跳包中包含当前Broker信息(IP+端口等)以及存储所有topic信息。注册成功后，namesrv集群中就有Topic跟Broker的映射关系
+3. 收发消息前，先创建topic，创建topic时需要指定该topic要存储在哪些Broker上。也可以在发送消息时自动创建Topic
+4. Producer发送消息，启动时先跟Namesrv集群中的其中一台建立长连接，并从Namesrv中获取当前发送的Topic存在哪些Broker上，轮询从队列列表中选择一个队列，然后与队列所在的Broker建立长连接从而向Broker发消息
+5. Consumer跟Producer类似。跟其中一台Namesrv建立长连接，获取当前订阅Topic存在哪些Broker上，然后直接跟Broker建立连接通道，开始消费消息
diff --git a/docs/cn/concept.md b/docs/cn/concept.md
new file mode 100644
index 000000000..38036f22f
--- /dev/null
+++ b/docs/cn/concept.md
@@ -0,0 +1,39 @@
+# 基本概念
+## 消息模型（Message Model）
+
+<img src="./images/message_model_1.jpg" style="width: 560px"/>
+
+RocketMQ 消息模型如图1所示，主要由 Producer、Broker、Consumer 三部分组成，其中Producer 负责生产消息，Consumer 负责消费消息，Broker 负责存储消息。Broker 在实际部署过程中对应一台服务器，每个 Broker 可以存储多个Topic的消息，每个Topic的消息也可以分片存储于不同的 Broker。Message Queue 用于存储消息的物理地址，每个Topic中的消息地址存储于多个 Message Queue 中。ConsumerGroup 由多个Consumer 实例构成。
+## 消息生产者（Producer）
+ 负责生产消息，一般由业务系统负责生产消息。一个消息生产者会把业务应用系统里产生的消息发送到broker服务器。RocketMQ提供多种发送方式，同步发送、异步发送、顺序发送、单向发送。同步和异步方式均需要Broker返回确认信息，单向发送不需要。
+## 消息消费者（Consumer）
+ 负责消费消息，一般是后台系统负责异步消费。一个消息消费者会从Broker服务器拉取消息、并将其提供给应用程序。从用户应用的角度而言提供了两种消费形式：拉取式消费、推动式消费。
+## 主题（Topic）
+ 表示一类消息的集合，每个主题包含若干条消息，每条消息只能属于一个主题，是RocketMQ进行消息订阅的基本单位。
+##代理服务器（Broker Server）
+消息中转角色，负责存储消息、转发消息。代理服务器在RocketMQ系统中负责接收从生产者发送来的消息并存储、同时为消费者的拉取请求作准备。代理服务器也存储消息相关的元数据，包括消费者组、消费进度偏移和主题和队列消息等。
+## 名字服务（Name Server）
+ 名称服务充当路由消息的提供者。生产者或消费者能够通过名字服务查找各主题相应的Broker IP列表。多个Namesrv实例组成集群，但相互独立，没有信息交换。
+## 拉取式消费（Pull Consumer）
+ Consumer消费的一种类型，应用通常主动调用Consumer的拉消息方法从Broker服务器拉消息、主动权由应用控制。一旦获取了批量消息，应用就会启动消费过程。
+## 推动式消费（Push Consumer）
+ Consumer消费的一种类型，该模式下Broker收到数据后会主动推送给消费端，该消费模式一般实时性较高。
+## 生产者组（Producer Group）
+ 同一类Producer的集合，这类Producer发送同一类消息且发送逻辑一致。如果发送的是事物消息且原始生产者在发送之后崩溃，则Broker服务器会联系同一生产者组的其他生产者实例以提交或回溯消费。
+## 消费者组（Consumer Group）
+ 同一类Consumer的集合，这类Consumer通常消费同一类消息且消费逻辑一致。消费者组使得在消息消费方面，实现负载均衡和容错的目标变得非常容易。要注意的是，消费者组的消费者实例必须订阅完全相同的Topic。RocketMQ 支持两种消息模式：集群消费（Clustering）和广播消费（Broadcasting）。
+## 集群消费（Clustering）
+集群消费模式下,相同Consumer Group的每个Consumer实例平均分摊消息。
+## 广播消费（Broadcasting）
+广播消费模式下，相同Consumer Group的每个Consumer实例都接收全量的消息。
+## 普通顺序消息（Normal Ordered Message）
+普通顺序消费模式下，消费者通过同一个消费队列收到的消息是有顺序的，不同消息队列收到的消息则可能是无顺序的。
+## 严格顺序消息（Strictly Ordered Message）
+严格顺序消息模式下，消费者收到的所有消息均是有顺序的。
+## 代理服务器（Broker Server）
+消息中转角色，负责存储消息、转发消息。代理服务器在RocketMQ系统中负责接收从生产者发送来的消息并存储、同时为消费者的拉取请求作准备。代理服务器也存储消息相关的元数据，包括消费者组、消费进度偏移和主题和队列消息等。
+## 消息（Message）
+消息系统所传输信息的物理载体，生产和消费数据的最小单位，每条消息必须属于一个主题。RocketMQ中每个消息拥有唯一的Message ID，且可以携带具有业务标识的Key。系统提供了通过Message ID和Key查询消息的功能。
+## 标签（Tag）
+ 为消息设置的标志，用于同一主题下区分不同类型的消息。来自同一业务单元的消息，可以根据不同业务目的在同一主题下设置不同标签。标签能够有效地保持代码的清晰度和连贯性，并优化RocketMQ提供的查询系统。消费者可以根据Tag实现对不同子主题的不同消费逻辑，实现更好的扩展性。
+
diff --git a/docs/cn/image/rocketmq_architecture_1.png b/docs/cn/image/rocketmq_architecture_1.png
new file mode 100644
index 000000000..548c8752a
Binary files /dev/null and b/docs/cn/image/rocketmq_architecture_1.png differ
diff --git a/docs/cn/image/rocketmq_architecture_2.png b/docs/cn/image/rocketmq_architecture_2.png
new file mode 100644
index 000000000..b2ab8d34c
Binary files /dev/null and b/docs/cn/image/rocketmq_architecture_2.png differ
diff --git a/docs/cn/image/rocketmq_architecture_3.png b/docs/cn/image/rocketmq_architecture_3.png
new file mode 100644
index 000000000..aa74ed7ac
Binary files /dev/null and b/docs/cn/image/rocketmq_architecture_3.png differ
diff --git a/docs/cn/index.md b/docs/cn/index.md
new file mode 100644
index 000000000..e69de29bb
diff --git a/docs/en/index.md b/docs/en/index.md
new file mode 100644
index 000000000..e69de29bb
diff --git a/example/pom.xml b/example/pom.xml
index 28dfe922f..1a4065770 100644
--- a/example/pom.xml
+++ b/example/pom.xml
@@ -53,5 +53,10 @@
<artifactId>rocketmq-openmessaging</artifactId>
<version>4.4.0-SNAPSHOT</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-acl</artifactId>
+ <version>4.4.0-SNAPSHOT</version>
+ </dependency>
</dependencies>
</project>
diff --git a/example/src/main/java/org/apache/rocketmq/example/simple/AclClient.java b/example/src/main/java/org/apache/rocketmq/example/simple/AclClient.java
new file mode 100644
index 000000000..0c97cd332
--- /dev/null
+++ b/example/src/main/java/org/apache/rocketmq/example/simple/AclClient.java
@@ -0,0 +1,168 @@
+/*
+ * 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.example.simple;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.rocketmq.acl.common.AclClientRPCHook;
+import org.apache.rocketmq.acl.common.SessionCredentials;
+import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.PullResult;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
+import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.remoting.RPCHook;
+import org.apache.rocketmq.remoting.common.RemotingHelper;
+
+
+public class AclClient {
+
+ private static final Map<MessageQueue, Long> OFFSE_TABLE = new HashMap<MessageQueue, Long>();
+
+ private static final String ACL_ACCESS_KEY = "RocketMQ";
+
+ private static final String ACL_SECRET_KEY = "1234567";
+
+ public static void main(String[] args) throws MQClientException, InterruptedException {
+ producer();
+ pushConsumer();
+ pullConsumer();
+ }
+
+ public static void producer() throws MQClientException {
+ DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName", getAclRPCHook());
+ producer.setNamesrvAddr("127.0.0.1:9876");
+ producer.start();
+
+ for (int i = 0; i < 128; i++)
+ try {
+ {
+ Message msg = new Message("TopicTest",
+ "TagA",
+ "OrderID188",
+ "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
+ SendResult sendResult = producer.send(msg);
+ System.out.printf("%s%n", sendResult);
+ }
+
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+
+ producer.shutdown();
+ }
+
+ public static void pushConsumer() throws MQClientException {
+
+ DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_5", getAclRPCHook(), new AllocateMessageQueueAveragely());
+ consumer.setNamesrvAddr("127.0.0.1:9876");
+ consumer.subscribe("TopicTest", "*");
+ consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
+ // Wrong time format 2017_0422_221800
+ consumer.setConsumeTimestamp("20180422221800");
+ consumer.registerMessageListener(new MessageListenerConcurrently() {
+
+ @Override
+ public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
+ System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
+ printBody(msgs);
+ return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+ }
+ });
+ consumer.start();
+ System.out.printf("Consumer Started.%n");
+ }
+
+ public static void pullConsumer() throws MQClientException {
+ DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("please_rename_unique_group_name_6", getAclRPCHook());
+ consumer.setNamesrvAddr("127.0.0.1:9876");
+ consumer.start();
+
+ Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("TopicTest");
+ for (MessageQueue mq : mqs) {
+ System.out.printf("Consume from the queue: %s%n", mq);
+ SINGLE_MQ:
+ while (true) {
+ try {
+ PullResult pullResult =
+ consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32);
+ System.out.printf("%s%n", pullResult);
+ putMessageQueueOffset(mq, pullResult.getNextBeginOffset());
+ printBody(pullResult);
+ switch (pullResult.getPullStatus()) {
+ case FOUND:
+ break;
+ case NO_MATCHED_MSG:
+ break;
+ case NO_NEW_MSG:
+ break SINGLE_MQ;
+ case OFFSET_ILLEGAL:
+ break;
+ default:
+ break;
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+ }
+
+ consumer.shutdown();
+ }
+
+ private static void printBody(PullResult pullResult) {
+ printBody(pullResult.getMsgFoundList());
+ }
+
+ private static void printBody(List<MessageExt> msg) {
+ if (msg == null || msg.size() == 0)
+ return;
+ for (MessageExt m : msg) {
+ if (m != null) {
+ System.out.printf("msgId : %s body : %s \n\r", m.getMsgId(), new String(m.getBody()));
+ }
+ }
+ }
+
+ private static long getMessageQueueOffset(MessageQueue mq) {
+ Long offset = OFFSE_TABLE.get(mq);
+ if (offset != null)
+ return offset;
+
+ return 0;
+ }
+
+ private static void putMessageQueueOffset(MessageQueue mq, long offset) {
+ OFFSE_TABLE.put(mq, offset);
+ }
+
+ static RPCHook getAclRPCHook() {
+ return new AclClientRPCHook(new SessionCredentials(ACL_ACCESS_KEY,ACL_SECRET_KEY));
+ }
+}
diff --git a/example/src/main/java/org/apache/rocketmq/example/simple/Producer.java b/example/src/main/java/org/apache/rocketmq/example/simple/Producer.java
index 7b504dd2a..448f8ee9f 100644
--- a/example/src/main/java/org/apache/rocketmq/example/simple/Producer.java
+++ b/example/src/main/java/org/apache/rocketmq/example/simple/Producer.java
@@ -26,7 +26,6 @@
public static void main(String[] args) throws MQClientException, InterruptedException {
DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
-
producer.start();
for (int i = 0; i < 128; i++)
diff --git a/example/src/main/java/org/apache/rocketmq/example/simple/PullConsumer.java b/example/src/main/java/org/apache/rocketmq/example/simple/PullConsumer.java
index efffa36d5..8aec7e309 100644
--- a/example/src/main/java/org/apache/rocketmq/example/simple/PullConsumer.java
+++ b/example/src/main/java/org/apache/rocketmq/example/simple/PullConsumer.java
@@ -29,10 +29,10 @@
public static void main(String[] args) throws MQClientException {
DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("please_rename_unique_group_name_5");
-
+ consumer.setNamesrvAddr("127.0.0.1:9876");
consumer.start();
- Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("TopicTest1");
+ Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("broker-a");
for (MessageQueue mq : mqs) {
System.out.printf("Consume from the queue: %s%n", mq);
SINGLE_MQ:
diff --git a/example/src/main/java/org/apache/rocketmq/example/simple/PullConsumerTest.java b/example/src/main/java/org/apache/rocketmq/example/simple/PullConsumerTest.java
index 16108b8c6..f12595a90 100644
--- a/example/src/main/java/org/apache/rocketmq/example/simple/PullConsumerTest.java
+++ b/example/src/main/java/org/apache/rocketmq/example/simple/PullConsumerTest.java
@@ -24,6 +24,7 @@
public class PullConsumerTest {
public static void main(String[] args) throws MQClientException {
DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("please_rename_unique_group_name_5");
+ consumer.setNamesrvAddr("127.0.0.1:9876");
consumer.start();
try {
diff --git a/example/src/main/java/org/apache/rocketmq/example/simple/PushConsumer.java b/example/src/main/java/org/apache/rocketmq/example/simple/PushConsumer.java
index c6c7e39d1..abbfbdffc 100644
--- a/example/src/main/java/org/apache/rocketmq/example/simple/PushConsumer.java
+++ b/example/src/main/java/org/apache/rocketmq/example/simple/PushConsumer.java
@@ -29,10 +29,10 @@
public static void main(String[] args) throws InterruptedException, MQClientException {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_JODIE_1");
- consumer.subscribe("Jodie_topic_1023", "*");
+ consumer.subscribe("TopicTest", "*");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
//wrong time format 2017_0422_221800
- consumer.setConsumeTimestamp("20170422221800");
+ consumer.setConsumeTimestamp("20181109221800");
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
diff --git a/example/src/main/java/org/apache/rocketmq/example/tracemessage/TraceProducer.java b/example/src/main/java/org/apache/rocketmq/example/tracemessage/TraceProducer.java
new file mode 100644
index 000000000..fb8e37fd2
--- /dev/null
+++ b/example/src/main/java/org/apache/rocketmq/example/tracemessage/TraceProducer.java
@@ -0,0 +1,49 @@
+/*
+ * 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.example.tracemessage;
+
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.remoting.common.RemotingHelper;
+
+public class TraceProducer {
+ public static void main(String[] args) throws MQClientException, InterruptedException {
+
+ DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName",true);
+ producer.start();
+
+ for (int i = 0; i < 128; i++)
+ try {
+ {
+ Message msg = new Message("TopicTest",
+ "TagA",
+ "OrderID188",
+ "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
+ SendResult sendResult = producer.send(msg);
+ System.out.printf("%s%n", sendResult);
+ }
+
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+
+ producer.shutdown();
+ }
+}
diff --git a/example/src/main/java/org/apache/rocketmq/example/tracemessage/TracePushConsumer.java b/example/src/main/java/org/apache/rocketmq/example/tracemessage/TracePushConsumer.java
new file mode 100644
index 000000000..473351963
--- /dev/null
+++ b/example/src/main/java/org/apache/rocketmq/example/tracemessage/TracePushConsumer.java
@@ -0,0 +1,48 @@
+/*
+ * 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.example.tracemessage;
+
+import java.util.List;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
+import org.apache.rocketmq.common.message.MessageExt;
+
+public class TracePushConsumer {
+ public static void main(String[] args) throws InterruptedException, MQClientException {
+ // Here,we use the default message track trace topic name
+ DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_JODIE_1",true);
+ consumer.subscribe("TopicTest", "*");
+ consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
+ // Wrong time format 2017_0422_221800
+ consumer.setConsumeTimestamp("20181109221800");
+ consumer.registerMessageListener(new MessageListenerConcurrently() {
+
+ @Override
+ public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
+ System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
+ return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+ }
+ });
+ consumer.start();
+ System.out.printf("Consumer Started.%n");
+ }
+}
diff --git a/pom.xml b/pom.xml
index 0a8fef875..5de0465b6 100644
--- a/pom.xml
+++ b/pom.xml
@@ -16,7 +16,8 @@
limitations under the License.
-->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<parent>
<groupId>org.apache</groupId>
@@ -125,6 +126,7 @@
<module>distribution</module>
<module>openmessaging</module>
<module>logging</module>
+ <module>acl</module>
</modules>
<build>
@@ -157,7 +159,7 @@
</executions>
<configuration>
<rules>
- <banCircularDependencies />
+ <banCircularDependencies/>
</rules>
<fail>true</fail>
</configuration>
@@ -214,9 +216,6 @@
<execution>
<id>generate-effective-dependencies-pom</id>
<phase>generate-resources</phase>
- <goals>
- <goal>effective-pom</goal>
- </goals>
<configuration>
<output>${project.build.directory}/effective-pom/effective-dependencies.xml</output>
</configuration>
@@ -257,8 +256,10 @@
<exclude>src/test/resources/certs/*</exclude>
<exclude>src/test/**/*.log</exclude>
<exclude>src/test/resources/META-INF/service/*</exclude>
+ <exclude>src/main/resources/META-INF/service/*</exclude>
<exclude>*/target/**</exclude>
<exclude>*/*.iml</exclude>
+ <exclude>docs/**</exclude>
</excludes>
</configuration>
</plugin>
@@ -521,6 +522,11 @@
<artifactId>rocketmq-example</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>rocketmq-acl</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
@@ -581,6 +587,16 @@
<artifactId>log4j</artifactId>
<version>1.2.17</version>
</dependency>
+ <dependency>
+ <groupId>org.yaml</groupId>
+ <artifactId>snakeyaml</artifactId>
+ <version>1.19</version>
+ </dependency>
+ <dependency>
+ <groupId>commons-codec</groupId>
+ <artifactId>commons-codec</artifactId>
+ <version>1.9</version>
+ </dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java
index 9b026403d..d190e00f4 100644
--- a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java
+++ b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java
@@ -23,6 +23,7 @@
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslHandler;
import java.net.SocketAddress;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
@@ -35,6 +36,8 @@
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.ChannelEventListener;
import org.apache.rocketmq.remoting.InvokeCallback;
import org.apache.rocketmq.remoting.RPCHook;
@@ -45,8 +48,6 @@
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException;
-import org.apache.rocketmq.logging.InternalLogger;
-import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RemotingSysResponseCode;
@@ -95,6 +96,13 @@
*/
protected volatile SslContext sslContext;
+ /**
+ * custom rpc hooks
+ */
+ protected List<RPCHook> rpcHooks = new ArrayList<RPCHook>();
+
+
+
static {
NettyLogger.initNettyLogger();
}
@@ -158,6 +166,23 @@ public void processMessageReceived(ChannelHandlerContext ctx, RemotingCommand ms
}
}
+ protected void doBeforeRpcHooks(String addr, RemotingCommand request) {
+ if (rpcHooks.size() > 0) {
+ for (RPCHook rpcHook: rpcHooks) {
+ rpcHook.doBeforeRequest(addr, request);
+ }
+ }
+ }
+
+ protected void doAfterRpcHooks(String addr, RemotingCommand request, RemotingCommand response) {
+ if (rpcHooks.size() > 0) {
+ for (RPCHook rpcHook: rpcHooks) {
+ rpcHook.doAfterResponse(addr, request, response);
+ }
+ }
+ }
+
+
/**
* Process incoming request command issued by remote peer.
*
@@ -174,15 +199,9 @@ public void processRequestCommand(final ChannelHandlerContext ctx, final Remotin
@Override
public void run() {
try {
- RPCHook rpcHook = NettyRemotingAbstract.this.getRPCHook();
- if (rpcHook != null) {
- rpcHook.doBeforeRequest(RemotingHelper.parseChannelRemoteAddr(ctx.channel()), cmd);
- }
-
+ doBeforeRpcHooks(RemotingHelper.parseChannelRemoteAddr(ctx.channel()), cmd);
final RemotingCommand response = pair.getObject1().processRequest(ctx, cmd);
- if (rpcHook != null) {
- rpcHook.doAfterResponse(RemotingHelper.parseChannelRemoteAddr(ctx.channel()), cmd, response);
- }
+ doAfterRpcHooks(RemotingHelper.parseChannelRemoteAddr(ctx.channel()), cmd, response);
if (!cmd.isOnewayRPC()) {
if (response != null) {
@@ -314,12 +333,29 @@ public void run() {
}
}
+
+
/**
* Custom RPC hook.
+ * Just be compatible with the previous version, use getRPCHooks instead.
+ */
+ @Deprecated
+ protected RPCHook getRPCHook() {
+ if (rpcHooks.size() > 0) {
+ return rpcHooks.get(0);
+ }
+ return null;
+ }
+
+ /**
+ * Custom RPC hooks.
*
- * @return RPC hook if specified; null otherwise.
+ * @return RPC hooks if specified; null otherwise.
*/
- public abstract RPCHook getRPCHook();
+ public List<RPCHook> getRPCHooks() {
+ return rpcHooks;
+ }
+
/**
* This method specifies thread pool to use while invoking callback methods.
diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingClient.java b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingClient.java
index 33c2eed8d..fc9df37c6 100644
--- a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingClient.java
+++ b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingClient.java
@@ -53,6 +53,8 @@
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.ChannelEventListener;
import org.apache.rocketmq.remoting.InvokeCallback;
import org.apache.rocketmq.remoting.RPCHook;
@@ -64,8 +66,6 @@
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException;
-import org.apache.rocketmq.logging.InternalLogger;
-import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
public class NettyRemotingClient extends NettyRemotingAbstract implements RemotingClient {
@@ -94,7 +94,6 @@
private ExecutorService callbackExecutor;
private final ChannelEventListener channelEventListener;
private DefaultEventExecutorGroup defaultEventExecutorGroup;
- private RPCHook rpcHook;
public NettyRemotingClient(final NettyClientConfig nettyClientConfig) {
this(nettyClientConfig, null);
@@ -283,7 +282,9 @@ public void closeChannel(final String addr, final Channel channel) {
@Override
public void registerRPCHook(RPCHook rpcHook) {
- this.rpcHook = rpcHook;
+ if (rpcHook != null && !rpcHooks.contains(rpcHook)) {
+ rpcHooks.add(rpcHook);
+ }
}
public void closeChannel(final Channel channel) {
@@ -357,6 +358,8 @@ public void updateNameServerAddressList(List<String> addrs) {
}
}
+
+
@Override
public RemotingCommand invokeSync(String addr, final RemotingCommand request, long timeoutMillis)
throws InterruptedException, RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException {
@@ -364,17 +367,13 @@ public RemotingCommand invokeSync(String addr, final RemotingCommand request, lo
final Channel channel = this.getAndCreateChannel(addr);
if (channel != null && channel.isActive()) {
try {
- if (this.rpcHook != null) {
- this.rpcHook.doBeforeRequest(addr, request);
- }
+ doBeforeRpcHooks(addr, request);
long costTime = System.currentTimeMillis() - beginStartTime;
if (timeoutMillis < costTime) {
throw new RemotingTimeoutException("invokeSync call timeout");
}
RemotingCommand response = this.invokeSyncImpl(channel, request, timeoutMillis - costTime);
- if (this.rpcHook != null) {
- this.rpcHook.doAfterResponse(RemotingHelper.parseChannelRemoteAddr(channel), request, response);
- }
+ doAfterRpcHooks(RemotingHelper.parseChannelRemoteAddr(channel), request, response);
return response;
} catch (RemotingSendRequestException e) {
log.warn("invokeSync: send request exception, so close the channel[{}]", addr);
@@ -522,9 +521,7 @@ public void invokeAsync(String addr, RemotingCommand request, long timeoutMillis
final Channel channel = this.getAndCreateChannel(addr);
if (channel != null && channel.isActive()) {
try {
- if (this.rpcHook != null) {
- this.rpcHook.doBeforeRequest(addr, request);
- }
+ doBeforeRpcHooks(addr, request);
long costTime = System.currentTimeMillis() - beginStartTime;
if (timeoutMillis < costTime) {
throw new RemotingTooMuchRequestException("invokeAsync call timeout");
@@ -547,9 +544,7 @@ public void invokeOneway(String addr, RemotingCommand request, long timeoutMilli
final Channel channel = this.getAndCreateChannel(addr);
if (channel != null && channel.isActive()) {
try {
- if (this.rpcHook != null) {
- this.rpcHook.doBeforeRequest(addr, request);
- }
+ doBeforeRpcHooks(addr, request);
this.invokeOnewayImpl(channel, request, timeoutMillis);
} catch (RemotingSendRequestException e) {
log.warn("invokeOneway: send request exception, so close the channel[{}]", addr);
@@ -592,10 +587,6 @@ public ChannelEventListener getChannelEventListener() {
return channelEventListener;
}
- @Override
- public RPCHook getRPCHook() {
- return this.rpcHook;
- }
@Override
public ExecutorService getCallbackExecutor() {
diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingServer.java b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingServer.java
index 198484251..c2f3ba48d 100644
--- a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingServer.java
+++ b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingServer.java
@@ -47,6 +47,8 @@
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.ChannelEventListener;
import org.apache.rocketmq.remoting.InvokeCallback;
import org.apache.rocketmq.remoting.RPCHook;
@@ -58,8 +60,6 @@
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException;
-import org.apache.rocketmq.logging.InternalLogger;
-import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
public class NettyRemotingServer extends NettyRemotingAbstract implements RemotingServer {
@@ -75,7 +75,6 @@
private final Timer timer = new Timer("ServerHouseKeepingService", true);
private DefaultEventExecutorGroup defaultEventExecutorGroup;
- private RPCHook rpcHook;
private int port = 0;
@@ -266,7 +265,9 @@ public void shutdown() {
@Override
public void registerRPCHook(RPCHook rpcHook) {
- this.rpcHook = rpcHook;
+ if (rpcHook != null && !rpcHooks.contains(rpcHook)) {
+ rpcHooks.add(rpcHook);
+ }
}
@Override
@@ -318,10 +319,6 @@ public ChannelEventListener getChannelEventListener() {
return channelEventListener;
}
- @Override
- public RPCHook getRPCHook() {
- return this.rpcHook;
- }
@Override
public ExecutorService getCallbackExecutor() {
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 02aa84a3e..8d60321ed 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
@@ -604,7 +604,7 @@ public void setDefaultQueryMaxNum(int defaultQueryMaxNum) {
}
/**
- * Enable transient commitLog store poll only if transientStorePoolEnable is true and the FlushDiskType is
+ * Enable transient commitLog store pool only if transientStorePoolEnable is true and the FlushDiskType is
* ASYNC_FLUSH
*
* @return <tt>true</tt> or <tt>false</tt>
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 57b6999c4..6e47cc279 100644
--- a/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java
+++ b/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java
@@ -30,6 +30,7 @@
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.UtilAll;
+import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.store.config.FlushDiskType;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.config.StorePathConfigHelper;
@@ -39,6 +40,7 @@
import org.junit.Test;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
public class DefaultMessageStoreTest {
@@ -269,6 +271,29 @@ public void testRecover() throws Exception {
}
}
+ @Test
+ public void testlookMessageByOffset() throws Exception{
+ String topic = "LookMessageByOffset %02d";
+ int topicLen = String.format(topic,0).getBytes().length;
+ int bodyLen = StoreMessage.getBytes().length;
+ MessageBody = StoreMessage.getBytes();
+ for (int i = 0; i < 100; i++) {
+ MessageExtBrokerInner messageExtBrokerInner = buildMessage();
+ messageExtBrokerInner.setTopic(String.format(topic,i));
+ messageExtBrokerInner.setQueueId(0);
+ messageStore.putMessage(messageExtBrokerInner);
+ }
+
+ MessageExt messageExt = messageStore.lookMessageByOffset(0 );
+ assertEquals(messageExt.getTopic(),"LookMessageByOffset 00");
+ //(91+bodyLen+topicLen)*50
+ messageExt = messageStore.lookMessageByOffset((91+topicLen+bodyLen) * 50);
+ assertEquals(messageExt.getTopic(),"LookMessageByOffset 50");
+
+ messageExt = messageStore.lookMessageByOffset((91+topicLen+bodyLen) * 99);
+ assertEquals(messageExt.getTopic(),"LookMessageByOffset 99");
+ }
+
private void damageCommitlog(long offset) throws Exception {
MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
File file = new File(messageStoreConfig.getStorePathCommitLog() + File.separator + "00000000000000000000");
diff --git a/test/src/test/java/org/apache/rocketmq/test/base/BaseConf.java b/test/src/test/java/org/apache/rocketmq/test/base/BaseConf.java
index 5027a3cce..780bd4750 100644
--- a/test/src/test/java/org/apache/rocketmq/test/base/BaseConf.java
+++ b/test/src/test/java/org/apache/rocketmq/test/base/BaseConf.java
@@ -19,9 +19,12 @@
import java.util.ArrayList;
import java.util.List;
+
import org.apache.log4j.Logger;
import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.common.MQVersion;
import org.apache.rocketmq.namesrv.NamesrvController;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.test.client.rmq.RMQAsyncSendProducer;
import org.apache.rocketmq.test.client.rmq.RMQNormalConsumer;
import org.apache.rocketmq.test.client.rmq.RMQNormalProducer;
@@ -48,6 +51,7 @@
private static Logger log = Logger.getLogger(BaseConf.class);
static {
+ System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
namesrvController = IntegrationTestBase.createAndStartNamesrv();
nsAddr = "127.0.0.1:" + namesrvController.getNettyServerConfig().getListenPort();
brokerController1 = IntegrationTestBase.createAndStartBroker(nsAddr);
diff --git a/test/src/test/java/org/apache/rocketmq/test/client/consumer/filter/SqlFilterIT.java b/test/src/test/java/org/apache/rocketmq/test/client/consumer/filter/SqlFilterIT.java
index 79f15dcc9..a0f6555ce 100644
--- a/test/src/test/java/org/apache/rocketmq/test/client/consumer/filter/SqlFilterIT.java
+++ b/test/src/test/java/org/apache/rocketmq/test/client/consumer/filter/SqlFilterIT.java
@@ -17,12 +17,18 @@
package org.apache.rocketmq.test.client.consumer.filter;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
import org.apache.log4j.Logger;
+import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
+import org.apache.rocketmq.client.consumer.PullResult;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.test.base.BaseConf;
-import org.apache.rocketmq.test.client.consumer.broadcast.BaseBroadCastIT;
-import org.apache.rocketmq.test.client.consumer.broadcast.normal.NormalMsgTwoSameGroupConsumerIT;
-import org.apache.rocketmq.test.client.rmq.RMQBroadCastConsumer;
import org.apache.rocketmq.test.client.rmq.RMQNormalProducer;
import org.apache.rocketmq.test.client.rmq.RMQSqlConsumer;
import org.apache.rocketmq.test.factory.ConsumerFactory;
@@ -39,12 +45,14 @@
private static Logger logger = Logger.getLogger(SqlFilterIT.class);
private RMQNormalProducer producer = null;
private String topic = null;
+ private static final Map<MessageQueue, Long> OFFSE_TABLE = new HashMap<MessageQueue, Long>();
@Before
public void setUp() {
topic = initTopic();
logger.info(String.format("use topic: %s;", topic));
producer = getProducer(nsAddr, topic);
+ OFFSE_TABLE.clear();
}
@After
@@ -71,4 +79,65 @@ public void testFilterConsumer() throws Exception {
assertThat(consumer.getListener().getAllMsgBody().size()).isEqualTo(msgSize * 2);
}
+
+ @Test
+ public void testFilterPullConsumer() throws Exception {
+ int msgSize = 16;
+
+ String group = initConsumerGroup();
+ MessageSelector selector = MessageSelector.bySql("(TAGS is not null and TAGS in ('TagA', 'TagB'))");
+ DefaultMQPullConsumer consumer = new DefaultMQPullConsumer(group);
+ consumer.setNamesrvAddr(nsAddr);
+ consumer.start();
+ Thread.sleep(3000);
+ producer.send("TagA", msgSize);
+ producer.send("TagB", msgSize);
+ producer.send("TagC", msgSize);
+ Assert.assertEquals("Not all sent succeeded", msgSize * 3, producer.getAllUndupMsgBody().size());
+
+ List<String> receivedMessage = new ArrayList<>(2);
+ Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues(topic);
+ for (MessageQueue mq : mqs) {
+ SINGLE_MQ:
+ while (true) {
+ try {
+ PullResult pullResult =
+ consumer.pull(mq, selector, getMessageQueueOffset(mq), 32);
+ putMessageQueueOffset(mq, pullResult.getNextBeginOffset());
+ switch (pullResult.getPullStatus()) {
+ case FOUND:
+ List<MessageExt> msgs = pullResult.getMsgFoundList();
+ for (MessageExt msg : msgs) {
+ receivedMessage.add(new String(msg.getBody()));
+ }
+ break;
+ case NO_MATCHED_MSG:
+ break;
+ case NO_NEW_MSG:
+ break SINGLE_MQ;
+ case OFFSET_ILLEGAL:
+ break;
+ default:
+ break;
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+ }
+
+ assertThat(receivedMessage.size()).isEqualTo(msgSize * 2);
+ }
+
+ private static long getMessageQueueOffset(MessageQueue mq) {
+ Long offset = OFFSE_TABLE.get(mq);
+ if (offset != null)
+ return offset;
+
+ return 0;
+ }
+
+ private static void putMessageQueueOffset(MessageQueue mq, long offset) {
+ OFFSE_TABLE.put(mq, offset);
+ }
}
diff --git a/tools/pom.xml b/tools/pom.xml
index dc0e256ed..a4a8630b1 100644
--- a/tools/pom.xml
+++ b/tools/pom.xml
@@ -15,7 +15,8 @@
limitations under the License.
-->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-all</artifactId>
@@ -36,6 +37,10 @@
<groupId>${project.groupId}</groupId>
<artifactId>rocketmq-client</artifactId>
</dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>rocketmq-acl</artifactId>
+ </dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>rocketmq-store</artifactId>
@@ -60,5 +65,9 @@
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.yaml</groupId>
+ <artifactId>snakeyaml</artifactId>
+ </dependency>
</dependencies>
</project>
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/MQAdminStartup.java b/tools/src/main/java/org/apache/rocketmq/tools/command/MQAdminStartup.java
index 6a51b7b4b..2ca60aa86 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/MQAdminStartup.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/MQAdminStartup.java
@@ -19,13 +19,16 @@
import ch.qos.logback.classic.LoggerContext;
import ch.qos.logback.classic.joran.JoranConfigurator;
import ch.qos.logback.core.joran.spi.JoranException;
-
import java.util.ArrayList;
import java.util.List;
-
+import com.alibaba.fastjson.JSONObject;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.PosixParser;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.acl.common.AclClientRPCHook;
+import org.apache.rocketmq.acl.common.AclUtils;
+import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.common.MQVersion;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.remoting.RPCHook;
@@ -129,7 +132,7 @@ public static void main0(String[] args, RPCHook rpcHook) {
System.setProperty(MixAll.NAMESRV_ADDR_PROPERTY, namesrvAddr);
}
- cmd.execute(commandLine, options, rpcHook);
+ cmd.execute(commandLine, options, getAclRPCHook());
} else {
System.out.printf("The sub command %s not exist.%n", args[0]);
}
@@ -157,7 +160,7 @@ public static void initCommand() {
initCommand(new QueryMsgByKeySubCommand());
initCommand(new QueryMsgByUniqueKeySubCommand());
initCommand(new QueryMsgByOffsetSubCommand());
-
+
initCommand(new PrintMessageSubCommand());
initCommand(new PrintMessageByQueueCommand());
initCommand(new SendMsgStatusCommand());
@@ -211,7 +214,6 @@ private static void initLogback() throws JoranException {
private static void printHelp() {
System.out.printf("The most commonly used mqadmin commands are:%n");
-
for (SubCommand cmd : subCommandList) {
System.out.printf(" %-20s %s%n", cmd.commandName(), cmd.commandDesc());
}
@@ -243,4 +245,25 @@ private static SubCommand findSubCommand(final String name) {
public static void initCommand(SubCommand command) {
subCommandList.add(command);
}
+
+ public static RPCHook getAclRPCHook() {
+ String fileHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY, System.getenv(MixAll.ROCKETMQ_HOME_ENV));
+ String fileName = "/conf/tools.yml";
+ JSONObject yamlDataObject = AclUtils.getYamlDataObject(fileHome + fileName ,
+ JSONObject.class);
+
+ if (yamlDataObject == null || yamlDataObject.isEmpty()) {
+ System.out.printf(" Cannot find conf file %s, acl is not be enabled.%n" ,fileHome + fileName);
+ return null;
+ }
+
+ String accessKey = yamlDataObject.getString("accessKey");
+ String secretKey = yamlDataObject.getString("secretKey");
+
+ if (StringUtils.isBlank(accessKey) || StringUtils.isBlank(secretKey)) {
+ System.out.printf("AccessKey or secretKey is blank, the acl is not enabled.%n");
+ return null;
+ }
+ return new AclClientRPCHook(new SessionCredentials(accessKey,secretKey));
+ }
}
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/consumer/ConsumerProgressSubCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/consumer/ConsumerProgressSubCommand.java
index a1b3c1a22..b946ee141 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/consumer/ConsumerProgressSubCommand.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/consumer/ConsumerProgressSubCommand.java
@@ -155,7 +155,7 @@ public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) t
}
System.out.printf("%n");
- System.out.printf("Consume TPS: %s%n", consumeStats.getConsumeTps());
+ System.out.printf("Consume TPS: %.2f%n", consumeStats.getConsumeTps());
System.out.printf("Diff Total: %d%n", diffTotal);
} else {
System.out.printf("%-32s %-6s %-24s %-5s %-14s %-7s %s%n",
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services