This is an automated email from the ASF dual-hosted git repository.
RongtongJin pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new 391f8ab009 feat: add mqadmin getTopicConfig command (#11473)
391f8ab009 is described below
commit 391f8ab00980908af7604f2534c3826e44b0c3c4
Author: zhang hongyuan <[email protected]>
AuthorDate: Thu Oct 22 17:22:17 2026 +0800
feat: add mqadmin getTopicConfig command (#11473)
Signed-off-by: zhang hongyuan <[email protected]>
---
docs/cn/operation.md | 20 ++
docs/en/CLITools.md | 20 ++
.../rocketmq/tools/command/MQAdminStartup.java | 2 +
.../command/topic/GetTopicConfigSubCommand.java | 194 +++++++++++++++
.../topic/GetTopicConfigSubCommandTest.java | 269 +++++++++++++++++++++
5 files changed, 505 insertions(+)
diff --git a/docs/cn/operation.md b/docs/cn/operation.md
index a38ea0d594..690a922921 100644
--- a/docs/cn/operation.md
+++ b/docs/cn/operation.md
@@ -281,6 +281,26 @@ RocketMQ 5.0 开始支持自动主从切换的模式，可参考以下文档
<td height=57 class=xl65 width=149 style='height:43.0pt;width:112pt'>-n</td>
<td class=xl66 width=159 style='width:119pt'>NameServer 服务地址，格式 ip:port</td>
</tr>
+ <tr height=23 style='height:17.0pt'>
+ <td rowspan=4 height=137 class=xl68 width=163 style='border-bottom:1.0pt;
+ height:103.0pt;border-top:none;width:122pt'>getTopicConfig</td>
+ <td rowspan=4 class=xl70 width=135 style='border-bottom:1.0pt;
+ border-top:none;width:101pt'>查看 Broker 上实际存储的 Topic 配置。用法：getTopicConfig -t
<topic> [-b <brokerAddr> | -c <cluster>]；未指定 -b 或 -c 时查询所有
Broker。</td>
+ <td class=xl65 width=149 style='width:112pt'>-t</td>
+ <td class=xl66 width=159 style='width:119pt'>topic 名称</td>
+ </tr>
+ <tr height=23 style='height:17.0pt'>
+ <td height=23 class=xl65 width=149 style='height:17.0pt;width:112pt'>-b</td>
+ <td class=xl66 width=159 style='width:119pt'>Broker 地址，格式 ip:port</td>
+ </tr>
+ <tr height=23 style='height:17.0pt'>
+ <td height=23 class=xl65 width=149 style='height:17.0pt;width:112pt'>-c</td>
+ <td class=xl66 width=159 style='width:119pt'>集群名称</td>
+ </tr>
+ <tr height=23 style='height:17.0pt'>
+ <td height=23 class=xl65 width=149 style='height:17.0pt;width:112pt'>-n</td>
+ <td class=xl66 width=159 style='width:119pt'>NameServer 服务地址，格式 ip:port</td>
+ </tr>
<tr height=23 style='height:17.0pt'>
<td rowspan=3 height=103 class=xl68 width=163 style='border-bottom:1.0pt;
height:77.0pt;border-top:none;width:122pt'>topicClusterList</td>
diff --git a/docs/en/CLITools.md b/docs/en/CLITools.md
index 15c593346b..cd944a793f 100644
--- a/docs/en/CLITools.md
+++ b/docs/en/CLITools.md
@@ -126,6 +126,26 @@ Before introducing the mqadmin management tool, the
following points need to be
<td height=57 class=xl65 width=149 style='height:43.0pt;width:112pt'>-n</td>
<td class=xl66 width=159 style='width:119pt'>Declare the service address of
the nameServer, and the option format is ip:port</td>
</tr>
+ <tr height=23 style='height:17.0pt'>
+ <td rowspan=4 height=137 class=xl68 width=163 style='border-bottom:1.0pt;
+ height:103.0pt;border-top:none;width:122pt'>getTopicConfig</td>
+ <td rowspan=4 class=xl70 width=135 style='border-bottom:1.0pt;
+ border-top:none;width:101pt'>View the actual topic configuration stored on a
broker. Usage: getTopicConfig -t <topic> [-b <brokerAddr> | -c
<cluster>]. If neither -b nor -c is specified, all brokers are
queried.</td>
+ <td class=xl65 width=149 style='width:112pt'>-t</td>
+ <td class=xl66 width=159 style='width:119pt'>Used to specify the name of the
topic</td>
+ </tr>
+ <tr height=23 style='height:17.0pt'>
+ <td height=23 class=xl65 width=149 style='height:17.0pt;width:112pt'>-b</td>
+ <td class=xl66 width=159 style='width:119pt'>Declare the address of the
broker and format as ip:port</td>
+ </tr>
+ <tr height=23 style='height:17.0pt'>
+ <td height=23 class=xl65 width=149 style='height:17.0pt;width:112pt'>-c</td>
+ <td class=xl66 width=159 style='width:119pt'>Declare the cluster name</td>
+ </tr>
+ <tr height=23 style='height:17.0pt'>
+ <td height=23 class=xl65 width=149 style='height:17.0pt;width:112pt'>-n</td>
+ <td class=xl66 width=159 style='width:119pt'>Declare the service address of
the nameServer, and the option format is ip:port</td>
+ </tr>
<tr height=23 style='height:17.0pt'>
<td rowspan=3 height=103 class=xl68 width=163 style='border-bottom:1.0pt;
height:77.0pt;border-top:none;width:122pt'>topicClusterList</td>
diff --git
a/tools/src/main/java/org/apache/rocketmq/tools/command/MQAdminStartup.java
b/tools/src/main/java/org/apache/rocketmq/tools/command/MQAdminStartup.java
index 14df64632b..9ee68b1cc1 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/MQAdminStartup.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/MQAdminStartup.java
@@ -113,6 +113,7 @@ import
org.apache.rocketmq.tools.command.queue.QueryConsumeQueueCommand;
import org.apache.rocketmq.tools.command.stats.StatsAllSubCommand;
import org.apache.rocketmq.tools.command.topic.AllocateMQSubCommand;
import org.apache.rocketmq.tools.command.topic.DeleteTopicSubCommand;
+import org.apache.rocketmq.tools.command.topic.GetTopicConfigSubCommand;
import org.apache.rocketmq.tools.command.topic.RemappingStaticTopicSubCommand;
import org.apache.rocketmq.tools.command.topic.TopicClusterSubCommand;
import org.apache.rocketmq.tools.command.topic.TopicListSubCommand;
@@ -258,6 +259,7 @@ public class MQAdminStartup {
initCommand(new UpdateNamesrvConfigCommand());
initCommand(new GetBrokerConfigCommand());
initCommand(new GetConsumerConfigSubCommand());
+ initCommand(new GetTopicConfigSubCommand());
initCommand(new QueryConsumeQueueCommand());
initCommand(new SendMessageCommand());
diff --git
a/tools/src/main/java/org/apache/rocketmq/tools/command/topic/GetTopicConfigSubCommand.java
b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/GetTopicConfigSubCommand.java
new file mode 100644
index 0000000000..c89a414561
--- /dev/null
+++
b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/GetTopicConfigSubCommand.java
@@ -0,0 +1,194 @@
+/*
+ * 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.topic;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.OptionGroup;
+import org.apache.commons.cli.Options;
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.common.TopicConfig;
+import org.apache.rocketmq.common.constant.PermName;
+import org.apache.rocketmq.remoting.RPCHook;
+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.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import
org.apache.rocketmq.remoting.protocol.statictopic.TopicConfigAndQueueMapping;
+import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
+import org.apache.rocketmq.tools.command.CommandUtil;
+import org.apache.rocketmq.tools.command.SubCommand;
+import org.apache.rocketmq.tools.command.SubCommandException;
+
+/**
+ * Prints the topic configuration stored on a broker. Unlike {@code
topicRoute}, which only
+ * exposes the routing view, this command returns the complete configuration
persisted by the
+ * broker, including attributes that affect custom message types.
+ */
+public class GetTopicConfigSubCommand implements SubCommand {
+ private static final String FIELD_FORMAT = "%s%-40s= %s%n";
+
+ @Override
+ public String commandName() {
+ return "getTopicConfig";
+ }
+
+ @Override
+ public String commandDesc() {
+ return "Get topic config from a broker or cluster.";
+ }
+
+ @Override
+ public Options buildCommandlineOptions(final Options options) {
+ Option topicOption = new Option("t", "topic", true, "topic name");
+ topicOption.setRequired(true);
+ options.addOption(topicOption);
+
+ // Both values identify where to read from. Keeping them in one group
prevents a user
+ // from accidentally querying a broker while believing a cluster was
selected.
+ OptionGroup targetGroup = new OptionGroup();
+ targetGroup.addOption(new Option("b", "brokerAddr", true, "get topic
config from which broker"));
+ targetGroup.addOption(new Option("c", "clusterName", true, "get topic
config from which cluster"));
+ options.addOptionGroup(targetGroup);
+
+ return options;
+ }
+
+ @Override
+ public void execute(final CommandLine commandLine, final Options options,
+ final RPCHook rpcHook) throws SubCommandException {
+ DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
+
defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
+
+ try {
+ String topic = commandLine.getOptionValue('t').trim();
+ if (commandLine.hasOption('n')) {
+
defaultMQAdminExt.setNamesrvAddr(commandLine.getOptionValue('n').trim());
+ }
+ defaultMQAdminExt.start();
+
+ if (commandLine.hasOption('b')) {
+ String brokerAddr = commandLine.getOptionValue('b').trim();
+ getAndPrint(defaultMQAdminExt, topic, brokerAddr,
+ String.format("============%s============\n", brokerAddr));
+ } else if (commandLine.hasOption('c')) {
+ String clusterName = commandLine.getOptionValue('c').trim();
+ printClusterConfig(defaultMQAdminExt, topic, clusterName);
+ } else {
+ printAllBrokerConfigs(defaultMQAdminExt, topic);
+ }
+ } catch (Exception e) {
+ throw new SubCommandException(this.getClass().getSimpleName() + "
command failed", e);
+ } finally {
+ defaultMQAdminExt.shutdown();
+ }
+ }
+
+ private void printClusterConfig(final DefaultMQAdminExt defaultMQAdminExt,
final String topic,
+ final String clusterName) throws Exception {
+ Map<String, List<String>> masterAndSlaveMap =
+ CommandUtil.fetchMasterAndSlaveDistinguish(defaultMQAdminExt,
clusterName);
+ for (Entry<String, List<String>> entry : masterAndSlaveMap.entrySet())
{
+ String masterAddr = entry.getKey();
+ if (!CommandUtil.NO_MASTER_PLACEHOLDER.equals(masterAddr)) {
+ getAndPrint(defaultMQAdminExt, topic, masterAddr,
+ String.format("============Master: %s============\n",
masterAddr));
+ }
+
+ for (String slaveAddr : entry.getValue()) {
+ getAndPrint(defaultMQAdminExt, topic, slaveAddr,
+ String.format("============My Master: %s=====Slave:
%s============\n",
+ masterAddr, slaveAddr));
+ }
+ }
+ }
+
+ private void printAllBrokerConfigs(final DefaultMQAdminExt
defaultMQAdminExt, final String topic)
+ throws Exception {
+ ClusterInfo clusterInfo = defaultMQAdminExt.examineBrokerClusterInfo();
+ if (clusterInfo == null || clusterInfo.getBrokerAddrTable() == null) {
+ return;
+ }
+
+ // When no explicit target is given, use the same master address
selection as the other
+ // mqadmin commands so a single logical broker is queried once while
its slaves are not
+ // mistaken for independent broker entries.
+ for (Entry<String, BrokerData> entry :
clusterInfo.getBrokerAddrTable().entrySet()) {
+ BrokerData brokerData = entry.getValue();
+ if (brokerData == null || brokerData.getBrokerAddrs() == null
+ || brokerData.getBrokerAddrs().isEmpty()) {
+ continue;
+ }
+ String brokerAddr = brokerData.selectBrokerAddr();
+ if (brokerAddr != null) {
+ getAndPrint(defaultMQAdminExt, topic, brokerAddr,
+ String.format("============%s:%s============\n",
entry.getKey(), brokerAddr));
+ }
+ }
+ }
+
+ private void getAndPrint(final DefaultMQAdminExt defaultMQAdminExt, final
String topic,
+ final String brokerAddr, final String printPrefix) throws
InterruptedException,
+ RemotingConnectException, RemotingTimeoutException,
RemotingSendRequestException,
+ MQBrokerException {
+ if (brokerAddr == null ||
CommandUtil.NO_MASTER_PLACEHOLDER.equals(brokerAddr)) {
+ return;
+ }
+
+ TopicConfig topicConfig;
+ try {
+ topicConfig = defaultMQAdminExt.examineTopicConfig(brokerAddr,
topic);
+ } catch (MQBrokerException e) {
+ // A topic may exist only on some brokers in a cluster. Treat that
broker as absent
+ // instead of failing the whole query and hiding the valid
configurations.
+ if (e.getResponseCode() == ResponseCode.TOPIC_NOT_EXIST) {
+ return;
+ }
+ throw e;
+ }
+
+ if (topicConfig == null) {
+ return;
+ }
+
+ System.out.print(printPrefix);
+ printField("topicName", topicConfig.getTopicName());
+ printField("readQueueNums", topicConfig.getReadQueueNums());
+ printField("writeQueueNums", topicConfig.getWriteQueueNums());
+ printField("perm", String.format("%s (%d)",
PermName.perm2String(topicConfig.getPerm()),
+ topicConfig.getPerm()));
+ printField("topicFilterType", topicConfig.getTopicFilterType());
+ printField("topicSysFlag", topicConfig.getTopicSysFlag());
+ printField("order", topicConfig.isOrder());
+ printField("attributes", topicConfig.getAttributes());
+
+ if (topicConfig instanceof TopicConfigAndQueueMapping) {
+ printField("mappingDetail",
+ ((TopicConfigAndQueueMapping) topicConfig).getMappingDetail());
+ }
+ System.out.printf("%n");
+ }
+
+ private void printField(final String name, final Object value) {
+ System.out.printf(FIELD_FORMAT, "", name, value == null ? "" : value);
+ }
+}
diff --git
a/tools/src/test/java/org/apache/rocketmq/tools/command/topic/GetTopicConfigSubCommandTest.java
b/tools/src/test/java/org/apache/rocketmq/tools/command/topic/GetTopicConfigSubCommandTest.java
new file mode 100644
index 0000000000..fae7389a3d
--- /dev/null
+++
b/tools/src/test/java/org/apache/rocketmq/tools/command/topic/GetTopicConfigSubCommandTest.java
@@ -0,0 +1,269 @@
+/*
+ * 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.topic;
+
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.commons.cli.AlreadySelectedException;
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.DefaultParser;
+import org.apache.commons.cli.MissingOptionException;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.ParseException;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.TopicConfig;
+import org.apache.rocketmq.common.TopicFilterType;
+import org.apache.rocketmq.common.attribute.TopicMessageType;
+import org.apache.rocketmq.common.constant.PermName;
+import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import
org.apache.rocketmq.remoting.protocol.statictopic.TopicConfigAndQueueMapping;
+import
org.apache.rocketmq.remoting.protocol.statictopic.TopicQueueMappingDetail;
+import org.apache.rocketmq.srvutil.ServerUtil;
+import org.apache.rocketmq.tools.command.SubCommandException;
+import org.apache.rocketmq.tools.command.server.ServerResponseMocker;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class GetTopicConfigSubCommandTest {
+ private final ByteArrayOutputStream outContent = new
ByteArrayOutputStream();
+ private final PrintStream originalOut = System.out;
+ private final List<ServerResponseMocker> mockers = new ArrayList<>();
+
+ @Before
+ public void setUp() {
+ outContent.reset();
+ System.setOut(new PrintStream(outContent));
+ }
+
+ @After
+ public void tearDown() {
+ for (ServerResponseMocker mocker : mockers) {
+ mocker.shutdown();
+ }
+ mockers.clear();
+ System.setOut(originalOut);
+ outContent.reset();
+ }
+
+ @Test
+ public void testCommandNameAndOptions() throws ParseException {
+ GetTopicConfigSubCommand command = new GetTopicConfigSubCommand();
+ Assert.assertEquals("getTopicConfig", command.commandName());
+ Assert.assertTrue(command.commandDesc().contains("topic config"));
+
+ Options options = command.buildCommandlineOptions(new Options());
+ Assert.assertTrue(options.hasOption("t"));
+ Assert.assertTrue(options.hasOption("b"));
+ Assert.assertTrue(options.hasOption("c"));
+
+ DefaultParser parser = new DefaultParser();
+ CommandLine brokerLine = parser.parse(options,
+ new String[] {"-t", "TopicTest", "-b", "127.0.0.1:10911"});
+ Assert.assertEquals("TopicTest", brokerLine.getOptionValue('t'));
+ Assert.assertEquals("127.0.0.1:10911", brokerLine.getOptionValue('b'));
+ Assert.assertFalse(brokerLine.hasOption('c'));
+
+ CommandLine clusterLine = parser.parse(options,
+ new String[] {"--topic=TopicTest",
"--clusterName=DefaultCluster"});
+ Assert.assertTrue(clusterLine.hasOption('c'));
+ Assert.assertFalse(clusterLine.hasOption('b'));
+ Assert.assertEquals("DefaultCluster", clusterLine.getOptionValue('c'));
+ }
+
+ @Test
+ public void testTopicIsRequired() {
+ GetTopicConfigSubCommand command = new GetTopicConfigSubCommand();
+ Options options = command.buildCommandlineOptions(new Options());
+ try {
+ new DefaultParser().parse(options, new String[] {"-b",
"127.0.0.1:10911"});
+ Assert.fail("The topic option should be required");
+ } catch (ParseException expected) {
+ // Commons CLI reports the missing option by its short key ("t")
rather than the
+ // user-facing "-t" spelling, so the exception type and payload
are the stable checks.
+ Assert.assertTrue(expected instanceof MissingOptionException);
+ Assert.assertTrue(((MissingOptionException)
expected).getMissingOptions().contains("t"));
+ }
+ }
+
+ @Test
+ public void testBrokerAndClusterAreMutuallyExclusive() {
+ GetTopicConfigSubCommand command = new GetTopicConfigSubCommand();
+ Options options = command.buildCommandlineOptions(new Options());
+ try {
+ new DefaultParser().parse(options, new String[] {"-t",
"TopicTest", "-b",
+ "127.0.0.1:10911", "-c", "DefaultCluster"});
+ Assert.fail("Only one broker target should be accepted");
+ } catch (ParseException expected) {
+ Assert.assertTrue(expected instanceof AlreadySelectedException);
+ }
+ }
+
+ @Test
+ public void testExecuteWithBrokerAddressPrintsAllConfigurationFields()
+ throws SubCommandException {
+ TopicConfig config = newConfig("TopicTest", 8, 4, PermName.PERM_READ |
PermName.PERM_WRITE);
+ config.setTopicMessageType(TopicMessageType.LITE);
+ config.setLiteTopicExpiration(30);
+ ServerResponseMocker broker = startBroker("broker-a", config);
+
+ String output = execute("-t", "TopicTest", "-b", "127.0.0.1:" +
broker.listenPort());
+
+ Assert.assertTrue(output.contains("============127.0.0.1:" +
broker.listenPort()
+ + "============"));
+ assertField(output, "topicName", "TopicTest");
+ assertField(output, "readQueueNums", "8");
+ assertField(output, "writeQueueNums", "4");
+ assertField(output, "perm", "RW- (6)");
+ assertField(output, "topicFilterType", "MULTI_TAG");
+ assertField(output, "topicSysFlag", "3");
+ assertField(output, "order", "true");
+ Assert.assertTrue(output.contains("message.type=LITE"));
+ Assert.assertTrue(output.contains("lite.topic.expiration=30"));
+ Assert.assertTrue(output.contains("mappingDetail"));
+ Assert.assertTrue(output.contains("epoch=7"));
+ }
+
+ @Test
+ public void testExecuteWithClusterQueriesEveryBroker() throws
SubCommandException {
+ ServerResponseMocker brokerA = startBroker("broker-a",
newConfig("TopicTest", 8, 4,
+ PermName.PERM_READ | PermName.PERM_WRITE));
+ ServerResponseMocker brokerB = startBroker("broker-b",
newConfig("TopicTest", 16, 2,
+ PermName.PERM_WRITE));
+
+ Map<String, Integer> brokerPorts = new HashMap<>();
+ brokerPorts.put("broker-a", brokerA.listenPort());
+ brokerPorts.put("broker-b", brokerB.listenPort());
+ ServerResponseMocker nameServer = startNameServer("DefaultCluster",
brokerPorts);
+
+ String output = execute("-t", "TopicTest", "-c", "DefaultCluster",
"-n",
+ "127.0.0.1:" + nameServer.listenPort());
+
+ Assert.assertTrue(output.contains("Master: 127.0.0.1:" +
brokerA.listenPort()));
+ Assert.assertTrue(output.contains("Master: 127.0.0.1:" +
brokerB.listenPort()));
+ assertField(output, "readQueueNums", "8");
+ assertField(output, "readQueueNums", "16");
+ assertField(output, "writeQueueNums", "4");
+ assertField(output, "writeQueueNums", "2");
+ }
+
+ @Test
+ public void testExecuteWithoutTargetEnumeratesAllBrokers() throws
SubCommandException {
+ ServerResponseMocker brokerA = startBroker("broker-a",
newConfig("TopicTest", 8, 4,
+ PermName.PERM_READ | PermName.PERM_WRITE));
+ ServerResponseMocker brokerB = startBroker("broker-b",
newConfig("TopicTest", 16, 2,
+ PermName.PERM_READ));
+
+ Map<String, Integer> brokerPorts = new HashMap<>();
+ brokerPorts.put("broker-a", brokerA.listenPort());
+ brokerPorts.put("broker-b", brokerB.listenPort());
+ ServerResponseMocker nameServer = startNameServer("DefaultCluster",
brokerPorts);
+
+ String output = execute("-t", "TopicTest", "-n",
+ "127.0.0.1:" + nameServer.listenPort());
+
+ Assert.assertTrue(output.contains("broker-a:127.0.0.1:" +
brokerA.listenPort()));
+ Assert.assertTrue(output.contains("broker-b:127.0.0.1:" +
brokerB.listenPort()));
+ assertField(output, "readQueueNums", "8");
+ assertField(output, "readQueueNums", "16");
+ }
+
+ @Test
+ public void testBrokerWithoutTopicConfigIsSkipped() throws
SubCommandException {
+ ServerResponseMocker broker = startBroker("broker-a", null);
+
+ String output = execute("-t", "MissingTopic", "-b",
+ "127.0.0.1:" + broker.listenPort());
+
+ Assert.assertFalse(output.contains("topicName"));
+ Assert.assertFalse(output.contains("readQueueNums"));
+ Assert.assertFalse(output.contains("============"));
+ }
+
+ private String execute(String... subargs) throws SubCommandException {
+ GetTopicConfigSubCommand command = new GetTopicConfigSubCommand();
+ Options options = ServerUtil.buildCommandlineOptions(new Options());
+ CommandLine commandLine = ServerUtil.parseCmdLine("mqadmin " +
command.commandName(),
+ subargs, command.buildCommandlineOptions(options), new
DefaultParser());
+ command.execute(commandLine, options, null);
+ return outContent.toString();
+ }
+
+ private TopicConfig newConfig(String topic, int readQueueNums, int
writeQueueNums, int perm) {
+ TopicConfig config = new TopicConfig(topic, readQueueNums,
writeQueueNums, perm);
+ config.setTopicFilterType(TopicFilterType.MULTI_TAG);
+ config.setTopicSysFlag(3);
+ config.setOrder(true);
+ return config;
+ }
+
+ private ServerResponseMocker startBroker(String brokerName, TopicConfig
config) {
+ byte[] body = null;
+ if (config != null) {
+ TopicQueueMappingDetail mappingDetail = new
TopicQueueMappingDetail(
+ config.getTopicName(), config.getReadQueueNums(), brokerName,
7L);
+ body = RemotingSerializable.encode(new
TopicConfigAndQueueMapping(config, mappingDetail));
+ }
+
+ ServerResponseMocker mocker = ServerResponseMocker.startServer(body);
+ mockers.add(mocker);
+ return mocker;
+ }
+
+ private ServerResponseMocker startNameServer(String clusterName,
+ Map<String, Integer> brokerPorts) {
+ ClusterInfo clusterInfo = new ClusterInfo();
+ HashMap<String, BrokerData> brokerAddrTable = new HashMap<>();
+ Set<String> brokerNames = new HashSet<>();
+
+ for (Map.Entry<String, Integer> entry : brokerPorts.entrySet()) {
+ HashMap<Long, String> brokerAddrs = new HashMap<>();
+ brokerAddrs.put(MixAll.MASTER_ID, "127.0.0.1:" + entry.getValue());
+
+ BrokerData brokerData = new BrokerData();
+ brokerData.setBrokerName(entry.getKey());
+ brokerData.setCluster(clusterName);
+ brokerData.setBrokerAddrs(brokerAddrs);
+ brokerAddrTable.put(entry.getKey(), brokerData);
+ brokerNames.add(entry.getKey());
+ }
+
+ HashMap<String, Set<String>> clusterAddrTable = new HashMap<>();
+ clusterAddrTable.put(clusterName, brokerNames);
+ clusterInfo.setBrokerAddrTable(brokerAddrTable);
+ clusterInfo.setClusterAddrTable(clusterAddrTable);
+
+ ServerResponseMocker mocker =
ServerResponseMocker.startServer(clusterInfo.encode());
+ mockers.add(mocker);
+ return mocker;
+ }
+
+ private void assertField(String output, String name, String expected) {
+ String expectedLine = String.format("%-40s= %s", name, expected);
+ Assert.assertTrue("Expected field line [" + expectedLine + "] in
output:\n" + output,
+ output.contains(expectedLine));
+ }
+}
