This is an automated email from the ASF dual-hosted git repository.
RongtongJin pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new 5a6e349739 [ISSUE #11306] fix(tools): compare long counters without
truncating to int (#11274)
5a6e349739 is described below
commit 5a6e349739bf2430d3d33250f8b1fcf92928405f
Author: zhang hongyuan <[email protected]>
AuthorDate: Thu Oct 8 14:46:23 2026 +0800
[ISSUE #11306] fix(tools): compare long counters without truncating to int
(#11274)
GroupConsumeInfo.compareTo, TagCountBean.compareTo and
QueryMsgByUniqueKeySubCommand.queryById subtracted two long values and cast
the difference to int, so a difference beyond Integer.MAX_VALUE reversed the
order and broke the comparator contract that TimSort checks. Use
Long.compare / Comparator.comparingLong instead.
Signed-off-by: zhang hongyuan <[email protected]>
---
.../consumer/ConsumerProgressSubCommand.java | 6 +-
.../message/PrintMessageByQueueCommand.java | 6 +-
.../message/QueryMsgByUniqueKeySubCommand.java | 12 ++-
.../consumer/ConsumerProgressSubCommandTest.java | 78 +++++++++++++++++++
.../message/PrintMessageByQueueCommandTest.java | 88 ++++++++++++++++++++++
.../message/QueryMsgByUniqueKeySubCommandTest.java | 29 +++++++
6 files changed, 216 insertions(+), 3 deletions(-)
diff --git
a/tools/src/main/java/org/apache/rocketmq/tools/command/consumer/ConsumerProgressSubCommand.java
b/tools/src/main/java/org/apache/rocketmq/tools/command/consumer/ConsumerProgressSubCommand.java
index b638dcf61f..40fe52c6bc 100644
---
a/tools/src/main/java/org/apache/rocketmq/tools/command/consumer/ConsumerProgressSubCommand.java
+++
b/tools/src/main/java/org/apache/rocketmq/tools/command/consumer/ConsumerProgressSubCommand.java
@@ -349,7 +349,11 @@ class GroupConsumeInfo implements
Comparable<GroupConsumeInfo> {
return o.count - this.count;
}
- return (int) (o.diffTotal - diffTotal);
+ // diffTotal is a message lag that can exceed Integer.MAX_VALUE, so
compare the
+ // longs directly. Casting the subtraction to int wraps around for
large gaps,
+ // which both reverses the reported ordering and breaks the Comparable
contract
+ // that Collections.sort (TimSort) relies on.
+ return Long.compare(o.diffTotal, diffTotal);
}
public int getConsumeTps() {
diff --git
a/tools/src/main/java/org/apache/rocketmq/tools/command/message/PrintMessageByQueueCommand.java
b/tools/src/main/java/org/apache/rocketmq/tools/command/message/PrintMessageByQueueCommand.java
index 0418e88a70..9933189ddd 100644
---
a/tools/src/main/java/org/apache/rocketmq/tools/command/message/PrintMessageByQueueCommand.java
+++
b/tools/src/main/java/org/apache/rocketmq/tools/command/message/PrintMessageByQueueCommand.java
@@ -250,7 +250,11 @@ public class PrintMessageByQueueCommand implements
SubCommand {
@Override
public int compareTo(final TagCountBean o) {
- return (int) (o.getCount().get() - this.count.get());
+ // Tag counts are longs that may exceed Integer.MAX_VALUE; the
previous
+ // (int) (o.count - this.count) form truncated the difference,
which
+ // inverted the order and violated the Comparable contract used by
+ // Collections.sort. Long.compare keeps the full 64-bit ordering.
+ return Long.compare(o.getCount().get(), this.count.get());
}
}
}
diff --git
a/tools/src/main/java/org/apache/rocketmq/tools/command/message/QueryMsgByUniqueKeySubCommand.java
b/tools/src/main/java/org/apache/rocketmq/tools/command/message/QueryMsgByUniqueKeySubCommand.java
index 8518a04e68..34b794999a 100644
---
a/tools/src/main/java/org/apache/rocketmq/tools/command/message/QueryMsgByUniqueKeySubCommand.java
+++
b/tools/src/main/java/org/apache/rocketmq/tools/command/message/QueryMsgByUniqueKeySubCommand.java
@@ -20,6 +20,7 @@ import java.io.DataOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
+import java.util.Comparator;
import java.util.List;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Option;
@@ -42,6 +43,15 @@ public class QueryMsgByUniqueKeySubCommand implements
SubCommand {
private DefaultMQAdminExt defaultMQAdminExt;
+ /**
+ * Orders messages by their store timestamp. The lambda it replaces
subtracted two longs
+ * and cast the result to int, which wrapped around whenever the
timestamps were more than
+ * Integer.MAX_VALUE milliseconds (about 24.8 days) apart, mis-sorting the
output and
+ * breaking the comparator contract that TimSort enforces.
+ */
+ static final Comparator<MessageExt> STORE_TIMESTAMP_COMPARATOR =
+ Comparator.comparingLong(MessageExt::getStoreTimestamp);
+
private DefaultMQAdminExt createMQAdminExt(RPCHook rpcHook) throws
SubCommandException {
if (this.defaultMQAdminExt != null) {
return defaultMQAdminExt;
@@ -73,7 +83,7 @@ public class QueryMsgByUniqueKeySubCommand implements
SubCommand {
if (list == null || list.size() == 0) {
return;
}
- list.sort((o1, o2) -> (int) (o1.getStoreTimestamp() -
o2.getStoreTimestamp()));
+ list.sort(STORE_TIMESTAMP_COMPARATOR);
for (int i = 0; i < (showAll ? list.size() : 1); i++) {
showMessage(admin, list.get(i), i);
}
diff --git
a/tools/src/test/java/org/apache/rocketmq/tools/command/consumer/ConsumerProgressSubCommandTest.java
b/tools/src/test/java/org/apache/rocketmq/tools/command/consumer/ConsumerProgressSubCommandTest.java
index 15c3fa777a..9b8a064d5c 100644
---
a/tools/src/test/java/org/apache/rocketmq/tools/command/consumer/ConsumerProgressSubCommandTest.java
+++
b/tools/src/test/java/org/apache/rocketmq/tools/command/consumer/ConsumerProgressSubCommandTest.java
@@ -16,7 +16,11 @@
*/
package org.apache.rocketmq.tools.command.consumer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.DefaultParser;
import org.apache.commons.cli.Options;
@@ -28,6 +32,7 @@ import org.apache.rocketmq.tools.command.SubCommandException;
import org.apache.rocketmq.tools.command.server.NameServerMocker;
import org.apache.rocketmq.tools.command.server.ServerResponseMocker;
import org.junit.After;
+import org.junit.Assert;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
@@ -81,4 +86,77 @@ public class ConsumerProgressSubCommandTest {
// start broker
return ServerResponseMocker.startServer(consumeStats.encode());
}
+
+ @Test
+ public void testCompareToIsAntisymmetricAcrossIntegerMaxValue() {
+ GroupConsumeInfo zeroLag = buildGroupConsumeInfo("zero-lag", 0L);
+ GroupConsumeInfo overIntLag = buildGroupConsumeInfo("over-int-lag",
(long) Integer.MAX_VALUE + 1L);
+ GroupConsumeInfo maxLag = buildGroupConsumeInfo("max-lag",
Long.MAX_VALUE);
+
+ // A subtraction-based comparator wraps for gaps beyond
Integer.MAX_VALUE and would
+ // then report both a < b and b < a; assert the antisymmetric contract
explicitly.
+ Assert.assertTrue(zeroLag.compareTo(overIntLag) > 0);
+ Assert.assertTrue(overIntLag.compareTo(zeroLag) < 0);
+ Assert.assertTrue(overIntLag.compareTo(maxLag) > 0);
+ Assert.assertTrue(maxLag.compareTo(overIntLag) < 0);
+ Assert.assertEquals(0,
overIntLag.compareTo(buildGroupConsumeInfo("over-int-lag-copy", (long)
Integer.MAX_VALUE + 1L)));
+ Assert.assertEquals(0,
maxLag.compareTo(buildGroupConsumeInfo("max-lag-copy", Long.MAX_VALUE)));
+ }
+
+ @Test
+ public void testCollectionsSortOrdersGroupsByLagDescending() {
+ GroupConsumeInfo zeroLag = buildGroupConsumeInfo("zero-lag", 0L);
+ GroupConsumeInfo overIntLag = buildGroupConsumeInfo("over-int-lag",
(long) Integer.MAX_VALUE + 1L);
+ GroupConsumeInfo maxLag = buildGroupConsumeInfo("max-lag",
Long.MAX_VALUE);
+
+ List<GroupConsumeInfo> groups = new ArrayList<>();
+ groups.add(zeroLag);
+ groups.add(maxLag);
+ groups.add(overIntLag);
+ Collections.sort(groups);
+
+ Assert.assertEquals(Arrays.asList(maxLag, overIntLag, zeroLag),
groups);
+ }
+
+ @Test
+ public void testSortWithLagValuesSpanningIntAndLongRanges() {
+ List<GroupConsumeInfo> groups = new ArrayList<>();
+ groups.add(buildGroupConsumeInfo("lag-zero", 0L));
+ groups.add(buildGroupConsumeInfo("lag-small", 1024L));
+ groups.add(buildGroupConsumeInfo("lag-over-int", (long)
Integer.MAX_VALUE + 1L));
+ groups.add(buildGroupConsumeInfo("lag-max", Long.MAX_VALUE));
+
+ Collections.sort(groups);
+
+ // Every adjacent pair must be ordered descending, which only holds
when the
+ // comparison never wraps around for gaps larger than
Integer.MAX_VALUE.
+ for (int i = 0; i < groups.size() - 1; i++) {
+ Assert.assertTrue(groups.get(i).getDiffTotal() > groups.get(i +
1).getDiffTotal());
+ Assert.assertTrue(groups.get(i).compareTo(groups.get(i + 1)) < 0);
+ }
+ Assert.assertEquals("lag-max", groups.get(0).getGroup());
+ Assert.assertEquals("lag-zero", groups.get(groups.size() -
1).getGroup());
+ }
+
+ @Test
+ public void testCompareToPrefersMoreConsumersBeforeLag() {
+ GroupConsumeInfo singleConsumer =
buildGroupConsumeInfo("single-consumer", Long.MAX_VALUE);
+ GroupConsumeInfo twoConsumers = buildGroupConsumeInfo("two-consumers",
0L);
+ twoConsumers.setCount(2);
+
+ // The count comparison short-circuits the diffTotal comparison:
groups with more
+ // live connections are always ordered first, whatever their lag is.
+ Assert.assertTrue(singleConsumer.compareTo(twoConsumers) > 0);
+ Assert.assertTrue(twoConsumers.compareTo(singleConsumer) < 0);
+ }
+
+ private static GroupConsumeInfo buildGroupConsumeInfo(String group, long
diffTotal) {
+ GroupConsumeInfo info = new GroupConsumeInfo();
+ info.setGroup(group);
+ // Keep the connection count identical so compareTo falls through to
the diffTotal
+ // comparison that is under test.
+ info.setCount(1);
+ info.setDiffTotal(diffTotal);
+ return info;
+ }
}
diff --git
a/tools/src/test/java/org/apache/rocketmq/tools/command/message/PrintMessageByQueueCommandTest.java
b/tools/src/test/java/org/apache/rocketmq/tools/command/message/PrintMessageByQueueCommandTest.java
new file mode 100644
index 0000000000..ac5f8a29a0
--- /dev/null
+++
b/tools/src/test/java/org/apache/rocketmq/tools/command/message/PrintMessageByQueueCommandTest.java
@@ -0,0 +1,88 @@
+/*
+ * 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.tools.command.message;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class PrintMessageByQueueCommandTest {
+
+ @Test
+ public void testCompareToIsAntisymmetricBeyondIntegerMaxValue() {
+ PrintMessageByQueueCommand.TagCountBean zero =
newTagCountBean("tag-zero", 0L);
+ PrintMessageByQueueCommand.TagCountBean overInt =
newTagCountBean("tag-over-int", (long) Integer.MAX_VALUE + 1L);
+
+ // (int) (o.count - this.count) wraps for this pair and returns a
negative value in
+ // both directions, so assert the sign of the comparison is properly
inverted.
+ Assert.assertTrue(zero.compareTo(overInt) > 0);
+ Assert.assertTrue(overInt.compareTo(zero) < 0);
+ Assert.assertEquals(Integer.signum(overInt.compareTo(zero)),
-Integer.signum(zero.compareTo(overInt)));
+ }
+
+ @Test
+ public void testCollectionsSortOrdersTagCountsDescending() {
+ PrintMessageByQueueCommand.TagCountBean zero =
newTagCountBean("tag-zero", 0L);
+ PrintMessageByQueueCommand.TagCountBean overInt =
newTagCountBean("tag-over-int", (long) Integer.MAX_VALUE + 1L);
+ PrintMessageByQueueCommand.TagCountBean max =
newTagCountBean("tag-max", Long.MAX_VALUE);
+
+ List<PrintMessageByQueueCommand.TagCountBean> beans = new
ArrayList<>();
+ beans.add(zero);
+ beans.add(max);
+ beans.add(overInt);
+ Collections.sort(beans);
+
+ Assert.assertEquals(Arrays.asList("tag-max", "tag-over-int",
"tag-zero"), tagsOf(beans));
+ }
+
+ @Test
+ public void testEqualCountsCompareEqual() {
+ PrintMessageByQueueCommand.TagCountBean first =
newTagCountBean("tag-first", (long) Integer.MAX_VALUE + 1L);
+ PrintMessageByQueueCommand.TagCountBean second =
newTagCountBean("tag-second", (long) Integer.MAX_VALUE + 1L);
+
+ // Equal counts must produce an equal comparison in both directions,
otherwise
+ // TimSort treats the two entries as an inconsistent pair.
+ Assert.assertEquals(0, first.compareTo(second));
+ Assert.assertEquals(0, second.compareTo(first));
+ }
+
+ @Test
+ public void testCompareToHandlesLongMaxValueCounts() {
+ PrintMessageByQueueCommand.TagCountBean overInt =
newTagCountBean("tag-over-int", (long) Integer.MAX_VALUE + 1L);
+ PrintMessageByQueueCommand.TagCountBean max =
newTagCountBean("tag-max", Long.MAX_VALUE);
+
+ // The largest possible gap between two long counters must still order
correctly.
+ Assert.assertTrue(overInt.compareTo(max) > 0);
+ Assert.assertTrue(max.compareTo(overInt) < 0);
+ }
+
+ private static PrintMessageByQueueCommand.TagCountBean
newTagCountBean(String tag, long count) {
+ return new PrintMessageByQueueCommand.TagCountBean(tag, new
AtomicLong(count));
+ }
+
+ private static List<String>
tagsOf(List<PrintMessageByQueueCommand.TagCountBean> beans) {
+ List<String> tags = new ArrayList<>(beans.size());
+ for (PrintMessageByQueueCommand.TagCountBean bean : beans) {
+ tags.add(bean.getTag());
+ }
+ return tags;
+ }
+}
diff --git
a/tools/src/test/java/org/apache/rocketmq/tools/command/message/QueryMsgByUniqueKeySubCommandTest.java
b/tools/src/test/java/org/apache/rocketmq/tools/command/message/QueryMsgByUniqueKeySubCommandTest.java
index b24bd22db8..1a535dec42 100644
---
a/tools/src/test/java/org/apache/rocketmq/tools/command/message/QueryMsgByUniqueKeySubCommandTest.java
+++
b/tools/src/test/java/org/apache/rocketmq/tools/command/message/QueryMsgByUniqueKeySubCommandTest.java
@@ -19,6 +19,7 @@ package org.apache.rocketmq.tools.command.message;
import java.lang.reflect.Field;
import java.net.InetSocketAddress;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -58,6 +59,7 @@ import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExtImpl;
import org.apache.rocketmq.tools.command.SubCommandException;
import org.assertj.core.util.Lists;
+import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
@@ -253,4 +255,31 @@ public class QueryMsgByUniqueKeySubCommandTest {
cmd.execute(commandLine, options, null);
}
+
+ @Test
+ public void
testStoreTimestampComparatorKeepsOlderMessageFirstAcrossIntRange() {
+ MessageExt newer = new MessageExt();
+ newer.setMsgId("newer-message");
+ newer.setStoreTimestamp(System.currentTimeMillis());
+
+ MessageExt older = new MessageExt();
+ older.setMsgId("older-message");
+ // More than Integer.MAX_VALUE milliseconds (roughly 24.8 days)
separate the two
+ // messages, so the old (int) (o1 - o2) lambda wrapped and reversed
their order.
+ long gap = (long) Integer.MAX_VALUE + 1L;
+ older.setStoreTimestamp(newer.getStoreTimestamp() - gap);
+
+ // The comparator itself must report the older message as smaller in
both directions.
+
Assert.assertTrue(QueryMsgByUniqueKeySubCommand.STORE_TIMESTAMP_COMPARATOR.compare(older,
newer) < 0);
+
Assert.assertTrue(QueryMsgByUniqueKeySubCommand.STORE_TIMESTAMP_COMPARATOR.compare(newer,
older) > 0);
+
+ List<MessageExt> messages = new ArrayList<>();
+ messages.add(newer);
+ messages.add(older);
+ Collections.sort(messages,
QueryMsgByUniqueKeySubCommand.STORE_TIMESTAMP_COMPARATOR);
+
+ Assert.assertSame(older, messages.get(0));
+ Assert.assertSame(newer, messages.get(1));
+ Assert.assertEquals(0,
QueryMsgByUniqueKeySubCommand.STORE_TIMESTAMP_COMPARATOR.compare(older, older));
+ }
}
