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 5155a18 feat: add consumer stack diagnostics API (#521)
5155a18 is described below
commit 5155a180abcc520948eb4e0c03b28bc01133985e
Author: aias00 <[email protected]>
AuthorDate: Thu Jul 23 23:21:46 2026 -0700
feat: add consumer stack diagnostics API (#521)
Add consumer thread stack diagnostics endpoint (GET
/api/groups/{name}/instances/{clientId}/stack) with Provider/Stub/Service
layering.
---
.../group/ConsumerDiagnosticsProvider.java | 22 ++++++
.../group/ConsumerDiagnosticsProviderStub.java | 41 +++++++++++
.../instance/group/ConsumerDiagnosticsService.java | 41 +++++++++++
.../instance/group/ConsumerGroupController.java | 8 +++
.../instance/group/ConsumerStackTraceVO.java | 38 +++++++++++
.../instance/group/ConsumerThreadStackVO.java | 38 +++++++++++
.../group/ConsumerDiagnosticsServiceTest.java | 76 +++++++++++++++++++++
.../group/ConsumerGroupControllerTest.java | 79 ++++++++++++++++++++++
8 files changed, 343 insertions(+)
diff --git
a/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerDiagnosticsProvider.java
b/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerDiagnosticsProvider.java
new file mode 100644
index 0000000..cf84348
--- /dev/null
+++
b/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerDiagnosticsProvider.java
@@ -0,0 +1,22 @@
+/*
+ * 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 com.rocketmq.studio.instance.group;
+
+public interface ConsumerDiagnosticsProvider {
+ ConsumerStackTraceVO getConsumerStack(String groupName, String clientId);
+}
diff --git
a/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerDiagnosticsProviderStub.java
b/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerDiagnosticsProviderStub.java
new file mode 100644
index 0000000..34f7f28
--- /dev/null
+++
b/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerDiagnosticsProviderStub.java
@@ -0,0 +1,41 @@
+/*
+ * 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 com.rocketmq.studio.instance.group;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.time.LocalDateTime;
+import java.util.List;
+
+@Slf4j
+@Component
+public class ConsumerDiagnosticsProviderStub implements
ConsumerDiagnosticsProvider {
+
+ @Override
+ public ConsumerStackTraceVO getConsumerStack(String groupName, String
clientId) {
+ log.warn("ConsumerDiagnosticsProviderStub.getConsumerStack called -
returning empty stack");
+ return ConsumerStackTraceVO.builder()
+ .groupName(groupName)
+ .clientId(clientId)
+ .capturedAt(LocalDateTime.now())
+ .threadCount(0)
+ .threads(List.of())
+ .build();
+ }
+}
diff --git
a/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerDiagnosticsService.java
b/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerDiagnosticsService.java
new file mode 100644
index 0000000..8bf35f8
--- /dev/null
+++
b/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerDiagnosticsService.java
@@ -0,0 +1,41 @@
+/*
+ * 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 com.rocketmq.studio.instance.group;
+
+import com.rocketmq.studio.common.exception.BusinessException;
+import lombok.RequiredArgsConstructor;
+import org.springframework.http.HttpStatus;
+import org.springframework.stereotype.Service;
+import org.springframework.util.StringUtils;
+
+@Service
+@RequiredArgsConstructor
+public class ConsumerDiagnosticsService {
+
+ private final ConsumerDiagnosticsProvider diagnosticsProvider;
+
+ public ConsumerStackTraceVO getConsumerStack(String groupName, String
clientId) {
+ if (!StringUtils.hasText(groupName)) {
+ throw new BusinessException(HttpStatus.BAD_REQUEST.value(),
"groupName is required");
+ }
+ if (!StringUtils.hasText(clientId)) {
+ throw new BusinessException(HttpStatus.BAD_REQUEST.value(),
"clientId is required");
+ }
+ return diagnosticsProvider.getConsumerStack(groupName, clientId);
+ }
+}
diff --git
a/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerGroupController.java
b/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerGroupController.java
index 395744f..9ac96f8 100644
---
a/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerGroupController.java
+++
b/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerGroupController.java
@@ -37,6 +37,7 @@ import java.util.Map;
public class ConsumerGroupController {
private final MetadataService metadataService;
+ private final ConsumerDiagnosticsService consumerDiagnosticsService;
@GetMapping
public Result<List<ConsumerGroupVO>> listConsumerGroups(
@@ -60,6 +61,13 @@ public class ConsumerGroupController {
return Result.ok(metadataService.getGroupSubscriptions(name));
}
+ @GetMapping("/{name}/instances/{clientId}/stack")
+ public Result<ConsumerStackTraceVO> getConsumerStack(
+ @PathVariable String name,
+ @PathVariable String clientId) {
+ return Result.ok(consumerDiagnosticsService.getConsumerStack(name,
clientId));
+ }
+
@PostMapping("/create")
public Result<ConsumerGroupVO> createConsumerGroup(@RequestBody
ConsumerGroupVO group) {
return Result.ok(metadataService.createConsumerGroup(group));
diff --git
a/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerStackTraceVO.java
b/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerStackTraceVO.java
new file mode 100644
index 0000000..6ed26af
--- /dev/null
+++
b/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerStackTraceVO.java
@@ -0,0 +1,38 @@
+/*
+ * 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 com.rocketmq.studio.instance.group;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.time.LocalDateTime;
+import java.util.List;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class ConsumerStackTraceVO {
+ private String groupName;
+ private String clientId;
+ private LocalDateTime capturedAt;
+ private int threadCount;
+ private List<ConsumerThreadStackVO> threads;
+}
diff --git
a/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerThreadStackVO.java
b/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerThreadStackVO.java
new file mode 100644
index 0000000..3af17b3
--- /dev/null
+++
b/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerThreadStackVO.java
@@ -0,0 +1,38 @@
+/*
+ * 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 com.rocketmq.studio.instance.group;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.List;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class ConsumerThreadStackVO {
+ private String threadName;
+ private long threadId;
+ private String state;
+ private long blockedTime;
+ private long waitedTime;
+ private List<String> stackTrace;
+}
diff --git
a/server/src/test/java/com/rocketmq/studio/instance/group/ConsumerDiagnosticsServiceTest.java
b/server/src/test/java/com/rocketmq/studio/instance/group/ConsumerDiagnosticsServiceTest.java
new file mode 100644
index 0000000..7ca5720
--- /dev/null
+++
b/server/src/test/java/com/rocketmq/studio/instance/group/ConsumerDiagnosticsServiceTest.java
@@ -0,0 +1,76 @@
+/*
+ * 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 com.rocketmq.studio.instance.group;
+
+import com.rocketmq.studio.common.exception.BusinessException;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.time.LocalDateTime;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class ConsumerDiagnosticsServiceTest {
+
+ @Mock
+ private ConsumerDiagnosticsProvider diagnosticsProvider;
+
+ @InjectMocks
+ private ConsumerDiagnosticsService diagnosticsService;
+
+ @Test
+ void getConsumerStackShouldDelegateToProvider() {
+ ConsumerStackTraceVO stackTrace = ConsumerStackTraceVO.builder()
+ .groupName("cg-orders")
+ .clientId("client-1")
+ .capturedAt(LocalDateTime.now())
+ .threadCount(0)
+ .threads(List.of())
+ .build();
+
+ when(diagnosticsProvider.getConsumerStack("cg-orders",
"client-1")).thenReturn(stackTrace);
+
+ ConsumerStackTraceVO result =
diagnosticsService.getConsumerStack("cg-orders", "client-1");
+
+ assertThat(result.getGroupName()).isEqualTo("cg-orders");
+ assertThat(result.getClientId()).isEqualTo("client-1");
+ verify(diagnosticsProvider).getConsumerStack("cg-orders", "client-1");
+ }
+
+ @Test
+ void getConsumerStackShouldRejectBlankGroupName() {
+ assertThatThrownBy(() -> diagnosticsService.getConsumerStack(" ",
"client-1"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("groupName is required");
+ }
+
+ @Test
+ void getConsumerStackShouldRejectBlankClientId() {
+ assertThatThrownBy(() ->
diagnosticsService.getConsumerStack("cg-orders", " "))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("clientId is required");
+ }
+}
diff --git
a/server/src/test/java/com/rocketmq/studio/instance/group/ConsumerGroupControllerTest.java
b/server/src/test/java/com/rocketmq/studio/instance/group/ConsumerGroupControllerTest.java
new file mode 100644
index 0000000..be45453
--- /dev/null
+++
b/server/src/test/java/com/rocketmq/studio/instance/group/ConsumerGroupControllerTest.java
@@ -0,0 +1,79 @@
+/*
+ * 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 com.rocketmq.studio.instance.group;
+
+import com.rocketmq.studio.instance.topic.MetadataService;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import
org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
+import org.springframework.boot.test.autoconfigure.web.servlet.WebMvcTest;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.test.web.servlet.MockMvc;
+
+import java.time.LocalDateTime;
+import java.util.List;
+
+import static org.mockito.Mockito.when;
+import static
org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
+import static
org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
+import static
org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+
+@WebMvcTest(ConsumerGroupController.class)
+@AutoConfigureMockMvc(addFilters = false)
+class ConsumerGroupControllerTest {
+
+ @Autowired
+ private MockMvc mockMvc;
+
+ @MockBean
+ private MetadataService metadataService;
+
+ @MockBean
+ private ConsumerDiagnosticsService consumerDiagnosticsService;
+
+ @Test
+ void getConsumerStackShouldReturnStackTrace() throws Exception {
+ ConsumerThreadStackVO thread = ConsumerThreadStackVO.builder()
+ .threadName("ConsumeMessageThread_1")
+ .threadId(42L)
+ .state("RUNNABLE")
+ .blockedTime(0L)
+ .waitedTime(5L)
+
.stackTrace(List.of("org.apache.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService.run"))
+ .build();
+ ConsumerStackTraceVO stackTrace = ConsumerStackTraceVO.builder()
+ .groupName("cg-orders")
+ .clientId("client-1")
+ .capturedAt(LocalDateTime.of(2026, 7, 23, 12, 0))
+ .threadCount(1)
+ .threads(List.of(thread))
+ .build();
+
+ when(consumerDiagnosticsService.getConsumerStack("cg-orders",
"client-1")).thenReturn(stackTrace);
+
+ mockMvc.perform(get("/api/groups/cg-orders/instances/client-1/stack"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.code").value(200))
+ .andExpect(jsonPath("$.data.groupName").value("cg-orders"))
+ .andExpect(jsonPath("$.data.clientId").value("client-1"))
+ .andExpect(jsonPath("$.data.threadCount").value(1))
+
.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"));
+ }
+}
