diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/ExtraInfoUtil.java b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/ExtraInfoUtil.java index bba6063f61f..ebb70590447 100644 --- a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/ExtraInfoUtil.java +++ b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/ExtraInfoUtil.java @@ -42,25 +42,25 @@ public static String[] split(String extraInfo) { return extraInfo.split(MessageConst.KEY_SEPARATOR); } - public static Long getCkQueueOffset(String[] extraInfoStrs) { + public static long getCkQueueOffset(String[] extraInfoStrs) { if (extraInfoStrs == null || extraInfoStrs.length < 1) { throw new IllegalArgumentException("getCkQueueOffset fail, extraInfoStrs length " + (extraInfoStrs == null ? 0 : extraInfoStrs.length)); } - return Long.valueOf(extraInfoStrs[0]); + return Long.parseLong(extraInfoStrs[0]); } - public static Long getPopTime(String[] extraInfoStrs) { + public static long getPopTime(String[] extraInfoStrs) { if (extraInfoStrs == null || extraInfoStrs.length < 2) { throw new IllegalArgumentException("getPopTime fail, extraInfoStrs length " + (extraInfoStrs == null ? 0 : extraInfoStrs.length)); } - return Long.valueOf(extraInfoStrs[1]); + return Long.parseLong(extraInfoStrs[1]); } - public static Long getInvisibleTime(String[] extraInfoStrs) { + public static long getInvisibleTime(String[] extraInfoStrs) { if (extraInfoStrs == null || extraInfoStrs.length < 3) { throw new IllegalArgumentException("getInvisibleTime fail, extraInfoStrs length " + (extraInfoStrs == null ? 0 : extraInfoStrs.length)); } - return Long.valueOf(extraInfoStrs[2]); + return Long.parseLong(extraInfoStrs[2]); } public static int getReviveQid(String[] extraInfoStrs) { @@ -216,18 +216,27 @@ public static Map> parseMsgOffsetInfo(String msgOffsetInfo) { } for (String one : array) { - String[] split = one.split(MessageConst.KEY_SEPARATOR); - if (split.length != 3) { - throw new IllegalArgumentException("parse msgOffsetMap error, " + msgOffsetMap); + long separators = locateEntrySeparators(one); + if (separators < 0) { + throw new IllegalArgumentException("parse msgOffsetInfo error, " + msgOffsetInfo); } - String key = split[0] + "@" + split[1]; + int sep1 = (int) (separators >>> 32); + int sep2 = (int) separators; + String key = buildEntryKey(one, sep1, sep2); if (msgOffsetMap.containsKey(key)) { - throw new IllegalArgumentException("parse msgOffsetMap error, duplicate, " + msgOffsetMap); + throw new IllegalArgumentException("parse msgOffsetInfo error, duplicate, " + msgOffsetInfo); } - msgOffsetMap.put(key, new ArrayList<>(8)); - String[] msgOffsets = split[2].split(","); - for (String msgOffset : msgOffsets) { - msgOffsetMap.get(key).add(Long.valueOf(msgOffset)); + List msgOffsets = new ArrayList<>(8); + msgOffsetMap.put(key, msgOffsets); + int start = sep2 + 1; + while (start < one.length()) { + int comma = one.indexOf(',', start); + int end = comma < 0 ? one.length() : comma; + msgOffsets.add(Long.valueOf(one.substring(start, end))); + if (comma < 0) { + break; + } + start = comma + 1; } } @@ -247,15 +256,17 @@ public static Map parseStartOffsetInfo(String startOffsetInfo) { } for (String one : array) { - String[] split = one.split(MessageConst.KEY_SEPARATOR); - if (split.length != 3) { + long separators = locateEntrySeparators(one); + if (separators < 0) { throw new IllegalArgumentException("parse startOffsetInfo error, " + startOffsetInfo); } - String key = split[0] + "@" + split[1]; + int sep1 = (int) (separators >>> 32); + int sep2 = (int) separators; + String key = buildEntryKey(one, sep1, sep2); if (startOffsetMap.containsKey(key)) { throw new IllegalArgumentException("parse startOffsetInfo error, duplicate, " + startOffsetInfo); } - startOffsetMap.put(key, Long.valueOf(split[2])); + startOffsetMap.put(key, Long.valueOf(one.substring(sep2 + 1))); } return startOffsetMap; @@ -274,20 +285,48 @@ public static Map parseOrderCountInfo(String orderCountInfo) { } for (String one : array) { - String[] split = one.split(MessageConst.KEY_SEPARATOR); - if (split.length != 3) { + long separators = locateEntrySeparators(one); + if (separators < 0) { throw new IllegalArgumentException("parse orderCountInfo error, " + orderCountInfo); } - String key = split[0] + "@" + split[1]; + int sep1 = (int) (separators >>> 32); + int sep2 = (int) separators; + String key = buildEntryKey(one, sep1, sep2); if (startOffsetMap.containsKey(key)) { throw new IllegalArgumentException("parse orderCountInfo error, duplicate, " + orderCountInfo); } - startOffsetMap.put(key, Integer.valueOf(split[2])); + startOffsetMap.put(key, Integer.valueOf(one.substring(sep2 + 1))); } return startOffsetMap; } + /** + * Locates the two {@link MessageConst#KEY_SEPARATOR} positions of an entry laid out as + * {@code retryFlag queueId value}, packed as {@code (sep1 << 32) | sep2}. Returns a negative + * value when the entry does not have exactly three non-empty fields. This is stricter than + * the previous split-based validation, which half-accepted entries with empty fields or + * trailing separators; such corrupt entries are now rejected with the same + * {@code IllegalArgumentException} the callers already throw for other malformed shapes. + */ + private static long locateEntrySeparators(String one) { + int sep1 = one.indexOf(MessageConst.KEY_SEPARATOR); + if (sep1 <= 0) { + return -1; + } + int sep2 = one.indexOf(MessageConst.KEY_SEPARATOR, sep1 + 1); + if (sep2 <= sep1 + 1 || sep2 >= one.length() - 1 + || one.indexOf(MessageConst.KEY_SEPARATOR, sep2 + 1) >= 0) { + return -1; + } + return ((long) sep1 << 32) | sep2; + } + + private static String buildEntryKey(String one, int sep1, int sep2) { + return new StringBuilder(sep2 + 1) + .append(one, 0, sep1).append('@').append(one, sep1 + 1, sep2).toString(); + } + public static List parseLiteOrderCountInfo(String orderCountInfo, int msgCount) { if (StringUtils.isEmpty(orderCountInfo)) { return null; diff --git a/remoting/src/test/java/org/apache/rocketmq/remoting/protocol/header/ExtraInfoUtilTest.java b/remoting/src/test/java/org/apache/rocketmq/remoting/protocol/header/ExtraInfoUtilTest.java index 8081f386cb6..bd6f4f5d269 100644 --- a/remoting/src/test/java/org/apache/rocketmq/remoting/protocol/header/ExtraInfoUtilTest.java +++ b/remoting/src/test/java/org/apache/rocketmq/remoting/protocol/header/ExtraInfoUtilTest.java @@ -16,10 +16,15 @@ */ package org.apache.rocketmq.remoting.protocol.header; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; import java.util.Map; +import org.apache.rocketmq.common.KeyBuilder; import org.junit.Test; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; public class ExtraInfoUtilTest { @@ -43,4 +48,50 @@ public void testOrderCountInfo() { assertEquals(queueIdCount, orderCountInfo.get(queueIdKey)); assertEquals(queueOffsetCount, orderCountInfo.get(queueOffsetKey)); } + + @Test + public void testStartOffsetInfoRoundTrip() { + String retryTopic = KeyBuilder.buildPopRetryTopicV2("TOPIC", "GROUP"); + StringBuilder sb = new StringBuilder(); + ExtraInfoUtil.buildStartOffsetInfo(sb, "TOPIC", 0, 100L); + ExtraInfoUtil.buildStartOffsetInfo(sb, retryTopic, 1, 200L); + + Map map = ExtraInfoUtil.parseStartOffsetInfo(sb.toString()); + assertEquals(Long.valueOf(100L), map.get(ExtraInfoUtil.getStartOffsetInfoMapKey("TOPIC", 0))); + assertEquals(Long.valueOf(200L), map.get(ExtraInfoUtil.getStartOffsetInfoMapKey(retryTopic, 1))); + } + + @Test + public void testMsgOffsetInfoRoundTrip() { + StringBuilder sb = new StringBuilder(); + ExtraInfoUtil.buildMsgOffsetInfo(sb, "TOPIC", 0, Arrays.asList(1L, 2L, 3L)); + ExtraInfoUtil.buildMsgOffsetInfo(sb, "TOPIC", 1, Collections.singletonList(9L)); + + Map> map = ExtraInfoUtil.parseMsgOffsetInfo(sb.toString()); + assertEquals(Arrays.asList(1L, 2L, 3L), map.get(ExtraInfoUtil.getStartOffsetInfoMapKey("TOPIC", 0))); + assertEquals(Collections.singletonList(9L), map.get(ExtraInfoUtil.getStartOffsetInfoMapKey("TOPIC", 1))); + } + + @Test + public void testExtraInfoGetters() { + String extraInfo = ExtraInfoUtil.buildExtraInfo(100L, 1690000000000L, 60000L, 3, "TOPIC", "broker-a", 2, 101L); + String[] parts = ExtraInfoUtil.split(extraInfo); + + assertEquals(100L, ExtraInfoUtil.getCkQueueOffset(parts)); + assertEquals(1690000000000L, ExtraInfoUtil.getPopTime(parts)); + assertEquals(60000L, ExtraInfoUtil.getInvisibleTime(parts)); + assertEquals(3, ExtraInfoUtil.getReviveQid(parts)); + assertEquals("broker-a", ExtraInfoUtil.getBrokerName(parts)); + assertEquals(2, ExtraInfoUtil.getQueueId(parts)); + assertEquals(101L, ExtraInfoUtil.getQueueOffset(parts)); + } + + @Test + public void testParseMalformedEntriesThrow() { + for (String bad : new String[] {"0 1", "0 1 2 3", " 0 1", "0 1", "0 1 "}) { + assertThrows(IllegalArgumentException.class, () -> ExtraInfoUtil.parseStartOffsetInfo(bad)); + assertThrows(IllegalArgumentException.class, () -> ExtraInfoUtil.parseOrderCountInfo(bad)); + assertThrows(IllegalArgumentException.class, () -> ExtraInfoUtil.parseMsgOffsetInfo(bad)); + } + } }