broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java

@@ -350,7 +350,7 @@ public class BrokerOuterAPI {
         return changedList;
     }
 
-    public TopicConfigSerializeWrapper getAllTopicConfig(
+    public TopicConfigAndMappingSerializeWrapper getAllTopicConfig(
         final String addr) throws RemotingConnectException, RemotingSendRequestException,
         RemotingTimeoutException, InterruptedException, MQBrokerException {
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_TOPIC_CONFIG, null);
@@ -359,7 +359,7 @@ public class BrokerOuterAPI {
         assert response != null;
         switch (response.getCode()) {
             case ResponseCode.SUCCESS: {
-                return TopicConfigSerializeWrapper.decode(response.getBody(), TopicConfigSerializeWrapper.class);
+                return TopicConfigSerializeWrapper.decode(response.getBody(), TopicConfigAndMappingSerializeWrapper.class);
             }
             default:
                 break;

broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java

@@ -20,6 +20,7 @@ import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONObject;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.util.concurrent.CompleteFuture;
 import org.apache.rocketmq.acl.AccessValidator;
 import org.apache.rocketmq.acl.plain.PlainAccessValidator;
 import org.apache.rocketmq.broker.BrokerController;
@@ -29,6 +30,7 @@ import org.apache.rocketmq.broker.filter.ConsumerFilterData;
 import org.apache.rocketmq.broker.filter.ExpressionMessageFilter;
 import org.apache.rocketmq.broker.transaction.queue.TransactionalMessageUtil;
 import org.apache.rocketmq.common.AclConfig;
+import org.apache.rocketmq.common.BrokerConfig;
 import org.apache.rocketmq.common.MQVersion;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.PlainAccessConfig;
@@ -62,6 +64,7 @@ import org.apache.rocketmq.common.protocol.body.QueryConsumeQueueResponseBody;
 import org.apache.rocketmq.common.protocol.body.QueryConsumeTimeSpanBody;
 import org.apache.rocketmq.common.protocol.body.QueryCorrectionOffsetBody;
 import org.apache.rocketmq.common.protocol.body.QueueTimeSpan;
+import org.apache.rocketmq.common.protocol.body.TopicConfigAndMappingSerializeWrapper;
 import org.apache.rocketmq.common.protocol.body.TopicList;
 import org.apache.rocketmq.common.protocol.body.UnlockBatchRequestBody;
 import org.apache.rocketmq.common.protocol.header.CloneGroupOffsetRequestHeader;
@@ -103,6 +106,9 @@ import org.apache.rocketmq.common.protocol.header.ViewBrokerStatsDataRequestHead
 import org.apache.rocketmq.common.protocol.header.filtersrv.RegisterFilterServerRequestHeader;
 import org.apache.rocketmq.common.protocol.header.filtersrv.RegisterFilterServerResponseHeader;
 import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
+import org.apache.rocketmq.common.rpc.RpcClient;
+import org.apache.rocketmq.common.rpc.RpcClientUtils;
+import org.apache.rocketmq.common.rpc.RpcException;
 import org.apache.rocketmq.common.rpc.RpcRequest;
 import org.apache.rocketmq.common.rpc.RpcResponse;
 import org.apache.rocketmq.common.statictopic.LogicQueueMappingItem;
@@ -146,16 +152,23 @@ import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Future;
+import java.util.function.BiConsumer;
 
 import static org.apache.rocketmq.remoting.protocol.RemotingCommand.buildErrorResponse;
 
 public class AdminBrokerProcessor extends AsyncNettyRequestProcessor implements NettyRequestProcessor {
     private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
     private final BrokerController brokerController;
+    private final RpcClient rpcClient;
+    private final BrokerConfig brokerConfig;
 
     public AdminBrokerProcessor(final BrokerController brokerController) {
         this.brokerController = brokerController;
+        this.brokerConfig = brokerController.getBrokerConfig();
+        this.rpcClient = brokerController.getBrokerOuterAPI().getRpcClient();
     }
 
     @Override
@@ -533,7 +546,16 @@ public class AdminBrokerProcessor extends AsyncNettyRequestProcessor implements
         // final GetAllTopicConfigResponseHeader responseHeader =
         // (GetAllTopicConfigResponseHeader) response.readCustomHeader();
 
-        String content = this.brokerController.getTopicConfigManager().encode();
+        TopicConfigAndMappingSerializeWrapper topicConfigAndMappingSerializeWrapper = new TopicConfigAndMappingSerializeWrapper();
+
+        topicConfigAndMappingSerializeWrapper.setDataVersion(this.brokerController.getTopicConfigManager().getDataVersion());
+        topicConfigAndMappingSerializeWrapper.setTopicConfigTable(this.brokerController.getTopicConfigManager().getTopicConfigTable());
+
+        topicConfigAndMappingSerializeWrapper.setMappingDataVersion(this.brokerController.getTopicQueueMappingManager().getDataVersion());
+        topicConfigAndMappingSerializeWrapper.setTopicQueueMappingDetailMap(this.brokerController.getTopicQueueMappingManager().getTopicQueueMappingTable());
+
+
+        String content = topicConfigAndMappingSerializeWrapper.toJson();
         if (content != null && content.length() > 0) {
             try {
                 response.setBody(content.getBytes(MixAll.DEFAULT_CHARSET));
@@ -644,7 +666,7 @@ public class AdminBrokerProcessor extends AsyncNettyRequestProcessor implements
                         break;
                     }
                 } else {
-                    requestHeader.setPhysical(true);
+                    requestHeader.setLo(false);
                     requestHeader.setTimestamp(timestamp);
                     requestHeader.setQueueId(item.getQueueId());
                     requestHeader.setBname(item.getBname());
@@ -714,7 +736,7 @@ public class AdminBrokerProcessor extends AsyncNettyRequestProcessor implements
             assert maxItem != null;
             assert maxItem.getLogicOffset() >= 0;
             requestHeader.setBname(maxItem.getBname());
-            requestHeader.setPhysical(true);
+            requestHeader.setLo(false);
             requestHeader.setQueueId(mappingItem.getQueueId());
 
             long maxPhysicalOffset = Long.MAX_VALUE;
@@ -764,22 +786,22 @@ public class AdminBrokerProcessor extends AsyncNettyRequestProcessor implements
         return response;
     }
 
-    private RemotingCommand rewriteRequestForStaticTopic(GetMinOffsetRequestHeader requestHeader, TopicQueueMappingContext mappingContext)  {
+    private CompletableFuture<RpcResponse> handleGetMinOffsetForStaticTopic(RpcRequest request, TopicQueueMappingContext mappingContext)  {
         if (mappingContext.getMappingDetail() == null) {
             return null;
         }
-
         TopicQueueMappingDetail mappingDetail = mappingContext.getMappingDetail();
         if (!mappingContext.isLeader()) {
             //this may not
-            return buildErrorResponse(ResponseCode.NOT_LEADER_FOR_QUEUE, String.format("%s-%d does not exit in request process of current broker %s", mappingContext.getTopic(), mappingContext.getGlobalId(), mappingDetail.getBname()));
+            return CompletableFuture.completedFuture(new RpcResponse(new RpcException(ResponseCode.NOT_LEADER_FOR_QUEUE,
+                    String.format("%s-%d is not leader in broker %s, request code %d", mappingContext.getTopic(), mappingContext.getGlobalId(), mappingDetail.getBname(), request.getCode()))));
         };
-
+        GetMinOffsetRequestHeader requestHeader = (GetMinOffsetRequestHeader) request.getHeader();
         LogicQueueMappingItem mappingItem = TopicQueueMappingUtils.findLogicQueueMappingItem(mappingContext.getMappingItemList(), 0L, true);
         assert  mappingItem != null;
         try {
             requestHeader.setBname(mappingItem.getBname());
-            requestHeader.setPhysical(true);
+            requestHeader.setLo(false);
             requestHeader.setQueueId(mappingItem.getQueueId());
             long physicalOffset;
             //run in local
@@ -796,38 +818,40 @@ public class AdminBrokerProcessor extends AsyncNettyRequestProcessor implements
             }
             long offset = mappingItem.computeStaticQueueOffsetLoosely(physicalOffset);
 
-            final RemotingCommand response = RemotingCommand.createResponseCommand(GetMinOffsetResponseHeader.class);
-            final GetMinOffsetResponseHeader responseHeader = (GetMinOffsetResponseHeader) response.readCustomHeader();
+            final GetMinOffsetResponseHeader responseHeader = new GetMinOffsetResponseHeader();
             responseHeader.setOffset(offset);
-            response.setCode(ResponseCode.SUCCESS);
-            response.setRemark(null);
-            return response;
+            return CompletableFuture.completedFuture(new RpcResponse(ResponseCode.SUCCESS, responseHeader, null));
         } catch (Throwable t) {
             log.error("rewriteRequestForStaticTopic failed", t);
-            return buildErrorResponse(ResponseCode.SYSTEM_ERROR, t.getMessage());
+            return CompletableFuture.completedFuture(new RpcResponse(new RpcException(ResponseCode.SYSTEM_ERROR, t.getMessage(), t)));
         }
     }
 
-    private RemotingCommand getMinOffset(ChannelHandlerContext ctx,
-        RemotingCommand request) throws RemotingCommandException {
-        final RemotingCommand response = RemotingCommand.createResponseCommand(GetMinOffsetResponseHeader.class);
-        final GetMinOffsetResponseHeader responseHeader = (GetMinOffsetResponseHeader) response.readCustomHeader();
-        final GetMinOffsetRequestHeader requestHeader =
-            (GetMinOffsetRequestHeader) request.decodeCommandCustomHeader(GetMinOffsetRequestHeader.class);
-
-
+    private CompletableFuture<RpcResponse>  handleGetMinOffset(RpcRequest request) {
+        assert request.getCode() == RequestCode.GET_MIN_OFFSET;
+        GetMinOffsetRequestHeader requestHeader = (GetMinOffsetRequestHeader) request.getHeader();
         TopicQueueMappingContext mappingContext = this.brokerController.getTopicQueueMappingManager().buildTopicQueueMappingContext(requestHeader, false);
-        RemotingCommand rewriteResult = rewriteRequestForStaticTopic(requestHeader, mappingContext);
+        CompletableFuture<RpcResponse> rewriteResult = handleGetMinOffsetForStaticTopic(request, mappingContext);
         if (rewriteResult != null) {
             return rewriteResult;
         }
-
+        final GetMinOffsetResponseHeader responseHeader = new GetMinOffsetResponseHeader();
         long offset = this.brokerController.getMessageStore().getMinOffsetInQueue(requestHeader.getTopic(), requestHeader.getQueueId());
-
         responseHeader.setOffset(offset);
-        response.setCode(ResponseCode.SUCCESS);
-        response.setRemark(null);
-        return response;
+        return CompletableFuture.completedFuture(new RpcResponse(ResponseCode.SUCCESS, responseHeader, null));
+    }
+
+    private RemotingCommand getMinOffset(ChannelHandlerContext ctx,
+        RemotingCommand request) throws RemotingCommandException {
+        final GetMinOffsetRequestHeader requestHeader =
+            (GetMinOffsetRequestHeader) request.decodeCommandCustomHeader(GetMinOffsetRequestHeader.class);
+        try {
+            CompletableFuture<RpcResponse> responseFuture = handleGetMinOffset(new RpcRequest(RequestCode.GET_MIN_OFFSET, requestHeader, null));
+            RpcResponse  rpcResponse = responseFuture.get();
+            return RpcClientUtils.createCommandForRpcResponse(rpcResponse);
+        } catch (Throwable t) {
+            return buildErrorResponse(ResponseCode.SYSTEM_ERROR, t.getMessage());
+        }
     }
 
     private RemotingCommand rewriteRequestForStaticTopic(GetEarliestMsgStoretimeRequestHeader requestHeader, TopicQueueMappingContext mappingContext)  {
@@ -843,7 +867,7 @@ public class AdminBrokerProcessor extends AsyncNettyRequestProcessor implements
         assert mappingItem != null;
         try {
             requestHeader.setBname(mappingItem.getBname());
-            requestHeader.setPhysical(true);
+            requestHeader.setLo(false);
             RpcRequest rpcRequest = new RpcRequest(RequestCode.GET_EARLIEST_MSG_STORETIME, requestHeader, null);
             //TODO check if it is in current broker
             RpcResponse rpcResponse = this.brokerController.getBrokerOuterAPI().getRpcClient().invoke(rpcRequest, this.brokerController.getBrokerConfig().getForwardTimeout()).get();
@@ -998,6 +1022,70 @@ public class AdminBrokerProcessor extends AsyncNettyRequestProcessor implements
         return response;
     }
 
+    private RpcResponse handleGetTopicStatsInfoForStaticTopic(RpcRequest request, TopicQueueMappingContext mappingContext) {
+        try {
+            assert request.getCode() == RequestCode.GET_TOPIC_STATS_INFO;
+            if (mappingContext.getMappingDetail() == null) {
+                return null;
+            }
+            final GetTopicStatsInfoRequestHeader requestHeader = (GetTopicStatsInfoRequestHeader) request.getHeader();
+            String topic = requestHeader.getTopic();
+            TopicQueueMappingDetail mappingDetail = mappingContext.getMappingDetail();
+            Map<Integer, LogicQueueMappingItem[]> qidItemMap = new HashMap<>();
+            Set<String> brokers = new HashSet<>();
+            mappingDetail.getHostedQueues().forEach((qid, items) -> {
+                if (TopicQueueMappingUtils.checkIfLeader(items, mappingDetail)) {
+                    LogicQueueMappingItem[] itemPair = new LogicQueueMappingItem[2];
+                    itemPair[0] = TopicQueueMappingUtils.findLogicQueueMappingItem(items, 0, true);
+                    itemPair[1] = TopicQueueMappingUtils.findLogicQueueMappingItem(items, Long.MAX_VALUE, true);
+                    assert itemPair[0] != null && itemPair[1] != null;
+                    qidItemMap.put(qid, itemPair);
+                    brokers.add(itemPair[0].getBname());
+                    brokers.add(itemPair[1].getBname());
+                }
+            });
+            Map<String, TopicStatsTable> statsTable = new HashMap<>();
+            for (String broker: brokers) {
+                GetTopicStatsInfoRequestHeader header = new GetTopicStatsInfoRequestHeader();
+                header.setTopic(topic);
+                header.setBname(broker);
+                header.setLo(false);
+                RpcRequest rpcRequest = new RpcRequest(RequestCode.GET_TOPIC_STATS_INFO, header, null);
+                RpcResponse rpcResponse = rpcClient.invoke(rpcRequest, brokerConfig.getForwardTimeout()).get();
+                if (rpcResponse.getException() != null) {
+                    throw rpcResponse.getException();
+                }
+                statsTable.put(broker, (TopicStatsTable) rpcResponse.getBody());
+            }
+            TopicStatsTable topicStatsTable = new TopicStatsTable();
+            qidItemMap.forEach( (qid, itemPair) -> {
+                LogicQueueMappingItem minItem = itemPair[0];
+                LogicQueueMappingItem maxItem = itemPair[1];
+                TopicOffset minTopicOffset = statsTable.get(minItem.getBname()).getOffsetTable().get(new MessageQueue(topic, minItem.getBname(), minItem.getQueueId()));
+                TopicOffset maxTopicOffset = statsTable.get(maxItem.getBname()).getOffsetTable().get(new MessageQueue(topic, maxItem.getBname(), maxItem.getQueueId()));
+
+                assert  minTopicOffset != null && maxTopicOffset != null;
+
+                long min = minItem.computeStaticQueueOffsetLoosely(minTopicOffset.getMinOffset());
+                if (min < 0)
+                    min = 0;
+                long max = maxItem.computeStaticQueueOffsetStrictly(maxTopicOffset.getMaxOffset());
+                if (max < 0)
+                    max = 0;
+                long timestamp = maxTopicOffset.getLastUpdateTimestamp();
+
+                TopicOffset topicOffset = new TopicOffset();
+                topicOffset.setMinOffset(min);
+                topicOffset.setMaxOffset(max);
+                topicOffset.setLastUpdateTimestamp(timestamp);
+                topicStatsTable.getOffsetTable().put(new MessageQueue(topic, MixAll.LOGICAL_QUEUE_MOCK_BROKER_NAME, qid), topicOffset);
+            });
+            return new RpcResponse(ResponseCode.SUCCESS, null, topicStatsTable);
+        } catch (Throwable t) {
+            return new RpcResponse(new RpcException(ResponseCode.SYSTEM_ERROR, t.getMessage(), t));
+        }
+    }
+
     private RemotingCommand getTopicStatsInfo(ChannelHandlerContext ctx,
         RemotingCommand request) throws RemotingCommandException {
         final RemotingCommand response = RemotingCommand.createResponseCommand(null);
@@ -1011,8 +1099,17 @@ public class AdminBrokerProcessor extends AsyncNettyRequestProcessor implements
             response.setRemark("topic[" + topic + "] not exist");
             return response;
         }
-
         TopicStatsTable topicStatsTable = new TopicStatsTable();
+        TopicQueueMappingContext mappingContext = this.brokerController.getTopicQueueMappingManager().buildTopicQueueMappingContext(requestHeader, false);
+        RpcResponse rpcResponse = handleGetTopicStatsInfoForStaticTopic(new RpcRequest(RequestCode.GET_TOPIC_STATS_INFO, requestHeader, null), mappingContext);
+        if (rpcResponse != null) {
+            if (rpcResponse.getException() != null) {
+                return RpcClientUtils.createCommandForRpcResponse(rpcResponse);
+            } else {
+                topicStatsTable.getOffsetTable().putAll(((TopicStatsTable)rpcResponse.getBody()).getOffsetTable());
+            }
+        }
+
         for (int i = 0; i < topicConfig.getWriteQueueNums(); i++) {
             MessageQueue mq = new MessageQueue();
             mq.setTopic(topic);
@@ -1909,7 +2006,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));

broker/src/main/java/org/apache/rocketmq/broker/processor/ConsumerManageProcessor.java

@@ -153,11 +153,8 @@ public class ConsumerManageProcessor extends AsyncNettyRequestProcessor implemen
             //by default, it is -1
             long offset = -1;
             //double read, first from leader, then from second leader
-            for (int i = 1; i <= 2; i++) {
-                if (itemList.size() - i < 0) {
-                    break;
-                }
-                LogicQueueMappingItem mappingItem = itemList.get(itemList.size() - i);
+            for (int i = itemList.size() - 1; i >= 0; i--) {
+                LogicQueueMappingItem mappingItem = itemList.get(i);
                 if (mappingItem.getBname().equals(mappingDetail.getBname())) {
                     offset = this.brokerController.getConsumerOffsetManager().queryOffset(requestHeader.getConsumerGroup(), requestHeader.getTopic(), mappingItem.getQueueId());
                     if (offset >= 0) {
@@ -170,7 +167,7 @@ public class ConsumerManageProcessor extends AsyncNettyRequestProcessor implemen
                     //maybe we need to reconstruct an object
                     requestHeader.setBname(mappingItem.getBname());
                     requestHeader.setQueueId(mappingItem.getQueueId());
-                    requestHeader.setPhysical(true);
+                    requestHeader.setLo(false);
                     requestHeader.setSetZeroIfNotFound(false);
                     RpcRequest rpcRequest = new RpcRequest(RequestCode.QUERY_CONSUMER_OFFSET, requestHeader, null);
                     RpcResponse rpcResponse = this.brokerController.getBrokerOuterAPI().getRpcClient().invoke(rpcRequest, this.brokerController.getBrokerConfig().getForwardTimeout()).get();
@@ -179,7 +176,8 @@ public class ConsumerManageProcessor extends AsyncNettyRequestProcessor implemen
                     }
                     if (rpcResponse.getCode() == ResponseCode.SUCCESS) {
                         offset = ((QueryConsumerOffsetResponseHeader) rpcResponse.getHeader()).getOffset();
-                    } else if (rpcResponse.getCode() == ResponseCode.PULL_NOT_FOUND){
+                        break;
+                    } else if (rpcResponse.getCode() == ResponseCode.QUERY_NOT_FOUND){
                         continue;
                     } else {
                         //this should not happen

broker/src/main/java/org/apache/rocketmq/broker/processor/PullMessageProcessor.java

@@ -134,18 +134,14 @@ public class PullMessageProcessor extends AsyncNettyRequestProcessor implements
                     && requestHeader.getMaxMsgNums() != null) {
                 requestHeader.setMaxMsgNums((int) Math.min(mappingItem.getEndOffset() - mappingItem.getStartOffset(), requestHeader.getMaxMsgNums()));
             }
-            int sysFlag = requestHeader.getSysFlag();
-            if (!mappingContext.isLeader()) {
-                sysFlag = PullSysFlag.clearCommitOffsetFlag(sysFlag);
-                requestHeader.setSysFlag(sysFlag);
-            }
 
             if (mappingDetail.getBname().equals(bname)) {
                 //just let it go, do the local pull process
                 return null;
             }
 
-            requestHeader.setPhysical(true);
+            int sysFlag = requestHeader.getSysFlag();
+            requestHeader.setLo(false);
             requestHeader.setBname(bname);
             sysFlag = PullSysFlag.clearSuspendFlag(sysFlag);
             sysFlag = PullSysFlag.clearCommitOffsetFlag(sysFlag);
@@ -189,11 +185,13 @@ public class PullMessageProcessor extends AsyncNettyRequestProcessor implements
             long minOffset = responseHeader.getMinOffset();
             long maxOffset = responseHeader.getMaxOffset();
             int responseCode = code;
+
             //consider the following situations
             // 1. read from slave, currently not supported
             // 2. the middle queue is truncated because of deleting commitlog
             if (code != ResponseCode.SUCCESS) {
                 //note the currentItem maybe both the leader and  the earliest
+                boolean isRevised = false;
                 if (leaderItem.getGen() == currentItem.getGen()) {
                     //read the leader
                     if (requestOffset > maxOffset) {
@@ -228,6 +226,7 @@ public class PullMessageProcessor extends AsyncNettyRequestProcessor implements
                         //just move to another item
                         LogicQueueMappingItem nextItem = TopicQueueMappingUtils.findNext(mappingContext.getMappingItemList(), currentItem, true);
                         if (nextItem != null) {
+                            isRevised = true;
                             currentItem = nextItem;
                             nextBeginOffset = currentItem.getStartOffset();
                             minOffset = currentItem.getStartOffset();
@@ -244,7 +243,8 @@ public class PullMessageProcessor extends AsyncNettyRequestProcessor implements
                 }
 
                 //read from the middle item, ignore the PULL_OFFSET_MOVED
-                if (leaderItem.getGen() != currentItem.getGen()
+                if (!isRevised
+                    && leaderItem.getGen() != currentItem.getGen()
                     && earlistItem.getGen() != currentItem.getGen()) {
                     if (requestOffset < minOffset) {
                         nextBeginOffset = minOffset;
@@ -289,7 +289,6 @@ public class PullMessageProcessor extends AsyncNettyRequestProcessor implements
                 return null;
             }
         } catch (Throwable t) {
-            t.printStackTrace();
             return buildErrorResponse(ResponseCode.SYSTEM_ERROR, t.getMessage());
         }
     }
@@ -443,6 +442,7 @@ public class PullMessageProcessor extends AsyncNettyRequestProcessor implements
             return response;
         }
 
+
         MessageFilter messageFilter;
         if (this.brokerController.getBrokerConfig().isFilterSupportRetry()) {
             messageFilter = new ExpressionForRetryMessageFilter(subscriptionData, consumerFilterData,

broker/src/main/java/org/apache/rocketmq/broker/slave/SlaveSynchronize.java

@@ -21,6 +21,7 @@ import org.apache.rocketmq.broker.BrokerController;
 import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.protocol.body.TopicConfigAndMappingSerializeWrapper;
 import org.apache.rocketmq.logging.InternalLogger;
 import org.apache.rocketmq.logging.InternalLoggerFactory;
 import org.apache.rocketmq.common.protocol.body.ConsumerOffsetSerializeWrapper;
@@ -56,7 +57,7 @@ public class SlaveSynchronize {
         String masterAddrBak = this.masterAddr;
         if (masterAddrBak != null && !masterAddrBak.equals(brokerController.getBrokerAddr())) {
             try {
-                TopicConfigSerializeWrapper topicWrapper =
+                TopicConfigAndMappingSerializeWrapper topicWrapper =
                     this.brokerController.getBrokerOuterAPI().getAllTopicConfig(masterAddrBak);
                 if (!this.brokerController.getTopicConfigManager().getDataVersion()
                     .equals(topicWrapper.getDataVersion())) {
@@ -67,9 +68,17 @@ public class SlaveSynchronize {
                     this.brokerController.getTopicConfigManager().getTopicConfigTable()
                         .putAll(topicWrapper.getTopicConfigTable());
                     this.brokerController.getTopicConfigManager().persist();
-
-                    log.info("Update slave topic config from master, {}", masterAddrBak);
                 }
+                if (topicWrapper.getTopicQueueMappingDetailMap() != null
+                        && !topicWrapper.getMappingDataVersion().equals(this.brokerController.getTopicQueueMappingManager().getDataVersion())) {
+                    this.brokerController.getTopicQueueMappingManager().getDataVersion()
+                            .assignNewOne(topicWrapper.getMappingDataVersion());
+                    this.brokerController.getTopicQueueMappingManager().getTopicQueueMappingTable().clear();
+                    this.brokerController.getTopicQueueMappingManager().getTopicQueueMappingTable()
+                            .putAll(topicWrapper.getTopicQueueMappingDetailMap());
+                    this.brokerController.getTopicQueueMappingManager().persist();
+                }
+                log.info("Update slave topic config from master, {}", masterAddrBak);
             } catch (Exception e) {
                 log.error("SyncTopicConfig Exception, {}", masterAddrBak, e);
             }

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();

broker/src/main/java/org/apache/rocketmq/broker/topic/TopicQueueMappingManager.java

@@ -30,6 +30,7 @@ import org.apache.rocketmq.common.protocol.header.GetTopicConfigRequestHeader;
 import org.apache.rocketmq.common.rpc.RpcRequest;
 import org.apache.rocketmq.common.rpc.RpcResponse;
 import org.apache.rocketmq.common.rpc.TopicQueueRequestHeader;
+import org.apache.rocketmq.common.rpc.TopicRequestHeader;
 import org.apache.rocketmq.common.statictopic.LogicQueueMappingItem;
 import org.apache.rocketmq.common.statictopic.TopicConfigAndQueueMapping;
 import org.apache.rocketmq.common.statictopic.TopicQueueMappingContext;
@@ -190,28 +191,37 @@ public class TopicQueueMappingManager extends ConfigManager {
         return dataVersion;
     }
 
-    public TopicQueueMappingContext buildTopicQueueMappingContext(TopicQueueRequestHeader requestHeader) {
+    public TopicQueueMappingContext buildTopicQueueMappingContext(TopicRequestHeader requestHeader) {
         return buildTopicQueueMappingContext(requestHeader, false);
     }
 
     //Do not return a null context
-    public TopicQueueMappingContext buildTopicQueueMappingContext(TopicQueueRequestHeader requestHeader, boolean selectOneWhenMiss) {
-        if (requestHeader.getPhysical() != null
-                && Boolean.TRUE.equals(requestHeader.getPhysical())) {
-            return new TopicQueueMappingContext(requestHeader.getTopic(), requestHeader.getQueueId(), null, null, null);
+    public TopicQueueMappingContext buildTopicQueueMappingContext(TopicRequestHeader requestHeader, boolean selectOneWhenMiss) {
+        //should disable logic queue explicitly, otherwise the old client may cause dirty data to newly created static topic
+        if (requestHeader.getLo() != null
+                && Boolean.FALSE.equals(requestHeader.getLo())) {
+            return new TopicQueueMappingContext(requestHeader.getTopic(), null, null, null, null);
         }
-        TopicQueueMappingDetail mappingDetail = getTopicQueueMapping(requestHeader.getTopic());
+        String topic = requestHeader.getTopic();
+        Integer globalId = null;
+        if (requestHeader instanceof  TopicQueueRequestHeader) {
+            globalId = ((TopicQueueRequestHeader) requestHeader).getQueueId();
+        }
+
+        TopicQueueMappingDetail mappingDetail = getTopicQueueMapping(topic);
         if (mappingDetail == null) {
             //it is not static topic
-            return new TopicQueueMappingContext(requestHeader.getTopic(), requestHeader.getQueueId(), null, null, null);
+            return new TopicQueueMappingContext(topic, null, null, null, null);
         }
-
         assert mappingDetail.getBname().equals(this.brokerController.getBrokerConfig().getBrokerName());
 
+        if (globalId == null) {
+            return new TopicQueueMappingContext(topic, null, mappingDetail, null, null);
+        }
+
         //If not find mappingItem, it encounters some errors
-        Integer globalId = requestHeader.getQueueId();
         if (globalId < 0 && !selectOneWhenMiss) {
-            return new TopicQueueMappingContext(requestHeader.getTopic(), globalId, mappingDetail, null, null);
+            return new TopicQueueMappingContext(topic, globalId, mappingDetail, null, null);
         }
 
         if (globalId < 0) {
@@ -224,7 +234,7 @@ public class TopicQueueMappingManager extends ConfigManager {
             }
         }
         if (globalId < 0) {
-            return new TopicQueueMappingContext(requestHeader.getTopic(), globalId,  mappingDetail, null, null);
+            return new TopicQueueMappingContext(topic, globalId,  mappingDetail, null, null);
         }
 
         List<LogicQueueMappingItem> mappingItemList = TopicQueueMappingDetail.getMappingInfo(mappingDetail, globalId);
@@ -233,7 +243,7 @@ public class TopicQueueMappingManager extends ConfigManager {
                 && mappingItemList.size() > 0) {
             leaderItem = mappingItemList.get(mappingItemList.size() - 1);
         }
-        return new TopicQueueMappingContext(requestHeader.getTopic(), globalId, mappingDetail, mappingItemList, leaderItem);
+        return new TopicQueueMappingContext(topic, globalId, mappingDetail, mappingItemList, leaderItem);
     }
 
 

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);

common/src/main/java/org/apache/rocketmq/common/protocol/body/TopicConfigAndMappingSerializeWrapper.java

@@ -17,6 +17,8 @@
 
 package org.apache.rocketmq.common.protocol.body;
 
+import org.apache.rocketmq.common.DataVersion;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingDetail;
 import org.apache.rocketmq.common.statictopic.TopicQueueMappingInfo;
 
 import java.util.Map;
@@ -25,6 +27,11 @@ import java.util.concurrent.ConcurrentHashMap;
 public class TopicConfigAndMappingSerializeWrapper extends TopicConfigSerializeWrapper {
     private Map<String/* topic */, TopicQueueMappingInfo> topicQueueMappingInfoMap = new ConcurrentHashMap<String, TopicQueueMappingInfo>();
 
+    private Map<String/* topic */, TopicQueueMappingDetail> topicQueueMappingDetailMap = new ConcurrentHashMap<String, TopicQueueMappingDetail>();
+
+    private DataVersion mappingDataVersion = new DataVersion();
+
+
     public Map<String, TopicQueueMappingInfo> getTopicQueueMappingInfoMap() {
         return topicQueueMappingInfoMap;
     }
@@ -33,6 +40,22 @@ public class TopicConfigAndMappingSerializeWrapper extends TopicConfigSerializeW
         this.topicQueueMappingInfoMap = topicQueueMappingInfoMap;
     }
 
+    public Map<String, TopicQueueMappingDetail> getTopicQueueMappingDetailMap() {
+        return topicQueueMappingDetailMap;
+    }
+
+    public void setTopicQueueMappingDetailMap(Map<String, TopicQueueMappingDetail> topicQueueMappingDetailMap) {
+        this.topicQueueMappingDetailMap = topicQueueMappingDetailMap;
+    }
+
+    public DataVersion getMappingDataVersion() {
+        return mappingDataVersion;
+    }
+
+    public void setMappingDataVersion(DataVersion mappingDataVersion) {
+        this.mappingDataVersion = mappingDataVersion;
+    }
+
     public static TopicConfigAndMappingSerializeWrapper from(TopicConfigSerializeWrapper wrapper) {
         if (wrapper instanceof  TopicConfigAndMappingSerializeWrapper) {
             return (TopicConfigAndMappingSerializeWrapper) wrapper;

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

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

@@ -17,12 +17,11 @@
 
 package org.apache.rocketmq.common.protocol.header;
 
-import org.apache.rocketmq.common.rpc.RpcRequestHeader;
-import org.apache.rocketmq.remoting.CommandCustomHeader;
+import org.apache.rocketmq.common.rpc.TopicRequestHeader;
 import org.apache.rocketmq.remoting.annotation.CFNotNull;
 import org.apache.rocketmq.remoting.exception.RemotingCommandException;
 
-public class GetTopicStatsInfoRequestHeader extends RpcRequestHeader {
+public class GetTopicStatsInfoRequestHeader extends TopicRequestHeader {
     @CFNotNull
     private String topic;
 

common/src/main/java/org/apache/rocketmq/common/rpc/RequestBuilder.java

@@ -25,7 +25,7 @@ public class RequestBuilder {
         }
         try {
             RpcRequestHeader requestHeader = (RpcRequestHeader) requestHeaderClass.newInstance();
-            requestHeader.setOneway(oneway);
+            requestHeader.setOway(oneway);
             requestHeader.setBname(destBrokerName);
             return requestHeader;
         } catch (Throwable t) {
@@ -37,26 +37,26 @@ public class RequestBuilder {
         return buildTopicQueueRequestHeader(requestCode, null, mq.getBrokerName(), mq.getTopic(), mq.getQueueId(), null);
     }
 
-    public static TopicQueueRequestHeader buildTopicQueueRequestHeader(int requestCode, MessageQueue mq, Boolean physical) {
-        return buildTopicQueueRequestHeader(requestCode, null, mq.getBrokerName(), mq.getTopic(), mq.getQueueId(), physical);
+    public static TopicQueueRequestHeader buildTopicQueueRequestHeader(int requestCode, MessageQueue mq, Boolean logic) {
+        return buildTopicQueueRequestHeader(requestCode, null, mq.getBrokerName(), mq.getTopic(), mq.getQueueId(), logic);
     }
 
-    public static TopicQueueRequestHeader buildTopicQueueRequestHeader(int requestCode, Boolean oneway, MessageQueue mq, Boolean physical) {
-        return buildTopicQueueRequestHeader(requestCode, oneway, mq.getBrokerName(), mq.getTopic(), mq.getQueueId(), physical);
+    public static TopicQueueRequestHeader buildTopicQueueRequestHeader(int requestCode, Boolean oneway, MessageQueue mq, Boolean logic) {
+        return buildTopicQueueRequestHeader(requestCode, oneway, mq.getBrokerName(), mq.getTopic(), mq.getQueueId(), logic);
     }
 
-    public static TopicQueueRequestHeader buildTopicQueueRequestHeader(int requestCode,  Boolean oneway, String destBrokerName, String topic, int queueId, Boolean physical) {
+    public static TopicQueueRequestHeader buildTopicQueueRequestHeader(int requestCode,  Boolean oneway, String destBrokerName, String topic, int queueId, Boolean logic) {
         Class requestHeaderClass = requestCodeMap.get(requestCode);
         if (requestHeaderClass == null) {
             throw new UnsupportedOperationException("unknown " + requestCode);
         }
         try {
             TopicQueueRequestHeader requestHeader = (TopicQueueRequestHeader) requestHeaderClass.newInstance();
-            requestHeader.setOneway(oneway);
+            requestHeader.setOway(oneway);
             requestHeader.setBname(destBrokerName);
             requestHeader.setTopic(topic);
             requestHeader.setQueueId(queueId);
-            requestHeader.setPhysical(physical);
+            requestHeader.setLo(logic);
             return requestHeader;
         } catch (Throwable t) {
             throw new RuntimeException(t);

common/src/main/java/org/apache/rocketmq/common/rpc/RpcClientImpl.java

@@ -214,6 +214,7 @@ public class RpcClientImpl implements RpcClient {
             }
             case ResponseCode.QUERY_NOT_FOUND: {
                 rpcResponsePromise.setSuccess(new RpcResponse(responseCommand.getCode(), null, null));
+                break;
             }
             default:{
                 rpcResponsePromise.setSuccess(new RpcResponse(new RpcException(responseCommand.getCode(), "unknown remote error")));

common/src/main/java/org/apache/rocketmq/common/rpc/RpcRequestHeader.java

@@ -20,13 +20,13 @@ import org.apache.rocketmq.remoting.CommandCustomHeader;
 
 public abstract class RpcRequestHeader implements CommandCustomHeader {
     //the namespace name
-    protected String namespace;
+    protected String ns;
     //if the data has been namespaced
-    protected Boolean namespaced;
+    protected Boolean nsd;
     //the abstract remote addr name, usually the physical broker name
     protected String bname;
-
-    protected Boolean oneway;
+    //oneway
+    protected Boolean oway;
 
     public String getBname() {
         return bname;
@@ -36,27 +36,27 @@ public abstract class RpcRequestHeader implements CommandCustomHeader {
         this.bname = bname;
     }
 
-    public Boolean getOneway() {
-        return oneway;
+    public Boolean getOway() {
+        return oway;
     }
 
-    public void setOneway(Boolean oneway) {
-        this.oneway = oneway;
+    public void setOway(Boolean oway) {
+        this.oway = oway;
     }
 
-    public String getNamespace() {
-        return namespace;
+    public String getNs() {
+        return ns;
     }
 
-    public void setNamespace(String namespace) {
-        this.namespace = namespace;
+    public void setNs(String ns) {
+        this.ns = ns;
     }
 
-    public Boolean getNamespaced() {
-        return namespaced;
+    public Boolean getNsd() {
+        return nsd;
     }
 
-    public void setNamespaced(Boolean namespaced) {
-        this.namespaced = namespaced;
+    public void setNsd(Boolean nsd) {
+        this.nsd = nsd;
     }
 }

common/src/main/java/org/apache/rocketmq/common/rpc/RpcResponse.java

@@ -34,7 +34,7 @@ public class RpcResponse   {
         this.body = body;
     }
 
-    RpcResponse(RpcException rpcException) {
+    public RpcResponse(RpcException rpcException) {
         this.code = rpcException.getErrorCode();
         this.exception = rpcException;
     }

common/src/main/java/org/apache/rocketmq/common/rpc/TopicQueueRequestHeader.java

@@ -16,20 +16,8 @@
  */
 package org.apache.rocketmq.common.rpc;
 
-public abstract class TopicQueueRequestHeader extends RpcRequestHeader {
-    //Physical or Logical
-    protected Boolean physical;
+public abstract class TopicQueueRequestHeader extends TopicRequestHeader {
 
-    public Boolean getPhysical() {
-        return physical;
-    }
-
-    public void setPhysical(Boolean physical) {
-        this.physical = physical;
-    }
-
-    public abstract String getTopic();
-    public abstract void setTopic(String topic);
     public abstract Integer getQueueId();
     public abstract void setQueueId(Integer queueId);
 

common/src/main/java/org/apache/rocketmq/common/rpc/TopicRequestHeader.java

+/*
+ * 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.rpc;
+
+public abstract class TopicRequestHeader extends RpcRequestHeader {
+    //logical
+    protected Boolean lo;
+
+    public abstract String getTopic();
+    public abstract void setTopic(String topic);
+
+    public Boolean getLo() {
+        return lo;
+    }
+    public void setLo(Boolean lo) {
+        this.lo = lo;
+    }
+}

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);
+    }
 
 }

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);
         }

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();

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);