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 f485de34 feat: add consumer client stack diagnostics (#1267)
f485de34 is described below
commit f485de34393313c8ccbaa4b96ba2bba1fc340638
Author: yx9o <[email protected]>
AuthorDate: Mon Aug 10 20:03:47 2026 +0800
feat: add consumer client stack diagnostics (#1267)
---
.../group/ConsumerDiagnosticsProvider.java | 2 +-
.../group/ConsumerDiagnosticsProviderStub.java | 4 +-
.../instance/group/ConsumerDiagnosticsService.java | 9 +-
.../instance/group/ConsumerGroupController.java | 5 +-
.../RocketMQConsumerDiagnosticsProvider.java | 174 +++++++++++++++++++++
.../group/ConsumerDiagnosticsProviderStubTest.java | 2 +-
.../group/ConsumerDiagnosticsServiceTest.java | 20 ++-
.../group/ConsumerGroupControllerTest.java | 7 +-
.../RocketMQConsumerDiagnosticsProviderTest.java | 145 +++++++++++++++++
web/src/api/metadata.test.ts | 20 +++
web/src/api/metadata.ts | 25 +++
.../pages/instance/__tests__/ConsumerPage.test.tsx | 54 +++++++
web/src/pages/instance/consumer.tsx | 125 +++++++++++++++
web/src/services/consumerService.test.ts | 10 ++
web/src/services/consumerService.ts | 18 +++
15 files changed, 602 insertions(+), 18 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerDiagnosticsProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerDiagnosticsProvider.java
index 87c6d271..6b5503d0 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerDiagnosticsProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerDiagnosticsProvider.java
@@ -18,5 +18,5 @@
package org.apache.rocketmq.studio.instance.group;
public interface ConsumerDiagnosticsProvider {
- ConsumerStackTraceVO getConsumerStack(String groupName, String clientId);
+ ConsumerStackTraceVO getConsumerStack(String instanceId, String groupName,
String clientId);
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerDiagnosticsProviderStub.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerDiagnosticsProviderStub.java
index 5f89438e..046a2c6b 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerDiagnosticsProviderStub.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerDiagnosticsProviderStub.java
@@ -26,9 +26,9 @@ import org.springframework.stereotype.Component;
public class ConsumerDiagnosticsProviderStub implements
ConsumerDiagnosticsProvider {
@Override
- public ConsumerStackTraceVO getConsumerStack(String groupName, String
clientId) {
+ public ConsumerStackTraceVO getConsumerStack(String instanceId, String
groupName, String clientId) {
log.warn("ConsumerDiagnosticsProviderStub.getConsumerStack called
without a real diagnostics provider. "
- + "groupName={}, clientId={}", groupName, clientId);
+ + "instanceId={}, groupName={}, clientId={}", instanceId,
groupName, clientId);
throw new BusinessException(501, "Consumer diagnostics provider is not
configured");
}
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerDiagnosticsService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerDiagnosticsService.java
index 7a8699b6..4b7315e4 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerDiagnosticsService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerDiagnosticsService.java
@@ -28,10 +28,11 @@ public class ConsumerDiagnosticsService {
private final ConsumerDiagnosticsProvider diagnosticsProvider;
- public ConsumerStackTraceVO getConsumerStack(String groupName, String
clientId) {
+ public ConsumerStackTraceVO getConsumerStack(String instanceId, String
groupName, String clientId) {
+ String normalizedInstanceId = normalizeOptional(instanceId);
String normalizedGroupName = normalizeRequired(groupName, "groupName");
String normalizedClientId = normalizeRequired(clientId, "clientId");
- return diagnosticsProvider.getConsumerStack(normalizedGroupName,
normalizedClientId);
+ return diagnosticsProvider.getConsumerStack(normalizedInstanceId,
normalizedGroupName, normalizedClientId);
}
private String normalizeRequired(String value, String fieldName) {
@@ -40,4 +41,8 @@ public class ConsumerDiagnosticsService {
}
return value.trim();
}
+
+ private String normalizeOptional(String value) {
+ return StringUtils.hasText(value) ? value.trim() : null;
+ }
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupController.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupController.java
index 4d7ddf70..49c2c6c5 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupController.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupController.java
@@ -71,8 +71,9 @@ public class ConsumerGroupController {
@GetMapping("/{name}/instances/{clientId}/stack")
public Result<ConsumerStackTraceVO> getConsumerStack(
@PathVariable String name,
- @PathVariable String clientId) {
- return Result.ok(consumerDiagnosticsService.getConsumerStack(name,
clientId));
+ @PathVariable String clientId,
+ @RequestParam(required = false) String instanceId) {
+ return
Result.ok(consumerDiagnosticsService.getConsumerStack(instanceId, name,
clientId));
}
@PostMapping("/create")
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQConsumerDiagnosticsProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQConsumerDiagnosticsProvider.java
new file mode 100644
index 00000000..6f094987
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQConsumerDiagnosticsProvider.java
@@ -0,0 +1,174 @@
+/*
+ * 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.provider.apache;
+
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
+import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
+import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.instance.group.ConsumerDiagnosticsProvider;
+import org.apache.rocketmq.studio.instance.group.ConsumerStackTraceVO;
+import org.apache.rocketmq.studio.instance.group.ConsumerThreadStackVO;
+import org.apache.rocketmq.tools.admin.MQAdminExt;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.annotation.Primary;
+import org.springframework.stereotype.Service;
+import org.springframework.util.StringUtils;
+
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * Reads live consumer diagnostics from RocketMQ and converts the raw client
jstack dump into
+ * structured rows that the Studio UI can render.
+ */
+@Slf4j
+@Service
+@Primary
+@RequiredArgsConstructor
+public class RocketMQConsumerDiagnosticsProvider implements
ConsumerDiagnosticsProvider {
+
+ private static final Pattern THREAD_HEADER =
+
Pattern.compile("^(?<name>.+?)\\s+TID:\\s+(?<id>\\d+)\\s+STATE:\\s+(?<state>\\S+)\\s*$");
+
+ private final RuntimeAdminClientResolver runtimeAdminClientResolver;
+ private final MqAdminExtFactory adminFactory;
+ private final RocketMQProperties properties;
+
+ @Override
+ public ConsumerStackTraceVO getConsumerStack(String instanceId, String
groupName, String clientId) {
+ if (StringUtils.hasText(instanceId)) {
+ return runtimeAdminClientResolver.execute(instanceId,
+ admin -> getConsumerStack(admin, groupName, clientId));
+ }
+ if (!StringUtils.hasText(properties.getNamesrvAddr())) {
+ throw new BusinessException(503, "RocketMQ admin not connected");
+ }
+ return adminFactory.execute(properties.getNamesrvAddr(), null,
+ admin -> getConsumerStack(admin, groupName, clientId));
+ }
+
+ private ConsumerStackTraceVO getConsumerStack(MQAdminExt admin, String
groupName, String clientId) {
+ try {
+ ConsumerRunningInfo runningInfo =
admin.getConsumerRunningInfo(groupName, clientId, true);
+ if (runningInfo == null) {
+ throw new BusinessException(404, "Consumer client not found: "
+ clientId);
+ }
+ List<ConsumerThreadStackVO> threads =
parseJstack(runningInfo.getJstack());
+ return ConsumerStackTraceVO.builder()
+ .groupName(groupName)
+ .clientId(clientId)
+ .capturedAt(LocalDateTime.now())
+ .threadCount(threads.size())
+ .threads(threads)
+ .build();
+ } catch (BusinessException e) {
+ throw e;
+ } catch (MQClientException e) {
+ if (e.getResponseCode() == ResponseCode.CONSUMER_NOT_ONLINE) {
+ throw new BusinessException(404, "Consumer client is not
online: " + clientId);
+ }
+ throw diagnosticsFailure(groupName, clientId, e);
+ } catch (Exception e) {
+ throw diagnosticsFailure(groupName, clientId, e);
+ }
+ }
+
+ private BusinessException diagnosticsFailure(String groupName, String
clientId, Exception exception) {
+ log.warn("Failed to get consumer stack, groupName={}, clientId={}: {}",
+ groupName, clientId, exception.getMessage());
+ return new BusinessException(502,
+ "Failed to get consumer stack for " + clientId + ": " +
rootMessage(exception));
+ }
+
+ private String rootMessage(Exception exception) {
+ Throwable cause = exception;
+ while (cause.getCause() != null) {
+ cause = cause.getCause();
+ }
+ return cause.getMessage() == null ? cause.toString() :
cause.getMessage();
+ }
+
+ private List<ConsumerThreadStackVO> parseJstack(String jstack) {
+ if (!StringUtils.hasText(jstack)) {
+ return List.of();
+ }
+
+ List<ConsumerThreadStackVO> threads = new ArrayList<>();
+ ThreadBuilder current = null;
+ for (String rawLine : jstack.split("\\R")) {
+ if (!StringUtils.hasText(rawLine)) {
+ continue;
+ }
+ Matcher header = THREAD_HEADER.matcher(rawLine);
+ if (header.matches()) {
+ if (current != null) {
+ threads.add(current.build());
+ }
+ current = new ThreadBuilder(
+ header.group("name").trim(),
+ Long.parseLong(header.group("id")),
+ header.group("state").trim());
+ continue;
+ }
+ if (current != null) {
+ current.addFrame(stripThreadNamePrefix(rawLine,
current.threadName()));
+ }
+ }
+ if (current != null) {
+ threads.add(current.build());
+ }
+ return threads;
+ }
+
+ private String stripThreadNamePrefix(String line, String threadName) {
+ if (line.startsWith(threadName)) {
+ return line.substring(threadName.length()).trim();
+ }
+ return line.trim();
+ }
+
+ private record ThreadBuilder(String threadName, long threadId, String
state, List<String> stackTrace) {
+
+ private ThreadBuilder(String threadName, long threadId, String state) {
+ this(threadName, threadId, state, new ArrayList<>());
+ }
+
+ private void addFrame(String frame) {
+ if (StringUtils.hasText(frame)) {
+ stackTrace.add(frame);
+ }
+ }
+
+ private ConsumerThreadStackVO build() {
+ return ConsumerThreadStackVO.builder()
+ .threadName(threadName)
+ .threadId(threadId)
+ .state(state)
+ .blockedTime(0)
+ .waitedTime(0)
+ .stackTrace(List.copyOf(stackTrace))
+ .build();
+ }
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerDiagnosticsProviderStubTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerDiagnosticsProviderStubTest.java
index 4f486440..264705d6 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerDiagnosticsProviderStubTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerDiagnosticsProviderStubTest.java
@@ -28,7 +28,7 @@ class ConsumerDiagnosticsProviderStubTest {
@Test
void getConsumerStackShouldFailWhenRealProviderIsMissing() {
- assertThatThrownBy(() -> provider.getConsumerStack("cg-orders",
"client-1"))
+ assertThatThrownBy(() -> provider.getConsumerStack("instance-a",
"cg-orders", "client-1"))
.isInstanceOf(BusinessException.class)
.hasMessage("Consumer diagnostics provider is not configured")
.satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(501));
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerDiagnosticsServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerDiagnosticsServiceTest.java
index 6867871b..1bd8c108 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerDiagnosticsServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerDiagnosticsServiceTest.java
@@ -51,13 +51,15 @@ class ConsumerDiagnosticsServiceTest {
.threads(List.of())
.build();
- when(diagnosticsProvider.getConsumerStack("cg-orders",
"client-1")).thenReturn(stackTrace);
+ when(diagnosticsProvider.getConsumerStack("instance-a", "cg-orders",
"client-1"))
+ .thenReturn(stackTrace);
- ConsumerStackTraceVO result =
diagnosticsService.getConsumerStack("cg-orders", "client-1");
+ ConsumerStackTraceVO result = diagnosticsService.getConsumerStack(
+ "instance-a", "cg-orders", "client-1");
assertThat(result.getGroupName()).isEqualTo("cg-orders");
assertThat(result.getClientId()).isEqualTo("client-1");
- verify(diagnosticsProvider).getConsumerStack("cg-orders", "client-1");
+ verify(diagnosticsProvider).getConsumerStack("instance-a",
"cg-orders", "client-1");
}
@Test
@@ -69,25 +71,27 @@ class ConsumerDiagnosticsServiceTest {
.threadCount(0)
.threads(List.of())
.build();
- when(diagnosticsProvider.getConsumerStack("cg-orders",
"client-1")).thenReturn(stackTrace);
+ when(diagnosticsProvider.getConsumerStack("instance-a", "cg-orders",
"client-1"))
+ .thenReturn(stackTrace);
- ConsumerStackTraceVO result = diagnosticsService.getConsumerStack("
cg-orders ", " client-1 ");
+ ConsumerStackTraceVO result = diagnosticsService.getConsumerStack(
+ " instance-a ", " cg-orders ", " client-1 ");
assertThat(result.getGroupName()).isEqualTo("cg-orders");
assertThat(result.getClientId()).isEqualTo("client-1");
- verify(diagnosticsProvider).getConsumerStack("cg-orders", "client-1");
+ verify(diagnosticsProvider).getConsumerStack("instance-a",
"cg-orders", "client-1");
}
@Test
void getConsumerStackShouldRejectBlankGroupName() {
- assertThatThrownBy(() -> diagnosticsService.getConsumerStack(" ",
"client-1"))
+ assertThatThrownBy(() ->
diagnosticsService.getConsumerStack("instance-a", " ", "client-1"))
.isInstanceOf(BusinessException.class)
.hasMessage("groupName is required");
}
@Test
void getConsumerStackShouldRejectBlankClientId() {
- assertThatThrownBy(() ->
diagnosticsService.getConsumerStack("cg-orders", " "))
+ assertThatThrownBy(() ->
diagnosticsService.getConsumerStack("instance-a", "cg-orders", " "))
.isInstanceOf(BusinessException.class)
.hasMessage("clientId is required");
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupControllerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupControllerTest.java
index d076715e..91b56491 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupControllerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupControllerTest.java
@@ -145,9 +145,11 @@ class ConsumerGroupControllerTest {
.threads(List.of(thread))
.build();
- when(consumerDiagnosticsService.getConsumerStack("cg-orders",
"client-1")).thenReturn(stackTrace);
+ when(consumerDiagnosticsService.getConsumerStack("instance-a",
"cg-orders", "client-1"))
+ .thenReturn(stackTrace);
- mockMvc.perform(get("/api/groups/cg-orders/instances/client-1/stack"))
+ mockMvc.perform(get("/api/groups/cg-orders/instances/client-1/stack")
+ .param("instanceId", "instance-a"))
.andExpect(status().isOk())
.andExpect(jsonPath("$.code").value(200))
.andExpect(jsonPath("$.data.groupName").value("cg-orders"))
@@ -156,6 +158,7 @@ class ConsumerGroupControllerTest {
.andExpect(jsonPath("$.data.threads[0].threadName").value("ConsumeMessageThread_1"))
.andExpect(jsonPath("$.data.threads[0].stackTrace[0]")
.value("org.apache.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService.run"));
+ verify(consumerDiagnosticsService).getConsumerStack("instance-a",
"cg-orders", "client-1");
}
@Test
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQConsumerDiagnosticsProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQConsumerDiagnosticsProviderTest.java
new file mode 100644
index 00000000..0b8a6371
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQConsumerDiagnosticsProviderTest.java
@@ -0,0 +1,145 @@
+/*
+ * 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.provider.apache;
+
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
+import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
+import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.instance.group.ConsumerStackTraceVO;
+import org.apache.rocketmq.tools.admin.MQAdminExt;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class RocketMQConsumerDiagnosticsProviderTest {
+
+ @Mock
+ private RuntimeAdminClientResolver runtimeAdminClientResolver;
+
+ @Mock
+ private MqAdminExtFactory adminFactory;
+
+ @Mock
+ private RocketMQProperties properties;
+
+ @Mock
+ private MQAdminExt adminExt;
+
+ private RocketMQConsumerDiagnosticsProvider provider;
+
+ @BeforeEach
+ void setUp() {
+
lenient().when(properties.getNamesrvAddr()).thenReturn("127.0.0.1:9876");
+ lenient().when(runtimeAdminClientResolver.execute(anyString(),
any())).thenAnswer(invocation ->
+
invocation.<MqAdminExtFactory.AdminAction<Object>>getArgument(1).apply(adminExt));
+ lenient().when(adminFactory.execute(anyString(), any(),
any())).thenAnswer(invocation ->
+
invocation.<MqAdminExtFactory.AdminAction<Object>>getArgument(2).apply(adminExt));
+ provider = new
RocketMQConsumerDiagnosticsProvider(runtimeAdminClientResolver, adminFactory,
properties);
+ }
+
+ @Test
+ void getConsumerStackShouldUseSelectedInstanceAndParseJstack() throws
Exception {
+ ConsumerRunningInfo runningInfo = new ConsumerRunningInfo();
+ runningInfo.setJstack("""
+ ConsumeMessageThread_1 TID: 12 STATE:
RUNNABLE
+ ConsumeMessageThread_1
org.apache.demo.OrderListener.consume(OrderListener.java:42)
+ ConsumeMessageThread_1
java.base/java.lang.Thread.run(Thread.java:1583)
+
+ PullMessageService TID: 13 STATE: WAITING
+ PullMessageService
java.base/jdk.internal.misc.Unsafe.park(Native Method)
+ """);
+ when(adminExt.getConsumerRunningInfo("cg-orders", "client-1",
true)).thenReturn(runningInfo);
+
+ ConsumerStackTraceVO result = provider.getConsumerStack("instance-a",
"cg-orders", "client-1");
+
+ assertThat(result.getGroupName()).isEqualTo("cg-orders");
+ assertThat(result.getClientId()).isEqualTo("client-1");
+ assertThat(result.getThreadCount()).isEqualTo(2);
+ assertThat(result.getThreads()).hasSize(2);
+
assertThat(result.getThreads().get(0).getThreadName()).isEqualTo("ConsumeMessageThread_1");
+ assertThat(result.getThreads().get(0).getThreadId()).isEqualTo(12);
+
assertThat(result.getThreads().get(0).getState()).isEqualTo("RUNNABLE");
+ assertThat(result.getThreads().get(0).getStackTrace())
+ .containsExactly(
+
"org.apache.demo.OrderListener.consume(OrderListener.java:42)",
+ "java.base/java.lang.Thread.run(Thread.java:1583)");
+ verify(runtimeAdminClientResolver).execute(eq("instance-a"), any());
+ verify(adminExt).getConsumerRunningInfo("cg-orders", "client-1", true);
+ verify(adminFactory, never()).execute(anyString(), any(), any());
+ }
+
+ @Test
+ void getConsumerStackShouldUseDefaultNameServerWhenInstanceIsBlank()
throws Exception {
+ ConsumerRunningInfo runningInfo = new ConsumerRunningInfo();
+ runningInfo.setJstack("");
+ when(adminExt.getConsumerRunningInfo("cg-orders", "client-1",
true)).thenReturn(runningInfo);
+
+ ConsumerStackTraceVO result = provider.getConsumerStack(null,
"cg-orders", "client-1");
+
+ assertThat(result.getThreadCount()).isZero();
+ verify(adminFactory).execute(eq("127.0.0.1:9876"), any(), any());
+ verify(runtimeAdminClientResolver, never()).execute(anyString(),
any());
+ }
+
+ @Test
+ void getConsumerStackShouldFailFastWhenDefaultAdminIsMissing() {
+ when(properties.getNamesrvAddr()).thenReturn("");
+
+ assertThatThrownBy(() -> provider.getConsumerStack(null, "cg-orders",
"client-1"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("RocketMQ admin not connected")
+ .satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(503));
+ }
+
+ @Test
+ void getConsumerStackShouldMapOfflineConsumerToNotFound() throws Exception
{
+ when(adminExt.getConsumerRunningInfo("cg-orders", "client-1", true))
+ .thenThrow(new
MQClientException(ResponseCode.CONSUMER_NOT_ONLINE, "consumer offline"));
+
+ assertThatThrownBy(() -> provider.getConsumerStack("instance-a",
"cg-orders", "client-1"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Consumer client is not online: client-1")
+ .satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(404));
+ }
+
+ @Test
+ void getConsumerStackShouldSurfaceAdminFailuresAsBadGateway() throws
Exception {
+ when(adminExt.getConsumerRunningInfo("cg-orders", "client-1", true))
+ .thenThrow(new MQClientException(1, "broker rejected
request"));
+
+ assertThatThrownBy(() -> provider.getConsumerStack("instance-a",
"cg-orders", "client-1"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("Failed to get consumer stack for
client-1")
+ .satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(502));
+ }
+}
diff --git a/web/src/api/metadata.test.ts b/web/src/api/metadata.test.ts
index 1d9d420a..dc77b50b 100644
--- a/web/src/api/metadata.test.ts
+++ b/web/src/api/metadata.test.ts
@@ -21,6 +21,7 @@ import client from './client';
import {
createTopic,
deleteTopic,
+ getConsumerStack,
getTopicConsumers,
getTopicRoutes,
listTopics,
@@ -63,6 +64,25 @@ describe('topic metadata API', () => {
await expect(getTopicConsumers(topicName,
'instance-a')).resolves.toEqual([]);
});
+ it('encodes consumer stack route parameters and passes instanceId', async ()
=> {
+ const stack = {
+ groupName: 'cg/orders',
+ clientId: 'client/10.0.0.1',
+ capturedAt: '2026-07-23T00:00:00Z',
+ threadCount: 0,
+ threads: [],
+ };
+ mock
+ .onGet('/groups/cg%2Forders/instances/client%2F10.0.0.1/stack', {
+ params: { instanceId: 'instance-a' },
+ })
+ .reply(200, { code: 200, data: stack });
+
+ await expect(getConsumerStack('cg/orders', 'client/10.0.0.1',
'instance-a')).resolves.toEqual(
+ stack,
+ );
+ });
+
it('persists topic creation, deletion, and sending through API endpoints',
async () => {
const topic = {
name: 'orders',
diff --git a/web/src/api/metadata.ts b/web/src/api/metadata.ts
index 9e02bced..966a71af 100644
--- a/web/src/api/metadata.ts
+++ b/web/src/api/metadata.ts
@@ -75,6 +75,23 @@ export interface ConsumerInstance {
topicLag: Record<string, number>;
}
+export interface ConsumerThreadStack {
+ threadName: string;
+ threadId: number;
+ state: string;
+ blockedTime: number;
+ waitedTime: number;
+ stackTrace: string[];
+}
+
+export interface ConsumerStackTrace {
+ groupName: string;
+ clientId: string;
+ capturedAt: string;
+ threadCount: number;
+ threads: ConsumerThreadStack[];
+}
+
export interface ConsumerGroupDetail extends ConsumerGroup {
instances: ConsumerInstance[];
}
@@ -193,6 +210,14 @@ export async function getConsumerSubscriptions(name:
string, instanceId?: string
return res.data.data;
}
+export async function getConsumerStack(name: string, clientId: string,
instanceId?: string) {
+ const res = await client.get<{ data: ConsumerStackTrace }>(
+
`/groups/${encodeURIComponent(name)}/instances/${encodeURIComponent(clientId)}/stack`,
+ { params: instanceId ? { instanceId } : {} },
+ );
+ return res.data.data;
+}
+
export async function createConsumerGroup(data: Partial<ConsumerGroup>) {
const res = await client.post<{ data: ConsumerGroup }>('/groups/create',
data);
return res.data.data;
diff --git a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
index 66605e8b..951b30ef 100644
--- a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
+++ b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
@@ -33,6 +33,7 @@ vi.mock('../../../services/consumerService', () => ({
deleteConsumerGroup: vi.fn(),
getConsumerGroup: vi.fn(),
getConsumerProgress: vi.fn(),
+ getConsumerStack: vi.fn(),
getConsumerSubscriptions: vi.fn(),
listConsumerGroups: vi.fn(),
resetConsumerOffset: vi.fn(),
@@ -143,6 +144,22 @@ describe('Consumer page', () => {
consistency: '一致',
},
]);
+ vi.mocked(consumerService.getConsumerStack).mockResolvedValue({
+ groupName: 'remote-cg',
+ clientId: 'client-1',
+ capturedAt: '2026-07-23T00:00:00Z',
+ threadCount: 1,
+ threads: [
+ {
+ threadName: 'ConsumeMessageThread_1',
+ threadId: 12,
+ state: 'RUNNABLE',
+ blockedTime: 0,
+ waitedTime: 0,
+ stackTrace:
['org.apache.demo.OrderListener.consume(OrderListener.java:42)'],
+ },
+ ],
+ });
instanceServiceMocks.listInstances.mockResolvedValue([
{
id: 'instance-1',
@@ -313,6 +330,43 @@ describe('Consumer page', () => {
);
});
+ it('loads a consumer client stack trace from the selected instance', async
() => {
+ vi.mocked(consumerService.listConsumerGroups).mockResolvedValue([
+ {
+ ...group,
+ instances: [
+ {
+ clientId: 'client-1',
+ protocol: 'Remoting',
+ address: '10.0.0.1:39210',
+ subscribedTopics: ['remote-topic'],
+ lastHeartbeat: '2026-07-23T00:00:00Z',
+ topicLag: {},
+ },
+ ],
+ },
+ ]);
+ const user = userEvent.setup();
+ renderWithProviders(<ConsumerPage />);
+
+ await user.click(await screen.findByRole('button', { name: /详情/ }));
+ await user.click(await screen.findByRole('tab', { name: /在线实例/ }));
+ await user.click(await screen.findByRole('button', { name: /线程栈/ }));
+
+ await waitFor(() =>
+ expect(consumerService.getConsumerStack).toHaveBeenCalledWith(
+ 'remote-cg',
+ 'client-1',
+ 'instance-1',
+ ),
+ );
+ expect(await screen.findByText('消费者线程栈')).toBeInTheDocument();
+ expect(screen.getByText('ConsumeMessageThread_1')).toBeInTheDocument();
+ expect(
+
screen.getByText('org.apache.demo.OrderListener.consume(OrderListener.java:42)'),
+ ).toBeInTheDocument();
+ });
+
it('highlights inconsistent subscriptions and refreshes the check result',
async () => {
vi.mocked(consumerService.getConsumerSubscriptions)
.mockResolvedValueOnce([
diff --git a/web/src/pages/instance/consumer.tsx
b/web/src/pages/instance/consumer.tsx
index a11b4cba..114264ad 100644
--- a/web/src/pages/instance/consumer.tsx
+++ b/web/src/pages/instance/consumer.tsx
@@ -65,6 +65,7 @@ import { formatDateTime } from '../../utils/format';
import type {
ConsumerGroup,
ConsumerInstance,
+ ConsumerStackTrace,
QueueProgress,
SubscriptionEntry,
} from '../../api/metadata';
@@ -73,6 +74,7 @@ import {
createConsumerGroup,
deleteConsumerGroup,
getConsumerProgress,
+ getConsumerStack,
getConsumerSubscriptions,
listConsumerGroups,
resetConsumerOffset,
@@ -216,6 +218,10 @@ const ConsumerPage = () => {
);
const [showOnlyInconsistent, setShowOnlyInconsistent] = useState(false);
const [progressByGroup, setProgressByGroup] = useState<Record<string,
QueueProgress[]>>({});
+ const [stackModalOpen, setStackModalOpen] = useState(false);
+ const [stackLoading, setStackLoading] = useState(false);
+ const [selectedStack, setSelectedStack] = useState<ConsumerStackTrace |
null>(null);
+ const [selectedStackClient, setSelectedStackClient] =
useState<ConsumerInstance | null>(null);
const importInputRef = useRef<HTMLInputElement>(null);
const [importModalOpen, setImportModalOpen] = useState(false);
const [importFilename, setImportFilename] = useState('');
@@ -343,6 +349,26 @@ const ConsumerPage = () => {
: selectedSubscriptions;
const selectedProgress = selectedGroup ?
(progressByGroup[selectedDiagnosticKey] ?? []) : [];
+ const openStackModal = async (consumerInstance: ConsumerInstance) => {
+ if (!selectedGroup) return;
+ setSelectedStackClient(consumerInstance);
+ setSelectedStack(null);
+ setStackModalOpen(true);
+ setStackLoading(true);
+ try {
+ const stack = await getConsumerStack(
+ selectedGroup.name,
+ consumerInstance.clientId,
+ selectedInstanceId || undefined,
+ );
+ setSelectedStack(stack);
+ } catch {
+ message.error(`客户端 ${consumerInstance.clientId} 线程栈获取失败`);
+ } finally {
+ setStackLoading(false);
+ }
+ };
+
const handleImportFile = async (file: File) => {
if (!selectedInstanceId) {
message.error('请先选择实例');
@@ -698,6 +724,20 @@ const ConsumerPage = () => {
</Text>
),
},
+ {
+ title: '诊断',
+ key: 'diagnostics',
+ width: 110,
+ render: (_: unknown, record: ConsumerInstance) => (
+ <Button
+ size="small"
+ icon={<ListBullets size={14} />}
+ onClick={() => void openStackModal(record)}
+ >
+ 线程栈
+ </Button>
+ ),
+ },
];
/* ═══════════════════════════════════════════
@@ -1231,6 +1271,91 @@ const ConsumerPage = () => {
)}
</Modal>
+ {/* ═══════════════════════════════════════════
+ Consumer Stack Modal
+ ═══════════════════════════════════════════ */}
+ <Modal
+ title={
+ <Space>
+ <ListBullets size={18} color="#1677ff" />
+ <span>消费者线程栈</span>
+ </Space>
+ }
+ open={stackModalOpen}
+ onCancel={() => {
+ setStackModalOpen(false);
+ setSelectedStack(null);
+ setSelectedStackClient(null);
+ }}
+ footer={null}
+ width={900}
+ destroyOnClose
+ >
+ <Space direction="vertical" size={16} style={{ width: '100%' }}>
+ <Descriptions bordered column={2} size="small">
+ <Descriptions.Item label="Group">
+ <Text strong>{selectedStack?.groupName ?? selectedGroup?.name ??
'-'}</Text>
+ </Descriptions.Item>
+ <Descriptions.Item label="Client ID">
+ <Text copyable>
+ {selectedStack?.clientId ?? selectedStackClient?.clientId ??
'-'}
+ </Text>
+ </Descriptions.Item>
+ <Descriptions.Item label="采集时间">
+ {selectedStack?.capturedAt ?
formatDateTime(selectedStack.capturedAt) : '-'}
+ </Descriptions.Item>
+ <Descriptions.Item label="线程数">{selectedStack?.threadCount ??
0}</Descriptions.Item>
+ </Descriptions>
+
+ {stackLoading ? (
+ <Table
+ loading
+ columns={[{ title: '线程', dataIndex: 'threadName', key:
'threadName' }]}
+ dataSource={[]}
+ pagination={false}
+ size="small"
+ />
+ ) : selectedStack && selectedStack.threads.length > 0 ? (
+ selectedStack.threads.map((thread) => (
+ <Card
+ key={`${thread.threadName}-${thread.threadId}`}
+ size="small"
+ title={
+ <Space>
+ <Text strong>{thread.threadName}</Text>
+ <Tag color="blue">TID {thread.threadId}</Tag>
+ <Tag color={thread.state === 'RUNNABLE' ? 'green' :
'orange'}>
+ {thread.state}
+ </Tag>
+ </Space>
+ }
+ >
+ <pre
+ style={{
+ margin: 0,
+ maxHeight: 240,
+ overflow: 'auto',
+ whiteSpace: 'pre-wrap',
+ wordBreak: 'break-word',
+ fontSize: 12,
+ lineHeight: 1.6,
+ }}
+ >
+ {thread.stackTrace.join('\n')}
+ </pre>
+ </Card>
+ ))
+ ) : (
+ <Alert
+ type="info"
+ showIcon
+ message="暂无线程栈数据"
+ description="客户端在线但没有返回可展示的 jstack 内容，或该客户端暂时无法采集线程信息。"
+ />
+ )}
+ </Space>
+ </Modal>
+
{/* ═══════════════════════════════════════════
Create Group Modal
═══════════════════════════════════════════ */}
diff --git a/web/src/services/consumerService.test.ts
b/web/src/services/consumerService.test.ts
index 27ef23b2..99328492 100644
--- a/web/src/services/consumerService.test.ts
+++ b/web/src/services/consumerService.test.ts
@@ -20,6 +20,7 @@ import {
createConsumerGroup,
getConsumerGroup,
getConsumerProgress,
+ getConsumerStack,
getConsumerSubscriptions,
listConsumerGroups,
} from './consumerService';
@@ -88,6 +89,15 @@ describe('consumer service mock data', () => {
expect(secondSubscriptions[0]).not.toBe(firstSubscriptions[0]);
});
+ it('returns an empty mock consumer stack trace', async () => {
+ const stack = await getConsumerStack('cg-order-notify', 'client-1');
+
+ expect(stack.groupName).toBe('cg-order-notify');
+ expect(stack.clientId).toBe('client-1');
+ expect(stack.threadCount).toBe(0);
+ expect(stack.threads).toEqual([]);
+ });
+
it('returns a copy after creating consumer groups', async () => {
const created = await createConsumerGroup({
name: 'cg-created-copy-test',
diff --git a/web/src/services/consumerService.ts
b/web/src/services/consumerService.ts
index dfabd684..d7f6ad09 100644
--- a/web/src/services/consumerService.ts
+++ b/web/src/services/consumerService.ts
@@ -4,6 +4,7 @@ import type {
ConsumerGroup,
ConsumerGroupQuery,
ConsumerGroupDetail,
+ ConsumerStackTrace,
QueueProgress,
ResetConsumerOffsetRequest,
SubscriptionEntry,
@@ -85,6 +86,23 @@ export async function getConsumerSubscriptions(
return metadataApi.getConsumerSubscriptions(name, instanceId);
}
+export async function getConsumerStack(
+ name: string,
+ clientId: string,
+ instanceId?: string,
+): Promise<ConsumerStackTrace> {
+ if (isMockMode()) {
+ return {
+ groupName: name,
+ clientId,
+ capturedAt: new Date().toISOString(),
+ threadCount: 0,
+ threads: [],
+ };
+ }
+ return metadataApi.getConsumerStack(name, clientId, instanceId);
+}
+
export async function createConsumerGroup(data: Partial<ConsumerGroup>):
Promise<ConsumerGroup> {
if (isMockMode()) {
const now = new Date().toISOString();
