提交 2047f94d 编写于 作者: L Li Zhanhui

[BugFix] When deleting topics, instruct both master and slave nodes to delete...

[BugFix] When deleting topics, instruct both master and slave nodes to delete it such that consume queues are all deleted
上级 2e488e53
......@@ -35,6 +35,9 @@ import org.apache.rocketmq.tools.admin.MQAdminExt;
public class CommandUtil {
private static final String ERROR_MESSAGE = "Make sure the specified clusterName exists or the name server " +
"connected to is correct.";
public static Map<String/*master addr*/, List<String>/*slave addr*/> fetchMasterAndSlaveDistinguish(
final MQAdminExt adminExt, final String clusterName)
throws InterruptedException, RemotingConnectException,
......@@ -46,8 +49,7 @@ public class CommandUtil {
Set<String> brokerNameSet = clusterInfoSerializeWrapper.getClusterAddrTable().get(clusterName);
if (brokerNameSet == null) {
System.out
.printf("[error] Make sure the specified clusterName exists or the nameserver which connected is correct.");
System.out.printf("[error] %s", ERROR_MESSAGE);
return masterAndSlaveMap;
}
......@@ -62,8 +64,7 @@ public class CommandUtil {
masterAndSlaveMap.put(masterAddr, new ArrayList<String>());
for (Long id : brokerData.getBrokerAddrs().keySet()) {
if (brokerData.getBrokerAddrs().get(id) == null
|| id.longValue() == MixAll.MASTER_ID) {
if (brokerData.getBrokerAddrs().get(id) == null || id == MixAll.MASTER_ID) {
continue;
}
......@@ -95,8 +96,7 @@ public class CommandUtil {
}
}
} else {
System.out
.printf("[error] Make sure the specified clusterName exists or the nameserver which connected is correct.");
System.out.printf("[error] %s", ERROR_MESSAGE);
}
return masterSet;
......@@ -105,26 +105,22 @@ public class CommandUtil {
public static Set<String> fetchMasterAndSlaveAddrByClusterName(final MQAdminExt adminExt, final String clusterName)
throws InterruptedException, RemotingConnectException, RemotingTimeoutException,
RemotingSendRequestException, MQBrokerException {
Set<String> masterSet = new HashSet<String>();
Set<String> brokerAddressSet = new HashSet<String>();
ClusterInfo clusterInfoSerializeWrapper = adminExt.examineBrokerClusterInfo();
Set<String> brokerNameSet = clusterInfoSerializeWrapper.getClusterAddrTable().get(clusterName);
if (brokerNameSet != null) {
for (String brokerName : brokerNameSet) {
BrokerData brokerData = clusterInfoSerializeWrapper.getBrokerAddrTable().get(brokerName);
if (brokerData != null) {
final Collection<String> addrs = brokerData.getBrokerAddrs().values();
masterSet.addAll(addrs);
brokerAddressSet.addAll(addrs);
}
}
} else {
System.out
.printf("[error] Make sure the specified clusterName exists or the nameserver which connected is correct.");
System.out.printf("[error] %s", ERROR_MESSAGE);
}
return masterSet;
return brokerAddressSet;
}
public static Set<String> fetchBrokerNameByClusterName(final MQAdminExt adminExt, final String clusterName)
......@@ -132,25 +128,23 @@ public class CommandUtil {
ClusterInfo clusterInfoSerializeWrapper = adminExt.examineBrokerClusterInfo();
Set<String> brokerNameSet = clusterInfoSerializeWrapper.getClusterAddrTable().get(clusterName);
if (brokerNameSet.isEmpty()) {
throw new Exception(
"Make sure the specified clusterName exists or the nameserver which connected is correct.");
throw new Exception(ERROR_MESSAGE);
}
return brokerNameSet;
}
public static String fetchBrokerNameByAddr(final MQAdminExt adminExt, final String addr) throws Exception {
ClusterInfo clusterInfoSerializeWrapper = adminExt.examineBrokerClusterInfo();
HashMap<String/* brokerName */, BrokerData> brokerAddrTable =
clusterInfoSerializeWrapper.getBrokerAddrTable();
HashMap<String/* brokerName */, BrokerData> brokerAddrTable = clusterInfoSerializeWrapper.getBrokerAddrTable();
Iterator<Map.Entry<String, BrokerData>> it = brokerAddrTable.entrySet().iterator();
while (it.hasNext()) {
Map.Entry<String, BrokerData> entry = it.next();
HashMap<Long, String> brokerAddrs = entry.getValue().getBrokerAddrs();
if (brokerAddrs.containsValue(addr))
if (brokerAddrs.containsValue(addr)) {
return entry.getKey();
}
throw new Exception(
"Make sure the specified broker addr exists or the nameserver which connected is correct.");
}
throw new Exception(ERROR_MESSAGE);
}
}
......@@ -38,8 +38,8 @@ public class DeleteTopicSubCommand implements SubCommand {
final String topic
) throws InterruptedException, MQBrokerException, RemotingException, MQClientException {
Set<String> masterSet = CommandUtil.fetchMasterAddrByClusterName(adminExt, clusterName);
adminExt.deleteTopicInBroker(masterSet, topic);
Set<String> brokerAddressSet = CommandUtil.fetchMasterAndSlaveAddrByClusterName(adminExt, clusterName);
adminExt.deleteTopicInBroker(brokerAddressSet, topic);
System.out.printf("delete topic [%s] from cluster [%s] success.%n", topic, clusterName);
Set<String> nameServerSet = null;
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册