[ISSUE #10582] Fix POP revive offset comparator overflow

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

@@ -53,6 +53,7 @@
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -725,7 +726,7 @@ ArrayList<PopCheckPoint> genSortList() {
                 return sortList;
             }
             sortList = new ArrayList<>(map.values());
-            sortList.sort((o1, o2) -> (int) (o1.getReviveOffset() - o2.getReviveOffset()));
+            sortList.sort(Comparator.comparingLong(PopCheckPoint::getReviveOffset));
             return sortList;
         }
     }

broker/src/test/java/org/apache/rocketmq/broker/processor/PopReviveServiceTest.java

@@ -442,6 +442,20 @@ public void testReviveMsgFromBatchAck() throws Throwable {
         assertEquals(1, commitOffsetCaptor.getValue().longValue());
     }
 
+    @Test
+    public void testGenSortListShouldSortLargeReviveOffsets() {
+        PopReviveService.ConsumeReviveObj consumeReviveObj = new PopReviveService.ConsumeReviveObj();
+        PopCheckPoint lowOffsetCk = buildPopCheckPoint(0, 0, 1);
+        PopCheckPoint highOffsetCk = buildPopCheckPoint(1, 0, (long) Integer.MAX_VALUE + 2);
+        consumeReviveObj.map.put("high", highOffsetCk);
+        consumeReviveObj.map.put("low", lowOffsetCk);
+
+        List<PopCheckPoint> sortList = consumeReviveObj.genSortList();
+
+        assertEquals(lowOffsetCk, sortList.get(0));
+        assertEquals(highOffsetCk, sortList.get(1));
+    }
+
     public static MessageExtBrokerInner buildBatchAckMsg(BatchAckMsg batchAckMsg, long deliverMs, long reviveOffset, long deliverTime) {
         MessageExtBrokerInner result = buildBatchAckInnerMessage(REVIVE_TOPIC, batchAckMsg, REVIVE_QUEUE_ID, STORE_HOST, deliverMs, PopMessageProcessor.genAckUniqueId(batchAckMsg));
         result.setQueueOffset(reviveOffset);

store/src/main/java/org/apache/rocketmq/store/pop/PopCheckPoint.java

@@ -212,6 +212,6 @@ public String toString() {
 
     @Override
     public int compareTo(PopCheckPoint o) {
-        return (int) (this.getStartOffset() - o.getStartOffset());
+        return Long.compare(this.getStartOffset(), o.getStartOffset());
     }
 }

store/src/test/java/org/apache/rocketmq/store/pop/PopCheckPointTest.java

@@ -0,0 +1,58 @@
+/*
+ * 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.store.pop;
+
+import java.util.TreeMap;
+import org.junit.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class PopCheckPointTest {
+
+    private static PopCheckPoint build(long startOffset) {
+        PopCheckPoint ck = new PopCheckPoint();
+        ck.setStartOffset(startOffset);
+        return ck;
+    }
+
+    @Test
+    public void testCompareToWithoutOverflow() {
+        PopCheckPoint low = build(0);
+        PopCheckPoint high = build((long) Integer.MAX_VALUE + 2L);
+
+        // The legacy (int)(a - b) comparator overflowed and reported the high
+        // offset as the smaller one. Long.compare must keep them ordered.
+        assertThat(low.compareTo(high)).isNegative();
+        assertThat(high.compareTo(low)).isPositive();
+        assertThat(low.compareTo(low)).isZero();
+    }
+
+    @Test
+    public void testTreeMapOrdersLargeStartOffsets() {
+        // PopReviveService keeps checkpoints in a TreeMap<PopCheckPoint, ...>,
+        // so a broken compareTo corrupts ordering of large offsets.
+        TreeMap<PopCheckPoint, Boolean> map = new TreeMap<>();
+        PopCheckPoint high = build((long) Integer.MAX_VALUE + 2L);
+        PopCheckPoint low = build(0);
+        map.put(high, true);
+        map.put(low, true);
+
+        assertThat(map.firstKey().getStartOffset()).isEqualTo(0L);
+        assertThat(map.lastKey().getStartOffset()).isEqualTo((long) Integer.MAX_VALUE + 2L);
+    }
+}