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 f7b6338cc feat(trace): persist custom trace topics in query history
(#2644)
f7b6338cc is described below
commit f7b6338ccbe6715a9de62917682a55ee6d5649d7
Author: xdz997 <[email protected]>
AuthorDate: Fri Sep 4 15:52:13 2026 +0800
feat(trace): persist custom trace topics in query history (#2644)
---
.../studio/instance/message/MessageService.java | 31 +++--
.../message/QueryHistorySchemaMigration.java | 5 +-
.../instance/message/QueryHistoryService.java | 21 +++-
.../instance/message/TraceQueryHistoryVO.java | 1 +
.../studio/persistence/entity/RmqTraceQuery.java | 7 ++
server/src/main/resources/db/schema.sql | 5 +-
.../instance/message/MessageServiceTest.java | 38 ++++++
.../message/QueryHistorySchemaMigrationTest.java | 82 +++++++++++++
.../QueryHistoryServiceIntegrationTest.java | 8 +-
.../instance/message/QueryHistoryServiceTest.java | 46 ++++++++
web/src/api/messageHistory.ts | 2 +
web/src/components/MessageQueryHistoryDrawer.tsx | 8 +-
.../__tests__/MessageQueryHistoryDrawer.test.tsx | 2 +
.../__tests__/MessagePageAsyncState.test.tsx | 127 +++++++++++++++++++++
web/src/pages/instance/message.tsx | 36 ++++--
web/src/utils/messageTraceTopicStorage.test.ts | 58 ++++++++++
web/src/utils/messageTraceTopicStorage.ts | 77 +++++++++++++
17 files changed, 525 insertions(+), 29 deletions(-)
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 544f72681..6b391c943 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
@@ -88,7 +88,7 @@ public class MessageService {
TraceRecordVO result = providerRegistry.byInstanceId(instanceId)
.map(provider -> provider.getMessageTrace(instanceId, msgId,
topic))
.orElseGet(() -> messageProvider.getMessageTrace(instanceId,
msgId, topic));
- recordTraceQuery(instanceId, msgId, topic, result);
+ recordTraceQuery(instanceId, msgId, topic, null, result);
return result;
}
@@ -131,16 +131,18 @@ public class MessageService {
if (!StringUtils.hasText(msgId)) {
throw new BusinessException(400, "msgId is required");
}
- if (!StringUtils.hasText(traceTopic)) {
+ String normalizedTraceTopic = normalizeOptional(traceTopic);
+ if (normalizedTraceTopic == null) {
// 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);
+ log.info("Getting message trace: msgId={}, topic={}, traceTopic={}",
msgId, topic,
+ normalizedTraceTopic);
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);
+ .map(provider -> provider.getMessageTrace(instanceId, msgId,
topic, normalizedTraceTopic))
+ .orElseGet(() -> messageProvider.getMessageTrace(instanceId,
msgId, topic, normalizedTraceTopic));
+ recordTraceQuery(instanceId, msgId, topic, normalizedTraceTopic,
result);
return result;
}
@@ -148,12 +150,14 @@ public class MessageService {
if (!StringUtils.hasText(key)) {
throw new BusinessException(400, "key is required");
}
- log.info("Getting message trace by key: key={}, topic={},
traceTopic={}", key, topic, traceTopic);
+ String normalizedTraceTopic = normalizeOptional(traceTopic);
+ log.info("Getting message trace by key: key={}, topic={},
traceTopic={}", key, topic,
+ normalizedTraceTopic);
// 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));
+ .map(provider -> provider.getMessageTraceByKey(instanceId,
key, topic, normalizedTraceTopic))
+ .orElseGet(() ->
messageProvider.getMessageTraceByKey(instanceId, key, topic,
normalizedTraceTopic));
}
private void recordMessageQuery(String instanceId, String topic, String
msgId, String tag,
String key, Long startTime, Long endTime,
List<MessageRecordVO> result) {
@@ -167,17 +171,22 @@ public class MessageService {
}
}
- private void recordTraceQuery(String instanceId, String msgId, String
topic, TraceRecordVO result) {
+ private void recordTraceQuery(String instanceId, String msgId, String
topic, String traceTopic,
+ TraceRecordVO result) {
int nodeCount = result == null || result.getNodes() == null ? 0 :
result.getNodes().size();
int consumerCount = result == null || result.getConsumerStatus() ==
null ? 0
: result.getConsumerStatus().size();
try {
- queryHistoryService.recordTraceQuery(instanceId, msgId, topic,
nodeCount, consumerCount);
+ queryHistoryService.recordTraceQuery(instanceId, msgId, topic,
traceTopic, nodeCount, consumerCount);
} catch (RuntimeException failure) {
log.warn("Failed to record trace query history: {}",
failure.getMessage());
}
}
+ private static String normalizeOptional(String value) {
+ return StringUtils.hasText(value) ? value.trim() : null;
+ }
+
private void validateTopicQueryWindow(String topic, String msgId, String
key, Long startTime, Long endTime) {
boolean hasTopic = StringUtils.hasText(topic);
boolean hasMessageId = StringUtils.hasText(msgId);
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistorySchemaMigration.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistorySchemaMigration.java
index fd13062b9..292745660 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistorySchemaMigration.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistorySchemaMigration.java
@@ -20,13 +20,14 @@ import java.sql.SQLException;
import java.sql.Statement;
import java.util.List;
-/** Adds query-history indexes to Studio databases created before the index
contract was added. */
+/** Adds query-history columns and indexes to Studio databases created before
the current schema. */
@Slf4j
@Component
@RequiredArgsConstructor
public class QueryHistorySchemaMigration implements ApplicationRunner {
private static final List<Column> COLUMNS = List.of(
- new Column("rmq_instance_message", "result_snapshot",
"MEDIUMTEXT"));
+ new Column("rmq_instance_message", "result_snapshot",
"MEDIUMTEXT"),
+ new Column("rmq_instance_trace", "trace_topic", "VARCHAR(255)"));
private static final List<Index> INDEXES = List.of(
new Index("rmq_instance_message", "idx_message_query_owner_lookup",
"queried_by, cluster_id, gmt_create, id"),
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistoryService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistoryService.java
index b028084dc..759888de4 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistoryService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistoryService.java
@@ -149,10 +149,16 @@ public class QueryHistoryService {
}
}
- public void recordTraceQuery(String clusterId, String msgId, String topic,
int nodeCount, int consumerCount) {
+ /**
+ * Records a message-id trace lookup together with the exact trace topic
used by the
+ * provider. The topic is optional because older/default lookups use the
provider default.
+ */
+ public void recordTraceQuery(String clusterId, String msgId, String topic,
String traceTopic,
+ int nodeCount, int consumerCount) {
RmqTraceQuery query = new RmqTraceQuery();
query.setMsgId(msgId);
query.setTopic(topic);
+ query.setTraceTopic(normalizeOptional(traceTopic));
query.setNodeCount(nodeCount);
query.setConsumerCount(consumerCount);
query.setClusterId(clusterId);
@@ -161,7 +167,12 @@ public class QueryHistoryService {
query.setGmtCreate(now);
query.setGmtModified(now);
traceQueryMapper.insert(query);
- log.debug("Trace query recorded: clusterId={} msgId={} topic={}",
clusterId, msgId, topic);
+ log.debug("Trace query recorded: clusterId={} msgId={} topic={}
traceTopic={}", clusterId, msgId,
+ topic, query.getTraceTopic());
+ }
+
+ public void recordTraceQuery(String clusterId, String msgId, String topic,
int nodeCount, int consumerCount) {
+ recordTraceQuery(clusterId, msgId, topic, null, nodeCount,
consumerCount);
}
public PageResult<MessageQueryHistoryVO> listMessageQueries(String
clusterId, String queryType,
@@ -194,6 +205,7 @@ public class QueryHistoryService {
.and(StringUtils.hasText(search), nested -> nested
.like("topic", pattern)
.or().like("msg_id", pattern)
+ .or().like("trace_topic", pattern)
.or().like("queried_by", pattern))
.orderByDesc("gmt_create", "id");
Page<RmqTraceQuery> result = traceQueryMapper.selectPage(new
Page<>(page, pageSize), query);
@@ -320,6 +332,7 @@ public class QueryHistoryService {
private static TraceQueryHistoryVO toTraceHistory(RmqTraceQuery query) {
return TraceQueryHistoryVO.builder()
.id(query.getId()).msgId(query.getMsgId()).topic(query.getTopic())
+ .traceTopic(query.getTraceTopic())
.nodeCount(query.getNodeCount() == null ? 0 :
query.getNodeCount())
.consumerCount(query.getConsumerCount() == null ? 0 :
query.getConsumerCount())
.clusterId(query.getClusterId()).queriedBy(query.getQueriedBy())
@@ -336,4 +349,8 @@ public class QueryHistoryService {
}
return search.replace("\\", "\\\\").replace("%", "\\%").replace("_",
"\\_");
}
+
+ private static String normalizeOptional(String value) {
+ return StringUtils.hasText(value) ? value.trim() : null;
+ }
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/TraceQueryHistoryVO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/TraceQueryHistoryVO.java
index 68413f6e1..85fc15e70 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/TraceQueryHistoryVO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/TraceQueryHistoryVO.java
@@ -17,6 +17,7 @@ public class TraceQueryHistoryVO {
private Long id;
private String msgId;
private String topic;
+ private String traceTopic;
private int nodeCount;
private int consumerCount;
private String clusterId;
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqTraceQuery.java
b/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqTraceQuery.java
index 7cab2007d..4374f216e 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqTraceQuery.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqTraceQuery.java
@@ -34,6 +34,13 @@ public class RmqTraceQuery {
private String topic;
+ /**
+ * The trace topic selected for the lookup. A null value means that the
provider default
+ * (normally {@code RMQ_SYS_TRACE_TOPIC}) was used. Keeping this nullable
preserves the
+ * meaning of history rows written before custom trace-topic support
existed.
+ */
+ private String traceTopic;
+
private Integer nodeCount;
private Integer consumerCount;
diff --git a/server/src/main/resources/db/schema.sql
b/server/src/main/resources/db/schema.sql
index 31f2e0938..4e21248cf 100644
--- a/server/src/main/resources/db/schema.sql
+++ b/server/src/main/resources/db/schema.sql
@@ -180,6 +180,7 @@ CREATE TABLE IF NOT EXISTS rmq_instance_trace (
`gmt_modified` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE
CURRENT_TIMESTAMP COMMENT '修改时间',
msg_id VARCHAR(128) NOT NULL,
topic VARCHAR(255),
+ trace_topic VARCHAR(255) COMMENT '自定义消息轨迹主题，空表示使用默认主题',
node_count INT DEFAULT 0,
consumer_count INT DEFAULT 0,
cluster_id VARCHAR(255),
@@ -428,5 +429,5 @@ ALTER TABLE rmq_instance_message MODIFY queried_by
VARCHAR(128);
ALTER TABLE rmq_instance_trace MODIFY queried_by VARCHAR(128);
ALTER TABLE rmq_operation_audit MODIFY operator VARCHAR(128);
--- Existing deployments are upgraded by AlertSchemaMigration after the
application
--- connects, because this schema is also parsed by H2 in the development
profile.
+-- Existing deployments are upgraded by the application schema-migration
runners after the
+-- application connects. This schema is also parsed by H2 in the development
profile.
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 b79acc8a2..c2683a7de 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
@@ -267,6 +267,26 @@ class MessageServiceTest {
verifyNoInteractions(fallback);
}
+ @Test
+ void recordsTheCustomTraceTopicUsedByMessageIdLookup() {
+ MessageProvider fallback = mock(MessageProvider.class);
+ InstanceProvider provider = mock(InstanceProvider.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.of(provider));
+ when(provider.getMessageTrace("instance-a", "msg-001", "orders",
"CUSTOM_TRACE"))
+ .thenReturn(trace);
+
+ assertThat(service.getMessageTrace("instance-a", "msg-001", "orders",
" CUSTOM_TRACE "))
+ .isSameAs(trace);
+
+ verify(provider).getMessageTrace("instance-a", "msg-001", "orders",
"CUSTOM_TRACE");
+ verify(history).recordTraceQuery("instance-a", "msg-001", "orders",
"CUSTOM_TRACE", 0, 0);
+ verifyNoInteractions(fallback);
+ }
+
@Test
void keyTraceLookupDoesNotRecordTraceQueryHistory() {
MessageProvider fallback = mock(MessageProvider.class);
@@ -285,4 +305,22 @@ class MessageServiceTest {
org.mockito.ArgumentMatchers.anyString(),
org.mockito.ArgumentMatchers.anyInt(),
org.mockito.ArgumentMatchers.anyInt());
}
+
+ @Test
+ void normalizesCustomTraceTopicForKeyLookup() {
+ 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",
"CUSTOM_TRACE"))
+ .thenReturn(trace);
+
+ assertThat(service.getMessageTraceByKey("instance-a", "ORDER-1",
"orders", " CUSTOM_TRACE "))
+ .isSameAs(trace);
+
+ verify(fallback).getMessageTraceByKey("instance-a", "ORDER-1",
"orders", "CUSTOM_TRACE");
+ verifyNoInteractions(history);
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistorySchemaMigrationTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistorySchemaMigrationTest.java
new file mode 100644
index 000000000..4292251e1
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistorySchemaMigrationTest.java
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.instance.message;
+
+import org.h2.jdbcx.JdbcDataSource;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.DefaultApplicationArguments;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.Statement;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class QueryHistorySchemaMigrationTest {
+
+ @Test
+ void addsTraceTopicToExistingHistoryWithoutDroppingRowsAndIsIdempotent()
throws Exception {
+ JdbcDataSource dataSource = new JdbcDataSource();
+
dataSource.setURL("jdbc:h2:mem:trace-history-schema-migration;MODE=MySQL;"
+ + "DB_CLOSE_DELAY=-1;DATABASE_TO_LOWER=TRUE");
+ dataSource.setUser("sa");
+ try (Connection connection = dataSource.getConnection(); Statement
statement = connection.createStatement()) {
+ statement.execute("CREATE TABLE rmq_instance_trace ("
+ + "id BIGINT PRIMARY KEY, msg_id VARCHAR(128) NOT NULL,
topic VARCHAR(255), "
+ + "node_count INT, consumer_count INT, cluster_id
VARCHAR(255), "
+ + "queried_by VARCHAR(128), gmt_create TIMESTAMP,
gmt_modified TIMESTAMP)");
+ statement.execute("INSERT INTO rmq_instance_trace (id, msg_id,
topic) "
+ + "VALUES (1, 'old-msg', 'orders')");
+ }
+
+ QueryHistorySchemaMigration migration = new
QueryHistorySchemaMigration(dataSource);
+ migration.run(new DefaultApplicationArguments());
+ migration.run(new DefaultApplicationArguments());
+
+ try (Connection connection = dataSource.getConnection(); Statement
statement = connection.createStatement()) {
+ try (ResultSet columns = statement.executeQuery("SELECT COUNT(*)
FROM information_schema.columns "
+ + "WHERE table_name = 'rmq_instance_trace' AND column_name
= 'trace_topic'")) {
+ columns.next();
+ assertThat(columns.getInt(1)).isEqualTo(1);
+ }
+ try (ResultSet rows = statement.executeQuery("SELECT msg_id,
trace_topic FROM rmq_instance_trace")) {
+ rows.next();
+ assertThat(rows.getString("msg_id")).isEqualTo("old-msg");
+ assertThat(rows.getString("trace_topic")).isNull();
+ assertThat(rows.next()).isFalse();
+ }
+ }
+ }
+
+ @Test
+ void skipsMissingHistoryTableSoCustomSchemaCanCreateItLater() throws
Exception {
+ JdbcDataSource dataSource = new JdbcDataSource();
+
dataSource.setURL("jdbc:h2:mem:trace-history-schema-missing;MODE=MySQL;"
+ + "DB_CLOSE_DELAY=-1;DATABASE_TO_LOWER=TRUE");
+ dataSource.setUser("sa");
+
+ QueryHistorySchemaMigration migration = new
QueryHistorySchemaMigration(dataSource);
+
+ migration.run(new DefaultApplicationArguments());
+
+ try (Connection connection = dataSource.getConnection(); Statement
statement = connection.createStatement();
+ ResultSet tables = connection.getMetaData().getTables(null,
null, "rmq_instance_trace",
+ new String[] {"TABLE"})) {
+ assertThat(tables.next()).isFalse();
+ }
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryServiceIntegrationTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryServiceIntegrationTest.java
index b369c9172..765735a7f 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryServiceIntegrationTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryServiceIntegrationTest.java
@@ -51,7 +51,7 @@ class QueryHistoryServiceIntegrationTest {
AuthenticatedUserContext.setUser(longUsername, true);
queryHistoryService.recordMessageQuery("qh-cluster", "TOPIC",
"qh-topic",
null, null, null, null, null, 3, null);
- queryHistoryService.recordTraceQuery("qh-cluster", "qh-msg-id",
"qh-topic", 2, 1);
+ queryHistoryService.recordTraceQuery("qh-cluster", "qh-msg-id",
"qh-topic", "qh-trace-topic", 2, 1);
PageResult<MessageQueryHistoryVO> messageHistory =
queryHistoryService.listMessageQueries("qh-cluster", null,
null, 1, 20);
@@ -60,8 +60,10 @@ class QueryHistoryServiceIntegrationTest {
PageResult<TraceQueryHistoryVO> traceHistory =
queryHistoryService.listTraceQueries("qh-cluster", null,
1, 20);
- assertThat(traceHistory.getItems()).anySatisfy(item ->
- assertThat(item.getQueriedBy()).isEqualTo(longUsername));
+ assertThat(traceHistory.getItems()).anySatisfy(item -> {
+ assertThat(item.getQueriedBy()).isEqualTo(longUsername);
+ assertThat(item.getTraceTopic()).isEqualTo("qh-trace-topic");
+ });
} finally {
AuthenticatedUserContext.clear();
messageQueryMapper.delete(new QueryWrapper<RmqMessageQuery>()
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryServiceTest.java
index b5a03a738..b29359f82 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryServiceTest.java
@@ -80,6 +80,52 @@ class QueryHistoryServiceTest {
verify(traceQueryMapper).insert(captor.capture());
assertThat(captor.getValue().getClusterId()).isEqualTo("cluster-a");
assertThat(captor.getValue().getQueriedBy()).isEqualTo(AuthenticatedUserContext.SYSTEM_ACTOR);
+ assertThat(captor.getValue().getTraceTopic()).isNull();
+ }
+
+ @Test
+ void recordsNormalizedCustomTraceTopic() {
+ AuthenticatedUserContext.setUsername("alice");
+
+ service.recordTraceQuery("cluster-a", "msg-1", "orders", "
CUSTOM_TRACE ", 2, 1);
+
+ ArgumentCaptor<RmqTraceQuery> captor =
ArgumentCaptor.forClass(RmqTraceQuery.class);
+ verify(traceQueryMapper).insert(captor.capture());
+
assertThat(captor.getValue().getTraceTopic()).isEqualTo("CUSTOM_TRACE");
+ assertThat(captor.getValue().getQueriedBy()).isEqualTo("alice");
+ }
+
+ @Test
+ void mapsCustomTraceTopicIntoHistoryView() {
+ AuthenticatedUserContext.setUsername("alice");
+ RmqTraceQuery entity = new RmqTraceQuery();
+ entity.setId(7L);
+ entity.setMsgId("msg-7");
+ entity.setTopic("orders");
+ entity.setTraceTopic("CUSTOM_TRACE");
+ entity.setNodeCount(3);
+ entity.setConsumerCount(2);
+ entity.setClusterId("cluster-a");
+ entity.setQueriedBy("alice");
+ entity.setGmtCreate(LocalDateTime.of(2026, 8, 5, 12, 0));
+ when(traceQueryMapper.selectPage(any(Page.class), any(Wrapper.class)))
+ .thenAnswer(invocation -> {
+ Page<RmqTraceQuery> result = invocation.getArgument(0);
+ result.setRecords(java.util.List.of(entity));
+ result.setTotal(1);
+ return result;
+ });
+
+ PageResult<TraceQueryHistoryVO> result =
service.listTraceQueries("cluster-a", "CUSTOM", 1, 20);
+
+ assertThat(result.getItems()).singleElement().satisfies(item -> {
+ assertThat(item.getTraceTopic()).isEqualTo("CUSTOM_TRACE");
+ assertThat(item.getMsgId()).isEqualTo("msg-7");
+ });
+
+ ArgumentCaptor<Wrapper<RmqTraceQuery>> queryCaptor =
ArgumentCaptor.forClass(Wrapper.class);
+ verify(traceQueryMapper).selectPage(any(Page.class),
queryCaptor.capture());
+
assertThat(queryCaptor.getValue().getCustomSqlSegment()).contains("trace_topic");
}
@Test
diff --git a/web/src/api/messageHistory.ts b/web/src/api/messageHistory.ts
index 538052a86..2c63a1063 100644
--- a/web/src/api/messageHistory.ts
+++ b/web/src/api/messageHistory.ts
@@ -32,6 +32,8 @@ export interface TraceQueryHistory {
id: number;
msgId: string;
topic?: string;
+ /** The custom trace topic used by the lookup; absent means the provider
default. */
+ traceTopic?: string;
nodeCount: number;
consumerCount: number;
clusterId?: string;
diff --git a/web/src/components/MessageQueryHistoryDrawer.tsx
b/web/src/components/MessageQueryHistoryDrawer.tsx
index 6914b28ce..fc25f3963 100644
--- a/web/src/components/MessageQueryHistoryDrawer.tsx
+++ b/web/src/components/MessageQueryHistoryDrawer.tsx
@@ -109,6 +109,12 @@ const MessageQueryHistoryDrawer = ({
const traceColumns: ColumnsType<TraceQueryHistory> = [
{ title: 'Message ID', dataIndex: 'msgId', ellipsis: true },
{ title: 'Topic', dataIndex: 'topic', ellipsis: true },
+ {
+ title: '轨迹 Topic',
+ dataIndex: 'traceTopic',
+ ellipsis: true,
+ render: (value?: string) => value?.trim() || '默认',
+ },
{ title: '轨迹节点', dataIndex: 'nodeCount', width: 90 },
{ title: '消费者', dataIndex: 'consumerCount', width: 90 },
{ title: '操作者', dataIndex: 'queriedBy', width: 110 },
@@ -124,7 +130,7 @@ const MessageQueryHistoryDrawer = ({
</Flex>
<Input.Search
allowClear
- placeholder="搜索 Topic、Message ID、Key 或操作者"
+ placeholder="搜索 Topic、轨迹 Topic、Message ID、Key 或操作者"
onSearch={(value) => {
setPage(1);
setSearch(value.trim());
diff --git a/web/src/components/__tests__/MessageQueryHistoryDrawer.test.tsx
b/web/src/components/__tests__/MessageQueryHistoryDrawer.test.tsx
index b1fad3a60..474234be4 100644
--- a/web/src/components/__tests__/MessageQueryHistoryDrawer.test.tsx
+++ b/web/src/components/__tests__/MessageQueryHistoryDrawer.test.tsx
@@ -60,6 +60,7 @@ describe('MessageQueryHistoryDrawer', () => {
id: 2,
msgId: 'msg-1',
topic: 'orders',
+ traceTopic: 'CUSTOM_TRACE',
nodeCount: 3,
consumerCount: 1,
queriedBy: 'bob',
@@ -86,6 +87,7 @@ describe('MessageQueryHistoryDrawer', () => {
);
await user.click(screen.getByRole('tab', { name: '轨迹查询' }));
expect(await screen.findByText('msg-1')).toBeInTheDocument();
+ expect(screen.getByText('CUSTOM_TRACE')).toBeInTheDocument();
await waitFor(() => expect(listTraceQueryHistory).toHaveBeenCalled());
});
diff --git a/web/src/pages/instance/__tests__/MessagePageAsyncState.test.tsx
b/web/src/pages/instance/__tests__/MessagePageAsyncState.test.tsx
index 260ee8f88..a8470739d 100644
--- a/web/src/pages/instance/__tests__/MessagePageAsyncState.test.tsx
+++ b/web/src/pages/instance/__tests__/MessagePageAsyncState.test.tsx
@@ -33,6 +33,11 @@ const serviceMocks = vi.hoisted(() => ({
const instanceFilterMocks = vi.hoisted(() => ({
useInstanceFilter: vi.fn(),
}));
+const historyMocks = vi.hoisted(() => ({
+ getQueryHistorySummary: vi.fn(),
+ listMessageQueryHistory: vi.fn(),
+ listTraceQueryHistory: vi.fn(),
+}));
vi.mock('../../../services/messageService', () => ({
...serviceMocks,
@@ -51,6 +56,8 @@ vi.mock('../../../services/messageService', () => ({
}));
vi.mock('../../../hooks/useInstanceFilter', () => instanceFilterMocks);
+vi.mock('../../../api/messageHistory', () => historyMocks);
+
vi.mock('../../../services/instanceService', () => ({
listInstances: vi.fn().mockResolvedValue([]),
}));
@@ -137,8 +144,22 @@ const selectTopic = async (user: ReturnType<typeof
userEvent.setup>) => {
describe('MessagePage async request ownership', () => {
beforeEach(() => {
vi.clearAllMocks();
+ localStorage.clear();
serviceMocks.getMessageTrace.mockResolvedValue(null);
serviceMocks.getMessageTraceByKey.mockResolvedValue(null);
+ historyMocks.getQueryHistorySummary.mockResolvedValue({ messageQueries: 0,
traceQueries: 0 });
+ historyMocks.listMessageQueryHistory.mockResolvedValue({
+ items: [],
+ total: 0,
+ page: 1,
+ size: 20,
+ });
+ historyMocks.listTraceQueryHistory.mockResolvedValue({
+ items: [],
+ total: 0,
+ page: 1,
+ size: 20,
+ });
instanceFilterMocks.useInstanceFilter.mockReturnValue({
selectedInstanceId: 1,
selectInstance: vi.fn(),
@@ -439,6 +460,112 @@ describe('MessagePage async request ownership', () => {
expect(within(dialog).getByText(/cg-notification/)).toBeInTheDocument();
});
+ it('remembers a custom trace topic per instance across page remounts', async
() => {
+
serviceMocks.queryMessages.mockResolvedValue([createMessage('remembered-message')]);
+ const user = userEvent.setup();
+ const firstRender = renderPage();
+ await selectTopic(user);
+
+ await user.click(screen.getByRole('button', { name: /^search查询$/ }));
+ const firstRow = await screen.findByRole('row', { name:
/remembered-message/ });
+ await user.click(within(firstRow).getByRole('button', { name: /轨迹/ }));
+ const firstDialog = await screen.findByRole('dialog', { name: '消息详情' });
+ const firstTraceTopicInput =
+ within(firstDialog).getByPlaceholderText('轨迹 Topic（留空使用默认）');
+ await user.type(firstTraceTopicInput, ' CUSTOM_TRACE ');
+
+ await waitFor(() => {
+
expect(localStorage.getItem('rocketmq-studio-message-trace-topic:1')).toBe('CUSTOM_TRACE');
+ });
+
+ firstRender.unmount();
+ renderPage();
+ await selectTopic(user);
+ await user.click(screen.getByRole('button', { name: /^search查询$/ }));
+ const secondRow = await screen.findByRole('row', { name:
/remembered-message/ });
+ await user.click(within(secondRow).getByRole('button', { name: /轨迹/ }));
+ const secondDialog = await screen.findByRole('dialog', { name: '消息详情' });
+
+ expect(within(secondDialog).getByPlaceholderText('轨迹
Topic（留空使用默认）')).toHaveValue(
+ 'CUSTOM_TRACE',
+ );
+ });
+
+ it('does not leak a stored custom trace topic between instances', async ()
=> {
+ localStorage.setItem('rocketmq-studio-message-trace-topic:1', 'TRACE_A');
+ localStorage.setItem('rocketmq-studio-message-trace-topic:2', 'TRACE_B');
+
serviceMocks.queryMessages.mockResolvedValue([createMessage('instance-message')]);
+ let currentInstanceId = 1;
+ instanceFilterMocks.useInstanceFilter.mockImplementation(() => ({
+ selectedInstanceId: currentInstanceId,
+ selectInstance: vi.fn(),
+ instanceOptions: [
+ { value: 1, label: 'Instance A' },
+ { value: 2, label: 'Instance B' },
+ ],
+ }));
+ const user = userEvent.setup();
+ const view = renderPage();
+ await selectTopic(user);
+ await user.click(screen.getByRole('button', { name: /^search查询$/ }));
+ const firstRow = await screen.findByRole('row', { name: /instance-message/
});
+ await user.click(within(firstRow).getByRole('button', { name: /轨迹/ }));
+ const firstDialog = await screen.findByRole('dialog', { name: '消息详情' });
+ expect(within(firstDialog).getByPlaceholderText('轨迹
Topic（留空使用默认）')).toHaveValue(
+ 'TRACE_A',
+ );
+
+ currentInstanceId = 2;
+ view.rerender(<MessagePageWithProviders />);
+ await selectTopic(user);
+ await user.click(screen.getByRole('button', { name: /^search查询$/ }));
+ const secondRow = await screen.findByRole('row', { name:
/instance-message/ });
+ await user.click(within(secondRow).getByRole('button', { name: /轨迹/ }));
+ const secondDialog = await screen.findByRole('dialog', { name: '消息详情' });
+ expect(within(secondDialog).getByPlaceholderText('轨迹
Topic（留空使用默认）')).toHaveValue(
+ 'TRACE_B',
+ );
+ });
+
+ it('restores a custom trace topic from history before opening the trace
again', async () => {
+ historyMocks.listTraceQueryHistory.mockResolvedValue({
+ items: [
+ {
+ id: 9,
+ msgId: 'history-message',
+ topic: 'orders',
+ traceTopic: 'CUSTOM_TRACE',
+ nodeCount: 1,
+ consumerCount: 0,
+ queriedBy: 'alice',
+ queriedAt: '2026-08-05T12:00:00Z',
+ },
+ ],
+ total: 1,
+ page: 1,
+ size: 20,
+ });
+
serviceMocks.queryMessages.mockResolvedValue([createMessage('history-message')]);
+ const user = userEvent.setup();
+ renderPage();
+
+ await user.click(screen.getByRole('button', { name: /服务端历史/ }));
+ await user.click(await screen.findByRole('tab', { name: '轨迹查询' }));
+ await user.click(await screen.findByText('history-message'));
+
+ const row = await screen.findByRole('row', { name: /history-message/ });
+ await user.click(within(row).getByRole('button', { name: /轨迹/ }));
+
+ await waitFor(() => {
+ expect(serviceMocks.getMessageTrace).toHaveBeenLastCalledWith(
+ 'history-message',
+ 1,
+ 'topic-history-message',
+ 'CUSTOM_TRACE',
+ );
+ });
+ });
+
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 a90a84cf6..3d23be5ea 100644
--- a/web/src/pages/instance/message.tsx
+++ b/web/src/pages/instance/message.tsx
@@ -71,6 +71,10 @@ import {
import { listTopics } from '../../services/topicService';
import { useInstanceFilter } from '../../hooks/useInstanceFilter';
import { downloadBlob } from '../../utils/download';
+import {
+ readMessageTraceTopic,
+ writeMessageTraceTopic,
+} from '../../utils/messageTraceTopicStorage';
import { tableScrollX } from '../../utils/table';
import {
analyzeMessageTrace,
@@ -397,7 +401,9 @@ const MessagePageContent = ({
const [traceError, setTraceError] = useState<string | null>(null);
const [traceQueryMode, setTraceQueryMode] = useState<'msgid' |
'key'>('msgid');
const [traceQueryValue, setTraceQueryValue] = useState('');
- const [customTraceTopic, setCustomTraceTopic] = useState('');
+ const [customTraceTopic, setCustomTraceTopic] = useState(() =>
+ readMessageTraceTopic(selectedInstanceId),
+ );
const [historyDrawerOpen, setHistoryDrawerOpen] = useState(false);
const [directConsumeOpen, setDirectConsumeOpen] = useState(false);
const [directConsumeGroup, setDirectConsumeGroup] = useState('');
@@ -416,6 +422,10 @@ const MessagePageContent = ({
[],
);
+ useEffect(() => {
+ writeMessageTraceTopic(selectedInstanceId, customTraceTopic);
+ }, [customTraceTopic, selectedInstanceId]);
+
const currentQueryParams: MessageQuery =
queryMode === 'topic'
? { topic: selectedTopic, startTime: dateRange[0].valueOf(), endTime:
dateRange[1].valueOf() }
@@ -562,6 +572,7 @@ const MessagePageContent = ({
handleQueryModeChange('msgid');
setSelectedTopic(record.topic);
setMsgIdInput(record.msgId);
+ setCustomTraceTopic(record.traceTopic?.trim() || '');
setHistoryDrawerOpen(false);
void executeQuery('msgid', { topic: record.topic, msgId: record.msgId });
};
@@ -577,15 +588,24 @@ const MessagePageContent = ({
setTraceError(null);
setTraceQueryMode('msgid');
setTraceQueryValue(record.msgId);
- const cacheKey = JSON.stringify([selectedInstanceId, record.topic,
record.msgId]);
+ const normalizedTraceTopic = customTraceTopic.trim();
+ const cacheKey = JSON.stringify([
+ selectedInstanceId,
+ record.topic,
+ record.msgId,
+ normalizedTraceTopic,
+ ]);
let traceRequest = traceCacheRef.current.get(cacheKey);
if (!traceRequest) {
- traceRequest = getMessageTrace(record.msgId, selectedInstanceId,
record.topic).catch(
- (error) => {
- traceCacheRef.current.delete(cacheKey);
- throw error;
- },
- );
+ traceRequest = getMessageTrace(
+ record.msgId,
+ selectedInstanceId,
+ record.topic,
+ normalizedTraceTopic,
+ ).catch((error) => {
+ traceCacheRef.current.delete(cacheKey);
+ throw error;
+ });
traceCacheRef.current.set(cacheKey, traceRequest);
}
try {
diff --git a/web/src/utils/messageTraceTopicStorage.test.ts
b/web/src/utils/messageTraceTopicStorage.test.ts
new file mode 100644
index 000000000..637bf1f7e
--- /dev/null
+++ b/web/src/utils/messageTraceTopicStorage.test.ts
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import { readMessageTraceTopic, writeMessageTraceTopic } from
'./messageTraceTopicStorage';
+
+describe('message trace topic storage', () => {
+ beforeEach(() => {
+ localStorage.clear();
+ });
+
+ afterEach(() => {
+ vi.restoreAllMocks();
+ });
+
+ it('stores normalized topics independently for each instance', () => {
+ writeMessageTraceTopic('instance/a', ' CUSTOM_TRACE ');
+ writeMessageTraceTopic('instance-b', 'OTHER_TRACE');
+
+ expect(readMessageTraceTopic('instance/a')).toBe('CUSTOM_TRACE');
+ expect(readMessageTraceTopic('instance-b')).toBe('OTHER_TRACE');
+ expect(readMessageTraceTopic('instance-c')).toBe('');
+ });
+
+ it('removes blank topics so the provider default is restored', () => {
+ writeMessageTraceTopic('instance-a', 'CUSTOM_TRACE');
+
+ writeMessageTraceTopic('instance-a', ' ');
+
+ expect(readMessageTraceTopic('instance-a')).toBe('');
+ });
+
+ it('treats denied browser storage as an optional preference', () => {
+ vi.spyOn(Storage.prototype, 'getItem').mockImplementation(() => {
+ throw new DOMException('storage denied', 'SecurityError');
+ });
+ vi.spyOn(Storage.prototype, 'setItem').mockImplementation(() => {
+ throw new DOMException('storage denied', 'SecurityError');
+ });
+
+ expect(() => writeMessageTraceTopic('instance-a',
'CUSTOM_TRACE')).not.toThrow();
+ expect(readMessageTraceTopic('instance-a')).toBe('');
+ });
+});
diff --git a/web/src/utils/messageTraceTopicStorage.ts
b/web/src/utils/messageTraceTopicStorage.ts
new file mode 100644
index 000000000..85345f436
--- /dev/null
+++ b/web/src/utils/messageTraceTopicStorage.ts
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+const STORAGE_KEY_PREFIX = 'rocketmq-studio-message-trace-topic:';
+
+const normalizeInstanceId = (instanceId: string | number | undefined): string
| undefined => {
+ if (instanceId === undefined || instanceId === null) return undefined;
+ const normalized = String(instanceId).trim();
+ return normalized || undefined;
+};
+
+const storageKey = (instanceId: string | number | undefined): string |
undefined => {
+ const normalized = normalizeInstanceId(instanceId);
+ return normalized ? `${STORAGE_KEY_PREFIX}${encodeURIComponent(normalized)}`
: undefined;
+};
+
+const getStorage = (): Storage | null => {
+ if (typeof window === 'undefined') return null;
+ try {
+ return window.localStorage;
+ } catch {
+ // Browsers can deny storage access in private mode or under a restrictive
policy.
+ return null;
+ }
+};
+
+/**
+ * Reads the last custom trace topic used for one instance.
+ * A missing, blank, or unreadable value intentionally means provider default.
+ */
+export const readMessageTraceTopic = (instanceId: string | number |
undefined): string => {
+ const key = storageKey(instanceId);
+ const storage = getStorage();
+ if (!key || !storage) return '';
+ try {
+ return storage.getItem(key)?.trim() || '';
+ } catch {
+ return '';
+ }
+};
+
+/**
+ * Persists a normalized custom trace topic without making browser storage a
prerequisite for
+ * querying traces. Blank values remove the per-instance preference and
restore default behavior.
+ */
+export const writeMessageTraceTopic = (
+ instanceId: string | number | undefined,
+ traceTopic: string | undefined,
+): void => {
+ const key = storageKey(instanceId);
+ const storage = getStorage();
+ if (!key || !storage) return;
+ const normalized = traceTopic?.trim() || '';
+ try {
+ if (normalized) {
+ storage.setItem(key, normalized);
+ } else {
+ storage.removeItem(key);
+ }
+ } catch {
+ // Storage failure must not block an otherwise valid trace request.
+ }
+};
