This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new a7e26555 fix(trace): parse batched contexts and EndTransaction state
(#1504)
a7e26555 is described below
commit a7e265558a0f83a800de2062d93ce30651977dd8
Author: majialong <[email protected]>
AuthorDate: Tue Aug 11 20:30:42 2026 +0800
fix(trace): parse batched contexts and EndTransaction state (#1504)
---
.../provider/apache/RocketMQMessageProvider.java | 32 ++++++++-------
.../apache/RocketMQMessageProviderTest.java | 45 +++++++++++++---------
2 files changed, 44 insertions(+), 33 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java
index 64ad563c..01433857 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java
@@ -20,6 +20,7 @@ import org.apache.rocketmq.client.QueryResult;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.PullStatus;
+import org.apache.rocketmq.client.trace.TraceConstants;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageId;
@@ -69,7 +70,6 @@ import java.util.Set;
public class RocketMQMessageProvider implements MessageProvider {
private static final String TRACE_TOPIC = "RMQ_SYS_TRACE_TOPIC";
- private static final char FIELD_SEPARATOR = '\u0001';
private static final int KEY_QUERY_MAX = 64;
private static final int TRACE_QUERY_MAX = 64;
private static final int DEFAULT_TOPIC_LIMIT = 200;
@@ -334,8 +334,9 @@ public class RocketMQMessageProvider implements
MessageProvider {
}
/**
- * Parse a trace message body. Each line is one trace context whose fields
are separated by
- * the SOH character ({@code \u0001}); the first field is the trace type.
+ * Parse a trace message body. Trace contexts are separated by STX ({@code
\u0002}) and the
+ * fields in each context are separated by SOH ({@code \u0001}); the first
field is the trace
+ * type.
*/
private void parseTraceBody(byte[] body, String targetMsgId,
List<TraceNodeVO> nodes,
List<ConsumerStatusVO> consumerStatus) {
@@ -343,17 +344,17 @@ public class RocketMQMessageProvider implements
MessageProvider {
return;
}
String data = new String(body, StandardCharsets.UTF_8);
- for (String line : data.split("\n")) {
- if (!StringUtils.hasText(line)) {
+ for (String context :
data.split(String.valueOf(TraceConstants.FIELD_SPLITOR))) {
+ if (!StringUtils.hasText(context)) {
continue;
}
- String[] fields = line.split(String.valueOf(FIELD_SEPARATOR), -1);
+ String[] fields =
context.split(String.valueOf(TraceConstants.CONTENT_SPLITOR), -1);
if (fields.length == 0) {
continue;
}
String traceType = fields[0].trim();
- // The message id column differs by trace type: Pub/EndTransaction
encode the trace bean
- // (msgId at index 5), while SubAfter in RocketMQ 5.3.3 places
msgId at index 2.
+ // The message ID column differs by trace type: Pub/EndTransaction
place msgId at
+ // index 5, while SubAfter places it at index 2 in RocketMQ 5.5.0.
int msgIdIndex = "SubAfter".equals(traceType) ? 2 : 5;
if (!targetMsgId.equals(field(fields, msgIdIndex))) {
continue;
@@ -375,12 +376,13 @@ public class RocketMQMessageProvider implements
MessageProvider {
break;
}
} catch (Exception e) {
- log.debug("Skipping unparseable trace line: {}",
e.getMessage());
+ log.debug("Skipping unparseable trace context: {}",
e.getMessage());
}
}
}
- // Pub layout (RocketMQ 5.3.3 TraceDataEncoder): type, time, region,
group, topic, msgId,
+ // Pub layout (RocketMQ 5.5.0 TraceDataEncoder):
+ // type, time, region, group, topic, msgId,
// tags, keys, storeHost, bodyLength, costTime, msgType,
offsetMsgId, isSuccess
private TraceNodeVO buildProduceNode(String[] f) {
return TraceNodeVO.builder()
@@ -392,7 +394,8 @@ public class RocketMQMessageProvider implements
MessageProvider {
.build();
}
- // SubAfter layout (RocketMQ 5.3.3 TraceDataEncoder): type, requestId,
msgId, costTime,
+ // SubAfter layout (RocketMQ 5.5.0 TraceDataEncoder):
+ // type, requestId, msgId, costTime,
// isSuccess, keys, contextCode, timeStamp, groupName. The
trailing
// timeStamp/groupName columns may be absent when the trace
has no region info,
// so lookups tolerate short lines.
@@ -415,15 +418,16 @@ public class RocketMQMessageProvider implements
MessageProvider {
.build();
}
- // EndTransaction layout (RocketMQ 5.3.3): type, time, region, group,
topic, msgId, tags,
- // keys, storeHost, bodyLength, costTime, msgType,
transactionId, txState
+ // EndTransaction layout (RocketMQ 5.5.0 TraceDataEncoder):
+ // type, time, region, group, topic, msgId, tags,
+ // keys, storeHost, msgType, transactionId, txState,
fromTransactionCheck
private TraceNodeVO buildTransactionNode(String[] f) {
return TraceNodeVO.builder()
.title("endTransaction")
.timestamp(parseLong(field(f, 1)))
.status("finish")
.costTime(0L)
- .description("group=" + field(f, 3) + ", transactionState=" +
field(f, 13))
+ .description("group=" + field(f, 3) + ", transactionState=" +
field(f, 11))
.build();
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java
index cc57db38..159572c1 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java
@@ -20,6 +20,7 @@ import org.apache.rocketmq.client.QueryResult;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.PullStatus;
+import org.apache.rocketmq.client.trace.TraceConstants;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
@@ -206,24 +207,20 @@ class RocketMQMessageProviderTest {
assertThat(record.isBodyTruncated()).isTrue();
}
- @Test
- void getMessageTraceParsesPubAndSubAfterPerRocketMq533Layout() throws
Exception {
- // Field order follows RocketMQ 5.3.3 TraceDataEncoder: Pub = type,
time, region, group,
+ void getMessageTraceParsesBatchedPubAndSubAfterContexts() throws Exception
{
+ // Field order follows RocketMQ 5.5.0 TraceDataEncoder:
+ // Pub = type, time, region, group,
// topic, msgId, tags, keys, storeHost, bodyLength, costTime, msgType,
offsetMsgId, isSuccess.
// SubAfter = type, requestId, msgId, costTime, isSuccess, keys,
contextCode, timeStamp,
// groupName.
- String pub = "Pub" + '' + "1000" + '' + "cn" + '' + "prod-group"
- + '' + "TopicA" + '' + "msg-123" + '' + "tag1" + '' +
"key1"
- + '' + "broker:10911" + '' + "15" + '' + "50" + '' + "0"
- + '' + "offset-1" + '' + "true";
- String subAfter = "SubAfter" + '' + "req-1" + '' + "msg-123" + '' +
"20"
- + '' + "true" + '' + "key1" + '' + "3" + '' + "3000"
- + '' + "cons-group";
- String otherMessage = "SubAfter" + '' + "req-2" + '' + "other-msg" +
'' + "5"
- + '' + "false" + '' + "key-other" + '' + "0" + '' + "0"
- + '' + "other-group";
+ String pub = traceContext("Pub", "1000", "cn", "prod-group", "TopicA",
"msg-123",
+ "tag1", "key1", "broker:10911", "15", "50", "0", "offset-1",
"true");
+ String subAfter = traceContext("SubAfter", "req-1", "msg-123", "20",
"true", "key1",
+ "3", "3000", "cons-group");
+ String otherMessage = traceContext("SubAfter", "req-2", "other-msg",
"5", "false",
+ "key-other", "0", "0", "other-group");
MessageExt traceMessage = new MessageExt();
- traceMessage.setBody(String.join("\n", pub, subAfter,
otherMessage).getBytes(StandardCharsets.UTF_8));
+ traceMessage.setBody(traceBody(pub, subAfter,
otherMessage).getBytes(StandardCharsets.UTF_8));
QueryResult queryResult = new QueryResult(0L, List.of(traceMessage));
when(adminExt.queryMessage(anyString(), anyString(), anyInt(),
anyLong(), anyLong()))
.thenReturn(queryResult);
@@ -252,10 +249,8 @@ class RocketMQMessageProviderTest {
@Test
void getMessageTraceParsesEndTransactionState() throws Exception {
- String body = "EndTransaction" + '' + "2000" + '' + "cn" + '' +
"tx-group"
- + '' + "TopicA" + '' + "msg-tx" + '' + "tag2" + '' + "key2"
- + '' + "broker:10911" + '' + "10" + '' + "40" + '' + "0"
- + '' + "tx-1" + '' + "COMMIT_MESSAGE";
+ String body = traceBody(traceContext("EndTransaction", "2000", "cn",
"tx-group", "TopicA",
+ "msg-tx", "tag2", "key2", "broker:10911", "0", "tx-1",
"COMMIT_MESSAGE", "false"));
MessageExt traceMessage = new MessageExt();
traceMessage.setBody(body.getBytes(StandardCharsets.UTF_8));
QueryResult queryResult = new QueryResult(0L, List.of(traceMessage));
@@ -267,7 +262,10 @@ class RocketMQMessageProviderTest {
assertThat(record.getNodes()).hasSize(1);
TraceNodeVO transaction = record.getNodes().get(0);
assertThat(transaction.getTitle()).isEqualTo("endTransaction");
-
assertThat(transaction.getDescription()).contains("tx-group").contains("COMMIT_MESSAGE");
+ assertThat(transaction.getDescription())
+ .contains("tx-group")
+ .contains("transactionState=COMMIT_MESSAGE")
+ .doesNotContain("transactionState=false");
assertThat(record.getConsumerStatus()).isEmpty();
}
@@ -283,4 +281,13 @@ class RocketMQMessageProviderTest {
verify(queryHistoryService, never()).recordTraceQuery(anyString(),
anyString(), any(), anyInt(), anyInt());
}
+
+ private static String traceContext(String... fields) {
+ return String.join(String.valueOf(TraceConstants.CONTENT_SPLITOR),
fields);
+ }
+
+ private static String traceBody(String... contexts) {
+ return String.join(String.valueOf(TraceConstants.FIELD_SPLITOR),
contexts)
+ + TraceConstants.FIELD_SPLITOR;
+ }
}
