Polish the rewrite logic for SendProcessor

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

@@ -17,6 +17,7 @@
 package org.apache.rocketmq.broker.processor;
 
 import java.net.SocketAddress;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
@@ -126,16 +127,28 @@ public class SendMessageProcessor extends AbstractSendMessageProcessor implement
         try {
             TopicQueueMappingDetail mappingDetail = this.brokerController.getTopicQueueMappingManager().getTopicQueueMapping(requestHeader.getTopic());
             if (mappingDetail == null) {
+                //it is not static topic
                 return null;
             }
-            if (!mappingDetail.getCurrIdMap().containsKey(requestHeader.getQueueId())) {
-               return buildErrorResponse(ResponseCode.NOT_LEADER_FOR_QUEUE, String.format("%s-%d does not exit in current broker %s", requestHeader.getTopic(), requestHeader.getQueueId(), this.brokerController.getBrokerConfig().getBrokerName()));
+            Integer phyQueueId = null;
+            //compatible with the old logic, but it fact, this should not happen
+            if (requestHeader.getQueueId() < 0) {
+                Iterator<Map.Entry<Integer, Integer>> it  = mappingDetail.getCurrIdMap().entrySet().iterator();
+                if (it.hasNext()) {
+                    phyQueueId = it.next().getValue();
+                }
+            } else {
+                phyQueueId = mappingDetail.getCurrIdMap().get(requestHeader.getQueueId());
+            }
+            if (phyQueueId == null) {
+                return buildErrorResponse(ResponseCode.NOT_LEADER_FOR_QUEUE, String.format("%s-%d does not exit in request process of current broker %s", requestHeader.getTopic(), requestHeader.getQueueId(), this.brokerController.getBrokerConfig().getBrokerName()));
+            } else {
+                requestHeader.setQueueId(phyQueueId);
+                return null;
             }
-            requestHeader.setQueueId(mappingDetail.getCurrIdMap().get(requestHeader.getQueueId()));
         } catch (Throwable t) {
             return buildErrorResponse(ResponseCode.SYSTEM_ERROR, t.getMessage());
         }
-        return null;
     }
 
     private RemotingCommand rewriteResponseForStaticTopic(String topic, SendMessageResponseHeader responseHeader) {
@@ -144,14 +157,16 @@ public class SendMessageProcessor extends AbstractSendMessageProcessor implement
             if (mappingDetail == null) {
                 return null;
             }
-            if (!mappingDetail.getCurrIdMap().containsKey(responseHeader.getQueueId())) {
-                return buildErrorResponse(ResponseCode.NOT_LEADER_FOR_QUEUE, String.format("%s-%d does not exit in current broker %s", topic, responseHeader.getQueueId(), this.brokerController.getBrokerConfig().getBrokerName()));
+            Integer globalId = mappingDetail.getCurrIdMapRevert().get(responseHeader.getQueueId());
+            if (globalId == null) {
+                return buildErrorResponse(ResponseCode.NOT_LEADER_FOR_QUEUE, String.format("%s-%d does not exist in response process of current broker %s", topic, responseHeader.getQueueId(), this.brokerController.getBrokerConfig().getBrokerName()));
             }
-            long staticLogicOffset = mappingDetail.convertToLogicOffset(responseHeader.getQueueId(), responseHeader.getQueueOffset());
+            long staticLogicOffset = mappingDetail.convertToLogicOffset(globalId, responseHeader.getQueueOffset());
             if (staticLogicOffset < 0) {
                 return buildErrorResponse(ResponseCode.NOT_LEADER_FOR_QUEUE, String.format("%s-%d convert offset error in current broker %s", topic, responseHeader.getQueueId(), this.brokerController.getBrokerConfig().getBrokerName()));
 
             }
+            responseHeader.setQueueId(globalId);
             responseHeader.setQueueOffset(staticLogicOffset);
         } catch (Throwable t) {
             return buildErrorResponse(ResponseCode.SYSTEM_ERROR, t.getMessage());

common/src/main/java/org/apache/rocketmq/common/TopicQueueMappingDetail.java

@@ -27,6 +27,9 @@ public class TopicQueueMappingDetail extends TopicQueueMappingInfo {
 
     // the mapping info in current broker, do not register to nameserver
     ConcurrentMap<Integer/*global id*/, ImmutableList<LogicQueueMappingItem>> hostedQueues = new ConcurrentHashMap<Integer, ImmutableList<LogicQueueMappingItem>>();
+    transient ConcurrentMap<Integer/*physicalId*/, Integer/*logicId*/> currIdMapRevert = new ConcurrentHashMap<Integer, Integer>();
+
+
 
     public TopicQueueMappingDetail(String topic, int totalQueues, String bname) {
         super(topic, totalQueues, bname);
@@ -45,6 +48,18 @@ public class TopicQueueMappingDetail extends TopicQueueMappingInfo {
     public void buildIdMap() {
         this.currIdMap = buildIdMap(LEVEL_0);
         this.prevIdMap = buildIdMap(LEVEL_1);
+        this.currIdMapRevert = revert(this.currIdMap);
+    }
+
+    public ConcurrentMap<Integer, Integer> revert(ConcurrentMap<Integer, Integer> original) {
+        if (original == null || original.isEmpty()) {
+            return new ConcurrentHashMap<Integer, Integer>();
+        }
+        ConcurrentMap<Integer, Integer> tmpIdMap = new ConcurrentHashMap<Integer, Integer>();
+        for (Map.Entry<Integer, Integer> entry: tmpIdMap.entrySet()) {
+            tmpIdMap.put(entry.getValue(), entry.getKey());
+        }
+        return tmpIdMap;
     }
 
     public ConcurrentMap<Integer, Integer> buildIdMap(int level) {
@@ -107,6 +122,14 @@ public class TopicQueueMappingDetail extends TopicQueueMappingInfo {
         return topicQueueMappingInfo;
     }
 
+    public ConcurrentMap<Integer, Integer> getCurrIdMapRevert() {
+        return currIdMapRevert;
+    }
+
+    public void setCurrIdMapRevert(ConcurrentMap<Integer, Integer> currIdMapRevert) {
+        this.currIdMapRevert = currIdMapRevert;
+    }
+
     public int getTotalQueues() {
         return totalQueues;
     }