Skip to content

Commit

Permalink
add cluster list to specified cluster
Browse files Browse the repository at this point in the history
  • Loading branch information
@lindzh
lindzh committed Aug 23, 2017
1 parent 53dcd8d commit aacb19c
Show file tree
Hide file tree
Showing 15 changed files with 121 additions and 37 deletions.
8 changes: 6 additions & 2 deletions client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -1167,10 +1168,13 @@ public Properties getBrokerConfig(final String addr, final long timeoutMillis)
throw new MQBrokerException(response.getCode(), response.getRemark());
}

public ClusterInfo getBrokerClusterInfo(
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;
Expand Down
42 changes: 42 additions & 0 deletions ...src/main/java/org/apache/rocketmq/common/protocol/header/GetClusterListRequestHeader.java
View file
Original file line number Diff line number Diff line change
@@ -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;
}
}
7 changes: 4 additions & 3 deletions namesrv/src/main/java/org/apache/rocketmq/namesrv/processor/DefaultRequestProcessor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,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;
Expand Down Expand Up @@ -302,10 +303,10 @@ 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);

byte[] content = this.namesrvController.getRouteInfoManager().getAllClusterInfo();
GetClusterListRequestHeader requestHeader = (GetClusterListRequestHeader)request.decodeCommandCustomHeader(GetClusterListRequestHeader.class);
byte[] content = this.namesrvController.getRouteInfoManager().getAllClusterInfo(requestHeader.getCluster());
response.setBody(content);

response.setCode(ResponseCode.SUCCESS);
Expand Down
37 changes: 32 additions & 5 deletions namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManager.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,8 @@
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.DataVersion;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.TopicConfig;
Expand Down Expand Up @@ -63,11 +65,36 @@ public RouteInfoManager() {
this.filterServerTable = new HashMap<String, List<String>>(256);
}

public byte[] getAllClusterInfo() {
ClusterInfo clusterInfoSerializeWrapper = new ClusterInfo();
clusterInfoSerializeWrapper.setBrokerAddrTable(this.brokerAddrTable);
clusterInfoSerializeWrapper.setClusterAddrTable(this.clusterAddrTable);
return clusterInfoSerializeWrapper.encode();
public byte[] getAllClusterInfo(String cluster) {
if (StringUtils.isNotBlank(cluster)) {
return getOneClusterInfo(cluster);
} else {
ClusterInfo clusterInfoSerializeWrapper = new ClusterInfo();
clusterInfoSerializeWrapper.setBrokerAddrTable(this.brokerAddrTable);
clusterInfoSerializeWrapper.setClusterAddrTable(this.clusterAddrTable);
return clusterInfoSerializeWrapper.encode();
}
}

private 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) {
Expand Down
2 changes: 1 addition & 1 deletion namesrv/src/test/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManagerTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ public void terminate() {

@Test
public void testGetAllClusterInfo() {
byte[] clusterInfo = routeInfoManager.getAllClusterInfo();
byte[] clusterInfo = routeInfoManager.getAllClusterInfo(null);
assertThat(clusterInfo).isNotNull();
}

Expand Down
2 changes: 1 addition & 1 deletion test/src/main/java/org/apache/rocketmq/test/util/MQAdmin.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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();
}
Expand Down
4 changes: 2 additions & 2 deletions tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -217,9 +217,9 @@ 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();
return defaultMQAdminExtImpl.examineBrokerClusterInfo(cluster);
}

@Override
Expand Down
12 changes: 6 additions & 6 deletions tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -270,9 +270,9 @@ 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(timeoutMillis);
return this.mqClientInstance.getMQClientAPIImpl().getBrokerClusterInfo(cluster,timeoutMillis);
}

@Override
Expand Down Expand Up @@ -626,7 +626,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);
Expand Down Expand Up @@ -666,7 +666,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);
Expand Down Expand Up @@ -832,7 +832,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()) {
Expand Down Expand Up @@ -894,7 +894,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();
Expand Down
2 changes: 1 addition & 1 deletion tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -97,7 +97,7 @@ ConsumeStats examineConsumeStats(final String consumerGroup,
final String topic) throws RemotingException, MQClientException,
InterruptedException, MQBrokerException;

ClusterInfo examineBrokerClusterInfo() throws InterruptedException, MQBrokerException, RemotingTimeoutException,
ClusterInfo examineBrokerClusterInfo(String cluster) throws InterruptedException, MQBrokerException, RemotingTimeoutException,
RemotingSendRequestException, RemotingConnectException;

TopicRouteData examineTopicRouteInfo(
Expand Down
10 changes: 5 additions & 5 deletions tools/src/main/java/org/apache/rocketmq/tools/command/CommandUtil.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down Expand Up @@ -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);

Expand Down Expand Up @@ -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);

Expand All @@ -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(
Expand All @@ -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();
Expand Down
2 changes: 1 addition & 1 deletion tools/src/main/java/org/apache/rocketmq/tools/command/cluster/CLusterSendMsgRTCommand.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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();

Expand Down
22 changes: 16 additions & 6 deletions tools/src/main/java/org/apache/rocketmq/tools/command/cluster/ClusterListSubCommand.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,10 @@ public Options buildCommandlineOptions(Options options) {
opt.setRequired(false);
options.addOption(opt);

opt = new Option("c", "cluster", true, "specify cluster to get only this cluster");
opt.setRequired(false);
options.addOption(opt);

return options;
}

Expand All @@ -74,6 +78,12 @@ public void execute(final CommandLine commandLine, final Options options,
printInterval = Long.parseLong(commandLine.getOptionValue('i')) * 1000;
}

String cluster = null;
boolean hasCluster = commandLine.hasOption('c');
if (hasCluster) {
cluster = commandLine.getOptionValue('c');
}

try {
defaultMQAdminExt.start();
long i = 0;
Expand All @@ -83,9 +93,9 @@ public void execute(final CommandLine commandLine, final Options options,
Thread.sleep(printInterval);
}
if (commandLine.hasOption('m')) {
this.printClusterMoreStats(defaultMQAdminExt);
this.printClusterMoreStats(cluster, defaultMQAdminExt);
} else {
this.printClusterBaseInfo(defaultMQAdminExt);
this.printClusterBaseInfo(cluster, defaultMQAdminExt);
}
}
while (enableInterval);
Expand All @@ -96,10 +106,10 @@ public void execute(final CommandLine commandLine, final Options options,
}
}

private void printClusterMoreStats(final DefaultMQAdminExt defaultMQAdminExt) throws RemotingConnectException,
private void printClusterMoreStats(final String cluster,final DefaultMQAdminExt defaultMQAdminExt) throws RemotingConnectException,
RemotingTimeoutException, RemotingSendRequestException, InterruptedException, MQBrokerException {

ClusterInfo clusterInfoSerializeWrapper = defaultMQAdminExt.examineBrokerClusterInfo();
ClusterInfo clusterInfoSerializeWrapper = defaultMQAdminExt.examineBrokerClusterInfo(cluster);

System.out.printf("%-16s %-32s %14s %14s %14s %14s%n",
"#Cluster Name",
Expand Down Expand Up @@ -165,11 +175,11 @@ private void printClusterMoreStats(final DefaultMQAdminExt defaultMQAdminExt) th
}
}

private void printClusterBaseInfo(
private void printClusterBaseInfo(final String cluster,
final DefaultMQAdminExt defaultMQAdminExt) throws RemotingConnectException, RemotingTimeoutException,
RemotingSendRequestException, InterruptedException, MQBrokerException {

ClusterInfo clusterInfoSerializeWrapper = defaultMQAdminExt.examineBrokerClusterInfo();
ClusterInfo clusterInfoSerializeWrapper = defaultMQAdminExt.examineBrokerClusterInfo(cluster);

System.out.printf("%-16s %-22s %-4s %-22s %-16s %19s %19s %10s %5s %6s%n",
"#Cluster Name",
Expand Down
2 changes: 1 addition & 1 deletion tools/src/main/java/org/apache/rocketmq/tools/command/topic/TopicListSubCommand.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,7 @@ public void execute(final CommandLine commandLine, final Options options,
try {
defaultMQAdminExt.start();
if (commandLine.hasOption('c')) {
ClusterInfo clusterInfo = defaultMQAdminExt.examineBrokerClusterInfo();
ClusterInfo clusterInfo = defaultMQAdminExt.examineBrokerClusterInfo(null);

System.out.printf("%-20s %-48s %-48s%n",
"#Cluster Name",
Expand Down
4 changes: 2 additions & 2 deletions tools/src/test/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -144,7 +144,7 @@ public static void init() throws Exception {
brokerAddrTable.put("broker-test", new BrokerData());
clusterInfo.setBrokerAddrTable(brokerAddrTable);
clusterInfo.setClusterAddrTable(new HashMap<String, Set<String>>());
when(mQClientAPIImpl.getBrokerClusterInfo(anyLong())).thenReturn(clusterInfo);
when(mQClientAPIImpl.getBrokerClusterInfo(anyString(), anyLong())).thenReturn(clusterInfo);
when(mQClientAPIImpl.cleanExpiredConsumeQueue(anyString(), anyLong())).thenReturn(true);

Set<String> clusterList = new HashSet<>();
Expand Down Expand Up @@ -257,7 +257,7 @@ public void testFetchBrokerRuntimeStats() throws InterruptedException, MQBrokerE

@Test
public void testExamineBrokerClusterInfo() throws InterruptedException, MQBrokerException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
ClusterInfo clusterInfo = defaultMQAdminExt.examineBrokerClusterInfo();
ClusterInfo clusterInfo = defaultMQAdminExt.examineBrokerClusterInfo(null);
HashMap<String, BrokerData> brokerList = clusterInfo.getBrokerAddrTable();
assertThat(brokerList.get("default-broker").getBrokerName()).isEqualTo("default-broker");
assertThat(brokerList.containsKey("broker-test")).isTrue();
Expand Down
2 changes: 1 addition & 1 deletion tools/src/test/java/org/apache/rocketmq/tools/command/CommandUtilTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -81,7 +81,7 @@ public void setup() throws MQClientException, NoSuchFieldException, IllegalAcces
clusterAddrTable.put("default-cluster", brokerSet);
clusterInfo.setBrokerAddrTable(brokerAddrTable);
clusterInfo.setClusterAddrTable(clusterAddrTable);
when(mQClientAPIImpl.getBrokerClusterInfo(anyLong())).thenReturn(clusterInfo);
when(mQClientAPIImpl.getBrokerClusterInfo(anyString(), anyLong())).thenReturn(clusterInfo);
when(mQClientAPIImpl.cleanExpiredConsumeQueue(anyString(), anyLong())).thenReturn(true);
}

Expand Down

0 comments on commit aacb19c

Please sign in to comment.