You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by st...@apache.org on 2017/01/20 09:45:09 UTC
[12/14] incubator-rocketmq git commit: [ROCKETMQ-57] Add license
[ROCKETMQ-57] Add license
Project: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/commit/0de84e20
Tree: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/tree/0de84e20
Diff: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/diff/0de84e20
Branch: refs/heads/master
Commit: 0de84e20fccea6391e3f50fce051be795deeccd7
Parents: 8399389 4291348
Author: stevenschew <st...@apache.org>
Authored: Thu Jan 19 21:00:54 2017 +0800
Committer: stevenschew <st...@apache.org>
Committed: Thu Jan 19 21:00:54 2017 +0800
----------------------------------------------------------------------
.travis.yml | 45 +++-
README.md | 2 +-
.../rocketmq/broker/BrokerTestHarness.java | 69 ------
.../broker/api/BrokerFastFailureTest.java | 61 -----
.../rocketmq/broker/api/SendMessageTest.java | 79 ------
.../offset/ConsumerOffsetManagerTest.java | 61 -----
.../broker/topic/TopicConfigManagerTest.java | 60 -----
.../consumer/DefaultMQPullConsumerTest.java | 152 ++++++++++++
.../consumer/DefaultMQPushConsumerTest.java | 178 +++++++++++++
.../store/LocalFileOffsetStoreTest.java | 3 +-
.../impl/factory/MQClientInstanceTest.java | 6 +-
.../client/producer/DefaultMQProducerTest.java | 248 +++++++++++++++++++
pom.xml | 76 +++++-
.../tools/admin/DefaultMQAdminExtTest.java | 16 ++
.../rocketmq/tools/command/CommandUtilTest.java | 16 ++
15 files changed, 728 insertions(+), 344 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/0de84e20/tools/src/test/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtTest.java
----------------------------------------------------------------------
diff --cc tools/src/test/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtTest.java
index ab6319f,0000000..5964cd1
mode 100644,000000..100644
--- a/tools/src/test/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtTest.java
+++ b/tools/src/test/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtTest.java
@@@ -1,240 -1,0 +1,256 @@@
++/*
++ * 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.tools.admin;
+
+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.impl.MQClientAPIImpl;
+import org.apache.rocketmq.client.impl.MQClientManager;
+import org.apache.rocketmq.client.impl.factory.MQClientInstance;
+import org.apache.rocketmq.common.namesrv.NamesrvUtil;
+import org.apache.rocketmq.common.protocol.body.*;
+import org.apache.rocketmq.common.protocol.route.BrokerData;
+import org.apache.rocketmq.common.protocol.route.TopicRouteData;
+import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
+import org.apache.rocketmq.remoting.exception.RemotingConnectException;
+import org.apache.rocketmq.remoting.exception.RemotingException;
+import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
+import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import java.io.UnsupportedEncodingException;
+import java.lang.reflect.Field;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
+public class DefaultMQAdminExtTest {
+ private DefaultMQAdminExtImpl defaultMQAdminExtImpl;
+ private MQClientInstance mqClientInstance = MQClientManager.getInstance().getAndCreateMQClientInstance(new ClientConfig());
+ private MQClientAPIImpl mQClientAPIImpl;
+ private Properties properties = new Properties();
+ private TopicList topicList = new TopicList();
+ private TopicRouteData topicRouteData = new TopicRouteData();
+ private KVTable kvTable = new KVTable();
+ private ClusterInfo clusterInfo = new ClusterInfo();
+
+ @Before
+ public void init() throws Exception {
+ mQClientAPIImpl = mock(MQClientAPIImpl.class);
+ DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt();
+ defaultMQAdminExtImpl = new DefaultMQAdminExtImpl(defaultMQAdminExt, 1000);
+
+ Field field = DefaultMQAdminExtImpl.class.getDeclaredField("mqClientInstance");
+ field.setAccessible(true);
+ field.set(defaultMQAdminExtImpl, mqClientInstance);
+ field = MQClientInstance.class.getDeclaredField("mQClientAPIImpl");
+ field.setAccessible(true);
+ field.set(mqClientInstance, mQClientAPIImpl);
+
+ properties.setProperty("maxMessageSize", "5000000");
+ properties.setProperty("flushDelayOffsetInterval", "15000");
+ properties.setProperty("serverSocketRcvBufSize", "655350");
+ when(mQClientAPIImpl.getBrokerConfig(anyString(), anyLong())).thenReturn(properties);
+
+ Set<String> topicSet = new HashSet<>();
+ topicSet.add("topic_one");
+ topicSet.add("topic_two");
+ topicList.setTopicList(topicSet);
+ when(mQClientAPIImpl.getTopicListFromNameServer(anyLong())).thenReturn(topicList);
+
+
+ List<BrokerData> brokerDatas = new ArrayList<>();
+ HashMap<Long, String> brokerAddrs = new HashMap<>();
+ brokerAddrs.put(1234l, "127.0.0.1:10911");
+ BrokerData brokerData = new BrokerData();
+ brokerData.setCluster("default-cluster");
+ brokerData.setBrokerName("default-broker");
+ brokerData.setBrokerAddrs(brokerAddrs);
+ brokerDatas.add(brokerData);
+ topicRouteData.setBrokerDatas(brokerDatas);
+ when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(topicRouteData);
+
+ HashMap<String, String> result = new HashMap<>();
+ result.put("id", "1234");
+ result.put("brokerName", "default-broker");
+ kvTable.setTable(result);
+ when(mQClientAPIImpl.getBrokerRuntimeInfo(anyString(), anyLong())).thenReturn(kvTable);
+
+ HashMap<String, BrokerData> brokerAddrTable = new HashMap<>();
+ brokerAddrTable.put("default-broker", brokerData);
+ brokerAddrTable.put("broker-test", new BrokerData());
+ clusterInfo.setBrokerAddrTable(brokerAddrTable);
+ when(mQClientAPIImpl.getBrokerClusterInfo(anyLong())).thenReturn(clusterInfo);
+ when(mQClientAPIImpl.cleanExpiredConsumeQueue(anyString(), anyLong())).thenReturn(true);
+
+ Set<String> clusterList = new HashSet<>();
+ clusterList.add("default-cluster-one");
+ clusterList.add("default-cluster-two");
+ when(mQClientAPIImpl.getClusterList(anyString(), anyLong())).thenReturn(clusterList);
+
+ GroupList groupList = new GroupList();
+ HashSet<String> groups = new HashSet<>();
+ groups.add("consumer-group-one");
+ groups.add("consumer-group-two");
+ groupList.setGroupList(groups);
+ when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(topicRouteData);
+ when(mQClientAPIImpl.queryTopicConsumeByWho(anyString(), anyString(), anyLong())).thenReturn(groupList);
+
+ SubscriptionGroupWrapper subscriptionGroupWrapper = new SubscriptionGroupWrapper();
+ ConcurrentHashMap<String, SubscriptionGroupConfig> subscriptions = new ConcurrentHashMap<>();
+ SubscriptionGroupConfig subscriptionGroupConfig = new SubscriptionGroupConfig();
+ subscriptionGroupConfig.setConsumeBroadcastEnable(true);
+ subscriptionGroupConfig.setBrokerId(1234);
+ subscriptionGroupConfig.setGroupName("Consumer-group-one");
+ subscriptions.put("Consumer-group-one", subscriptionGroupConfig);
+ subscriptionGroupWrapper.setSubscriptionGroupTable(subscriptions);
+ when(mQClientAPIImpl.getAllSubscriptionGroup(anyString(), anyLong())).thenReturn(subscriptionGroupWrapper);
+
+ String topicListConfig = "topicListConfig";
+ when(mQClientAPIImpl.getKVConfigValue(anyString(), anyString(), anyLong())).thenReturn(topicListConfig);
+
+ KVTable kvTable = new KVTable();
+ HashMap<String, String> kv = new HashMap<>();
+ kv.put("broker-name", "broker-one");
+ kv.put("cluster-name", "default-cluster");
+ kvTable.setTable(kv);
+ when(mQClientAPIImpl.getKVListByNamespace(anyString(), anyLong())).thenReturn(kvTable);
+ }
+
+ @After
+ public void terminate() throws Exception {
+ if (defaultMQAdminExtImpl != null)
+ defaultMQAdminExtImpl.shutdown();
+ }
+
+ @Test
+ public void testUpdateBrokerConfig() throws InterruptedException, RemotingConnectException, UnsupportedEncodingException, RemotingTimeoutException, MQBrokerException, RemotingSendRequestException {
+ Properties result = defaultMQAdminExtImpl.getBrokerConfig("127.0.0.1:10911");
+ assertThat(result.getProperty("maxMessageSize")).isEqualTo("5000000");
+ assertThat(result.getProperty("flushDelayOffsetInterval")).isEqualTo("15000");
+ assertThat(result.getProperty("serverSocketRcvBufSize")).isEqualTo("655350");
+ }
+
+
+ @Test
+ public void testFetchAllTopicList() throws RemotingException, MQClientException, InterruptedException {
+ TopicList topicList = defaultMQAdminExtImpl.fetchAllTopicList();
+ assertThat(topicList.getTopicList().size()).isEqualTo(2);
+ assertThat(topicList.getTopicList()).contains("topic_one");
+ }
+
+ @Test
+ public void testFetchBrokerRuntimeStats() throws InterruptedException, MQBrokerException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
+ KVTable brokerStats = defaultMQAdminExtImpl.fetchBrokerRuntimeStats("127.0.0.1:10911");
+ assertThat(brokerStats.getTable().get("id")).isEqualTo("1234");
+ assertThat(brokerStats.getTable().get("brokerName")).isEqualTo("default-broker");
+ }
+
+ @Test
+ public void testExamineBrokerClusterInfo() throws InterruptedException, MQBrokerException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
+ ClusterInfo clusterInfo = defaultMQAdminExtImpl.examineBrokerClusterInfo();
+ HashMap<String, BrokerData> brokerList = clusterInfo.getBrokerAddrTable();
+ assertThat(brokerList.get("default-broker").getBrokerName()).isEqualTo("default-broker");
+ assertThat(brokerList.containsKey("broker-test")).isTrue();
+
+ HashMap<String, Set<String>> clusterMap = new HashMap<>();
+ Set<String> brokers = new HashSet<>();
+ brokers.add("default-broker");
+ brokers.add("broker-test");
+ clusterMap.put("default-cluster", brokers);
+ ClusterInfo cInfo = mock(ClusterInfo.class);
+ when(cInfo.getClusterAddrTable()).thenReturn(clusterMap);
+ HashMap<String, Set<String>> clusterAddress = cInfo.getClusterAddrTable();
+ assertThat(clusterAddress.containsKey("default-cluster")).isTrue();
+ assertThat(clusterAddress.get("default-cluster").size()).isEqualTo(2);
+ }
+
+ @Test
+ public void testExamineTopicRouteInfo() throws RemotingException, MQClientException, InterruptedException {
+ TopicRouteData topicRouteData = defaultMQAdminExtImpl.examineTopicRouteInfo("UnitTest");
+ assertThat(topicRouteData.getBrokerDatas().get(0).getBrokerName()).isEqualTo("default-broker");
+ assertThat(topicRouteData.getBrokerDatas().get(0).getCluster()).isEqualTo("default-cluster");
+ }
+
+ @Test
+ public void testGetNameServerAddressList() {
+ List<String> result = new ArrayList<>();
+ result.add("default-name-one");
+ result.add("default-name-two");
+ when(mqClientInstance.getMQClientAPIImpl().getNameServerAddressList()).thenReturn(result);
+ List<String> nameList = defaultMQAdminExtImpl.getNameServerAddressList();
+ assertThat(nameList.get(0)).isEqualTo("default-name-one");
+ assertThat(nameList.get(1)).isEqualTo("default-name-two");
+ }
+
+ @Test
+ public void testPutKVConfig() throws RemotingException, MQClientException, InterruptedException {
+ String topicConfig = defaultMQAdminExtImpl.getKVConfig(NamesrvUtil.NAMESPACE_ORDER_TOPIC_CONFIG, "UnitTest");
+ assertThat(topicConfig).isEqualTo("topicListConfig");
+ KVTable kvs = defaultMQAdminExtImpl.getKVListByNamespace(NamesrvUtil.NAMESPACE_ORDER_TOPIC_CONFIG);
+ assertThat(kvs.getTable().get("broker-name")).isEqualTo("broker-one");
+ assertThat(kvs.getTable().get("cluster-name")).isEqualTo("default-cluster");
+ }
+
+
+ @Test
+ public void testQueryTopicConsumeByWho() throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
+ GroupList groupList = defaultMQAdminExtImpl.queryTopicConsumeByWho("UnitTest");
+ assertThat(groupList.getGroupList().contains("consumer-group-two")).isTrue();
+ }
+
+ @Test
+ public void testCleanExpiredConsumerQueueByAddr() throws InterruptedException, RemotingTimeoutException, MQClientException, RemotingSendRequestException, RemotingConnectException {
+ boolean clean = defaultMQAdminExtImpl.cleanExpiredConsumerQueueByAddr("127.0.0.1:10911");
+ assertThat(clean).isTrue();
+ }
+
+ @Test
+ public void testGetClusterList() throws InterruptedException, RemotingTimeoutException, MQClientException, RemotingSendRequestException, RemotingConnectException {
+ Set<String> clusterlist = defaultMQAdminExtImpl.getClusterList("UnitTest");
+ assertThat(clusterlist.contains("default-cluster-one")).isTrue();
+ assertThat(clusterlist.contains("default-cluster-two")).isTrue();
+ }
+
+ @Test
+ public void testFetchConsumeStatsInBroker() throws InterruptedException, RemotingTimeoutException, MQClientException, RemotingSendRequestException, RemotingConnectException {
+ ConsumeStatsList result = new ConsumeStatsList();
+ result.setBrokerAddr("127.0.0.1:10911");
+ when(mqClientInstance.getMQClientAPIImpl().fetchConsumeStatsInBroker("127.0.0.1:10911", false, 10000)).thenReturn(result);
+ ConsumeStatsList consumeStatsList = defaultMQAdminExtImpl.fetchConsumeStatsInBroker("127.0.0.1:10911", false, 10000);
+ assertThat(consumeStatsList.getBrokerAddr()).isEqualTo("127.0.0.1:10911");
+ }
+
+ @Test
+ public void testGetAllSubscriptionGroup() throws InterruptedException, MQBrokerException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
+ SubscriptionGroupWrapper subscriptionGroupWrapper = defaultMQAdminExtImpl.getAllSubscriptionGroup("127.0.0.1:10911", 10000);
+ assertThat(subscriptionGroupWrapper.getSubscriptionGroupTable().get("Consumer-group-one").getBrokerId()).isEqualTo(1234);
+ assertThat(subscriptionGroupWrapper.getSubscriptionGroupTable().get("Consumer-group-one").getGroupName()).isEqualTo("Consumer-group-one");
+ assertThat(subscriptionGroupWrapper.getSubscriptionGroupTable().get("Consumer-group-one").isConsumeBroadcastEnable()).isTrue();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/0de84e20/tools/src/test/java/org/apache/rocketmq/tools/command/CommandUtilTest.java
----------------------------------------------------------------------
diff --cc tools/src/test/java/org/apache/rocketmq/tools/command/CommandUtilTest.java
index 35c205e,0000000..ba58010
mode 100644,000000..100644
--- a/tools/src/test/java/org/apache/rocketmq/tools/command/CommandUtilTest.java
+++ b/tools/src/test/java/org/apache/rocketmq/tools/command/CommandUtilTest.java
@@@ -1,92 -1,0 +1,108 @@@
++/*
++ * 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.tools.command;
+
+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.impl.MQClientAPIImpl;
+import org.apache.rocketmq.client.impl.MQClientManager;
+import org.apache.rocketmq.client.impl.factory.MQClientInstance;
+import org.apache.rocketmq.common.protocol.body.ClusterInfo;
+import org.apache.rocketmq.common.protocol.route.BrokerData;
+import org.apache.rocketmq.remoting.exception.RemotingConnectException;
+import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
+import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
+import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
+import org.apache.rocketmq.tools.admin.DefaultMQAdminExtImpl;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.lang.reflect.Field;
+import java.util.*;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class CommandUtilTest {
+ private DefaultMQAdminExt defaultMQAdminExt;
+ private DefaultMQAdminExtImpl defaultMQAdminExtImpl;
+ private MQClientInstance mqClientInstance = MQClientManager.getInstance().getAndCreateMQClientInstance(new ClientConfig());
+ private MQClientAPIImpl mQClientAPIImpl;
+
+ @Before
+ public void setup() throws MQClientException, NoSuchFieldException, IllegalAccessException, InterruptedException, MQBrokerException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
+ defaultMQAdminExt = mock(DefaultMQAdminExt.class);
+ MQClientAPIImpl mQClientAPIImpl = mock(MQClientAPIImpl.class);
+ defaultMQAdminExtImpl = new DefaultMQAdminExtImpl(defaultMQAdminExt, 3000);
+
+ Field field = DefaultMQAdminExtImpl.class.getDeclaredField("mqClientInstance");
+ field.setAccessible(true);
+ field.set(defaultMQAdminExtImpl, mqClientInstance);
+ field = MQClientInstance.class.getDeclaredField("mQClientAPIImpl");
+ field.setAccessible(true);
+ field.set(mqClientInstance, mQClientAPIImpl);
+
+ ClusterInfo clusterInfo = new ClusterInfo();
+ HashMap<String, BrokerData> brokerAddrTable = new HashMap<>();
+ HashMap<String, Set<String>> clusterAddrTable = new HashMap<>();
+ HashMap<Long, String> brokerAddrs = new HashMap<>();
+ brokerAddrs.put(1234l, "127.0.0.1:10911");
+ BrokerData brokerData = new BrokerData();
+ brokerData.setBrokerName("default-broker");
+ brokerData.setCluster("default-cluster");
+ brokerData.setBrokerAddrs(brokerAddrs);
+ brokerAddrTable.put("default-broker", brokerData);
+ brokerAddrTable.put("broker-test", new BrokerData());
+ Set<String> brokerSet = new HashSet<>();
+ brokerSet.add("default-broker");
+ brokerSet.add("default-broker-one");
+ clusterAddrTable.put("default-cluster", brokerSet);
+ clusterInfo.setBrokerAddrTable(brokerAddrTable);
+ clusterInfo.setClusterAddrTable(clusterAddrTable);
+ when(mQClientAPIImpl.getBrokerClusterInfo(anyLong())).thenReturn(clusterInfo);
+ when(mQClientAPIImpl.cleanExpiredConsumeQueue(anyString(), anyLong())).thenReturn(true);
+ }
+
+ @After
+ public void shutdown() throws Exception {
+ }
+
+ @Test
+ public void testFetchMasterAndSlaveDistinguish() throws InterruptedException, MQBrokerException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
+ Map<String, List<String>> result = CommandUtil.fetchMasterAndSlaveDistinguish(defaultMQAdminExtImpl, "default-cluster");
+ assertThat(result.get(null).get(0)).isEqualTo("127.0.0.1:10911");
+ }
+
+ @Test
+ public void testFetchMasterAddrByClusterName() throws InterruptedException, MQBrokerException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
+ Set<String> result = CommandUtil.fetchMasterAddrByClusterName(defaultMQAdminExtImpl, "default-cluster");
+ assertThat(result.size()).isEqualTo(0);
+ }
+
+ @Test
+ public void testFetchBrokerNameByClusterName() throws Exception {
+ Set<String> result = CommandUtil.fetchBrokerNameByClusterName(defaultMQAdminExtImpl, "default-cluster");
+ assertThat(result.contains("default-broker")).isTrue();
+ assertThat(result.contains("default-broker-one")).isTrue();
+ assertThat(result.size()).isEqualTo(2);
+ }
+}