Merge 590ffb2 into 56b0983

client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java

@@ -87,6 +87,7 @@
 import org.apache.rocketmq.common.protocol.header.DeleteSubscriptionGroupRequestHeader;
 import org.apache.rocketmq.common.protocol.header.DeleteTopicRequestHeader;
 import org.apache.rocketmq.common.protocol.header.EndTransactionRequestHeader;
+import org.apache.rocketmq.common.protocol.header.GetClusterListRequestHeader;
 import org.apache.rocketmq.common.protocol.header.GetConsumeStatsInBrokerHeader;
 import org.apache.rocketmq.common.protocol.header.GetConsumeStatsRequestHeader;
 import org.apache.rocketmq.common.protocol.header.GetConsumerConnectionListRequestHeader;
@@ -1167,10 +1168,18 @@ public Properties getBrokerConfig(final String addr, final long timeoutMillis)
         throw new MQBrokerException(response.getCode(), response.getRemark());
     }
 
-    public ClusterInfo getBrokerClusterInfo(
+    public ClusterInfo getBrokerClusterInfo(final long timeoutMillis) throws InterruptedException, RemotingTimeoutException,
+        RemotingSendRequestException, RemotingConnectException, MQBrokerException {
+        return this.getBrokerClusterInfo("", timeoutMillis);
+    }
+
+    public ClusterInfo getBrokerClusterInfo(String cluster,
         final long timeoutMillis) throws InterruptedException, RemotingTimeoutException,
         RemotingSendRequestException, RemotingConnectException, MQBrokerException {
-        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_BROKER_CLUSTER_INFO, null);
+        GetClusterListRequestHeader requestHeader = new GetClusterListRequestHeader();
+        String pCluster = cluster != null ? cluster : "";
+        requestHeader.setCluster(pCluster);
+        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_BROKER_CLUSTER_INFO, requestHeader);
 
         RemotingCommand response = this.remotingClient.invokeSync(null, request, timeoutMillis);
         assert response != null;

common/src/main/java/org/apache/rocketmq/common/protocol/header/GetClusterListRequestHeader.java

@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.common.protocol.header;
+
+import org.apache.rocketmq.remoting.CommandCustomHeader;
+import org.apache.rocketmq.remoting.annotation.CFNullable;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+
+
+public class GetClusterListRequestHeader implements CommandCustomHeader {
+
+    @CFNullable
+    private String cluster;
+
+    @Override
+    public void checkFields() throws RemotingCommandException {
+
+    }
+
+    public String getCluster() {
+        return cluster;
+    }
+
+    public void setCluster(String cluster) {
+        this.cluster = cluster;
+    }
+}

namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/DefaultRequestProcessor.java

@@ -20,6 +20,8 @@
 import java.io.UnsupportedEncodingException;
 import java.util.Properties;
 import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.commons.lang3.StringUtils;
 import org.apache.rocketmq.common.MQVersion;
 import org.apache.rocketmq.common.MQVersion.Version;
 import org.apache.rocketmq.common.MixAll;
@@ -31,6 +33,7 @@
 import org.apache.rocketmq.common.protocol.ResponseCode;
 import org.apache.rocketmq.common.protocol.body.RegisterBrokerBody;
 import org.apache.rocketmq.common.protocol.body.TopicConfigSerializeWrapper;
+import org.apache.rocketmq.common.protocol.header.GetClusterListRequestHeader;
 import org.apache.rocketmq.common.protocol.header.GetTopicsByClusterRequestHeader;
 import org.apache.rocketmq.common.protocol.header.namesrv.DeleteKVConfigRequestHeader;
 import org.apache.rocketmq.common.protocol.header.namesrv.DeleteTopicInNamesrvRequestHeader;
@@ -302,11 +305,17 @@ public RemotingCommand getRouteInfoByTopic(ChannelHandlerContext ctx,
         return response;
     }
 
-    private RemotingCommand getBrokerClusterInfo(ChannelHandlerContext ctx, RemotingCommand request) {
+    private RemotingCommand getBrokerClusterInfo(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
         final RemotingCommand response = RemotingCommand.createResponseCommand(null);
+        GetClusterListRequestHeader requestHeader = (GetClusterListRequestHeader) request.decodeCommandCustomHeader(GetClusterListRequestHeader.class);
+        if (StringUtils.isNotBlank(requestHeader.getCluster())) {
+            byte[] content = this.namesrvController.getRouteInfoManager().getAllClusterInfo();
+            response.setBody(content);
+        } else {
+            byte[] content = this.namesrvController.getRouteInfoManager().getOneClusterInfo(requestHeader.getCluster());
+            response.setBody(content);
+        }
 
-        byte[] content = this.namesrvController.getRouteInfoManager().getAllClusterInfo();
-        response.setBody(content);
 
         response.setCode(ResponseCode.SUCCESS);
         response.setRemark(null);

namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManager.java

@@ -28,6 +28,7 @@
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
+
 import org.apache.rocketmq.common.DataVersion;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.TopicConfig;
@@ -70,6 +71,27 @@ public byte[] getAllClusterInfo() {
         return clusterInfoSerializeWrapper.encode();
     }
 
+    public byte[] getOneClusterInfo(String cluster) {
+        HashMap<String, Set<String>> clusterAddr = new HashMap<>();
+        HashMap<String, BrokerData> brokerAddr = new HashMap<>();
+
+        Set<String> brokers = clusterAddrTable.get(cluster);
+        if (brokers != null) {
+            for (String broker : brokers) {
+                BrokerData brokerData = brokerAddrTable.get(broker);
+                if (brokerData != null) {
+                    brokerAddr.put(broker, brokerData);
+                }
+            }
+            clusterAddr.put(cluster, brokers);
+        }
+
+        ClusterInfo info = new ClusterInfo();
+        info.setBrokerAddrTable(brokerAddr);
+        info.setClusterAddrTable(clusterAddr);
+        return info.encode();
+    }
+
     public void deleteTopic(final String topic) {
         try {
             try {

namesrv/src/test/java/org/apache/rocketmq/namesrv/processor/DefaultRequestProcessorTest.java

@@ -29,7 +29,9 @@
 import org.apache.rocketmq.common.namesrv.RegisterBrokerResult;
 import org.apache.rocketmq.common.protocol.RequestCode;
 import org.apache.rocketmq.common.protocol.ResponseCode;
+import org.apache.rocketmq.common.protocol.body.ClusterInfo;
 import org.apache.rocketmq.common.protocol.body.TopicConfigSerializeWrapper;
+import org.apache.rocketmq.common.protocol.header.GetClusterListRequestHeader;
 import org.apache.rocketmq.common.protocol.header.namesrv.DeleteKVConfigRequestHeader;
 import org.apache.rocketmq.common.protocol.header.namesrv.GetKVConfigRequestHeader;
 import org.apache.rocketmq.common.protocol.header.namesrv.GetKVConfigResponseHeader;
@@ -76,7 +78,9 @@ public void init() throws Exception {
         field.set(namesrvController, routeInfoManager);
         defaultRequestProcessor = new DefaultRequestProcessor(namesrvController);
 
-        registerRouteInfoManager();
+        for (int i = 10; i < 15; i++) {
+            registerRouteInfoManager("a" + i, i);
+        }
 
         logger = mock(Logger.class);
         when(logger.isInfoEnabled()).thenReturn(false);
@@ -257,20 +261,45 @@ private static void setFinalStatic(Field field, Object newValue) throws Exceptio
         field.set(null, newValue);
     }
 
-    private void registerRouteInfoManager() {
+    private void registerRouteInfoManager(String prefix, int id) {
+
         TopicConfigSerializeWrapper topicConfigSerializeWrapper = new TopicConfigSerializeWrapper();
         ConcurrentHashMap<String, TopicConfig> topicConfigConcurrentHashMap = new ConcurrentHashMap<>();
         TopicConfig topicConfig = new TopicConfig();
         topicConfig.setWriteQueueNums(8);
-        topicConfig.setTopicName("unit-test");
+        topicConfig.setTopicName("unit-test_" + prefix);
         topicConfig.setPerm(6);
         topicConfig.setReadQueueNums(8);
         topicConfig.setOrder(false);
         topicConfigConcurrentHashMap.put("unit-test", topicConfig);
         topicConfigSerializeWrapper.setTopicConfigTable(topicConfigConcurrentHashMap);
+
         Channel channel = mock(Channel.class);
-        RegisterBrokerResult registerBrokerResult = routeInfoManager.registerBroker("default-cluster", "127.0.0.1:10911", "default-broker", 1234, "127.0.0.1:1001",
+        RegisterBrokerResult registerBrokerResult = routeInfoManager.registerBroker("default-cluster_" + prefix, "127.0.0.1:120" + id, "default-broker_" + prefix, 1236 + id, "127.0.0.1:100" + id,
             topicConfigSerializeWrapper, new ArrayList<String>(), channel);
 
     }
+
+    private static RemotingCommand genClusterListCmd(String cluster) {
+        GetClusterListRequestHeader header = new GetClusterListRequestHeader();
+        header.setCluster(cluster);
+        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_BROKER_CLUSTER_INFO, header);
+        return request;
+    }
+
+    @Test
+    public void testClusterInfo() throws RemotingCommandException {
+        testClusterInfo(null, 5);
+        testClusterInfo("default-cluster_a10", 1);
+    }
+
+    public void testClusterInfo(String cluster, int want) throws RemotingCommandException {
+        ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
+        when(ctx.channel()).thenReturn(null);
+        RemotingCommand response = defaultRequestProcessor.processRequest(ctx, genClusterListCmd(cluster));
+        assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
+        assertThat(response.getRemark()).isNull();
+        ClusterInfo info = ClusterInfo.decode(response.getBody(), ClusterInfo.class);
+        assertThat(info.getClusterAddrTable().size() == want);
+    }
 }

test/src/main/java/org/apache/rocketmq/test/util/MQAdmin.java

@@ -107,7 +107,7 @@ public static ClusterInfo getCluster(String nameSrvAddr) {
         ClusterInfo clusterInfo = null;
         try {
             mqAdminExt.start();
-            clusterInfo = mqAdminExt.examineBrokerClusterInfo();
+            clusterInfo = mqAdminExt.examineBrokerClusterInfo(null);
         } catch (Exception e) {
             e.printStackTrace();
         }

tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java

@@ -217,8 +217,13 @@ public ConsumeStats examineConsumeStats(String consumerGroup,
     }
 
     @Override
-    public ClusterInfo examineBrokerClusterInfo() throws InterruptedException, RemotingConnectException, RemotingTimeoutException,
+    public ClusterInfo examineBrokerClusterInfo(String cluster) throws InterruptedException, RemotingConnectException, RemotingTimeoutException,
         RemotingSendRequestException, MQBrokerException {
+        return defaultMQAdminExtImpl.examineBrokerClusterInfo(cluster);
+    }
+
+    @Override
+    public ClusterInfo examineBrokerClusterInfo() throws InterruptedException, MQBrokerException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
         return defaultMQAdminExtImpl.examineBrokerClusterInfo();
     }
 

tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java

@@ -270,8 +270,13 @@ public ConsumeStats examineConsumeStats(String consumerGroup,
     }
 
     @Override
-    public ClusterInfo examineBrokerClusterInfo() throws InterruptedException, MQBrokerException, RemotingTimeoutException,
+    public ClusterInfo examineBrokerClusterInfo(String cluster) throws InterruptedException, MQBrokerException, RemotingTimeoutException,
         RemotingSendRequestException, RemotingConnectException {
+        return this.mqClientInstance.getMQClientAPIImpl().getBrokerClusterInfo(cluster,timeoutMillis);
+    }
+
+    @Override
+    public ClusterInfo examineBrokerClusterInfo() throws InterruptedException, MQBrokerException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
         return this.mqClientInstance.getMQClientAPIImpl().getBrokerClusterInfo(timeoutMillis);
     }
 
@@ -626,7 +631,7 @@ public boolean cleanExpiredConsumerQueue(
         RemotingTimeoutException, MQClientException, InterruptedException {
         boolean result = false;
         try {
-            ClusterInfo clusterInfo = examineBrokerClusterInfo();
+            ClusterInfo clusterInfo = examineBrokerClusterInfo(cluster);
             if (null == cluster || "".equals(cluster)) {
                 for (String targetCluster : clusterInfo.retrieveAllClusterNames()) {
                     result = cleanExpiredConsumerQueueByCluster(clusterInfo, targetCluster);
@@ -666,7 +671,7 @@ public boolean cleanUnusedTopic(String cluster) throws RemotingConnectException,
         RemotingTimeoutException, MQClientException, InterruptedException {
         boolean result = false;
         try {
-            ClusterInfo clusterInfo = examineBrokerClusterInfo();
+            ClusterInfo clusterInfo = examineBrokerClusterInfo(cluster);
             if (null == cluster || "".equals(cluster)) {
                 for (String targetCluster : clusterInfo.retrieveAllClusterNames()) {
                     result = cleanUnusedTopicByCluster(clusterInfo, targetCluster);
@@ -832,7 +837,7 @@ public boolean consumed(final MessageExt msg,
 
         ConsumeStats cstats = this.examineConsumeStats(group);
 
-        ClusterInfo ci = this.examineBrokerClusterInfo();
+        ClusterInfo ci = this.examineBrokerClusterInfo(null);
 
         Iterator<Entry<MessageQueue, OffsetWrapper>> it = cstats.getOffsetTable().entrySet().iterator();
         while (it.hasNext()) {
@@ -894,7 +899,7 @@ public Set<String> getTopicClusterList(
         final String topic) throws InterruptedException, MQBrokerException, MQClientException,
         RemotingException {
         Set<String> clusterSet = new HashSet<String>();
-        ClusterInfo clusterInfo = examineBrokerClusterInfo();
+        ClusterInfo clusterInfo = examineBrokerClusterInfo(null);
         TopicRouteData topicRouteData = examineTopicRouteInfo(topic);
         BrokerData brokerData = topicRouteData.getBrokerDatas().get(0);
         String brokerName = brokerData.getBrokerName();

tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java

@@ -97,6 +97,9 @@ ConsumeStats examineConsumeStats(final String consumerGroup,
         final String topic) throws RemotingException, MQClientException,
         InterruptedException, MQBrokerException;
 
+    ClusterInfo examineBrokerClusterInfo(String cluster) throws InterruptedException, MQBrokerException, RemotingTimeoutException,
+        RemotingSendRequestException, RemotingConnectException;
+
     ClusterInfo examineBrokerClusterInfo() throws InterruptedException, MQBrokerException, RemotingTimeoutException,
         RemotingSendRequestException, RemotingConnectException;
 

tools/src/main/java/org/apache/rocketmq/tools/command/CommandUtil.java

@@ -42,7 +42,7 @@ public class CommandUtil {
         MQBrokerException {
         Map<String, List<String>> masterAndSlaveMap = new HashMap<String, List<String>>(4);
 
-        ClusterInfo clusterInfoSerializeWrapper = adminExt.examineBrokerClusterInfo();
+        ClusterInfo clusterInfoSerializeWrapper = adminExt.examineBrokerClusterInfo(clusterName);
         Set<String> brokerNameSet = clusterInfoSerializeWrapper.getClusterAddrTable().get(clusterName);
 
         if (brokerNameSet == null) {
@@ -79,7 +79,7 @@ public static Set<String> fetchMasterAddrByClusterName(final MQAdminExt adminExt
         RemotingSendRequestException, MQBrokerException {
         Set<String> masterSet = new HashSet<String>();
 
-        ClusterInfo clusterInfoSerializeWrapper = adminExt.examineBrokerClusterInfo();
+        ClusterInfo clusterInfoSerializeWrapper = adminExt.examineBrokerClusterInfo(clusterName);
 
         Set<String> brokerNameSet = clusterInfoSerializeWrapper.getClusterAddrTable().get(clusterName);
 
@@ -107,7 +107,7 @@ public static Set<String> fetchMasterAndSlaveAddrByClusterName(final MQAdminExt
         RemotingSendRequestException, MQBrokerException {
         Set<String> masterSet = new HashSet<String>();
 
-        ClusterInfo clusterInfoSerializeWrapper = adminExt.examineBrokerClusterInfo();
+        ClusterInfo clusterInfoSerializeWrapper = adminExt.examineBrokerClusterInfo(clusterName);
 
         Set<String> brokerNameSet = clusterInfoSerializeWrapper.getClusterAddrTable().get(clusterName);
 
@@ -129,7 +129,7 @@ public static Set<String> fetchMasterAndSlaveAddrByClusterName(final MQAdminExt
 
     public static Set<String> fetchBrokerNameByClusterName(final MQAdminExt adminExt, final String clusterName)
         throws Exception {
-        ClusterInfo clusterInfoSerializeWrapper = adminExt.examineBrokerClusterInfo();
+        ClusterInfo clusterInfoSerializeWrapper = adminExt.examineBrokerClusterInfo(clusterName);
         Set<String> brokerNameSet = clusterInfoSerializeWrapper.getClusterAddrTable().get(clusterName);
         if (brokerNameSet.isEmpty()) {
             throw new Exception(
@@ -139,7 +139,7 @@ public static Set<String> fetchBrokerNameByClusterName(final MQAdminExt adminExt
     }
 
     public static String fetchBrokerNameByAddr(final MQAdminExt adminExt, final String addr) throws Exception {
-        ClusterInfo clusterInfoSerializeWrapper = adminExt.examineBrokerClusterInfo();
+        ClusterInfo clusterInfoSerializeWrapper = adminExt.examineBrokerClusterInfo(null);
         HashMap<String/* brokerName */, BrokerData> brokerAddrTable =
             clusterInfoSerializeWrapper.getBrokerAddrTable();
         Iterator<Map.Entry<String, BrokerData>> it = brokerAddrTable.entrySet().iterator();

tools/src/main/java/org/apache/rocketmq/tools/command/cluster/CLusterSendMsgRTCommand.java

@@ -92,7 +92,7 @@ public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) t
             defaultMQAdminExt.start();
             producer.start();
 
-            ClusterInfo clusterInfoSerializeWrapper = defaultMQAdminExt.examineBrokerClusterInfo();
+            ClusterInfo clusterInfoSerializeWrapper = defaultMQAdminExt.examineBrokerClusterInfo(null);
             HashMap<String, Set<String>> clusterAddr = clusterInfoSerializeWrapper
                 .getClusterAddrTable();