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 fb821990 feat: resolve message trace by instance endpoint (#1082)
fb821990 is described below
commit fb8219901f2c9862b114e87b9634baf447cf6dab
Author: aias00 <[email protected]>
AuthorDate: Thu Aug 6 01:25:21 2026 -0700
feat: resolve message trace by instance endpoint (#1082)
---
.../studio/instance/message/MessageController.java | 4 ++--
.../studio/instance/message/MessageProvider.java | 2 +-
.../studio/instance/message/MessageProviderStub.java | 2 +-
.../studio/instance/message/MessageService.java | 4 ++--
.../studio/rocketmq/RocketMQMessageProvider.java | 19 +++++++++++--------
.../instance/message/MessageControllerTest.java | 12 ++++++++++++
.../instance/message/MessageProviderStubTest.java | 2 +-
.../studio/rocketmq/RocketMQMessageProviderTest.java | 4 ++--
web/src/api/message.test.ts | 12 ++++++++----
web/src/api/message.ts | 3 ++-
web/src/pages/instance/message.tsx | 2 +-
web/src/services/messageService.ts | 7 +++++--
12 files changed, 48 insertions(+), 25 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 6c061a28..3b2ddbfb 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
@@ -46,7 +46,7 @@ public class MessageController {
}
@GetMapping("/{msgId}/trace")
- public Result<TraceRecordVO> getMessageTrace(@PathVariable String msgId) {
- return Result.ok(messageService.getMessageTrace(msgId));
+ public Result<TraceRecordVO> getMessageTrace(@PathVariable String msgId,
@RequestParam String instanceId) {
+ return Result.ok(messageService.getMessageTrace(instanceId, msgId));
}
}
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 744d4677..fe79a31e 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
@@ -23,5 +23,5 @@ public interface MessageProvider {
List<MessageRecordVO> queryMessages(String instanceId, String topic,
String msgId, String tag, String key, Long startTime,
Long endTime);
- TraceRecordVO getMessageTrace(String msgId);
+ TraceRecordVO getMessageTrace(String instanceId, String msgId);
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageProviderStub.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageProviderStub.java
index f4352a1d..932d7117 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageProviderStub.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageProviderStub.java
@@ -36,7 +36,7 @@ public class MessageProviderStub implements MessageProvider {
}
@Override
- public TraceRecordVO getMessageTrace(String msgId) {
+ public TraceRecordVO getMessageTrace(String instanceId, String msgId) {
log.warn("MessageProviderStub.getMessageTrace called but no real
message provider is configured");
throw unsupported();
}
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 549ce4f9..58412c06 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
@@ -39,9 +39,9 @@ public class MessageService {
return messageProvider.queryMessages(instanceId, topic, msgId, tag,
key, startTime, endTime);
}
- public TraceRecordVO getMessageTrace(String msgId) {
+ public TraceRecordVO getMessageTrace(String instanceId, String msgId) {
log.info("Getting message trace: msgId={}", msgId);
- return messageProvider.getMessageTrace(msgId);
+ return messageProvider.getMessageTrace(instanceId, msgId);
}
private void validateTopicQueryWindow(String topic, String msgId, String
key, Long startTime, Long endTime) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQMessageProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQMessageProvider.java
index a08b0e81..f23d7b7e 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQMessageProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQMessageProvider.java
@@ -24,8 +24,8 @@ import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageId;
import org.apache.rocketmq.common.message.MessageQueue;
-import org.apache.rocketmq.studio.common.domain.enums.DeliveryStatus;
import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
+import org.apache.rocketmq.studio.common.domain.enums.DeliveryStatus;
import org.apache.rocketmq.studio.instance.message.ConsumerStatusVO;
import org.apache.rocketmq.studio.instance.message.MessageProvider;
import org.apache.rocketmq.studio.instance.message.MessageRecordVO;
@@ -84,13 +84,16 @@ public class RocketMQMessageProvider implements
MessageProvider {
private final ObjectProvider<DefaultMQAdminExt> adminExtProvider;
private final RuntimeAdminClientResolver runtimeAdminClientResolver;
private final QueryHistoryService queryHistoryService;
+ private final RocketMQProperties properties;
public RocketMQMessageProvider(ObjectProvider<DefaultMQAdminExt>
adminExtProvider,
RuntimeAdminClientResolver
runtimeAdminClientResolver,
- QueryHistoryService queryHistoryService) {
+ QueryHistoryService queryHistoryService,
+ RocketMQProperties properties) {
this.adminExtProvider = adminExtProvider;
this.runtimeAdminClientResolver = runtimeAdminClientResolver;
this.queryHistoryService = queryHistoryService;
+ this.properties = properties;
}
@Override
@@ -239,12 +242,12 @@ public class RocketMQMessageProvider implements
MessageProvider {
}
@Override
- public TraceRecordVO getMessageTrace(String msgId) {
- DefaultMQAdminExt adminExt = adminExtProvider.getIfAvailable();
- if (adminExt == null) {
- log.warn("DefaultMQAdminExt is not configured, returning empty
trace");
- return emptyTrace();
- }
+ public TraceRecordVO getMessageTrace(String instanceId, String msgId) {
+ return runtimeAdminClientResolver.execute(instanceId,
+ adminExt -> getMessageTrace((DefaultMQAdminExt) adminExt,
msgId));
+ }
+
+ private TraceRecordVO getMessageTrace(DefaultMQAdminExt adminExt, String
msgId) {
long now = System.currentTimeMillis();
long begin = now - ONE_HOUR_MILLIS;
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 1b1ef293..36a87ee5 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
@@ -70,4 +70,16 @@ class MessageControllerTest {
verify(messageService).queryMessages(eq("instance-a"), eq("orders"),
isNull(), eq("created"), eq("order-1"),
eq(1784246400000L), eq(1784332800000L));
}
+
+ @Test
+ void messageTraceShouldPassInstanceId() throws Exception {
+ TraceRecordVO trace =
TraceRecordVO.builder().nodes(List.of()).consumerStatus(List.of()).build();
+ when(messageService.getMessageTrace("instance-a",
"msg-001")).thenReturn(trace);
+
+ mockMvc.perform(get("/api/messages/msg-001/trace").param("instanceId",
"instance-a"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.code").value(200));
+
+ verify(messageService).getMessageTrace("instance-a", "msg-001");
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageProviderStubTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageProviderStubTest.java
index bb8d4109..be2fa14f 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageProviderStubTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageProviderStubTest.java
@@ -37,7 +37,7 @@ class MessageProviderStubTest {
@Test
void getMessageTraceShouldFailExplicitlyWhenRealProviderIsMissing() {
- assertThatThrownBy(() -> provider.getMessageTrace("msg-001"))
+ assertThatThrownBy(() -> provider.getMessageTrace("instance-a",
"msg-001"))
.isInstanceOf(BusinessException.class)
.hasMessage("Message query provider is not configured")
.extracting("code")
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/rocketmq/RocketMQMessageProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/rocketmq/RocketMQMessageProviderTest.java
index 4c8e90c4..bd025e1f 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/rocketmq/RocketMQMessageProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/rocketmq/RocketMQMessageProviderTest.java
@@ -163,7 +163,7 @@ class RocketMQMessageProviderTest {
when(adminExt.queryMessage(anyString(), anyString(), anyInt(),
anyLong(), anyLong()))
.thenReturn(queryResult);
- TraceRecordVO record = provider.getMessageTrace("msg-123");
+ TraceRecordVO record = provider.getMessageTrace("instance-a",
"msg-123");
assertThat(record.getNodes()).hasSize(2);
TraceNodeVO produce = record.getNodes().get(0);
@@ -196,7 +196,7 @@ class RocketMQMessageProviderTest {
when(adminExt.queryMessage(anyString(), anyString(), anyInt(),
anyLong(), anyLong()))
.thenReturn(queryResult);
- TraceRecordVO record = provider.getMessageTrace("msg-tx");
+ TraceRecordVO record = provider.getMessageTrace("instance-a",
"msg-tx");
assertThat(record.getNodes()).hasSize(1);
TraceNodeVO transaction = record.getNodes().get(0);
diff --git a/web/src/api/message.test.ts b/web/src/api/message.test.ts
index 32107a02..38802c4d 100644
--- a/web/src/api/message.test.ts
+++ b/web/src/api/message.test.ts
@@ -107,9 +107,11 @@ describe('message API', () => {
},
],
};
- mock.onGet('/messages/msg-1/trace').reply(200, { code: 200, data: trace });
+ mock
+ .onGet('/messages/msg-1/trace', { params: { instanceId: 'instance-a' } })
+ .reply(200, { code: 200, data: trace });
- await expect(getMessageTrace('msg-1')).resolves.toEqual(trace);
+ await expect(getMessageTrace('msg-1',
'instance-a')).resolves.toEqual(trace);
});
it('encodes message IDs before requesting trace records', async () => {
@@ -117,8 +119,10 @@ describe('message API', () => {
nodes: [],
consumerStatus: [],
};
- mock.onGet('/messages/AC1E0A64%2F0000%202A9F%3A1/trace').reply(200, {
code: 200, data: trace });
+ mock
+ .onGet('/messages/AC1E0A64%2F0000%202A9F%3A1/trace', { params: {
instanceId: 'instance-a' } })
+ .reply(200, { code: 200, data: trace });
- await expect(getMessageTrace('AC1E0A64/0000
2A9F:1')).resolves.toEqual(trace);
+ await expect(getMessageTrace('AC1E0A64/0000 2A9F:1',
'instance-a')).resolves.toEqual(trace);
});
});
diff --git a/web/src/api/message.ts b/web/src/api/message.ts
index 57c17138..8847b157 100644
--- a/web/src/api/message.ts
+++ b/web/src/api/message.ts
@@ -77,9 +77,10 @@ export async function queryMessages(params: MessageQuery) {
return sortMessagesByStoreTimeDesc(res.data.data);
}
-export async function getMessageTrace(msgId: string) {
+export async function getMessageTrace(msgId: string, instanceId?: string) {
const res = await client.get<{ data: TraceRecord }>(
`/messages/${encodeURIComponent(msgId)}/trace`,
+ { params: { instanceId } },
);
return res.data.data;
}
diff --git a/web/src/pages/instance/message.tsx
b/web/src/pages/instance/message.tsx
index e7a59a6b..96bf564c 100644
--- a/web/src/pages/instance/message.tsx
+++ b/web/src/pages/instance/message.tsx
@@ -391,7 +391,7 @@ const MessagePage = () => {
setTraceLoading(true);
setTraceError(null);
try {
- const result = await getMessageTrace(record.msgId);
+ const result = await getMessageTrace(record.msgId, selectedInstanceId);
if (traceGenerationRef.current !== requestGeneration) return;
setTraceData(result);
setTraceError(null);
diff --git a/web/src/services/messageService.ts
b/web/src/services/messageService.ts
index 31ccf040..9e08e612 100644
--- a/web/src/services/messageService.ts
+++ b/web/src/services/messageService.ts
@@ -48,12 +48,15 @@ export async function queryMessages(params: MessageQuery):
Promise<MessageRecord
return messageApi.queryMessages(params);
}
-export async function getMessageTrace(msgId: string): Promise<TraceRecord |
null> {
+export async function getMessageTrace(
+ msgId: string,
+ instanceId?: string,
+): Promise<TraceRecord | null> {
if (isMockMode()) {
const trace = mockMessageTraces[msgId] as unknown as TraceRecord |
undefined;
return trace ? cloneTrace(trace) : null;
}
- return messageApi.getMessageTrace(msgId);
+ return messageApi.getMessageTrace(msgId, instanceId);
}
export async function listDLQGroups(instanceId: string): Promise<DLQGroup[]> {
