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
commit 628a6ccab48c2c560f130a533d2d81f17bd00ff4 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 00000000..cf843487 --- /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 00000000..34f7f286 --- /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 00000000..8bf35f85 --- /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 395744f9..9ac96f85 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 00000000..6ed26af7 --- /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 00000000..3af17b34 --- /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 00000000..7ca5720b --- /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 00000000..be454532 --- /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")); + } +}
