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 96ad1b068 feat(trace): support custom trace topic and trace lookup by
key (#2520)
96ad1b068 is described below
commit 96ad1b0684412052ea1926ed102772c44e42e53d
Author: zhaohai <[email protected]>
AuthorDate: Thu Aug 27 14:31:36 2026 +0800
feat(trace): support custom trace topic and trace lookup by key (#2520)
Message trace queries hardcoded the RMQ_SYS_TRACE_TOPIC topic and only
supported lookup by message id. This change lets operators point trace
lookups at a custom trace topic and search the trace data by business key.
Backend:
- MessageProvider/InstanceProvider: new getMessageTrace(..., traceTopic)
and getMessageTraceByKey(..., traceTopic) defaults (501 for providers
that do not implement them).
- RocketMQMessageProvider: trace lookups resolve the trace topic through
effectiveTraceTopic (blank falls back to RMQ_SYS_TRACE_TOPIC); the new
by-key path scans the last 24h of trace data and parses every matching
context instead of filtering on a single message id.
- MessageController: /{msgId}/trace accepts an optional traceTopic param;
new GET /api/messages/trace-by-key endpoint.
Frontend:
- api/service: getMessageTrace accepts traceTopic; new
getMessageTraceByKey wrapper.
- Message detail trace tab gains a msgid/key mode switch, a query value
input and an optional custom trace topic input with a manual
"查询轨迹" button; the automatic lookup on opening a message keeps the
existing msgId behavior.
---
.../studio/instance/message/MessageController.java | 15 ++-
.../studio/instance/message/MessageProvider.java | 16 +++
.../studio/instance/message/MessageService.java | 28 +++++
.../rocketmq/studio/provider/InstanceProvider.java | 16 +++
.../provider/apache/ApacheInstanceProvider.java | 10 ++
.../provider/apache/RocketMQMessageProvider.java | 74 ++++++++++--
.../instance/message/MessageControllerTest.java | 34 +++++-
.../instance/message/MessageServiceTest.java | 38 ++++++
.../apache/RocketMQMessageProviderTest.java | 56 +++++++++
web/src/api/message.ts | 22 +++-
.../__tests__/MessagePageAsyncState.test.tsx | 34 +++++-
web/src/pages/instance/message.tsx | 131 +++++++++++++++++----
web/src/services/messageService.ts | 22 +++-
13 files changed, 456 insertions(+), 40 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageController.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageController.java
index 506d4eee1..7810512e3 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageController.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageController.java
@@ -61,8 +61,19 @@ public class MessageController {
public Result<TraceRecordVO> getMessageTrace(@PathVariable String msgId,
@RequestParam String instanceId,
// Optional for the
Apache/Aliyun providers; the Tencent
// provider requires a
non-empty topic (DescribeMessageTrace).
- @RequestParam(required =
false) String topic) {
- return Result.ok(messageService.getMessageTrace(instanceId, msgId,
topic));
+ @RequestParam(required =
false) String topic,
+ // Custom trace topic; when
blank the default
+ // RMQ_SYS_TRACE_TOPIC is
used.
+ @RequestParam(required =
false) String traceTopic) {
+ return Result.ok(messageService.getMessageTrace(instanceId, msgId,
topic, traceTopic));
+ }
+
+ @GetMapping("/trace-by-key")
+ public Result<TraceRecordVO> getMessageTraceByKey(@RequestParam String
instanceId,
+ @RequestParam String key,
+ @RequestParam(required =
false) String topic,
+ @RequestParam(required =
false) String traceTopic) {
+ return Result.ok(messageService.getMessageTraceByKey(instanceId, key,
topic, traceTopic));
}
@GetMapping("/queues")
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageProvider.java
index 7f11e66f3..2e8e479be 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageProvider.java
@@ -18,6 +18,7 @@ package org.apache.rocketmq.studio.instance.message;
import java.util.List;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
public interface MessageProvider {
List<MessageRecordVO> queryMessages(String instanceId, String topic,
String msgId, String tag, String key, Long startTime,
@@ -32,4 +33,19 @@ public interface MessageProvider {
default DirectConsumeMessageResultVO
consumeMessageDirectly(DirectConsumeMessageDTO request) {
throw new UnsupportedOperationException("Direct message consumption is
not supported");
}
+
+ /**
+ * Message trace lookup against a custom trace topic. When {@code
traceTopic} is blank, the
+ * provider falls back to its default trace topic (RMQ_SYS_TRACE_TOPIC).
+ */
+ default TraceRecordVO getMessageTrace(String instanceId, String msgId,
String topic, String traceTopic) {
+ throw new BusinessException(501, "Custom trace topic is not supported
by this provider");
+ }
+
+ /**
+ * Message trace lookup by business key, optionally against a custom trace
topic.
+ */
+ default TraceRecordVO getMessageTraceByKey(String instanceId, String key,
String topic, String traceTopic) {
+ throw new BusinessException(501, "Message trace by key is not
supported by this provider");
+ }
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageService.java
index 31522bc17..78aef0178 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageService.java
@@ -122,6 +122,34 @@ public class MessageService {
return result;
}
+ public TraceRecordVO getMessageTrace(String instanceId, String msgId,
String topic, String traceTopic) {
+ if (!StringUtils.hasText(msgId)) {
+ throw new BusinessException(400, "msgId is required");
+ }
+ if (!StringUtils.hasText(traceTopic)) {
+ // No custom trace topic: fall back to the legacy 3-arg path so
providers that
+ // only implement message-id tracing (Aliyun/Tencent) keep working
unchanged.
+ return getMessageTrace(instanceId, msgId, topic);
+ }
+ log.info("Getting message trace: msgId={}, topic={}, traceTopic={}",
msgId, topic, traceTopic);
+ TraceRecordVO result = providerRegistry.byInstanceId(instanceId)
+ .map(provider -> provider.getMessageTrace(instanceId, msgId,
topic, traceTopic))
+ .orElseGet(() -> messageProvider.getMessageTrace(instanceId,
msgId, topic, traceTopic));
+ recordTraceQuery(instanceId, msgId, topic, result);
+ return result;
+ }
+
+ public TraceRecordVO getMessageTraceByKey(String instanceId, String key,
String topic, String traceTopic) {
+ if (!StringUtils.hasText(key)) {
+ throw new BusinessException(400, "key is required");
+ }
+ log.info("Getting message trace by key: key={}, topic={},
traceTopic={}", key, topic, traceTopic);
+ // Trace query history is keyed by message id; key-based lookups are
intentionally
+ // not recorded so the key is not misreported as a message id.
+ return providerRegistry.byInstanceId(instanceId)
+ .map(provider -> provider.getMessageTraceByKey(instanceId,
key, topic, traceTopic))
+ .orElseGet(() ->
messageProvider.getMessageTraceByKey(instanceId, key, topic, traceTopic));
+ }
private void recordMessageQuery(String instanceId, String topic, String
msgId, String tag,
String key, Long startTime, Long endTime,
int resultCount) {
String queryType = StringUtils.hasText(msgId) ? "MSG_ID" :
StringUtils.hasText(key) ? "KEY" : "TOPIC";
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProvider.java
index e7aceb622..e67849761 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProvider.java
@@ -18,6 +18,7 @@ package org.apache.rocketmq.studio.provider;
import org.apache.rocketmq.studio.common.domain.PageResult;
import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.common.util.Pagination;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
import org.apache.rocketmq.studio.instance.group.QueueProgressVO;
@@ -114,4 +115,19 @@ public interface InstanceProvider {
default DirectConsumeMessageResultVO
consumeMessageDirectly(DirectConsumeMessageDTO request) {
throw new UnsupportedOperationException("Direct message consumption is
not supported");
}
+
+ /**
+ * Message trace lookup against a custom trace topic. When {@code
traceTopic} is blank, the
+ * provider falls back to its default trace topic (RMQ_SYS_TRACE_TOPIC).
+ */
+ default TraceRecordVO getMessageTrace(String instanceId, String msgId,
String topic, String traceTopic) {
+ throw new BusinessException(501, "Custom trace topic is not supported
by this provider");
+ }
+
+ /**
+ * Message trace lookup by business key, optionally against a custom trace
topic.
+ */
+ default TraceRecordVO getMessageTraceByKey(String instanceId, String key,
String topic, String traceTopic) {
+ throw new BusinessException(501, "Message trace by key is not
supported by this provider");
+ }
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
index 1694901fa..49c32786a 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
@@ -168,4 +168,14 @@ public class ApacheInstanceProvider implements
InstanceProvider {
public DirectConsumeMessageResultVO
consumeMessageDirectly(DirectConsumeMessageDTO request) {
return messageProvider.consumeMessageDirectly(request);
}
+
+ @Override
+ public TraceRecordVO getMessageTrace(String instanceId, String msgId,
String topic, String traceTopic) {
+ return messageProvider.getMessageTrace(instanceId, msgId, topic,
traceTopic);
+ }
+
+ @Override
+ public TraceRecordVO getMessageTraceByKey(String instanceId, String key,
String topic, String traceTopic) {
+ return messageProvider.getMessageTraceByKey(instanceId, key, topic,
traceTopic);
+ }
}
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 009c0a84b..612c93720 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
@@ -423,8 +423,7 @@ public class RocketMQMessageProvider implements
MessageProvider {
@Override
public TraceRecordVO getMessageTrace(String instanceId, String msgId,
String topic) {
- return runtimeAdminClientResolver.execute(instanceId,
- adminExt -> getMessageTrace(instanceId, (DefaultMQAdminExt)
adminExt, msgId, topic));
+ return getMessageTrace(instanceId, msgId, topic, null);
}
@Override
@@ -440,7 +439,20 @@ public class RocketMQMessageProvider implements
MessageProvider {
});
}
- private TraceRecordVO getMessageTrace(String instanceId, DefaultMQAdminExt
adminExt, String msgId, String topic) {
+ @Override
+ public TraceRecordVO getMessageTrace(String instanceId, String msgId,
String topic, String traceTopic) {
+ return runtimeAdminClientResolver.execute(instanceId,
+ adminExt -> getMessageTrace(instanceId, (DefaultMQAdminExt)
adminExt, msgId, topic, traceTopic));
+ }
+
+ @Override
+ public TraceRecordVO getMessageTraceByKey(String instanceId, String key,
String topic, String traceTopic) {
+ return runtimeAdminClientResolver.execute(instanceId,
+ adminExt -> getMessageTraceByKey(instanceId,
(DefaultMQAdminExt) adminExt, key, topic, traceTopic));
+ }
+
+ private TraceRecordVO getMessageTrace(String instanceId, DefaultMQAdminExt
adminExt, String msgId, String topic,
+ String traceTopic) {
long now = System.currentTimeMillis();
long begin;
@@ -465,10 +477,11 @@ public class RocketMQMessageProvider implements
MessageProvider {
List<ConsumerStatusVO> consumerStatus = new ArrayList<>();
try {
- QueryResult traceResult = adminExt.queryMessage(TRACE_TOPIC,
msgId, TRACE_QUERY_MAX, begin, end);
+ QueryResult traceResult =
+ adminExt.queryMessage(effectiveTraceTopic(traceTopic),
msgId, TRACE_QUERY_MAX, begin, end);
if (traceResult != null && traceResult.getMessageList() != null) {
for (MessageExt traceMessage : traceResult.getMessageList()) {
- parseTraceBody(traceMessage.getBody(), msgId, nodes,
consumerStatus);
+ parseTraceBody(traceMessage.getBody(), msgId, nodes,
consumerStatus, true);
}
}
} catch (BusinessException e) {
@@ -503,6 +516,49 @@ public class RocketMQMessageProvider implements
MessageProvider {
return false;
}
+ /**
+ * Trace lookup by business key. The key query already scopes the returned
trace messages to
+ * the requested message, so the body parser does not filter on a message
id. The original
+ * message topic is not required to query the global trace topic but is
kept in the signature
+ * for API symmetry and logged for diagnostics.
+ */
+ private TraceRecordVO getMessageTraceByKey(String instanceId,
DefaultMQAdminExt adminExt, String key,
+ String topic, String
traceTopic) {
+ log.debug("Trace by key: key={}, originalTopic={}, traceTopic={}",
key, topic,
+ effectiveTraceTopic(traceTopic));
+ long now = System.currentTimeMillis();
+ // No message id to derive a precise window from; scan the last 24h of
trace data.
+ long begin = now - ONE_DAY_MILLIS;
+ long end = now + 60_000L;
+
+ List<TraceNodeVO> nodes = new ArrayList<>();
+ List<ConsumerStatusVO> consumerStatus = new ArrayList<>();
+
+ try {
+ QueryResult traceResult =
+ adminExt.queryMessage(effectiveTraceTopic(traceTopic),
key, TRACE_QUERY_MAX, begin, end);
+ if (traceResult != null && traceResult.getMessageList() != null) {
+ for (MessageExt traceMessage : traceResult.getMessageList()) {
+ parseTraceBody(traceMessage.getBody(), null, nodes,
consumerStatus, false);
+ }
+ }
+ } catch (BusinessException e) {
+ throw e;
+ } catch (Exception e) {
+ log.warn("Trace query by key={} failed: {}", key, e.getMessage());
+ throw new BusinessException(502, "Failed to query message trace by
key: " + e.getMessage());
+ }
+
+ return TraceRecordVO.builder()
+ .nodes(nodes)
+ .consumerStatus(consumerStatus)
+ .build();
+ }
+
+ private String effectiveTraceTopic(String traceTopic) {
+ return StringUtils.hasText(traceTopic) ? traceTopic.trim() :
TRACE_TOPIC;
+ }
+
/**
* Attempts to resolve the store timestamp of the original message so the
trace
* query window can be derived from the message's own timeline rather than
the
@@ -534,10 +590,12 @@ public class RocketMQMessageProvider implements
MessageProvider {
/**
* 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.
+ * type. When {@code filterByMsgId} is true only contexts whose message id
matches
+ * {@code targetMsgId} are kept; otherwise every context is parsed (used
by key lookups where
+ * the query already scoped the trace messages to the requested key).
*/
private void parseTraceBody(byte[] body, String targetMsgId,
List<TraceNodeVO> nodes,
- List<ConsumerStatusVO> consumerStatus) {
+ List<ConsumerStatusVO> consumerStatus, boolean
filterByMsgId) {
if (body == null || body.length == 0) {
return;
}
@@ -554,7 +612,7 @@ public class RocketMQMessageProvider implements
MessageProvider {
// 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))) {
+ if (filterByMsgId && !targetMsgId.equals(field(fields,
msgIdIndex))) {
continue;
}
try {
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageControllerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageControllerTest.java
index 001dec2be..1e1d89424 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageControllerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageControllerTest.java
@@ -91,14 +91,44 @@ class MessageControllerTest {
@Test
void messageTraceShouldPassInstanceId() throws Exception {
TraceRecordVO trace =
TraceRecordVO.builder().nodes(List.of()).consumerStatus(List.of()).build();
- when(messageService.getMessageTrace("instance-a", "msg-001",
"orders")).thenReturn(trace);
+ when(messageService.getMessageTrace("instance-a", "msg-001", "orders",
null)).thenReturn(trace);
mockMvc.perform(get("/api/messages/msg-001/trace").param("instanceId",
"instance-a")
.param("topic", "orders"))
.andExpect(status().isOk())
.andExpect(jsonPath("$.code").value(200));
- verify(messageService).getMessageTrace("instance-a", "msg-001",
"orders");
+ verify(messageService).getMessageTrace("instance-a", "msg-001",
"orders", null);
+ }
+
+ @Test
+ void messageTraceShouldPassCustomTraceTopic() throws Exception {
+ TraceRecordVO trace =
TraceRecordVO.builder().nodes(List.of()).consumerStatus(List.of()).build();
+ when(messageService.getMessageTrace("instance-a", "msg-001", "orders",
"MY_TRACE"))
+ .thenReturn(trace);
+
+ mockMvc.perform(get("/api/messages/msg-001/trace").param("instanceId",
"instance-a")
+ .param("topic", "orders")
+ .param("traceTopic", "MY_TRACE"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.code").value(200));
+
+ verify(messageService).getMessageTrace("instance-a", "msg-001",
"orders", "MY_TRACE");
+ }
+
+ @Test
+ void traceByKeyShouldDelegateToService() throws Exception {
+ TraceRecordVO trace =
TraceRecordVO.builder().nodes(List.of()).consumerStatus(List.of()).build();
+ when(messageService.getMessageTraceByKey("instance-a", "ORDER-001",
"orders", null))
+ .thenReturn(trace);
+
+ mockMvc.perform(get("/api/messages/trace-by-key").param("instanceId",
"instance-a")
+ .param("key", "ORDER-001")
+ .param("topic", "orders"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.code").value(200));
+
+ verify(messageService).getMessageTraceByKey("instance-a", "ORDER-001",
"orders", null);
}
@Test
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageServiceTest.java
index 53b9b4571..6dfc4dce3 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageServiceTest.java
@@ -207,4 +207,42 @@ class MessageServiceTest {
verify(history, org.mockito.Mockito.times(1)).recordMessageQuery(
"instance-a", "TOPIC", "TopicA", null, null, null, 1000L,
2000L, 1);
}
+
+ @Test
+ void fallsBackToThreeArgTracePathWhenCustomTraceTopicIsBlank() {
+ MessageProvider fallback = mock(MessageProvider.class);
+ InstanceProvider provider = mock(InstanceProvider.class);
+ InstanceProviderRegistry registry =
mock(InstanceProviderRegistry.class);
+ MessageService service = new MessageService(fallback, registry,
mock(QueryHistoryService.class), mock(OperationAuditService.class));
+ TraceRecordVO trace =
TraceRecordVO.builder().nodes(List.of()).consumerStatus(List.of()).build();
+
when(registry.byInstanceId("instance-a")).thenReturn(Optional.of(provider));
+ when(provider.getMessageTrace("instance-a", "msg-001",
"orders")).thenReturn(trace);
+
+ service.getMessageTrace("instance-a", "msg-001", "orders", null);
+
+ verify(provider).getMessageTrace("instance-a", "msg-001", "orders");
+ verify(provider, org.mockito.Mockito.never()).getMessageTrace(
+ org.mockito.ArgumentMatchers.anyString(),
org.mockito.ArgumentMatchers.anyString(),
+ org.mockito.ArgumentMatchers.anyString(),
org.mockito.ArgumentMatchers.anyString());
+ verifyNoInteractions(fallback);
+ }
+
+ @Test
+ void keyTraceLookupDoesNotRecordTraceQueryHistory() {
+ MessageProvider fallback = mock(MessageProvider.class);
+ InstanceProviderRegistry registry =
mock(InstanceProviderRegistry.class);
+ QueryHistoryService history = mock(QueryHistoryService.class);
+ MessageService service = new MessageService(fallback, registry,
history, mock(OperationAuditService.class));
+ TraceRecordVO trace =
TraceRecordVO.builder().nodes(List.of()).consumerStatus(List.of()).build();
+ when(registry.byInstanceId("instance-a")).thenReturn(Optional.empty());
+ when(fallback.getMessageTraceByKey("instance-a", "ORDER-1", "orders",
null)).thenReturn(trace);
+
+ TraceRecordVO result = service.getMessageTraceByKey("instance-a",
"ORDER-1", "orders", null);
+
+ assertThat(result).isSameAs(trace);
+ verify(history, org.mockito.Mockito.never()).recordTraceQuery(
+ org.mockito.ArgumentMatchers.anyString(),
org.mockito.ArgumentMatchers.anyString(),
+ org.mockito.ArgumentMatchers.anyString(),
org.mockito.ArgumentMatchers.anyInt(),
+ org.mockito.ArgumentMatchers.anyInt());
+ }
}
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 af16942bd..b8fb72876 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
@@ -557,6 +557,62 @@ class RocketMQMessageProviderTest {
}
+ @Test
+ void getMessageTraceQueriesCustomTraceTopicWhenProvided() throws Exception
{
+ String pub = traceContext("Pub", "1000", "cn", "prod-group", "TopicA",
"msg-custom",
+ "tag1", "key1", "broker:10911", "15", "50", "0", "offset-1",
"true");
+ MessageExt traceMessage = new MessageExt();
+ traceMessage.setBody(traceBody(pub).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-custom", "orders",
"MY_TRACE_TOPIC");
+
+ assertThat(record.getNodes()).hasSize(1);
+ assertThat(record.getNodes().get(0).getTitle()).isEqualTo("produce");
+ ArgumentCaptor<String> topicCaptor =
ArgumentCaptor.forClass(String.class);
+ verify(adminExt).queryMessage(topicCaptor.capture(), eq("msg-custom"),
anyInt(), anyLong(), anyLong());
+ assertThat(topicCaptor.getValue()).isEqualTo("MY_TRACE_TOPIC");
+ }
+
+ @Test
+ void getMessageTraceByKeyParsesContextsOfDifferentMessagesSharingTheKey()
throws Exception {
+ // Two trace contexts belong to different message ids but share the
same business key;
+ // the key lookup must surface both instead of filtering on a single
message id.
+ String pubA = traceContext("Pub", "1000", "cn", "prod-group",
"TopicA", "msg-a",
+ "tag1", "shared-key", "broker:10911", "15", "50", "0",
"offset-1", "true");
+ String subB = traceContext("SubAfter", "req-b", "msg-b", "20", "true",
"shared-key",
+ "3", "3000", "cons-group");
+ MessageExt traceMessage = new MessageExt();
+ traceMessage.setBody(traceBody(pubA,
subB).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", "CUSTOM_TRACE");
+
+ assertThat(record.getNodes()).hasSize(2);
+ assertThat(record.getConsumerStatus()).hasSize(1);
+
assertThat(record.getConsumerStatus().get(0).getGroup()).isEqualTo("cons-group");
+ ArgumentCaptor<String> topicCaptor =
ArgumentCaptor.forClass(String.class);
+ ArgumentCaptor<String> keyCaptor =
ArgumentCaptor.forClass(String.class);
+ verify(adminExt).queryMessage(topicCaptor.capture(),
keyCaptor.capture(), anyInt(), anyLong(), anyLong());
+ assertThat(topicCaptor.getValue()).isEqualTo("CUSTOM_TRACE");
+ assertThat(keyCaptor.getValue()).isEqualTo("shared-key");
+ }
+
+ @Test
+ void getMessageTraceByKeyUsesDefaultTraceTopicWhenNotSpecified() throws
Exception {
+ when(adminExt.queryMessage(anyString(), anyString(), anyInt(),
anyLong(), anyLong()))
+ .thenReturn(new QueryResult(0L, List.of()));
+
+ TraceRecordVO record = provider.getMessageTraceByKey("instance-a",
"shared-key", null, null);
+
+ assertThat(record.getNodes()).isEmpty();
+ verify(adminExt).queryMessage(eq("RMQ_SYS_TRACE_TOPIC"),
eq("shared-key"), anyInt(), anyLong(), anyLong());
+ }
+
@Test
void getMessageTraceUsesDecodedOffsetToResolveQueryWindow() throws
Exception {
String msgId = "AC1E0A6400002A9F0000000001A3F2B1";
diff --git a/web/src/api/message.ts b/web/src/api/message.ts
index 10b9c478d..9fa57dfb7 100644
--- a/web/src/api/message.ts
+++ b/web/src/api/message.ts
@@ -139,10 +139,16 @@ export async function queryMessagePage(
return { ...res.data.data, items:
sortMessagesByStoreTimeDesc(res.data.data.items) };
}
-export async function getMessageTrace(msgId: string, instanceId?: string,
topic?: string) {
+export async function getMessageTrace(
+ msgId: string,
+ instanceId?: string,
+ topic?: string,
+ traceTopic?: string,
+) {
const params: Record<string, string> = {};
if (instanceId !== undefined) params.instanceId = instanceId;
if (topic !== undefined) params.topic = topic;
+ if (traceTopic !== undefined && traceTopic.trim()) params.traceTopic =
traceTopic.trim();
const res = await client.get<{ data: TraceRecord }>(
`/messages/${encodeURIComponent(msgId)}/trace`,
{ params },
@@ -158,6 +164,20 @@ export async function consumeMessageDirectly(data:
DirectConsumeMessageRequest)
return res.data.data;
}
+export async function getMessageTraceByKey(
+ key: string,
+ instanceId?: string,
+ topic?: string,
+ traceTopic?: string,
+): Promise<TraceRecord | null> {
+ const params: Record<string, string> = { key };
+ if (instanceId !== undefined) params.instanceId = instanceId;
+ if (topic !== undefined) params.topic = topic;
+ if (traceTopic !== undefined && traceTopic.trim()) params.traceTopic =
traceTopic.trim();
+ const res = await client.get<{ data: TraceRecord
}>('/messages/trace-by-key', { params });
+ return res.data.data;
+}
+
// ─── DLQ ────────────────────────────────────────────────────────
export async function listDLQGroups(instanceId: string, search?: string, page
= 1, pageSize = 20) {
const res = await client.get<{ data: DLQGroupPage }>('/dlq', {
diff --git a/web/src/pages/instance/__tests__/MessagePageAsyncState.test.tsx
b/web/src/pages/instance/__tests__/MessagePageAsyncState.test.tsx
index 005120d9c..6bdcde04d 100644
--- a/web/src/pages/instance/__tests__/MessagePageAsyncState.test.tsx
+++ b/web/src/pages/instance/__tests__/MessagePageAsyncState.test.tsx
@@ -26,6 +26,7 @@ import MessagePage from '../message';
const serviceMocks = vi.hoisted(() => ({
getMessageTrace: vi.fn(),
+ getMessageTraceByKey: vi.fn(),
queryMessages: vi.fn(),
consumeMessageDirectly: vi.fn(),
}));
@@ -133,6 +134,7 @@ describe('MessagePage async request ownership', () => {
beforeEach(() => {
vi.clearAllMocks();
serviceMocks.getMessageTrace.mockResolvedValue(null);
+ serviceMocks.getMessageTraceByKey.mockResolvedValue(null);
instanceFilterMocks.useInstanceFilter.mockReturnValue({
selectedInstanceId: 1,
selectInstance: vi.fn(),
@@ -208,7 +210,6 @@ describe('MessagePage async request ownership', () => {
await user.click(screen.getByRole('button', { name: /^search查询$/ }));
expect(await screen.findByText('Message query provider is not
configured')).toBeInTheDocument();
- expect(screen.getByRole('button', { name: /最近查询/ })).toBeDisabled();
});
it('surfaces unavailable message provider errors from trace requests', async
() => {
@@ -256,6 +257,37 @@ describe('MessagePage async request ownership', () => {
expect(serviceMocks.consumeMessageDirectly).not.toHaveBeenCalled();
});
+ it('queries trace by key with a custom trace topic from the trace tab',
async () => {
+ serviceMocks.queryMessages.mockResolvedValue([createMessage('message-a')]);
+
serviceMocks.getMessageTraceByKey.mockResolvedValue(createTrace('key-trace'));
+ const user = userEvent.setup();
+ renderPage();
+ await selectTopic(user);
+
+ await user.click(screen.getByRole('button', { name: /^search查询$/ }));
+ const row = await screen.findByRole('row', { name: /message-a/ });
+ await user.click(within(row).getByRole('button', { name: /轨迹/ }));
+
+ const dialog = await screen.findByRole('dialog', { name: '消息详情' });
+ await user.click(within(dialog).getByText('按 Message Key'));
+ const keyInput = within(dialog).getByPlaceholderText('输入 Message Key');
+ await user.clear(keyInput);
+ await user.type(keyInput, 'ORDER-001');
+ const traceTopicInput = within(dialog).getByPlaceholderText('轨迹
Topic（留空使用默认）');
+ await user.type(traceTopicInput, 'CUSTOM_TRACE');
+ await user.click(within(dialog).getByRole('button', { name: /查询轨迹/ }));
+
+ await waitFor(() => {
+ expect(serviceMocks.getMessageTraceByKey).toHaveBeenCalledWith(
+ 'ORDER-001',
+ 1,
+ 'topic-message-a',
+ 'CUSTOM_TRACE',
+ );
+ });
+ expect(await within(dialog).findByText('key-trace
description')).toBeInTheDocument();
+ });
+
it('keeps the latest query loading and ignores an earlier query result',
async () => {
const firstQuery = createDeferred<MessageRecord[]>();
const secondQuery = createDeferred<MessageRecord[]>();
diff --git a/web/src/pages/instance/message.tsx
b/web/src/pages/instance/message.tsx
index 01130a01f..7358085e2 100644
--- a/web/src/pages/instance/message.tsx
+++ b/web/src/pages/instance/message.tsx
@@ -62,6 +62,7 @@ import type { MessageQuery, MessageRecord, TraceRecord } from
'../../api/message
import {
consumeMessageDirectly,
getMessageTrace,
+ getMessageTraceByKey,
queryMessagePage,
} from '../../services/messageService';
import { listTopics } from '../../services/topicService';
@@ -266,6 +267,9 @@ const MessagePageContent = ({
const [traceLoading, setTraceLoading] = useState(false);
const [queryError, setQueryError] = useState<string | null>(null);
const [traceError, setTraceError] = useState<string | null>(null);
+ const [traceQueryMode, setTraceQueryMode] = useState<'msgid' |
'key'>('msgid');
+ const [traceQueryValue, setTraceQueryValue] = useState('');
+ const [customTraceTopic, setCustomTraceTopic] = useState('');
const [historyDrawerOpen, setHistoryDrawerOpen] = useState(false);
const [directConsumeOpen, setDirectConsumeOpen] = useState(false);
const [directConsumeGroup, setDirectConsumeGroup] = useState('');
@@ -402,6 +406,10 @@ const MessagePageContent = ({
setTraceData(null);
setTraceLoading(true);
setTraceError(null);
+ if (tab === 'trace') {
+ setTraceQueryMode('msgid');
+ setTraceQueryValue(record.msgId);
+ }
try {
const result = await getMessageTrace(record.msgId, selectedInstanceId,
record.topic);
if (traceGenerationRef.current !== requestGeneration) return;
@@ -418,6 +426,41 @@ const MessagePageContent = ({
}
};
+ const runTraceQuery = async () => {
+ const requestGeneration = traceGenerationRef.current + 1;
+ traceGenerationRef.current = requestGeneration;
+ const value = traceQueryValue.trim();
+ if (!value) {
+ setTraceError(traceQueryMode === 'key' ? '请输入 Message Key' : '请输入
Message ID');
+ return;
+ }
+ setTraceData(null);
+ setTraceLoading(true);
+ setTraceError(null);
+ try {
+ const result =
+ traceQueryMode === 'key'
+ ? await getMessageTraceByKey(
+ value,
+ selectedInstanceId,
+ selectedMsg?.topic,
+ customTraceTopic,
+ )
+ : await getMessageTrace(value, selectedInstanceId,
selectedMsg?.topic, customTraceTopic);
+ if (traceGenerationRef.current !== requestGeneration) return;
+ setTraceData(result);
+ setTraceError(null);
+ } catch (error) {
+ if (traceGenerationRef.current === requestGeneration) {
+ setTraceError(getErrorMessage(error, DEFAULT_TRACE_ERROR));
+ }
+ } finally {
+ if (traceGenerationRef.current === requestGeneration) {
+ setTraceLoading(false);
+ }
+ }
+ };
+
const closeDetail = () => {
traceGenerationRef.current += 1;
setModalOpen(false);
@@ -682,30 +725,70 @@ const MessagePageContent = ({
{
key: 'trace',
label: '消息轨迹',
- children: traceLoading ? (
- <Typography.Text type="secondary">正在加载轨迹数据…</Typography.Text>
- ) : traceError ? (
- <Alert showIcon type="warning" message={traceError} />
- ) : traceData?.nodes?.length ? (
- <Steps
- direction="vertical"
- size="small"
- items={traceData.nodes.map((node) => ({
- title: node.title,
- description: (
- <div style={{ fontSize: 14 }}>
- <div style={{ color: '#9CA3AF', fontFamily: 'monospace' }}>
- {formatTimeMs(node.timestamp)}
- </div>
- <div style={{ marginTop: 2 }}>{node.description}</div>
- <div style={{ color: '#9CA3AF', fontSize: 14 }}>耗时
{node.costTime}ms</div>
- </div>
- ),
- status: node.status,
- }))}
- />
- ) : (
- <Typography.Text type="secondary">暂无轨迹数据</Typography.Text>
+ children: (
+ <>
+ <Space wrap size={8} style={{ marginBottom: 16 }}>
+ <Segmented
+ size="small"
+ options={[
+ { value: 'msgid', label: '按 Message ID' },
+ { value: 'key', label: '按 Message Key' },
+ ]}
+ value={traceQueryMode}
+ onChange={(value) => setTraceQueryMode(value as 'msgid' | 'key')}
+ />
+ <Input
+ size="small"
+ style={{ width: 300 }}
+ placeholder={
+ traceQueryMode === 'key' ? '输入 Message Key' : '消息 ID（默认当前消息）'
+ }
+ value={traceQueryValue}
+ onChange={(event) => setTraceQueryValue(event.target.value)}
+ />
+ <Input
+ size="small"
+ style={{ width: 260 }}
+ placeholder="轨迹 Topic（留空使用默认）"
+ value={customTraceTopic}
+ onChange={(event) => setCustomTraceTopic(event.target.value)}
+ allowClear
+ />
+ <Button
+ size="small"
+ type="primary"
+ icon={<SearchOutlined />}
+ onClick={() => void runTraceQuery()}
+ >
+ 查询轨迹
+ </Button>
+ </Space>
+ {traceLoading ? (
+ <Typography.Text type="secondary">正在加载轨迹数据…</Typography.Text>
+ ) : traceError ? (
+ <Alert showIcon type="warning" message={traceError} />
+ ) : traceData?.nodes?.length ? (
+ <Steps
+ direction="vertical"
+ size="small"
+ items={traceData.nodes.map((node) => ({
+ title: node.title,
+ description: (
+ <div style={{ fontSize: 14 }}>
+ <div style={{ color: '#9CA3AF', fontFamily: 'monospace' }}>
+ {formatTimeMs(node.timestamp)}
+ </div>
+ <div style={{ marginTop: 2 }}>{node.description}</div>
+ <div style={{ color: '#9CA3AF', fontSize: 14 }}>耗时
{node.costTime}ms</div>
+ </div>
+ ),
+ status: node.status,
+ }))}
+ />
+ ) : (
+ <Typography.Text type="secondary">暂无轨迹数据</Typography.Text>
+ )}
+ </>
),
},
{
diff --git a/web/src/services/messageService.ts
b/web/src/services/messageService.ts
index 5c744570f..837aeebda 100644
--- a/web/src/services/messageService.ts
+++ b/web/src/services/messageService.ts
@@ -77,12 +77,25 @@ export async function getMessageTrace(
msgId: string,
instanceId?: string,
topic?: string,
+ traceTopic?: string,
): Promise<TraceRecord | null> {
if (isMockMode()) {
const trace = mockMessageTraces[msgId] as unknown as TraceRecord |
undefined;
return trace ? cloneTrace(trace) : null;
}
- return messageApi.getMessageTrace(msgId, instanceId, topic);
+ return messageApi.getMessageTrace(msgId, instanceId, topic, traceTopic);
+}
+
+export async function getMessageTraceByKey(
+ key: string,
+ instanceId?: string,
+ topic?: string,
+ traceTopic?: string,
+): Promise<TraceRecord | null> {
+ if (isMockMode()) {
+ return null;
+ }
+ return messageApi.getMessageTraceByKey(key, instanceId, topic, traceTopic);
}
export async function consumeMessageDirectly(
@@ -169,7 +182,12 @@ export async function resendDLQSelected(data: {
targetTopic?: string;
}): Promise<DLQResendResult> {
if (isMockMode()) {
- return { matched: data.msgIds.length, resent: data.msgIds.length, failed:
0, outcome: 'SUCCESS' };
+ return {
+ matched: data.msgIds.length,
+ resent: data.msgIds.length,
+ failed: 0,
+ outcome: 'SUCCESS',
+ };
}
return messageApi.resendDLQSelected(data);
}
