[Studio] Add consumer stack diagnostics API

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);
+}

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();
+    }
+}

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);
+    }
+}

server/src/main/java/com/rocketmq/studio/instance/group/ConsumerGroupController.java

@@ -37,6 +37,7 @@
 public class ConsumerGroupController {
 
     private final MetadataService metadataService;
+    private final ConsumerDiagnosticsService consumerDiagnosticsService;
 
     @GetMapping
     public Result<List<ConsumerGroupVO>> listConsumerGroups(
@@ -60,6 +61,13 @@ public Result<List<SubscriptionEntryVO>> getGroupSubscriptions(@PathVariable Str
         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));

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;
+}

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;
+}

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");
+    }
+}

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"));
+    }
+}