This is an automated email from the ASF dual-hosted git repository.
dongeforever pushed a commit to branch 5.0.0-alpha-static-topic
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/5.0.0-alpha-static-topic by
this push:
new a9addc3 Add tests for command
a9addc3 is described below
commit a9addc3c293d878112ae3eb45fa00f75adf1f6c3
Author: dongeforever <[email protected]>
AuthorDate: Wed Dec 8 20:12:04 2021 +0800
Add tests for command
---
.../broker/processor/AdminBrokerProcessor.java | 2 +-
.../topic/TopicQueueMappingCleanService.java | 3 +-
.../rocketmq/client/impl/MQClientAPIImpl.java | 2 +-
.../header/GetTopicConfigRequestHeader.java | 12 +---
.../rocketmq/test/util/MQAdminTestUtils.java | 74 ++++++++++++++++++++++
.../rocketmq/test/statictopic/StaticTopicIT.java | 67 +++++++++++++++++---
.../tools/admin/DefaultMQAdminExtImpl.java | 1 +
.../topic/RemappingStaticTopicSubCommand.java | 8 ++-
.../command/topic/UpdateStaticTopicSubCommand.java | 3 +-
9 files changed, 147 insertions(+), 25 deletions(-)
diff --git
a/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
b/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
index fb4a711..e4f3697 100644
---
a/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
+++
b/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
@@ -1996,7 +1996,7 @@ public class AdminBrokerProcessor extends
AsyncNettyRequestProcessor implements
return response;
}
TopicQueueMappingDetail topicQueueMappingDetail = null;
- if (Boolean.TRUE.equals(requestHeader.getWithMapping())) {
+ if (Boolean.TRUE.equals(requestHeader.getLo())) {
topicQueueMappingDetail =
this.brokerController.getTopicQueueMappingManager().getTopicQueueMapping(requestHeader.getTopic());
}
String content = JSONObject.toJSONString(new
TopicConfigAndQueueMapping(topicConfig, topicQueueMappingDetail));
diff --git
a/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicQueueMappingCleanService.java
b/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicQueueMappingCleanService.java
index 05c5003..8374903 100644
---
a/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicQueueMappingCleanService.java
+++
b/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicQueueMappingCleanService.java
@@ -134,6 +134,7 @@ public class TopicQueueMappingCleanService extends
ServiceThread {
GetTopicStatsInfoRequestHeader header = new
GetTopicStatsInfoRequestHeader();
header.setTopic(topic);
header.setBname(broker);
+ header.setLo(false);
try {
RpcRequest rpcRequest = new
RpcRequest(RequestCode.GET_TOPIC_STATS_INFO, header, null);
RpcResponse rpcResponse =
rpcClient.invoke(rpcRequest, brokerConfig.getForwardTimeout()).get();
@@ -261,7 +262,7 @@ public class TopicQueueMappingCleanService extends
ServiceThread {
GetTopicConfigRequestHeader header = new
GetTopicConfigRequestHeader();
header.setTopic(topic);
header.setBname(broker);
- header.setWithMapping(true);
+ header.setLo(true);
try {
RpcRequest rpcRequest = new
RpcRequest(RequestCode.GET_TOPIC_CONFIG, header, null);
RpcResponse rpcResponse =
rpcClient.invoke(rpcRequest, brokerConfig.getForwardTimeout()).get();
diff --git
a/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
b/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
index 7538866..1de4fdb 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
@@ -2581,7 +2581,7 @@ public class MQClientAPIImpl {
RemotingTimeoutException, RemotingSendRequestException,
RemotingConnectException, MQBrokerException {
GetTopicConfigRequestHeader header = new GetTopicConfigRequestHeader();
header.setTopic(topic);
- header.setWithMapping(true);
+ header.setLo(true);
RemotingCommand request =
RemotingCommand.createRequestCommand(RequestCode.GET_TOPIC_CONFIG, header);
RemotingCommand response = this.remotingClient
.invokeSync(MixAll.brokerVIPChannel(this.clientConfig.isVipChannelEnabled(),
brokerAddr), request, timeoutMillis);
diff --git
a/common/src/main/java/org/apache/rocketmq/common/protocol/header/GetTopicConfigRequestHeader.java
b/common/src/main/java/org/apache/rocketmq/common/protocol/header/GetTopicConfigRequestHeader.java
index b282efa..b701df6 100644
---
a/common/src/main/java/org/apache/rocketmq/common/protocol/header/GetTopicConfigRequestHeader.java
+++
b/common/src/main/java/org/apache/rocketmq/common/protocol/header/GetTopicConfigRequestHeader.java
@@ -18,11 +18,12 @@
package org.apache.rocketmq.common.protocol.header;
import org.apache.rocketmq.common.rpc.RpcRequestHeader;
+import org.apache.rocketmq.common.rpc.TopicRequestHeader;
import org.apache.rocketmq.remoting.CommandCustomHeader;
import org.apache.rocketmq.remoting.annotation.CFNotNull;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
-public class GetTopicConfigRequestHeader extends RpcRequestHeader {
+public class GetTopicConfigRequestHeader extends TopicRequestHeader {
@Override
public void checkFields() throws RemotingCommandException {
}
@@ -30,7 +31,6 @@ public class GetTopicConfigRequestHeader extends
RpcRequestHeader {
@CFNotNull
private String topic;
- private Boolean withMapping;
/**
* @return the topic
@@ -45,12 +45,4 @@ public class GetTopicConfigRequestHeader extends
RpcRequestHeader {
public void setTopic(String topic) {
this.topic = topic;
}
-
- public Boolean getWithMapping() {
- return withMapping;
- }
-
- public void setWithMapping(Boolean withMapping) {
- this.withMapping = withMapping;
- }
}
\ No newline at end of file
diff --git
a/test/src/main/java/org/apache/rocketmq/test/util/MQAdminTestUtils.java
b/test/src/main/java/org/apache/rocketmq/test/util/MQAdminTestUtils.java
index 7eaf6a0..6784b76 100644
--- a/test/src/main/java/org/apache/rocketmq/test/util/MQAdminTestUtils.java
+++ b/test/src/main/java/org/apache/rocketmq/test/util/MQAdminTestUtils.java
@@ -22,9 +22,16 @@ import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ForkJoinPool;
+import java.util.stream.Collectors;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.PosixParser;
import org.apache.log4j.Logger;
+import org.apache.rocketmq.acl.common.AclUtils;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.admin.TopicStatsTable;
import org.apache.rocketmq.common.protocol.body.ClusterInfo;
import org.apache.rocketmq.common.protocol.route.BrokerData;
@@ -34,9 +41,16 @@ import
org.apache.rocketmq.common.statictopic.TopicQueueMappingUtils;
import org.apache.rocketmq.common.statictopic.TopicRemappingDetailWrapper;
import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.remoting.exception.RemotingException;
+import org.apache.rocketmq.srvutil.ServerUtil;
+import org.apache.rocketmq.test.client.rmq.RMQNormalConsumer;
+import org.apache.rocketmq.test.client.rmq.RMQNormalProducer;
+import org.apache.rocketmq.test.listener.rmq.concurrent.RMQNormalListener;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.apache.rocketmq.tools.admin.MQAdminUtils;
import org.apache.rocketmq.tools.command.CommandUtil;
+import org.apache.rocketmq.tools.command.MQAdminStartup;
+import org.apache.rocketmq.tools.command.topic.RemappingStaticTopicSubCommand;
+import org.apache.rocketmq.tools.command.topic.UpdateStaticTopicSubCommand;
public class MQAdminTestUtils {
private static Logger log = Logger.getLogger(MQAdminTestUtils.class);
@@ -231,5 +245,65 @@ public class MQAdminTestUtils {
}
+ public static void createStaticTopicWithCommand(String topic, int
queueNum, Set<String> brokers, String cluster, String nameservers) throws
Exception {
+ UpdateStaticTopicSubCommand cmd = new UpdateStaticTopicSubCommand();
+ Options options = ServerUtil.buildCommandlineOptions(new Options());
+ String[] args;
+ if (cluster != null) {
+ args = new String[] {
+ "-c", cluster,
+ "-t", topic,
+ "-qn", String.valueOf(queueNum),
+ "-n", nameservers
+ };
+ } else {
+ String brokerStr = String.join(",", brokers);
+ args = new String[] {
+ "-b", brokerStr,
+ "-t", topic,
+ "-qn", String.valueOf(queueNum),
+ "-n", nameservers
+ };
+ }
+ final CommandLine commandLine = ServerUtil.parseCmdLine("mqadmin " +
cmd.commandName(), args, cmd.buildCommandlineOptions(options), new
PosixParser());
+ if (null == commandLine) {
+ return;
+ }
+ if (commandLine.hasOption('n')) {
+ String namesrvAddr = commandLine.getOptionValue('n');
+ System.setProperty(MixAll.NAMESRV_ADDR_PROPERTY, namesrvAddr);
+ }
+ cmd.execute(commandLine, options, null);
+ }
+
+
+ public static void remappingStaticTopicWithCommand(String topic,
Set<String> brokers, String cluster, String nameservers) throws Exception {
+ RemappingStaticTopicSubCommand cmd = new
RemappingStaticTopicSubCommand();
+ Options options = ServerUtil.buildCommandlineOptions(new Options());
+ String[] args;
+ if (cluster != null) {
+ args = new String[] {
+ "-c", cluster,
+ "-t", topic,
+ "-n", nameservers
+ };
+ } else {
+ String brokerStr = String.join(",", brokers);
+ args = new String[] {
+ "-b", brokerStr,
+ "-t", topic,
+ "-n", nameservers
+ };
+ }
+ final CommandLine commandLine = ServerUtil.parseCmdLine("mqadmin " +
cmd.commandName(), args, cmd.buildCommandlineOptions(options), new
PosixParser());
+ if (null == commandLine) {
+ return;
+ }
+ if (commandLine.hasOption('n')) {
+ String namesrvAddr = commandLine.getOptionValue('n');
+ System.setProperty(MixAll.NAMESRV_ADDR_PROPERTY, namesrvAddr);
+ }
+ cmd.execute(commandLine, options, null);
+ }
}
diff --git
a/test/src/test/java/org/apache/rocketmq/test/statictopic/StaticTopicIT.java
b/test/src/test/java/org/apache/rocketmq/test/statictopic/StaticTopicIT.java
index e89dc3a..38811f6 100644
--- a/test/src/test/java/org/apache/rocketmq/test/statictopic/StaticTopicIT.java
+++ b/test/src/test/java/org/apache/rocketmq/test/statictopic/StaticTopicIT.java
@@ -2,7 +2,11 @@ package org.apache.rocketmq.test.statictopic;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.PosixParser;
import org.apache.log4j.Logger;
+import org.apache.rocketmq.acl.common.AclUtils;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.admin.ConsumeStats;
@@ -14,6 +18,7 @@ import
org.apache.rocketmq.common.statictopic.LogicQueueMappingItem;
import org.apache.rocketmq.common.statictopic.TopicConfigAndQueueMapping;
import org.apache.rocketmq.common.statictopic.TopicQueueMappingOne;
import org.apache.rocketmq.common.statictopic.TopicQueueMappingUtils;
+import org.apache.rocketmq.srvutil.ServerUtil;
import org.apache.rocketmq.test.base.BaseConf;
import org.apache.rocketmq.test.client.rmq.RMQNormalConsumer;
import org.apache.rocketmq.test.client.rmq.RMQNormalProducer;
@@ -23,10 +28,12 @@ import org.apache.rocketmq.test.util.MQRandomUtils;
import org.apache.rocketmq.test.util.VerifyUtils;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.apache.rocketmq.tools.admin.MQAdminUtils;
+import org.apache.rocketmq.tools.command.MQAdminStartup;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.FixMethodOrder;
+import org.junit.Ignore;
import org.junit.Test;
import java.util.ArrayList;
@@ -56,6 +63,51 @@ public class StaticTopicIT extends BaseConf {
defaultMQAdminExt.start();
}
+ @Test
+ public void testCommandsWithCluster() throws Exception {
+ //This case is used to mock the env to test the command manually
+ String topic = "static" + MQRandomUtils.getRandomTopic();
+ RMQNormalProducer producer = getProducer(nsAddr, topic);
+ RMQNormalConsumer consumer = getConsumer(nsAddr, topic, "*", new
RMQNormalListener());
+ int queueNum = 10;
+ int msgEachQueue = 100;
+
+ {
+ MQAdminTestUtils.createStaticTopicWithCommand(topic, queueNum,
null, clusterName, nsAddr);
+ sendMessagesAndCheck(producer, getBrokers(), topic, queueNum,
msgEachQueue, 0);
+ //consume and check
+ consumeMessagesAndCheck(producer, consumer, topic, queueNum,
msgEachQueue, 0, 1);
+ }
+ {
+ MQAdminTestUtils.remappingStaticTopicWithCommand(topic, null,
clusterName, nsAddr);
+ Thread.sleep(500);
+ sendMessagesAndCheck(producer, getBrokers(), topic, queueNum,
msgEachQueue, 100);
+ }
+ }
+
+ @Test
+ public void testCommandsWithBrokers() throws Exception {
+ //This case is used to mock the env to test the command manually
+ String topic = "static" + MQRandomUtils.getRandomTopic();
+ RMQNormalProducer producer = getProducer(nsAddr, topic);
+ RMQNormalConsumer consumer = getConsumer(nsAddr, topic, "*", new
RMQNormalListener());
+ int queueNum = 10;
+ int msgEachQueue = 100;
+ {
+ Set<String> brokers = ImmutableSet.of(broker1Name);
+ MQAdminTestUtils.createStaticTopicWithCommand(topic, queueNum,
brokers, null, nsAddr);
+ sendMessagesAndCheck(producer, brokers, topic, queueNum,
msgEachQueue, 0);
+ //consume and check
+ consumeMessagesAndCheck(producer, consumer, topic, queueNum,
msgEachQueue, 0, 1);
+ }
+ {
+ Set<String> brokers = ImmutableSet.of(broker2Name);
+ MQAdminTestUtils.remappingStaticTopicWithCommand(topic, brokers,
null, nsAddr);
+ Thread.sleep(500);
+ sendMessagesAndCheck(producer, brokers, topic, queueNum,
msgEachQueue, TopicQueueMappingUtils.DEFAULT_BLOCK_SEQ_SIZE);
+ consumeMessagesAndCheck(producer, consumer, topic, queueNum,
msgEachQueue, 0, 2);
+ }
+ }
@Test
public void testNoTargetBrokers() throws Exception {
@@ -89,7 +141,7 @@ public class StaticTopicIT extends BaseConf {
}
- private void sendMessagesAndCheck(RMQNormalProducer producer, Set<String>
targetBrokers, String topic, int queueNum, int msgEachQueue, int gen) throws
Exception {
+ private void sendMessagesAndCheck(RMQNormalProducer producer, Set<String>
targetBrokers, String topic, int queueNum, int msgEachQueue, long baseOffset)
throws Exception {
ClientMetadata clientMetadata =
MQAdminUtils.getBrokerAndTopicMetadata(topic, defaultMQAdminExt);
List<MessageQueue> messageQueueList = producer.getMessageQueue();
Assert.assertEquals(queueNum, messageQueueList.size());
@@ -109,12 +161,12 @@ public class StaticTopicIT extends BaseConf {
Thread.sleep(100);
for(MessageQueue messageQueue: messageQueueList) {
Assert.assertEquals(0, defaultMQAdminExt.minOffset(messageQueue));
- Assert.assertEquals(msgEachQueue + gen *
TopicQueueMappingUtils.DEFAULT_BLOCK_SEQ_SIZE,
defaultMQAdminExt.maxOffset(messageQueue));
+ Assert.assertEquals(msgEachQueue + baseOffset,
defaultMQAdminExt.maxOffset(messageQueue));
}
TopicStatsTable topicStatsTable =
defaultMQAdminExt.examineTopicStats(topic);
for(MessageQueue messageQueue: messageQueueList) {
Assert.assertEquals(0,
topicStatsTable.getOffsetTable().get(messageQueue).getMinOffset());
- Assert.assertEquals(msgEachQueue + gen *
TopicQueueMappingUtils.DEFAULT_BLOCK_SEQ_SIZE,
topicStatsTable.getOffsetTable().get(messageQueue).getMaxOffset());
+ Assert.assertEquals(msgEachQueue + baseOffset,
topicStatsTable.getOffsetTable().get(messageQueue).getMaxOffset());
}
}
@@ -214,7 +266,6 @@ public class StaticTopicIT extends BaseConf {
sendMessagesAndCheck(producer, targetBrokers, topic, queueNum,
msgEachQueue, 0);
consumeMessagesAndCheck(producer, consumer, topic, queueNum,
msgEachQueue, 0, 1);
}
-
System.out.println("=============================================================");
//remapping the static topic
{
Set<String> targetBrokers = ImmutableSet.of(broker2Name);
@@ -228,7 +279,7 @@ public class StaticTopicIT extends BaseConf {
Assert.assertEquals(TopicQueueMappingUtils.DEFAULT_BLOCK_SEQ_SIZE,
mappingOne.getItems().get(mappingOne.getItems().size() - 1).getLogicOffset());
}
Thread.sleep(500);
- sendMessagesAndCheck(producer, targetBrokers, topic, queueNum,
msgEachQueue, 1);
+ sendMessagesAndCheck(producer, targetBrokers, topic, queueNum,
msgEachQueue, TopicQueueMappingUtils.DEFAULT_BLOCK_SEQ_SIZE);
consumeMessagesAndCheck(producer, consumer, topic, queueNum,
msgEachQueue, 0, 2);
}
}
@@ -261,7 +312,7 @@ public class StaticTopicIT extends BaseConf {
MQAdminTestUtils.remappingStaticTopic(topic, targetBrokers,
defaultMQAdminExt);
//make the metadata
Thread.sleep(500);
- sendMessagesAndCheck(producer, targetBrokers, topic, queueNum,
msgEachQueue, i + 1);
+ sendMessagesAndCheck(producer, targetBrokers, topic, queueNum,
msgEachQueue, (i + 1) * TopicQueueMappingUtils.DEFAULT_BLOCK_SEQ_SIZE);
}
consumer = getConsumer(nsAddr, group, topic, "*", new
RMQNormalListener());
consumeMessagesAndCheck(producer, consumer, topic, queueNum,
msgEachQueue, 1, brokers.size());
@@ -289,7 +340,7 @@ public class StaticTopicIT extends BaseConf {
MQAdminTestUtils.remappingStaticTopic(topic, targetBrokers,
defaultMQAdminExt);
//leave the time to refresh the metadata
Thread.sleep(500);
- sendMessagesAndCheck(producer, targetBrokers, topic, queueNum,
msgEachQueue, 1);
+ sendMessagesAndCheck(producer, targetBrokers, topic, queueNum,
msgEachQueue, 1 * TopicQueueMappingUtils.DEFAULT_BLOCK_SEQ_SIZE);
}
//remapping to broker3Name
@@ -298,7 +349,7 @@ public class StaticTopicIT extends BaseConf {
MQAdminTestUtils.remappingStaticTopic(topic, targetBrokers,
defaultMQAdminExt);
//leave the time to refresh the metadata
Thread.sleep(500);
- sendMessagesAndCheck(producer, targetBrokers, topic, queueNum,
msgEachQueue, 2);
+ sendMessagesAndCheck(producer, targetBrokers, topic, queueNum,
msgEachQueue, 2 * TopicQueueMappingUtils.DEFAULT_BLOCK_SEQ_SIZE);
}
// 1 -> 2 -> 3, currently 1 should not has any mappings
diff --git
a/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java
b/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java
index 7b0450c..d1f0fcd 100644
---
a/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java
+++
b/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java
@@ -268,6 +268,7 @@ public class DefaultMQAdminExtImpl implements MQAdminExt,
MQAdminExtInner {
}
}
+
if (topicStatsTable.getOffsetTable().isEmpty()) {
throw new MQClientException("Not found the topic stats info",
null);
}
diff --git
a/tools/src/main/java/org/apache/rocketmq/tools/command/topic/RemappingStaticTopicSubCommand.java
b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/RemappingStaticTopicSubCommand.java
index 9a6f15a..a3b757a 100644
---
a/tools/src/main/java/org/apache/rocketmq/tools/command/topic/RemappingStaticTopicSubCommand.java
+++
b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/RemappingStaticTopicSubCommand.java
@@ -20,13 +20,12 @@ 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.MQAdmin;
import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.protocol.body.ClusterInfo;
+import org.apache.rocketmq.common.rpc.ClientMetadata;
import org.apache.rocketmq.common.statictopic.TopicConfigAndQueueMapping;
import org.apache.rocketmq.common.statictopic.TopicQueueMappingOne;
import org.apache.rocketmq.common.statictopic.TopicQueueMappingUtils;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
-import org.apache.rocketmq.common.rpc.ClientMetadata;
import org.apache.rocketmq.common.statictopic.TopicRemappingDetailWrapper;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.srvutil.ServerUtil;
@@ -91,6 +90,7 @@ public class RemappingStaticTopicSubCommand implements
SubCommand {
defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
try {
+ defaultMQAdminExt.start();
String topic = commandLine.getOptionValue('t').trim();
String mapFileName = commandLine.getOptionValue('f').trim();
@@ -137,6 +137,7 @@ public class RemappingStaticTopicSubCommand implements
SubCommand {
Set<String> targetBrokers = new HashSet<>();
try {
+ defaultMQAdminExt.start();
if ((!commandLine.hasOption("b") && !commandLine.hasOption('c'))) {
ServerUtil.printCommandLineHelp("mqadmin " +
this.commandName(), options);
return;
@@ -148,6 +149,7 @@ public class RemappingStaticTopicSubCommand implements
SubCommand {
|| clusterInfo.getClusterAddrTable().isEmpty()) {
throw new RuntimeException("The Cluster info is empty");
}
+ clientMetadata.refreshClusterInfo(clusterInfo);
{
if (commandLine.hasOption("b")) {
String brokerStrs = commandLine.getOptionValue("b").trim();
diff --git
a/tools/src/main/java/org/apache/rocketmq/tools/command/topic/UpdateStaticTopicSubCommand.java
b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/UpdateStaticTopicSubCommand.java
index 958922e..410381e 100644
---
a/tools/src/main/java/org/apache/rocketmq/tools/command/topic/UpdateStaticTopicSubCommand.java
+++
b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/UpdateStaticTopicSubCommand.java
@@ -92,9 +92,9 @@ public class UpdateStaticTopicSubCommand implements
SubCommand {
RPCHook rpcHook) throws SubCommandException {
DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
- ClientMetadata clientMetadata = new ClientMetadata();
try {
+ defaultMQAdminExt.start();
String topic = commandLine.getOptionValue('t').trim();
String mapFileName = commandLine.getOptionValue('f').trim();
String mapData = MixAll.file2String(mapFileName);
@@ -139,6 +139,7 @@ public class UpdateStaticTopicSubCommand implements
SubCommand {
Set<String> targetBrokers = new HashSet<>();
try {
+ defaultMQAdminExt.start();
if ((!commandLine.hasOption("b") && !commandLine.hasOption('c'))
|| !commandLine.hasOption("qn")) {
ServerUtil.printCommandLineHelp("mqadmin " +
this.commandName(), options);
