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 89e6f2615 fix(trace): preserve send status in legacy producer traces
(#6620)
89e6f2615 is described below
commit 89e6f2615285212d7040f3f11be2c1c54c58adcb
Author: Xiao Yang <[email protected]>
AuthorDate: Thu Oct 15 18:37:19 2026 +0800
fix(trace): preserve send status in legacy producer traces (#6620)
---
.../provider/apache/RocketMQMessageProvider.java | 9 +++-
.../apache/RocketMQMessageProviderTest.java | 57 ++++++++++++++++++++++
2 files changed, 65 insertions(+), 1 deletion(-)
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 b14f5386c..4afed6fb9 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
@@ -878,11 +878,18 @@ public class RocketMQMessageProvider implements
MessageProvider {
// Pub layout (RocketMQ 5.5.0 TraceDataEncoder):
// type, time, region, group, topic, msgId,
// tags, keys, storeHost, bodyLength, costTime, msgType,
offsetMsgId, isSuccess
+ // Legacy 13-field records omit offsetMsgId; newer records may append
clientHost.
private TraceNodeVO buildProduceNode(String[] f) {
+ // Match the SDK's split behavior when selecting the layout: trailing
empty columns do not count.
+ int fieldCount = f.length;
+ while (fieldCount > 0 && f[fieldCount - 1].isEmpty()) {
+ fieldCount--;
+ }
+ int successIndex = fieldCount == 13 ? 12 : 13;
return TraceNodeVO.builder()
.title("produce")
.timestamp(parseLong(field(f, 1)))
- .status(parseBoolean(field(f, 13)) ? "finish" : "failed")
+ .status(parseBoolean(field(f, successIndex)) ? "finish" :
"failed")
.costTime(parseLong(field(f, 10)))
.description("producer=" + field(f, 3) + ", storeHost=" +
field(f, 8))
.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 f244e2e82..ff4f09fb4 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
@@ -66,6 +66,8 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
@@ -917,6 +919,61 @@ class RocketMQMessageProviderTest {
assertThat(record.isBodyTruncated()).isTrue();
}
+ @ParameterizedTest
+ @CsvSource({"13,true,false", "13,false,false", "14,true,false",
"14,false,false",
+ "15,true,false", "15,false,false", "13,true,true",
"13,false,true",
+ "14,true,true", "14,false,true", "15,true,true",
"15,false,true"})
+ void getMessageTraceShouldMatchSdkProducerTraceLayoutsTest(int fieldCount,
boolean success,
+ boolean
emptyOptionalFields) throws Exception {
+ List<String> fields = new ArrayList<>(List.of("Pub", "1000", "cn",
"prod-group", "TopicA", "msg-123",
+ "tag1", "key1", "broker:10911", "15", "50", "0"));
+ if (fieldCount >= 14) {
+ fields.add(fieldCount == 14 && emptyOptionalFields ? "" :
"offset-1");
+ }
+ fields.add(Boolean.toString(success));
+ if (fieldCount == 15) {
+ fields.add(emptyOptionalFields ? "" : "producer-host:1000");
+ }
+ String context = traceContext(fields.toArray(String[]::new));
+ if (fieldCount == 13 && emptyOptionalFields) {
+ context += TraceConstants.CONTENT_SPLITOR;
+ }
+ String body = traceBody(context);
+
assertThat(TraceDataEncoder.decoderFromTraceDataString(body).getFirst().isSuccess()).isEqualTo(success);
+ MessageExt traceMessage = new MessageExt();
+ traceMessage.setBody(body.getBytes(StandardCharsets.UTF_8));
+ when(adminExt.queryMessage(anyString(), anyString(), anyInt(),
anyLong(), anyLong()))
+ .thenReturn(new QueryResult(0L, List.of(traceMessage)));
+
+ TraceRecordVO record = provider.getMessageTrace("instance-a",
"msg-123", "orders");
+
+ assertThat(record.getNodes()).hasSize(1);
+ TraceNodeVO produce = record.getNodes().getFirst();
+ assertThat(produce.getTitle()).isEqualTo("produce");
+ assertThat(produce.getStatus()).isEqualTo(success ? "finish" :
"failed");
+ assertThat(produce.getTimestamp()).isEqualTo(1000L);
+ assertThat(produce.getCostTime()).isEqualTo(50L);
+ assertThat(produce.getDescription()).isEqualTo("producer=prod-group,
storeHost=broker:10911");
+ }
+
+ @Test
+ void
getMessageTraceByKeyShouldPreserveLegacyProducerSuccessAndFilterOtherKeysTest()
throws Exception {
+ String matching = traceContext("Pub", "1000", "cn", "prod-group",
"TopicA", "msg-123",
+ "tag1", "shared-key", "broker:10911", "15", "50", "0", "true");
+ String unrelated = traceContext("Pub", "2000", "cn", "other-group",
"TopicA", "msg-other",
+ "tag1", "other-key", "broker:10911", "15", "50", "0", "true");
+ MessageExt traceMessage = new MessageExt();
+ traceMessage.setBody(traceBody(matching,
unrelated).getBytes(StandardCharsets.UTF_8));
+ when(adminExt.queryMessage(anyString(), anyString(), anyInt(),
anyLong(), anyLong()))
+ .thenReturn(new QueryResult(0L, List.of(traceMessage)));
+
+ TraceRecordVO record = provider.getMessageTraceByKey("instance-a",
"shared-key", "orders", null);
+
+ assertThat(record.getNodes()).hasSize(1);
+
assertThat(record.getNodes().getFirst().getStatus()).isEqualTo("finish");
+
assertThat(record.getNodes().getFirst().getDescription()).contains("prod-group");
+ }
+
void getMessageTraceParsesBatchedPubAndSubAfterContexts() throws Exception
{
// Field order follows RocketMQ 5.5.0 TraceDataEncoder:
// Pub = type, time, region, group,
