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 2871f3bf fix: scope topic diagnostics to selected instance (#1128)
2871f3bf is described below
commit 2871f3bfc29c76e8078aae84e1b2543f690140fe
Author: aias00 <[email protected]>
AuthorDate: Thu Aug 6 23:56:34 2026 -0700
fix: scope topic diagnostics to selected instance (#1128)
---
.../studio/instance/topic/MetadataService.java | 2 +-
.../provider/apache/ApacheInstanceProvider.java | 2 +-
.../studio/provider/apache/MetadataProvider.java | 4 +-
.../provider/apache/RocketMQMetadataProvider.java | 193 +++++++++++----------
.../studio/instance/topic/MetadataServiceTest.java | 14 ++
.../studio/instance/topic/TopicControllerTest.java | 16 ++
.../apache/RocketMQMetadataProviderTest.java | 30 +++-
web/src/api/metadata.test.ts | 12 +-
.../pages/instance/__tests__/TopicPage.test.tsx | 20 ++-
9 files changed, 191 insertions(+), 102 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
index 04b80abe..fa7b1eae 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
@@ -88,7 +88,7 @@ public class MetadataService {
// broker routing does not apply to serverless cloud instances
return List.of();
}
- return metadataProvider.getTopicRoutes(name);
+ return metadataProvider.getTopicRoutes(instanceId, name);
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
index a9e4236a..4915303a 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
@@ -85,7 +85,7 @@ public class ApacheInstanceProvider implements
InstanceProvider {
@Override
public List<TopicConsumerVO> getTopicConsumers(String instanceId, String
topicName) {
- return metadataProvider.getTopicConsumers(topicName);
+ return metadataProvider.getTopicConsumers(instanceId, topicName);
}
@Override
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/MetadataProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/MetadataProvider.java
index e5ccef5c..173e7c8a 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/MetadataProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/MetadataProvider.java
@@ -28,8 +28,8 @@ import java.util.List;
public interface MetadataProvider {
List<TopicVO> listTopics(String clusterId, String type, String search);
List<ConsumerGroupVO> listConsumerGroups(String clusterId, String search);
- List<BrokerRouteVO> getTopicRoutes(String name);
- List<TopicConsumerVO> getTopicConsumers(String name);
+ List<BrokerRouteVO> getTopicRoutes(String instanceId, String name);
+ List<TopicConsumerVO> getTopicConsumers(String instanceId, String name);
List<QueueProgressVO> getGroupProgress(String name);
List<SubscriptionEntryVO> getGroupSubscriptions(String name);
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
index 90233979..b17703bb 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
@@ -28,6 +28,8 @@ import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.remoting.protocol.route.QueueData;
import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
+import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
+import org.apache.rocketmq.tools.admin.MQAdminExt;
import org.apache.rocketmq.studio.common.domain.enums.ConsumeType;
import org.apache.rocketmq.studio.common.domain.enums.TopicPerm;
import org.apache.rocketmq.studio.common.domain.enums.TopicType;
@@ -83,6 +85,7 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
private final RocketMQProperties properties;
private final RmqTopicMapper topicMapper;
private final RmqGroupMapper groupMapper;
+ private final RuntimeAdminClientResolver runtimeAdminClientResolver;
/** Whether a default NameServer is configured and live queries are
therefore possible. */
private boolean hasAdmin() {
@@ -192,125 +195,133 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
}
@Override
- public List<BrokerRouteVO> getTopicRoutes(String name) {
+ public List<BrokerRouteVO> getTopicRoutes(String instanceId, String name) {
+ if (StringUtils.hasText(instanceId)) {
+ return runtimeAdminClientResolver.execute(instanceId, admin ->
getTopicRoutes(admin, name));
+ }
if (!hasAdmin()) {
return Collections.emptyList();
}
+ return adminExecute(admin -> getTopicRoutes(admin, name));
+ }
- return adminExecute(admin -> {
- try {
- TopicRouteData routeData = admin.examineTopicRouteInfo(name);
- if (routeData == null) {
- return Collections.emptyList();
- }
+ private List<BrokerRouteVO> getTopicRoutes(MQAdminExt admin, String name) {
+ try {
+ TopicRouteData routeData = admin.examineTopicRouteInfo(name);
+ if (routeData == null) {
+ return Collections.emptyList();
+ }
- Map<String, BrokerData> brokerDataMap = new HashMap<>();
- if (routeData.getBrokerDatas() != null) {
- for (BrokerData bd : routeData.getBrokerDatas()) {
- brokerDataMap.put(bd.getBrokerName(), bd);
- }
+ Map<String, BrokerData> brokerDataMap = new HashMap<>();
+ if (routeData.getBrokerDatas() != null) {
+ for (BrokerData bd : routeData.getBrokerDatas()) {
+ brokerDataMap.put(bd.getBrokerName(), bd);
}
+ }
- List<BrokerRouteVO> routes = new ArrayList<>();
- if (routeData.getQueueDatas() != null) {
- for (QueueData qd : routeData.getQueueDatas()) {
- BrokerData bd = brokerDataMap.get(qd.getBrokerName());
- String brokerAddr = "";
- if (bd != null && bd.getBrokerAddrs() != null &&
!bd.getBrokerAddrs().isEmpty()) {
- brokerAddr =
bd.getBrokerAddrs().get(MixAll.MASTER_ID);
- if (brokerAddr == null) {
- brokerAddr =
bd.getBrokerAddrs().values().iterator().next();
- }
+ List<BrokerRouteVO> routes = new ArrayList<>();
+ if (routeData.getQueueDatas() != null) {
+ for (QueueData qd : routeData.getQueueDatas()) {
+ BrokerData bd = brokerDataMap.get(qd.getBrokerName());
+ String brokerAddr = "";
+ if (bd != null && bd.getBrokerAddrs() != null &&
!bd.getBrokerAddrs().isEmpty()) {
+ brokerAddr = bd.getBrokerAddrs().get(MixAll.MASTER_ID);
+ if (brokerAddr == null) {
+ brokerAddr =
bd.getBrokerAddrs().values().iterator().next();
}
-
- routes.add(BrokerRouteVO.builder()
- .brokerName(qd.getBrokerName())
- .brokerAddr(brokerAddr)
- .writeQueues(qd.getWriteQueueNums())
- .readQueues(qd.getReadQueueNums())
- .perm(mapPerm(qd.getPerm()))
- .build());
}
+
+ routes.add(BrokerRouteVO.builder()
+ .brokerName(qd.getBrokerName())
+ .brokerAddr(brokerAddr)
+ .writeQueues(qd.getWriteQueueNums())
+ .readQueues(qd.getReadQueueNums())
+ .perm(mapPerm(qd.getPerm()))
+ .build());
}
- return routes;
- } catch (Exception e) {
- log.warn("Failed to get routes for topic {}: {}", name,
e.getMessage());
- return Collections.emptyList();
}
- });
+ return routes;
+ } catch (Exception e) {
+ log.warn("Failed to get routes for topic {}: {}", name,
e.getMessage());
+ return Collections.emptyList();
+ }
}
@Override
- public List<TopicConsumerVO> getTopicConsumers(String name) {
+ public List<TopicConsumerVO> getTopicConsumers(String instanceId, String
name) {
+ if (StringUtils.hasText(instanceId)) {
+ return runtimeAdminClientResolver.execute(instanceId, admin ->
getTopicConsumers(admin, name));
+ }
if (!hasAdmin()) {
return Collections.emptyList();
}
+ return adminExecute(admin -> getTopicConsumers(admin, name));
+ }
- return adminExecute(admin -> {
- try {
- // Ask the broker who consumes this topic instead of scanning
every subscription
- // group, which floods the result with system groups.
- GroupList groupList = admin.queryTopicConsumeByWho(name);
- Set<String> subscribingGroups = new HashSet<>();
- if (groupList != null && groupList.getGroupList() != null) {
- for (String group : groupList.getGroupList()) {
- if (!isSystemConsumerGroup(group)) {
- subscribingGroups.add(group);
- }
+ private List<TopicConsumerVO> getTopicConsumers(MQAdminExt admin, String
name) {
+ try {
+ // Ask the broker who consumes this topic instead of scanning
every subscription
+ // group, which floods the result with system groups.
+ GroupList groupList = admin.queryTopicConsumeByWho(name);
+ Set<String> subscribingGroups = new HashSet<>();
+ if (groupList != null && groupList.getGroupList() != null) {
+ for (String group : groupList.getGroupList()) {
+ if (!isSystemConsumerGroup(group)) {
+ subscribingGroups.add(group);
}
}
+ }
- List<TopicConsumerVO> consumers = new ArrayList<>();
- for (String group : subscribingGroups) {
- try {
- ConsumeStats stats = admin.examineConsumeStats(group,
name);
- long diffTotal = 0;
- double consumeTps = 0;
- if (stats != null && stats.getOffsetTable() != null) {
- for (Map.Entry<MessageQueue, OffsetWrapper> entry
: stats.getOffsetTable().entrySet()) {
- OffsetWrapper ow = entry.getValue();
- diffTotal += Math.max(0, ow.getBrokerOffset()
- ow.getConsumerOffset());
- }
- consumeTps = stats.getConsumeTps();
+ List<TopicConsumerVO> consumers = new ArrayList<>();
+ for (String group : subscribingGroups) {
+ try {
+ ConsumeStats stats = admin.examineConsumeStats(group,
name);
+ long diffTotal = 0;
+ double consumeTps = 0;
+ if (stats != null && stats.getOffsetTable() != null) {
+ for (Map.Entry<MessageQueue, OffsetWrapper> entry :
stats.getOffsetTable().entrySet()) {
+ OffsetWrapper ow = entry.getValue();
+ diffTotal += Math.max(0, ow.getBrokerOffset() -
ow.getConsumerOffset());
}
+ consumeTps = stats.getConsumeTps();
+ }
- ConsumeType consumeType = ConsumeType.CLUSTERING;
- String messageModel = "CLUSTERING";
- try {
- ConsumerConnection conn =
admin.examineConsumerConnectionInfo(group);
- if (conn != null && conn.getMessageModel() !=
null) {
- messageModel = conn.getMessageModel().name();
- consumeType = parseConsumeType(messageModel);
- }
- } catch (Exception ignored) {
- // group may be offline
+ ConsumeType consumeType = ConsumeType.CLUSTERING;
+ String messageModel = "CLUSTERING";
+ try {
+ ConsumerConnection conn =
admin.examineConsumerConnectionInfo(group);
+ if (conn != null && conn.getMessageModel() != null) {
+ messageModel = conn.getMessageModel().name();
+ consumeType = parseConsumeType(messageModel);
}
-
- consumers.add(TopicConsumerVO.builder()
- .group(group)
- .consumeType(consumeType)
- .messageModel(messageModel)
- .consumeTps(consumeTps)
- .diffTotal(diffTotal)
- .build());
} catch (Exception ignored) {
- // stats unavailable for this group, still list it
below without numbers
- consumers.add(TopicConsumerVO.builder()
- .group(group)
- .consumeType(ConsumeType.CLUSTERING)
- .messageModel("CLUSTERING")
- .consumeTps(0)
- .diffTotal(0)
- .build());
+ // group may be offline
}
+
+ consumers.add(TopicConsumerVO.builder()
+ .group(group)
+ .consumeType(consumeType)
+ .messageModel(messageModel)
+ .consumeTps(consumeTps)
+ .diffTotal(diffTotal)
+ .build());
+ } catch (Exception ignored) {
+ // stats unavailable for this group, still list it below
without numbers
+ consumers.add(TopicConsumerVO.builder()
+ .group(group)
+ .consumeType(ConsumeType.CLUSTERING)
+ .messageModel("CLUSTERING")
+ .consumeTps(0)
+ .diffTotal(0)
+ .build());
}
- consumers.sort((a, b) ->
a.getGroup().compareToIgnoreCase(b.getGroup()));
- return consumers;
- } catch (Exception e) {
- log.warn("Failed to get consumers for topic {}: {}", name,
e.getMessage());
- return Collections.emptyList();
}
- });
+ consumers.sort((a, b) ->
a.getGroup().compareToIgnoreCase(b.getGroup()));
+ return consumers;
+ } catch (Exception e) {
+ log.warn("Failed to get consumers for topic {}: {}", name,
e.getMessage());
+ return Collections.emptyList();
+ }
}
private boolean isSystemConsumerGroup(String group) {
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
index caa06436..4d6d02e1 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
@@ -157,6 +157,20 @@ class MetadataServiceTest {
verify(apacheProvider).deleteTopic(null, "topic-to-delete");
}
+ @Test
+ void topicRuntimeDiagnosticsShouldDelegateWithSelectedInstance() {
+ BrokerRouteVO route =
BrokerRouteVO.builder().brokerName("broker-a").build();
+ TopicConsumerVO consumer =
TopicConsumerVO.builder().group("cg-orders").build();
+ when(metadataProvider.getTopicRoutes("instance-a",
"orders")).thenReturn(List.of(route));
+ when(apacheProvider.getTopicConsumers("instance-a",
"orders")).thenReturn(List.of(consumer));
+
+ assertThat(metadataService.getTopicRoutes("instance-a",
"orders")).containsExactly(route);
+ assertThat(metadataService.getTopicConsumers("instance-a",
"orders")).containsExactly(consumer);
+
+ verify(metadataProvider).getTopicRoutes("instance-a", "orders");
+ verify(apacheProvider).getTopicConsumers("instance-a", "orders");
+ }
+
@Test
void sendMessageShouldReturnResult() {
SendMessageDTO request = SendMessageDTO.builder()
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/TopicControllerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/TopicControllerTest.java
index 6da4af95..4cdd8317 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/TopicControllerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/TopicControllerTest.java
@@ -87,6 +87,22 @@ class TopicControllerTest {
verify(metadataService).listTopics(isNull(), eq("cluster-1"),
eq("NORMAL"), eq("test"));
}
+ @Test
+ void topicRuntimeDiagnosticsShouldPassSelectedInstance() throws Exception {
+ when(metadataService.getTopicRoutes("instance-a",
"orders")).thenReturn(List.of());
+ when(metadataService.getTopicConsumers("instance-a",
"orders")).thenReturn(List.of());
+
+ mockMvc.perform(get("/api/topics/orders/routes").param("instanceId",
"instance-a"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.code").value(200));
+
mockMvc.perform(get("/api/topics/orders/consumers").param("instanceId",
"instance-a"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.code").value(200));
+
+ verify(metadataService).getTopicRoutes("instance-a", "orders");
+ verify(metadataService).getTopicConsumers("instance-a", "orders");
+ }
+
@Test
void createTopicShouldReturnCreatedTopic() throws Exception {
TopicVO input = new TopicVO();
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
index e7cf16a1..55991581 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
@@ -18,6 +18,9 @@ package org.apache.rocketmq.studio.provider.apache;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
import org.apache.rocketmq.studio.common.domain.enums.ConsumeType;
+import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
+import org.apache.rocketmq.studio.instance.topic.BrokerRouteVO;
+import org.apache.rocketmq.studio.instance.topic.TopicConsumerVO;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
import org.apache.rocketmq.studio.persistence.entity.RmqGroup;
import org.apache.rocketmq.studio.persistence.mapper.RmqGroupMapper;
@@ -31,7 +34,9 @@ import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
@@ -43,13 +48,16 @@ class RocketMQMetadataProviderTest {
@Mock
private RmqGroupMapper groupMapper;
+ @Mock
+ private RuntimeAdminClientResolver runtimeAdminClientResolver;
+
/**
* Builds a provider without a configured NameServer, mirroring the former
absent admin bean:
* DB-backed listings work while live enrichment is skipped.
*/
private RocketMQMetadataProvider newProvider() {
return new RocketMQMetadataProvider(mock(MqAdminExtFactory.class), new
RocketMQProperties(),
- topicMapper, groupMapper);
+ topicMapper, groupMapper, runtimeAdminClientResolver);
}
@Test
@@ -85,4 +93,24 @@ class RocketMQMetadataProviderTest {
assertThat(groups).hasSize(1);
assertThat(groups.get(0).getConsumeType()).isEqualTo(ConsumeType.CLUSTERING);
}
+
+ @Test
+ void getTopicRoutesShouldUseSelectedInstanceRuntimeClient() {
+ List<BrokerRouteVO> routes =
List.of(BrokerRouteVO.builder().brokerName("broker-a").build());
+ when(runtimeAdminClientResolver.execute(eq("instance-a"),
any())).thenReturn(routes);
+ RocketMQMetadataProvider provider = newProvider();
+
+ assertThat(provider.getTopicRoutes("instance-a",
"orders")).containsExactlyElementsOf(routes);
+ verify(runtimeAdminClientResolver).execute(eq("instance-a"), any());
+ }
+
+ @Test
+ void getTopicConsumersShouldUseSelectedInstanceRuntimeClient() {
+ List<TopicConsumerVO> consumers =
List.of(TopicConsumerVO.builder().group("cg-orders").build());
+ when(runtimeAdminClientResolver.execute(eq("instance-a"),
any())).thenReturn(consumers);
+ RocketMQMetadataProvider provider = newProvider();
+
+ assertThat(provider.getTopicConsumers("instance-a",
"orders")).containsExactlyElementsOf(consumers);
+ verify(runtimeAdminClientResolver).execute(eq("instance-a"), any());
+ }
}
diff --git a/web/src/api/metadata.test.ts b/web/src/api/metadata.test.ts
index 3a57c1bc..d004f440 100644
--- a/web/src/api/metadata.test.ts
+++ b/web/src/api/metadata.test.ts
@@ -52,11 +52,15 @@ describe('topic metadata API', () => {
it('encodes topic names used in path segments', async () => {
const topicName = '%DLQ%cg-order';
- mock.onGet('/topics/%25DLQ%25cg-order/routes').reply(200, { code: 200,
data: [] });
- mock.onGet('/topics/%25DLQ%25cg-order/consumers').reply(200, { code: 200,
data: [] });
+ mock
+ .onGet('/topics/%25DLQ%25cg-order/routes', { params: { instanceId:
'instance-a' } })
+ .reply(200, { code: 200, data: [] });
+ mock
+ .onGet('/topics/%25DLQ%25cg-order/consumers', { params: { instanceId:
'instance-a' } })
+ .reply(200, { code: 200, data: [] });
- await expect(getTopicRoutes(topicName)).resolves.toEqual([]);
- await expect(getTopicConsumers(topicName)).resolves.toEqual([]);
+ await expect(getTopicRoutes(topicName, 'instance-a')).resolves.toEqual([]);
+ await expect(getTopicConsumers(topicName,
'instance-a')).resolves.toEqual([]);
});
it('persists topic creation, deletion, and sending through API endpoints',
async () => {
diff --git a/web/src/pages/instance/__tests__/TopicPage.test.tsx
b/web/src/pages/instance/__tests__/TopicPage.test.tsx
index 9fb9cc26..eaffb8ff 100644
--- a/web/src/pages/instance/__tests__/TopicPage.test.tsx
+++ b/web/src/pages/instance/__tests__/TopicPage.test.tsx
@@ -187,7 +187,23 @@ describe('TopicPage', () => {
it('keeps the current table page after opening and closing topic details',
async () => {
const user = userEvent.setup();
- renderWithProviders();
+ instanceServiceMocks.listInstances.mockResolvedValue([
+ {
+ id: 'instance-a',
+ name: 'Instance A',
+ type: 'DIRECT',
+ endpoint: '127.0.0.1:9876',
+ remark: '',
+ topicCount: 25,
+ consumerGroupCount: 0,
+ createdAt: '2026-01-01T00:00:00Z',
+ updatedAt: '2026-01-01T00:00:00Z',
+ },
+ ]);
+ topicServiceMocks.listTopics.mockResolvedValue(
+ buildTopics(25).map((topic) => ({ ...topic, instanceId: 'instance-a' })),
+ );
+ renderWithProviders('/instance/instance-a/topic');
expect(await screen.findByText('topic-01')).toBeInTheDocument();
@@ -200,7 +216,7 @@ describe('TopicPage', () => {
await user.click(screen.getAllByRole('button', { name: /详情/ })[0]);
await waitFor(() =>
-
expect(topicServiceMocks.getTopicRoutes).toHaveBeenCalledWith('topic-21',
'instance-proxy-1'),
+
expect(topicServiceMocks.getTopicRoutes).toHaveBeenCalledWith('topic-21',
'instance-a'),
);
const closeButton = document.querySelector('.ant-modal-close');
