Merge branch 'master' of https://github.com/apache/rocketmq

.travis.yml

@@ -1,3 +1,5 @@
+dist: trusty
+
 notifications:
   email:
     recipients:

CONTRIBUTING.md

@@ -10,7 +10,7 @@ Nor is code the only way to contribute to the project. We strongly value documen
 To submit a change for inclusion, please do the following:
 
 #### If the change is non-trivial please include some unit tests that cover the new functionality.
-#### If you are introducing a completely new feature or API it is a good idea to start a wiki and get consensus on the basic design first.
+#### If you are introducing a completely new feature or API it is a good idea to start a [RIP](https://github.com/apache/rocketmq/wiki/RocketMQ-Improvement-Proposal) and get consensus on the basic design first.
 #### It is our job to follow up on patches in a timely fashion. Nag us if we aren't doing our job (sometimes we drop things).
 
 ## Becoming a Committer
@@ -19,9 +19,8 @@ We are always interested in adding new contributors. What we look for are series
 
 Nowadays,we have several important contribution points:
 #### Wiki & JavaDoc
-#### RocketMQ Console
 #### RocketMQ SDK(C++\.Net\Php\Python\Go\Node.js)
-#### RocketMQ MySQL(Oracle\PostgreSQL\Redis\MongoDB\HBase\MSSQL) Replicator
+#### RocketMQ Connectors
 
 ##### Prerequisite
 If you want to contribute the above listing points, you must abide our some prerequisites:

README.md

@@ -51,3 +51,23 @@ We always welcome new contributions, whether for trivial cleanups, [big new feat
 [Apache License, Version 2.0](http://www.apache.org/licenses/LICENSE-2.0.html) Copyright (C) Apache Software Foundation
 
 
+----------
+## Export Control Notice
+This distribution includes cryptographic software. The country in which you currently reside may have
+restrictions on the import, possession, use, and/or re-export to another country, of encryption software.
+BEFORE using any encryption software, please check your country's laws, regulations and policies concerning
+the import, possession, or use, and re-export of encryption software, to see if this is permitted. See
+<http://www.wassenaar.org/> for more information.
+
+The U.S. Government Department of Commerce, Bureau of Industry and Security (BIS), has classified this
+software as Export Commodity Control Number (ECCN) 5D002.C.1, which includes information security software
+using or performing cryptographic functions with asymmetric algorithms. The form and manner of this Apache
+Software Foundation distribution makes it eligible for export under the License Exception ENC Technology
+Software Unrestricted (TSU) exception (see the BIS Export Administration Regulations, Section 740.13) for
+both object code and source code.
+
+The following provides more details on the included cryptographic software:
+
+This software uses Apache Commons Crypto (https://commons.apache.org/proper/commons-crypto/) to
+support authentication, and encryption and decryption of data sent across the network between
+services.

acl/pom.xml

@@ -13,7 +13,7 @@
     <parent>
         <groupId>org.apache.rocketmq</groupId>
         <artifactId>rocketmq-all</artifactId>
-        <version>4.5.0</version>
+        <version>4.5.1</version>
     </parent>
     <artifactId>rocketmq-acl</artifactId>
     <name>rocketmq-acl ${project.version}</name>

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

@@ -16,19 +16,27 @@
  */
 package org.apache.rocketmq.acl.common;
 
+import com.alibaba.fastjson.JSONObject;
 import java.io.File;
 import java.io.FileInputStream;
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.Map;
 import java.util.SortedMap;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
+import org.apache.rocketmq.remoting.RPCHook;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 import org.yaml.snakeyaml.Yaml;
 
 import static org.apache.rocketmq.acl.common.SessionCredentials.CHARSET;
 
 public class AclUtils {
 
+    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME);
+
     public static byte[] combineRequestContent(RemotingCommand request, SortedMap<String, String> fieldsMap) {
         try {
             StringBuilder sb = new StringBuilder("");
@@ -124,17 +132,44 @@ public static <T> T getYamlDataObject(String path, Class<T> clazz) {
         try {
             fis = new FileInputStream(new File(path));
             return ymal.loadAs(fis, clazz);
+        } catch (FileNotFoundException ignore) {
+            return null;
         } catch (Exception e) {
-            throw new AclException(String.format("The  file for Plain mode was not found , paths %s", path), e);
+            throw new AclException(e.getMessage());
         } finally {
             if (fis != null) {
                 try {
                     fis.close();
-                } catch (IOException e) {
-                    throw new AclException("close transport fileInputStream Exception", e);
+                } catch (IOException ignore) {
                 }
             }
         }
     }
 
+    public static RPCHook getAclRPCHook(String fileName) {
+        JSONObject yamlDataObject = null;
+        try {
+            yamlDataObject = AclUtils.getYamlDataObject(fileName,
+                JSONObject.class);
+        } catch (Exception e) {
+            log.error("convert yaml file to data object error, ",e);
+            return null;
+        }
+
+        if (yamlDataObject == null || yamlDataObject.isEmpty()) {
+            log.warn("Cannot find conf file :{}, acl isn't be enabled." ,fileName);
+            return null;
+        }
+
+        String accessKey = yamlDataObject.getString("accessKey");
+        String secretKey = yamlDataObject.getString("secretKey");
+
+        if (StringUtils.isBlank(accessKey) || StringUtils.isBlank(secretKey)) {
+            log.warn("AccessKey or secretKey is blank, the acl is not enabled.");
+
+            return null;
+        }
+        return new AclClientRPCHook(new SessionCredentials(accessKey,secretKey));
+    }
+
 }

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

@@ -53,11 +53,13 @@ public AccessResource parse(RemotingCommand request, String remoteAddr) {
             accessResource.setWhiteRemoteAddress(remoteAddr);
         }
 
+        accessResource.setRequestCode(request.getCode());
+
         if (request.getExtFields() == null) {
-            throw new AclException("request's extFields value is null");
+            //If request's extFields is null,then return accessResource directly(users can use whiteAddress pattern)
+            //The following logic codes depend on the request's extFields not to be null.
+            return accessResource;
         }
-
-        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));
@@ -116,6 +118,7 @@ public AccessResource parse(RemotingCommand request, String remoteAddr) {
         } 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()) {

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

@@ -16,10 +16,12 @@
  */
 package org.apache.rocketmq.acl.common;
 
+import com.alibaba.fastjson.JSONObject;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.remoting.RPCHook;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -133,9 +135,34 @@ public void getYamlDataObjectTest() {
         Assert.assertFalse(map.isEmpty());
     }
 
+    @Test
+    public void getYamlDataIgnoreFileNotFoundExceptionTest() {
+
+        JSONObject yamlDataObject = AclUtils.getYamlDataObject("plain_acl.yml", JSONObject.class);
+        Assert.assertTrue(yamlDataObject == null);
+    }
+
     @Test(expected = Exception.class)
-    public void getYamlDataObjectExceptionTest() {
+    public void getYamlDataExceptionTest() {
 
-        AclUtils.getYamlDataObject("plain_acl.yml", Map.class);
+        AclUtils.getYamlDataObject("src/test/resources/conf/plain_acl_format_error.yml", Map.class);
     }
+
+    @Test
+    public void getAclRPCHookTest() {
+
+        RPCHook errorContRPCHook = AclUtils.getAclRPCHook("src/test/resources/conf/plain_acl_format_error.yml");
+        Assert.assertNull(errorContRPCHook);
+
+        RPCHook noFileRPCHook = AclUtils.getAclRPCHook("src/test/resources/plain_acl_format_error1.yml");
+        Assert.assertNull(noFileRPCHook);
+
+        RPCHook emptyContRPCHook = AclUtils.getAclRPCHook("src/test/resources/conf/plain_acl_null.yml");
+        Assert.assertNull(emptyContRPCHook);
+
+        RPCHook incompleteContRPCHook = AclUtils.getAclRPCHook("src/test/resources/conf/plain_acl_incomplete.yml");
+        Assert.assertNull(incompleteContRPCHook);
+    }
+
+
 }

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

@@ -116,7 +116,7 @@ public void validateSendMessageV2Test() {
         plainAccessValidator.validate(accessResource);
     }
 
-    @Test(expected = AclException.class)
+    @Test
     public void validateForAdminCommandWithOutAclRPCHook() {
         RemotingCommand consumerOffsetAdminRequest = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_CONSUMER_OFFSET, null);
         plainAccessValidator.parse(consumerOffsetAdminRequest, "192.168.0.1:9876");
@@ -284,4 +284,17 @@ public void validateErrorSecretKeyTest() {
         PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "192.168.1.1");
         plainAccessValidator.validate(accessResource);
     }
+
+    @Test
+    public void validateGetAllTopicConfigTest() {
+        String whiteRemoteAddress = "192.168.0.1";
+        RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_TOPIC_CONFIG, null);
+
+        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), whiteRemoteAddress);
+        plainAccessValidator.validate(accessResource);
+    }
 }

acl/src/test/resources/conf/plain_acl_format_error.yml

@@ -0,0 +1,26 @@
+# 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
+
+date 2015-02-01
+accounts:
+  - name: Jai
+accounts:
+- accessKey: RocketMQ
+  secretKey: 12345678
+  whiteRemoteAddress: 192.168.0.*
+  admin: false
+

acl/src/test/resources/conf/plain_acl_incomplete.yml

@@ -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.
+
+## suggested format
+
+- accessKey: rocketmq2
+  secretKey: 
+  whiteRemoteAddress: 192.168.1.*
+  # if it is admin, it could access all resources
+  admin: true

broker/pom.xml

@@ -13,7 +13,7 @@
     <parent>
         <groupId>org.apache.rocketmq</groupId>
         <artifactId>rocketmq-all</artifactId>
-        <version>4.5.0</version>
+        <version>4.5.1</version>
     </parent>
 
     <modelVersion>4.0.0</modelVersion>

broker/src/main/java/org/apache/rocketmq/broker/mqtrace/ConsumeMessageContext.java

@@ -35,6 +35,7 @@ public class ConsumeMessageContext {
     private BrokerStatsManager.StatsType commercialRcvStats;
     private int commercialRcvTimes;
     private int commercialRcvSize;
+    private String namespace;
 
     public String getConsumerGroup() {
         return consumerGroup;
@@ -147,4 +148,12 @@ public int getCommercialRcvSize() {
     public void setCommercialRcvSize(final int commercialRcvSize) {
         this.commercialRcvSize = commercialRcvSize;
     }
+
+    public String getNamespace() {
+        return namespace;
+    }
+
+    public void setNamespace(String namespace) {
+        this.namespace = namespace;
+    }
 }

broker/src/main/java/org/apache/rocketmq/broker/mqtrace/SendMessageContext.java

@@ -40,11 +40,12 @@ public class SendMessageContext {
     private long bornTimeStamp;
     private MessageType msgType = MessageType.Trans_msg_Commit;
     private boolean isSuccess = false;
-    //For Commercial
+
     private String commercialOwner;
     private BrokerStatsManager.StatsType commercialSendStats;
     private int commercialSendSize;
     private int commercialSendTimes;
+    private String namespace;
 
     public boolean isSuccess() {
         return isSuccess;
@@ -229,4 +230,12 @@ public int getCommercialSendTimes() {
     public void setCommercialSendTimes(final int commercialSendTimes) {
         this.commercialSendTimes = commercialSendTimes;
     }
+
+    public String getNamespace() {
+        return namespace;
+    }
+
+    public void setNamespace(String namespace) {
+        this.namespace = namespace;
+    }
 }