Skip to content

[ISSUE #10973] Reduce allocation in ExtraInfoUtil parsing and primitive getters #10975

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wang-jiahua wants to merge 1 commit into
base: develop
Choose a base branch
from
+113 −23
Open

[ISSUE #10973] Reduce allocation in ExtraInfoUtil parsing and primitive getters #10975

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
85 changes: 62 additions & 23 deletions remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/ExtraInfoUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down Expand Up @@ -216,18 +216,27 @@ public static Map<String, List<Long>> 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<Long> 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;
}
}

Expand All @@ -247,15 +256,17 @@ public static Map<String, Long> 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;
Expand All @@ -274,20 +285,48 @@ public static Map<String, Integer> 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<Integer> parseLiteOrderCountInfo(String orderCountInfo, int msgCount) {
if (StringUtils.isEmpty(orderCountInfo)) {
return null;
Expand Down
51 changes: 51 additions & 0 deletions remoting/src/test/java/org/apache/rocketmq/remoting/protocol/header/ExtraInfoUtilTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {

Expand All @@ -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<String, Long> 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<String, List<Long>> 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));
}
}
}
Loading