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 206314b9a fix(ai): make topic and group tools work for cluster scope
and cloud data (#4115)
206314b9a is described below
commit 206314b9a3a9701683bb6462abad5739ab059a84
Author: Zhao Jianing <[email protected]>
AuthorDate: Wed Sep 9 20:47:09 2026 +0800
fix(ai): make topic and group tools work for cluster scope and cloud data
(#4115)
The AI resource tool chain was broken end to end: every one of
`rmq.topic.list` and `rmq.group.list` failed, and for cloud instances the
failure happened even once routing was fixed. Four separate defects, each fixed
here, plus one stale test that hid a contract drift.
**Routing.** Both handlers passed the tool's `cluster` input into the
`instanceId` slot — `listTopicsPage(clusterId, null, ...)` — so the request
never reached the `!hasText(instanceId) && hasText(clusterId)` cluster-scoped
branch. Because `cluster` is `required` with `minLength: 1` in the catalog, the
legacy branch was unreachable and both tools threw `404 Instance not found:
DefaultCluster` on every call. The arguments are now `(null, clusterId, ...)`.
Regression introduced by #3052.
**Output schema.** `rmq-tools.yaml` declared `cluster.proxies`,
`stats.totalProxies`, `stats.totalNameServers` and `items[].totalLag` as `type:
integer` while keeping them `required`, but the handlers deliberately emit
`null` to mean "unknown" — `totalLag` does so for `ConsumerLagResolver.UNKNOWN`
since #3988. `ToolGatewayService.validateOutput` preserves null keys through
`valueToTree`, so listing any group with an unknown lag raised
`IllegalStateException` and returned 500. Those fo [...]
**Cloud topic type and perm.** The Aliyun converter never set `perm` at
all, so every Aliyun topic carried a null perm and
`TopicListToolHandler.safeProjection` threw on its first `requiredEnumName`
call. Both cloud `toTopicType` implementations also returned null for a blank
or unrecognised type. They now guarantee non-null: `TopicPerm.RW` for Aliyun
(its ListTopics API returns no permission field, and console-created cloud
topics are read-write, matching the Tencent provider) and `T [...]
**Cloud consumer group subscription mode.** Neither cloud provider set
`subscriptionMode`, and Aliyun's `toConsumeType` returned null unless
`messageModel` was exactly "Clustering", so `requiredEnumName` threw for every
cloud consumer group. Both now set `SubscriptionMode.Push` — the pinned
`alibabacloud-rocketmq20220801` response body has no subscription-mode field at
all, and cloud TCP groups are push consumers — and `toConsumeType` returns
`CLUSTERING`.
**Test contract drift.**
`AliyunInstanceProviderTest.getGroupProgressShouldMapLagRowsTest` still
expected a per-topic row *and* a `broker="total"` aggregate row, but
`AliyunConverters.toQueueProgressRows` has only emitted the aggregate as a
fallback when there are no topic rows since #2907. The expectation was stale,
not the production code: emitting both would double-count lag, because
`CloudRocketMqBusinessMetricsCollector` sums `getDiffTotal()` across every row
it is given. The tes [...]
Folded in from #4117, #4134, #4135 and #4137, all by the same author and
all part of this one chain; those PRs are closed as superseded. Merging any
single one of them would not have made the tools usable.
---
.../ops/ai/tool/ConsumerGroupListToolHandler.java | 2 +-
.../studio/ops/ai/tool/TopicListToolHandler.java | 2 +-
.../studio/provider/alibaba/AliyunConverters.java | 26 ++++---
.../provider/tencent/TencentInstanceProvider.java | 13 +++-
.../src/main/resources/tool-catalog/rmq-tools.yaml | 8 +-
.../ai/tool/ConsumerGroupListToolHandlerTest.java | 75 ++++++++++++++++++
.../studio/ops/ai/tool/ToolGatewayServiceTest.java | 67 +++++++++++++++-
.../ops/ai/tool/TopicListToolHandlerTest.java | 78 +++++++++++++++++++
.../alibaba/AliyunInstanceProviderTest.java | 90 ++++++++++++++++++++--
.../tencent/TencentInstanceProviderTest.java | 22 ++++++
10 files changed, 355 insertions(+), 28 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/ConsumerGroupListToolHandler.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/ConsumerGroupListToolHandler.java
index fcaf58381..87792fed5 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/ConsumerGroupListToolHandler.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/ConsumerGroupListToolHandler.java
@@ -45,7 +45,7 @@ public class ConsumerGroupListToolHandler implements
ToolHandler {
String clusterId = (String) input.get("cluster");
String search = (String) input.get("search");
PageResult<ConsumerGroupVO> page =
metadataService.listConsumerGroupsPage(
- clusterId, null, search, ToolListPagination.page(input),
ToolListPagination.pageSize(input));
+ null, clusterId, search, ToolListPagination.page(input),
ToolListPagination.pageSize(input));
return ToolListPagination.pagedResult(page, page.getItems().stream()
.map(ConsumerGroupListToolHandler::safeProjection)
.toList());
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/TopicListToolHandler.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/TopicListToolHandler.java
index 910cff251..50f669f76 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/TopicListToolHandler.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/TopicListToolHandler.java
@@ -44,7 +44,7 @@ public class TopicListToolHandler implements ToolHandler {
String type = (String) input.get("type");
String search = (String) input.get("search");
PageResult<TopicVO> page = metadataService.listTopicsPage(
- clusterId, null, type, search, ToolListPagination.page(input),
ToolListPagination.pageSize(input));
+ null, clusterId, type, search, ToolListPagination.page(input),
ToolListPagination.pageSize(input));
return ToolListPagination.pagedResult(page, page.getItems().stream()
.map(TopicListToolHandler::safeProjection)
.toList());
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConverters.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConverters.java
index e8a49cf89..f658e48ce 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConverters.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConverters.java
@@ -29,6 +29,8 @@ import
com.aliyun.sdk.service.rocketmq20220801.models.ListTopicSubscriptionsResp
import com.aliyun.sdk.service.rocketmq20220801.models.ListTopicsResponseBody;
import org.apache.rocketmq.studio.common.domain.enums.ConsumeType;
import org.apache.rocketmq.studio.common.domain.enums.DeliveryStatus;
+import org.apache.rocketmq.studio.common.domain.enums.SubscriptionMode;
+import org.apache.rocketmq.studio.common.domain.enums.TopicPerm;
import org.apache.rocketmq.studio.common.domain.enums.TopicType;
import org.apache.rocketmq.studio.common.util.SubscriptionFilterModes;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
@@ -118,6 +120,9 @@ final class AliyunConverters {
vo.setName(data.getTopicName());
vo.setInstanceId(studioInstanceId);
vo.setType(toTopicType(data.getMessageType()));
+ // Aliyun's ListTopics API does not return permissions;
console-created cloud
+ // topics are read-write, matching the Tencent provider's mapping.
+ vo.setPerm(TopicPerm.RW);
vo.setRemark(data.getRemark());
vo.setGmtCreate(parseDateTime(data.getCreateTime()));
vo.setGmtModified(parseDateTime(data.getUpdateTime()));
@@ -127,8 +132,8 @@ final class AliyunConverters {
}
static TopicType toTopicType(String messageType) {
- if (messageType == null) {
- return null;
+ if (messageType == null || messageType.isBlank()) {
+ return TopicType.NORMAL;
}
switch (messageType.toUpperCase(Locale.ROOT)) {
case "NORMAL":
@@ -140,7 +145,10 @@ final class AliyunConverters {
case "TRANSACTION":
return TopicType.TRANSACTION;
default:
- return null;
+ // Unknown message types fall back to NORMAL so read paths (web
+ // detail, AI rmq.topic.list) never see a null type, matching
the
+ // Apache provider's parseTopicType fallback.
+ return TopicType.NORMAL;
}
}
@@ -157,22 +165,20 @@ final class AliyunConverters {
vo.setName(data.getConsumerGroupId());
vo.setInstanceId(studioInstanceId);
vo.setConsumeType(toConsumeType(data.getMessageModel()));
+ // Aliyun's messageModel carries the consume model, not the
subscription mode; cloud TCP
+ // consumer groups are push consumers. Read paths (web detail, AI
rmq.group.list) require
+ // a non-null subscriptionMode, mirroring the Apache provider
invariant.
+ vo.setSubscriptionMode(SubscriptionMode.Push);
vo.setGmtCreate(parseDateTime(data.getCreateTime()));
vo.setGmtModified(parseDateTime(data.getUpdateTime()));
return vo;
}
static ConsumeType toConsumeType(String messageModel) {
- if (messageModel == null) {
- return null;
- }
- if ("Clustering".equalsIgnoreCase(messageModel)) {
- return ConsumeType.CLUSTERING;
- }
if ("Broadcasting".equalsIgnoreCase(messageModel)) {
return ConsumeType.BROADCASTING;
}
- return null;
+ return ConsumeType.CLUSTERING;
}
static List<QueueProgressVO>
toQueueProgressRows(GetConsumerGroupLagResponseBody.Data data) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProvider.java
index 07c536404..9730693a2 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProvider.java
@@ -48,6 +48,7 @@ import org.apache.rocketmq.studio.common.domain.PageResult;
import org.apache.rocketmq.studio.common.domain.enums.ConsumeType;
import org.apache.rocketmq.studio.common.domain.enums.DeliveryStatus;
import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
+import org.apache.rocketmq.studio.common.domain.enums.SubscriptionMode;
import org.apache.rocketmq.studio.common.domain.enums.TopicPerm;
import org.apache.rocketmq.studio.common.domain.enums.TopicType;
import org.apache.rocketmq.studio.common.exception.BusinessException;
@@ -944,6 +945,9 @@ public class TencentInstanceProvider implements
InstanceProvider {
group.setClusterId(item.getClusterIdV4());
group.setNamespace(item.getNamespaceV4());
group.setConsumeType(toConsumeType(item.getConsumeMessageOrderly()));
+ // Tencent consumer groups are TCP push consumers; read paths (web
detail,
+ // AI rmq.group.list) require a non-null subscriptionMode.
+ group.setSubscriptionMode(SubscriptionMode.Push);
group.setDeliveryOrderType(item.getConsumeMessageOrderly() == null ||
!item.getConsumeMessageOrderly()
? "Concurrently" : "Orderly");
group.setRetryMaxTimes(toInt(item.getMaxRetryTimes()));
@@ -1036,12 +1040,15 @@ public class TencentInstanceProvider implements
InstanceProvider {
private static TopicType toTopicType(String raw) {
if (!StringUtils.hasText(raw)) {
- return null;
+ return TopicType.NORMAL;
}
try {
return TopicType.valueOf(raw.trim().toUpperCase(Locale.ROOT));
- } catch (IllegalArgumentException ignored) {
- return null;
+ } catch (IllegalArgumentException ex) {
+ // Unknown topic types fall back to NORMAL so read paths (web
detail,
+ // AI rmq.topic.list) never see a null type, matching the Apache
+ // provider's parseTopicType fallback.
+ return TopicType.NORMAL;
}
}
diff --git a/server/src/main/resources/tool-catalog/rmq-tools.yaml
b/server/src/main/resources/tool-catalog/rmq-tools.yaml
index 647467a18..e822461e9 100644
--- a/server/src/main/resources/tool-catalog/rmq-tools.yaml
+++ b/server/src/main/resources/tool-catalog/rmq-tools.yaml
@@ -126,7 +126,7 @@ tools:
brokers:
type: integer
proxies:
- type: integer
+ type: [integer, 'null']
topics:
type: integer
groups:
@@ -164,9 +164,9 @@ tools:
totalBrokers:
type: integer
totalProxies:
- type: integer
+ type: [integer, 'null']
totalNameServers:
- type: integer
+ type: [integer, 'null']
totalTopics:
type: integer
totalConsumerGroups:
@@ -332,7 +332,7 @@ tools:
onlineInstances:
type: integer
totalLag:
- type: integer
+ type: [integer, 'null']
subscribedTopics:
type: array
items:
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/ConsumerGroupListToolHandlerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/ConsumerGroupListToolHandlerTest.java
new file mode 100644
index 000000000..ec3701e88
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/ConsumerGroupListToolHandlerTest.java
@@ -0,0 +1,75 @@
+/*
+ * 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.ops.ai.tool;
+
+import org.apache.rocketmq.studio.common.domain.PageResult;
+import org.apache.rocketmq.studio.common.domain.enums.ConsumeType;
+import org.apache.rocketmq.studio.common.domain.enums.SubscriptionMode;
+import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
+import org.apache.rocketmq.studio.instance.topic.MetadataService;
+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.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class ConsumerGroupListToolHandlerTest {
+
+ @Mock
+ private MetadataService metadataService;
+
+ @InjectMocks
+ private ConsumerGroupListToolHandler handler;
+
+ @Test
+ void executeShouldRouteClusterToClusterScopedRead() {
+ ConsumerGroupVO group = new ConsumerGroupVO();
+ group.setName("cg-orders");
+ group.setClusterId("DefaultCluster");
+ group.setSubscriptionMode(SubscriptionMode.Push);
+ group.setConsumeType(ConsumeType.CLUSTERING);
+ group.setOnlineInstances(3);
+ group.setTotalLag(42L);
+ group.setSubscribedTopics(List.of("orders"));
+ group.setRetryMaxTimes(16);
+ when(metadataService.listConsumerGroupsPage(isNull(),
eq("DefaultCluster"), eq("order"), eq(1), eq(20)))
+ .thenReturn(PageResult.of(List.of(group), 1L, 1, 20));
+
+ Object result = handler.execute(Map.of("cluster", "DefaultCluster",
"search", "order"));
+
+ Map<?, ?> page = (Map<?, ?>) result;
+ assertThat(page.get("total")).isEqualTo(1L);
+ List<?> items = (List<?>) page.get("items");
+ assertThat(items).hasSize(1);
+ Map<?, ?> row = (Map<?, ?>) items.get(0);
+ assertThat(row.get("name")).isEqualTo("cg-orders");
+ assertThat(row.get("subscriptionMode")).isEqualTo("Push");
+ assertThat(row.get("consumeType")).isEqualTo("CLUSTERING");
+ assertThat(row.get("totalLag")).isEqualTo(42L);
+ verify(metadataService).listConsumerGroupsPage(isNull(),
eq("DefaultCluster"), eq("order"), eq(1), eq(20));
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/ToolGatewayServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/ToolGatewayServiceTest.java
index 1b88dde62..b78aec568 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/ToolGatewayServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/ToolGatewayServiceTest.java
@@ -41,6 +41,7 @@ import
org.apache.rocketmq.studio.ops.dashboard.ClusterOverviewVO;
import org.apache.rocketmq.studio.ops.dashboard.DashboardDataVO;
import org.apache.rocketmq.studio.ops.dashboard.DashboardService;
import org.apache.rocketmq.studio.ops.dashboard.DashboardStatsVO;
+import org.apache.rocketmq.studio.provider.apache.ConsumerLagResolver;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -54,6 +55,8 @@ import java.util.Map;
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.anyInt;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
@@ -339,6 +342,48 @@ class ToolGatewayServiceTest {
"tpsOut", 8L));
}
+ @Test
+ @SuppressWarnings("unchecked")
+ void executesDashboardSummaryWhenTopologyCountsAreUnavailable() {
+ // Mirrors the V5/degraded provider paths (RocketMQDashboardProvider)
and
+ //
DashboardControllerTest.getDashboardShouldPreserveUnavailableTopologyCounts:
+ // proxy/name-server counts are deliberately null when the topology is
unavailable.
+
when(dashboardService.getDashboard()).thenReturn(DashboardDataVO.builder()
+ .stats(DashboardStatsVO.builder()
+ .totalClusters(1)
+ .healthyClusters(1)
+ .totalBrokers(2)
+ .totalProxies(null)
+ .totalNameServers(null)
+ .totalTopics(3)
+ .totalConsumerGroups(4)
+ .build())
+ .clusters(List.of(ClusterOverviewVO.builder()
+ .id("cluster-v5")
+ .name("test")
+ .type(ClusterType.V5_PROXY_CLUSTER)
+ .status(ClusterStatus.healthy)
+ .brokers(2)
+ .proxies(null)
+ .topics(3)
+ .groups(4)
+ .tpsIn(7)
+ .tpsOut(8)
+ .version("5.2.0")
+ .build()))
+ .build());
+
+ Object output = gateway.execute(
+ "rmq.dashboard.summary", Map.of("cluster", "cluster-v5"));
+
+ Map<String, Object> result = (Map<String, Object>) output;
+ Map<String, Object> cluster = (Map<String, Object>)
result.get("cluster");
+ assertThat(cluster).containsEntry("proxies", null);
+ Map<String, Object> stats = (Map<String, Object>) result.get("stats");
+ assertThat(stats).containsEntry("totalProxies", null);
+ assertThat(stats).containsEntry("totalNameServers", null);
+ }
+
@Test
void rejectsDashboardSummaryWithoutRequiredClusterBeforeHandlerRuns() {
assertThatThrownBy(() -> gateway.execute("rmq.dashboard.summary",
Map.of()))
@@ -391,7 +436,7 @@ class ToolGatewayServiceTest {
when(clusterService.getCluster("cluster-v5")).thenReturn(cluster(ClusterType.V5_PROXY_CLUSTER));
TopicVO topic = topic();
topic.setRemark("do-not-expose");
- when(metadataService.listTopicsPage("cluster-v5", null, "NORMAL",
"order", 2, 20))
+ when(metadataService.listTopicsPage(null, "cluster-v5", "NORMAL",
"order", 2, 20))
.thenReturn(PageResult.of(List.of(topic), 101, 2, 20));
Object output = gateway.execute("rmq.topic.list", Map.of(
@@ -432,7 +477,7 @@ class ToolGatewayServiceTest {
when(clusterService.getCluster("cluster-v5")).thenReturn(cluster(ClusterType.V5_PROXY_CLUSTER));
ConsumerGroupVO group = consumerGroup();
group.setDelaySeconds(30);
- when(metadataService.listConsumerGroupsPage("cluster-v5", null,
"order", 2, 20))
+ when(metadataService.listConsumerGroupsPage(null, "cluster-v5",
"order", 2, 20))
.thenReturn(PageResult.of(List.of(group), 101, 2, 20));
Object output = gateway.execute("rmq.group.list", Map.of(
@@ -458,6 +503,24 @@ class ToolGatewayServiceTest {
assertThat(output.toString()).doesNotContain("delaySeconds");
}
+ @Test
+ @SuppressWarnings("unchecked")
+ void executesConsumerGroupListWhenLagIsUnknown() {
+
when(clusterService.getCluster("cluster-v5")).thenReturn(cluster(ClusterType.V5_PROXY_CLUSTER));
+ ConsumerGroupVO group = consumerGroup();
+ group.setTotalLag(ConsumerLagResolver.UNKNOWN);
+ when(metadataService.listConsumerGroupsPage(any(), any(), any(),
anyInt(), anyInt()))
+ .thenReturn(PageResult.of(List.of(group), 1, 1, 20));
+
+ Object output = gateway.execute("rmq.group.list", Map.of(
+ "cluster", "cluster-v5", "search", "order", "page", 1,
"pageSize", 20));
+
+ Map<String, Object> page = (Map<String, Object>) output;
+ List<Map<String, Object>> items = (List<Map<String, Object>>)
page.get("items");
+ assertThat(items).hasSize(1);
+ assertThat(items.get(0)).containsEntry("totalLag", null);
+ }
+
@Test
void rejectsConsumerGroupListWithoutAClusterBeforeHandlerRuns() {
assertThatThrownBy(() -> gateway.execute("rmq.group.list", Map.of()))
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/TopicListToolHandlerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/TopicListToolHandlerTest.java
new file mode 100644
index 000000000..73d7883f0
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/TopicListToolHandlerTest.java
@@ -0,0 +1,78 @@
+/*
+ * 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.ops.ai.tool;
+
+import org.apache.rocketmq.studio.common.domain.PageResult;
+import org.apache.rocketmq.studio.common.domain.enums.TopicPerm;
+import org.apache.rocketmq.studio.common.domain.enums.TopicType;
+import org.apache.rocketmq.studio.instance.topic.MetadataService;
+import org.apache.rocketmq.studio.instance.topic.TopicVO;
+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.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class TopicListToolHandlerTest {
+
+ @Mock
+ private MetadataService metadataService;
+
+ @InjectMocks
+ private TopicListToolHandler handler;
+
+ @Test
+ void executeShouldRouteClusterToClusterScopedRead() {
+ TopicVO topic = new TopicVO();
+ topic.setName("orders");
+ topic.setClusterId("DefaultCluster");
+ topic.setType(TopicType.NORMAL);
+ topic.setPerm(TopicPerm.RW);
+ topic.setWriteQueues(8);
+ topic.setReadQueues(8);
+ topic.setMessageCount(100L);
+ topic.setTps(1.5D);
+ topic.setConsumerGroupCount(2);
+ when(metadataService.listTopicsPage(isNull(), eq("DefaultCluster"),
eq("NORMAL"), eq("order"), eq(1), eq(20)))
+ .thenReturn(PageResult.of(List.of(topic), 1L, 1, 20));
+
+ Object result = handler.execute(Map.of(
+ "cluster", "DefaultCluster", "type", "NORMAL", "search",
"order"));
+
+ Map<?, ?> page = (Map<?, ?>) result;
+ assertThat(page.get("total")).isEqualTo(1L);
+ assertThat(page.get("page")).isEqualTo(1);
+ assertThat(page.get("size")).isEqualTo(20);
+ List<?> items = (List<?>) page.get("items");
+ assertThat(items).hasSize(1);
+ Map<?, ?> row = (Map<?, ?>) items.get(0);
+ assertThat(row.get("name")).isEqualTo("orders");
+ assertThat(row.get("type")).isEqualTo("NORMAL");
+ assertThat(row.get("perm")).isEqualTo("RW");
+ verify(metadataService).listTopicsPage(isNull(), eq("DefaultCluster"),
eq("NORMAL"), eq("order"), eq(1), eq(20));
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProviderTest.java
index fb45c4936..0ae1136cb 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProviderTest.java
@@ -37,7 +37,9 @@ import
com.aliyun.sdk.service.rocketmq20220801.models.ResetConsumeOffsetRequest;
import
com.aliyun.sdk.service.rocketmq20220801.models.ResetConsumeOffsetResponse;
import
com.aliyun.sdk.service.rocketmq20220801.models.ResetConsumeOffsetResponseBody;
import org.apache.rocketmq.studio.common.domain.enums.ConsumeType;
+import org.apache.rocketmq.studio.common.domain.enums.SubscriptionMode;
import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
+import org.apache.rocketmq.studio.common.domain.enums.TopicPerm;
import org.apache.rocketmq.studio.common.domain.enums.TopicType;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.instance.InstanceRepository;
@@ -129,11 +131,12 @@ class AliyunInstanceProviderTest {
assertThat(all).hasSize(3);
assertThat(all.get(0).getName()).isEqualTo("topic-normal");
assertThat(all.get(0).getType()).isEqualTo(TopicType.NORMAL);
+ assertThat(all.get(0).getPerm()).isEqualTo(TopicPerm.RW);
assertThat(all.get(0).getInstanceId()).isEqualTo(STUDIO_INSTANCE_PK);
assertThat(all.get(0).getWriteQueues()).isZero();
assertThat(all.get(0).getReadQueues()).isZero();
assertThat(all.get(0).getRemark()).isEqualTo("remark-topic-normal");
- assertThat(all.get(2).getType()).isNull();
+ assertThat(all.get(2).getType()).isEqualTo(TopicType.NORMAL);
List<TopicVO> fifos = provider.listTopics(STUDIO_INSTANCE_ID, "FIFO",
null);
@@ -141,6 +144,24 @@ class AliyunInstanceProviderTest {
assertThat(fifos.get(0).getType()).isEqualTo(TopicType.FIFO);
}
+ @Test
+ void listTopicsShouldGuaranteeTypeAndPermForAiToolProjectionTest() {
+ stubInstance();
+ stubCallThrough();
+
when(asyncClient.listTopics(any(ListTopicsRequest.class))).thenReturn(CompletableFuture.completedFuture(
+ topicsResponse(
+ topicRow("topic-untyped", null),
+ topicRow("topic-unknown", "NEW_TYPE"))));
+
+ List<TopicVO> topics = provider.listTopics(STUDIO_INSTANCE_ID, null,
null);
+
+ assertThat(topics).hasSize(2);
+ assertThat(topics).allSatisfy(topic -> {
+ assertThat(topic.getType()).isEqualTo(TopicType.NORMAL);
+ assertThat(topic.getPerm()).isEqualTo(TopicPerm.RW);
+ });
+ }
+
@Test
void listTopicsShouldTraversePastLegacyFivePageCapTest() {
stubInstance();
@@ -222,6 +243,37 @@ class AliyunInstanceProviderTest {
assertThat(groups.get(0).getName()).isEqualTo("GID_test");
assertThat(groups.get(0).getInstanceId()).isEqualTo(STUDIO_INSTANCE_PK);
assertThat(groups.get(0).getConsumeType()).isEqualTo(ConsumeType.CLUSTERING);
+
assertThat(groups.get(0).getSubscriptionMode()).isEqualTo(SubscriptionMode.Push);
+ }
+
+ @Test
+ void listConsumerGroupsShouldFallBackWhenMessageModelMissingTest() {
+ stubInstance();
+ stubCallThrough();
+ ListConsumerGroupsResponse response =
ListConsumerGroupsResponse.create().toBuilder()
+ .statusCode(200)
+ .body(ListConsumerGroupsResponseBody.builder()
+ .data(ListConsumerGroupsResponseBody.Data.builder()
+
.list(java.util.Arrays.asList(ListConsumerGroupsResponseBody.List.builder()
+ .consumerGroupId("GID_plain")
+ .status("RUNNING")
+ .build()))
+ .pageNumber(1L)
+ .pageSize(100L)
+ .totalCount(1L)
+ .build())
+ .build())
+ .build();
+ when(asyncClient.listConsumerGroups(any()))
+ .thenReturn(CompletableFuture.completedFuture(response));
+
+ List<ConsumerGroupVO> groups =
provider.listConsumerGroups(STUDIO_INSTANCE_ID, null);
+
+ assertThat(groups).singleElement().satisfies(group -> {
+ // read paths (web detail, AI rmq.group.list) require both enums
to be non-null
+
assertThat(group.getConsumeType()).isEqualTo(ConsumeType.CLUSTERING);
+
assertThat(group.getSubscriptionMode()).isEqualTo(SubscriptionMode.Push);
+ });
}
@Test
@@ -313,17 +365,41 @@ class AliyunInstanceProviderTest {
List<QueueProgressVO> rows =
provider.getGroupProgress(STUDIO_INSTANCE_ID, "GID_test");
- assertThat(rows).hasSize(2);
+ // with a topic breakdown the aggregate total row is dropped,
otherwise callers
+ // that sum the rows would report the same lag twice
+ assertThat(rows).hasSize(1);
QueueProgressVO topicRow = rows.stream()
.filter(row -> "topic:topic-a".equals(row.getBroker()))
.findFirst()
.orElseThrow();
assertThat(topicRow.getDiffTotal()).isEqualTo(42L);
- QueueProgressVO totalRow = rows.stream()
- .filter(row -> "total".equals(row.getBroker()))
- .findFirst()
- .orElseThrow();
- assertThat(totalRow.getDiffTotal()).isEqualTo(100L);
+ assertThat(rows).noneMatch(row -> "total".equals(row.getBroker()));
+ }
+
+ @Test
+ void getGroupProgressShouldFallBackToTotalRowWithoutTopicBreakdownTest() {
+ stubInstance();
+ stubCallThrough();
+ GetConsumerGroupLagResponse response =
GetConsumerGroupLagResponse.create().toBuilder()
+ .statusCode(200)
+ .body(GetConsumerGroupLagResponseBody.builder()
+ .data(GetConsumerGroupLagResponseBody.Data.builder()
+ .consumerGroupId("GID_test")
+
.totalLag(GetConsumerGroupLagResponseBody.TotalLag.builder()
+ .readyCount(100L)
+ .build())
+ .build())
+ .build())
+ .build();
+ when(asyncClient.getConsumerGroupLag(any()))
+ .thenReturn(CompletableFuture.completedFuture(response));
+
+ List<QueueProgressVO> rows =
provider.getGroupProgress(STUDIO_INSTANCE_ID, "GID_test");
+
+ assertThat(rows).singleElement().satisfies(row -> {
+ assertThat(row.getBroker()).isEqualTo("total");
+ assertThat(row.getDiffTotal()).isEqualTo(100L);
+ });
}
@Test
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProviderTest.java
index 7b2844ba9..2f7d98b4a 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProviderTest.java
@@ -44,8 +44,10 @@ import
com.tencentcloudapi.trocket.v20230308.models.SubscriptionData;
import com.tencentcloudapi.trocket.v20230308.models.TopicItem;
import com.tencentcloudapi.trocket.v20230308.TrocketClient;
import org.apache.rocketmq.studio.common.domain.enums.ConsumeType;
+import org.apache.rocketmq.studio.common.domain.enums.SubscriptionMode;
import org.apache.rocketmq.studio.common.domain.enums.DeliveryStatus;
import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
+import org.apache.rocketmq.studio.common.domain.enums.TopicPerm;
import org.apache.rocketmq.studio.common.domain.enums.TopicType;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.instance.InstanceRepository;
@@ -210,6 +212,25 @@ class TencentInstanceProviderTest {
assertThat(captor.getValue().getFilters()[1].getValues()).containsExactly("FIFO");
}
+ @Test
+ void listTopicsShouldFallBackToNormalTypeWhenTopicTypeMissingTest() throws
Exception {
+ when(client.DescribeTopicList(any())).thenAnswer(invocation -> {
+ DescribeTopicListResponse response = new
DescribeTopicListResponse();
+ response.setData(new TopicItem[]{
+ topicItem("orders-untyped", null, 4L),
+ topicItem("orders-unknown", "NEW_TYPE", 4L)});
+ return response;
+ });
+ DescribeTopicResponse detail = new DescribeTopicResponse();
+ when(client.DescribeTopic(any())).thenReturn(detail);
+
+ List<TopicVO> topics = provider.listTopics(STUDIO_INSTANCE_ID, null,
null);
+
+ assertThat(topics).hasSize(2);
+ assertThat(topics).allSatisfy(topic ->
assertThat(topic.getType()).isEqualTo(TopicType.NORMAL));
+ assertThat(topics).allSatisfy(topic ->
assertThat(topic.getPerm()).isEqualTo(TopicPerm.RW));
+ }
+
@Test
void listTopicsPageShouldUseTencentNativePaginationAndFiltersTest() throws
Exception {
TopicItem item = topicItem("orders-fifo-10000", "FIFO", 8L);
@@ -536,6 +557,7 @@ class TencentInstanceProviderTest {
assertThat(groups.get(0).getRetryMaxTimes()).isEqualTo(16);
assertThat(groups.get(0).getGmtCreate()).isNotNull();
assertThat(groups.get(0).getConsumeType()).isEqualTo(ConsumeType.CLUSTERING);
+
assertThat(groups.get(0).getSubscriptionMode()).isEqualTo(SubscriptionMode.Push);
assertThat(groups.get(0).getInstances()).isNotNull().isEmpty();
}
