[ISSUE #10580] Fix static topic epoch comparator overflow

remoting/src/main/java/org/apache/rocketmq/remoting/protocol/statictopic/TopicQueueMappingUtils.java

@@ -338,7 +338,7 @@ public static void  checkPhysicalQueueConsistence(Map<String, TopicConfigAndQueu
 
 
     public static Map<Integer, TopicQueueMappingOne> checkAndBuildMappingItems(List<TopicQueueMappingDetail> mappingDetailList, boolean replace, boolean checkConsistence) {
-        mappingDetailList.sort((o1, o2) -> (int) (o2.getEpoch() - o1.getEpoch()));
+        mappingDetailList.sort((o1, o2) -> Long.compare(o2.getEpoch(), o1.getEpoch()));
 
         int maxNum = 0;
         Map<Integer, TopicQueueMappingOne> globalIdMap = new HashMap<>();

remoting/src/main/java/org/apache/rocketmq/remoting/rpc/ClientMetadata.java

@@ -119,7 +119,7 @@ public static ConcurrentMap<MessageQueue, String> topicRouteData2EndpointsForSta
             Map<String, TopicQueueMappingInfo> topicQueueMappingInfoMap =  mapEntry.getValue();
             ConcurrentMap<MessageQueue, TopicQueueMappingInfo> mqEndPoints = new ConcurrentHashMap<>();
             List<Map.Entry<String, TopicQueueMappingInfo>> mappingInfos = new ArrayList<>(topicQueueMappingInfoMap.entrySet());
-            mappingInfos.sort((o1, o2) -> (int) (o2.getValue().getEpoch() - o1.getValue().getEpoch()));
+            mappingInfos.sort((o1, o2) -> Long.compare(o2.getValue().getEpoch(), o1.getValue().getEpoch()));
             int maxTotalNums = 0;
             long maxTotalNumOfEpoch = -1;
             for (Map.Entry<String, TopicQueueMappingInfo> entry : mappingInfos) {

remoting/src/test/java/org/apache/rocketmq/remoting/protocol/statictopic/TopicQueueMappingUtilsTest.java

@@ -81,6 +81,32 @@ private void testIdToBroker(Map<Integer, String> idToBroker, Map<String, Integer
         }
     }
 
+    @Test
+    public void testCheckAndBuildMappingItemsUsesLatestEpochWhenReplacingDuplicatedQueue() {
+        String topic = "static";
+        String oldBroker = "oldBroker";
+        String newBroker = "newBroker";
+        long oldEpoch = 0L;
+        long newEpoch = (long) Integer.MAX_VALUE + 1L;
+        List<TopicQueueMappingDetail> mappingDetails = new ArrayList<>();
+        mappingDetails.add(buildMappingDetail(topic, oldBroker, oldEpoch));
+        mappingDetails.add(buildMappingDetail(topic, newBroker, newEpoch));
+
+        Map<Integer, TopicQueueMappingOne> globalIdMap =
+            TopicQueueMappingUtils.checkAndBuildMappingItems(mappingDetails, true, false);
+
+        TopicQueueMappingOne mappingOne = globalIdMap.get(0);
+        Assert.assertEquals(newEpoch, mappingOne.getMappingDetail().getEpoch());
+        Assert.assertEquals(newBroker, mappingOne.getBname());
+    }
+
+    private TopicQueueMappingDetail buildMappingDetail(String topic, String brokerName, long epoch) {
+        TopicQueueMappingDetail mappingDetail = new TopicQueueMappingDetail(topic, 1, brokerName, epoch);
+        mappingDetail.getHostedQueues().put(0, new ArrayList<>(Collections.singletonList(
+            new LogicQueueMappingItem(0, 0, brokerName, 0, 0, -1, -1, -1))));
+        return mappingDetail;
+    }
+
     @Test
     public void testAllocator() {
         //stability

remoting/src/test/java/org/apache/rocketmq/remoting/rpc/ClientMetadataTest.java

@@ -20,6 +20,7 @@
 import org.apache.rocketmq.common.message.MessageQueue;
 import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
 import org.apache.rocketmq.remoting.protocol.statictopic.TopicQueueMappingInfo;
+import org.apache.rocketmq.remoting.protocol.statictopic.TopicQueueMappingUtils;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -120,4 +121,35 @@ public void testTopicRouteData2EndpointsForStaticTopic() {
         ConcurrentMap<MessageQueue, String> actual = ClientMetadata.topicRouteData2EndpointsForStaticTopic(defaultTopic, topicRouteData);
         assertEquals(1, actual.size());
     }
+
+    @Test
+    public void testTopicRouteData2EndpointsForStaticTopicUsesLatestEpoch() {
+        String scope = "scope";
+        String oldBroker = "oldBroker";
+        String newBroker = "newBroker";
+        long oldEpoch = 0L;
+        long newEpoch = (long) Integer.MAX_VALUE + 1L;
+        TopicRouteData topicRouteData = new TopicRouteData();
+        Map<String, TopicQueueMappingInfo> mappingInfos = new HashMap<>();
+        mappingInfos.put(oldBroker, buildMappingInfo(scope, oldBroker, oldEpoch));
+        mappingInfos.put(newBroker, buildMappingInfo(scope, newBroker, newEpoch));
+        topicRouteData.setTopicQueueMappingByBroker(mappingInfos);
+
+        ConcurrentMap<MessageQueue, String> actual =
+            ClientMetadata.topicRouteData2EndpointsForStaticTopic(defaultTopic, topicRouteData);
+
+        MessageQueue mq = new MessageQueue(defaultTopic, TopicQueueMappingUtils.getMockBrokerName(scope), 0);
+        assertEquals(newBroker, actual.get(mq));
+    }
+
+    private TopicQueueMappingInfo buildMappingInfo(String scope, String brokerName, long epoch) {
+        TopicQueueMappingInfo info = new TopicQueueMappingInfo();
+        info.setScope(scope);
+        info.setCurrIdMap(new ConcurrentHashMap<>());
+        info.getCurrIdMap().put(0, 0);
+        info.setTotalQueues(1);
+        info.setBname(brokerName);
+        info.setEpoch(epoch);
+        return info;
+    }
 }