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 a8d263eb fix: harden RocketMQ metadata semantics, client contracts and
admin error reporting (#1025)
a8d263eb is described below
commit a8d263eb8bd67166955785efeaa2fbbf1dca2199
Author: aias00 <[email protected]>
AuthorDate: Wed Aug 5 00:51:47 2026 -0700
fix: harden RocketMQ metadata semantics, client contracts and admin error
reporting (#1025)
Consolidates #1012, #1018, #1019, #1020, #1021 and #1025 into a single
change set:
- report namespace discovery as a structured 501 error (#1012)
- replace simulated alert rule editing with explicit warnings and export
server YAML verbatim (#1018)
- retain topic permission on partial updates and prefer the master broker
address (#1019)
- mark capped producer connection scans as partial and count client
instances once (#1020)
- filter trace contexts by message id (#1021)
- surface consumer group admin failures instead of swallowing them (#1025,
closes #965 #1028)
Verified locally: backend 612 tests pass, frontend 388 tests pass, deployed
and smoke-tested.
---
.../studio/cluster/client/ClientConnectionVO.java | 1 +
.../studio/instance/topic/MetadataService.java | 2 +-
.../studio/rocketmq/RocketMQAdminClientImpl.java | 52 +++++++--
.../studio/rocketmq/RocketMQClientProvider.java | 16 +--
.../studio/rocketmq/RocketMQMessageProvider.java | 3 +
.../studio/rocketmq/RocketMQMetadataProvider.java | 33 +++---
.../studio/instance/topic/MetadataServiceTest.java | 8 ++
.../instance/topic/NamespaceControllerTest.java | 53 +++++++++
.../rocketmq/RocketMQAdminClientImplTest.java | 85 ++++++++++++++
web/src/api/connections.ts | 3 +-
web/src/pages/cluster/clients.tsx | 47 +++++---
web/src/pages/studio/AlertManagement.tsx | 130 ++++-----------------
.../studio/__tests__/AlertManagement.test.tsx | 22 ++--
13 files changed, 293 insertions(+), 162 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ClientConnectionVO.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ClientConnectionVO.java
index 04fd7cc2..479dc2ce 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ClientConnectionVO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ClientConnectionVO.java
@@ -40,5 +40,6 @@ public class ClientConnectionVO {
private ClientLanguage language;
private String version;
private LocalDateTime connectedAt;
+ private boolean partial;
private String clusterName;
}
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 4470d7e6..4a829d5e 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
@@ -119,7 +119,7 @@ public class MetadataService {
public List<NamespaceVO> listNamespaces() {
- throw new UnsupportedOperationException("Not implemented");
+ throw new BusinessException(501, "Namespace discovery is not
implemented by the current metadata provider");
}
private String normalizeFilter(String value) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQAdminClientImpl.java
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQAdminClientImpl.java
index 616370e6..f83268ee 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQAdminClientImpl.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQAdminClientImpl.java
@@ -16,14 +16,17 @@
*/
package org.apache.rocketmq.studio.rocketmq;
+import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import
org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.common.domain.enums.TopicPerm;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
import org.apache.rocketmq.studio.instance.topic.AdminClient;
import org.apache.rocketmq.studio.instance.topic.SendMessageDTO;
@@ -122,8 +125,14 @@ public class RocketMQAdminClientImpl implements
AdminClient {
vo.setSubscribedTopics(new
ArrayList<>(conn.getSubscriptionTable().keySet()));
}
}
- } catch (Exception ignored) {
- // Group may be offline
+ } catch (MQClientException exception) {
+ if (exception.getResponseCode() ==
ResponseCode.CONSUMER_NOT_ONLINE) {
+ log.debug("Consumer group {} is offline", name);
+ return vo;
+ }
+ throw new BusinessException(502, "Failed to get consumer group: "
+ exception.getMessage());
+ } catch (Exception exception) {
+ throw new BusinessException(502, "Failed to get consumer group: "
+ exception.getMessage());
}
return vo;
}
@@ -139,6 +148,11 @@ public class RocketMQAdminClientImpl implements
AdminClient {
int readQueues = topic.getReadQueues() > 0 ? topic.getReadQueues() : 8;
try {
+ RmqTopic existing = topicMapper.selectOne(
+ new LambdaQueryWrapper<RmqTopic>().eq(RmqTopic::getName,
topicName));
+ TopicPerm effectivePerm = topic.getPerm() != null
+ ? topic.getPerm()
+ : existing == null ? TopicPerm.RW :
fromRocketMQPerm(existing.getPerm());
Set<String> brokerAddrs = getAllMasterBrokerAddrs();
if (brokerAddrs.isEmpty()) {
throw new BusinessException(500, "No broker available to
create topic");
@@ -148,7 +162,7 @@ public class RocketMQAdminClientImpl implements AdminClient
{
topicConfig.setTopicName(topicName);
topicConfig.setWriteQueueNums(writeQueues);
topicConfig.setReadQueueNums(readQueues);
- topicConfig.setPerm(6); // RW
+ topicConfig.setPerm(toRocketMQPerm(effectivePerm));
for (String addr : brokerAddrs) {
adminExt.createAndUpdateTopicConfig(addr, topicConfig);
@@ -174,7 +188,7 @@ public class RocketMQAdminClientImpl implements AdminClient
{
entity.setTopicType(topic.getType() != null ?
topic.getType().name() : "NORMAL");
entity.setReadQueueNums(readQueues);
entity.setWriteQueueNums(writeQueues);
- entity.setPerm(6);
+ entity.setPerm(topicConfig.getPerm());
if (StringUtils.hasText(topic.getRemark())) {
entity.setRemark(topic.getRemark());
}
@@ -213,6 +227,11 @@ public class RocketMQAdminClientImpl implements
AdminClient {
int readQueues = topic.getReadQueues() > 0 ? topic.getReadQueues() : 8;
try {
+ RmqTopic existing = topicMapper.selectOne(
+ new LambdaQueryWrapper<RmqTopic>().eq(RmqTopic::getName,
topicName));
+ TopicPerm effectivePerm = topic.getPerm() != null
+ ? topic.getPerm()
+ : existing == null ? TopicPerm.RW :
fromRocketMQPerm(existing.getPerm());
Set<String> brokerAddrs = getAllMasterBrokerAddrs();
if (brokerAddrs.isEmpty()) {
throw new BusinessException(500, "No broker available to
update topic");
@@ -222,18 +241,17 @@ public class RocketMQAdminClientImpl implements
AdminClient {
topicConfig.setTopicName(topicName);
topicConfig.setWriteQueueNums(writeQueues);
topicConfig.setReadQueueNums(readQueues);
- topicConfig.setPerm(6); // RW
+ topicConfig.setPerm(toRocketMQPerm(effectivePerm));
for (String addr : brokerAddrs) {
adminExt.createAndUpdateTopicConfig(addr, topicConfig);
}
// Update DB record
- RmqTopic existing = topicMapper.selectOne(
- new LambdaQueryWrapper<RmqTopic>().eq(RmqTopic::getName,
topicName));
if (existing != null) {
existing.setWriteQueueNums(writeQueues);
existing.setReadQueueNums(readQueues);
+ existing.setPerm(topicConfig.getPerm());
existing.setUpdatedAt(LocalDateTime.now());
topicMapper.updateById(existing);
}
@@ -489,4 +507,24 @@ public class RocketMQAdminClientImpl implements
AdminClient {
}
return "DefaultCluster";
}
+
+ private int toRocketMQPerm(TopicPerm perm) {
+ if (perm == TopicPerm.RO) {
+ return 4;
+ }
+ if (perm == TopicPerm.WO) {
+ return 2;
+ }
+ return 6;
+ }
+
+ private TopicPerm fromRocketMQPerm(Integer perm) {
+ if (perm != null && perm == 4) {
+ return TopicPerm.RO;
+ }
+ if (perm != null && perm == 2) {
+ return TopicPerm.WO;
+ }
+ return TopicPerm.RW;
+ }
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQClientProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQClientProvider.java
index 00ffc193..157e6794 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQClientProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQClientProvider.java
@@ -26,6 +26,7 @@ import org.apache.rocketmq.remoting.protocol.body.TopicList;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.studio.cluster.client.ClientConnectionVO;
import org.apache.rocketmq.studio.cluster.client.ClientProvider;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.common.domain.enums.ClientLanguage;
import org.apache.rocketmq.studio.common.domain.enums.ClientType;
import org.apache.rocketmq.studio.common.domain.enums.Protocol;
@@ -96,12 +97,11 @@ public class RocketMQClientProvider implements
ClientProvider {
}
int scanned = 0;
- for (String topic : topics) {
- if (isSystemTopic(topic)) {
- continue;
- }
+ boolean capped = false;
+ for (String topic : topics.stream().filter(topic ->
!isSystemTopic(topic)).sorted().toList()) {
if (scanned >= MAX_PRODUCER_TOPIC_SCAN) {
- log.info("Producer connection scan capped at {} non-system
topics", MAX_PRODUCER_TOPIC_SCAN);
+ log.warn("Producer connection scan capped at {} non-system
topics", MAX_PRODUCER_TOPIC_SCAN);
+ capped = true;
break;
}
scanned++;
@@ -120,6 +120,9 @@ public class RocketMQClientProvider implements
ClientProvider {
log.warn("Failed to examine producer connection for topic={},
skipping", topic, e);
}
}
+ if (capped) {
+ result.forEach(connection -> connection.setPartial(true));
+ }
return result;
}
@@ -210,8 +213,7 @@ public class RocketMQClientProvider implements
ClientProvider {
if (normalized.startsWith("cons")) {
return ClientType.Consumer;
}
- log.warn("Unknown client type filter: {}", type);
- return null;
+ throw new BusinessException(400, "Unknown client type filter: " +
type);
}
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQMessageProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQMessageProvider.java
index 3ea7aff8..b361fa2e 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQMessageProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQMessageProvider.java
@@ -275,6 +275,9 @@ public class RocketMQMessageProvider implements
MessageProvider {
if (fields.length == 0) {
continue;
}
+ if (!targetMsgId.equals(field(fields, 5))) {
+ continue;
+ }
try {
switch (fields[0].trim()) {
case "Pub":
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQMetadataProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQMetadataProvider.java
index e017d746..fe8b2c5d 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQMetadataProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQMetadataProvider.java
@@ -18,6 +18,7 @@ package org.apache.rocketmq.studio.rocketmq;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
import org.apache.rocketmq.remoting.protocol.admin.OffsetWrapper;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
@@ -215,7 +216,10 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
BrokerData bd = brokerDataMap.get(qd.getBrokerName());
String brokerAddr = "";
if (bd != null && bd.getBrokerAddrs() != null &&
!bd.getBrokerAddrs().isEmpty()) {
- brokerAddr =
bd.getBrokerAddrs().values().iterator().next();
+ brokerAddr = bd.getBrokerAddrs().get(MixAll.MASTER_ID);
+ if (brokerAddr == null) {
+ brokerAddr =
bd.getBrokerAddrs().values().iterator().next();
+ }
}
routes.add(BrokerRouteVO.builder()
@@ -271,11 +275,9 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
String messageModel = "CLUSTERING";
try {
ConsumerConnection conn =
adminExt.examineConsumerConnectionInfo(group);
- if (conn != null && conn.getConsumeType() != null) {
- messageModel = conn.getConsumeType().name();
- if (conn.getMessageModel() != null) {
- messageModel = conn.getMessageModel().name();
- }
+ if (conn != null && conn.getMessageModel() != null) {
+ messageModel = conn.getMessageModel().name();
+ consumeType = parseConsumeType(messageModel);
}
} catch (Exception ignored) {
// group may be offline
@@ -376,7 +378,7 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
.topic(sd.getTopic())
.expression(sd.getSubString())
.type(sd.getExpressionType())
-
.filterMode("CLASS_FILTER".equals(sd.getExpressionType()) ? "SQL" : "TAG")
+ .filterMode(filterMode(sd.getExpressionType()))
.build());
}
return subscriptions;
@@ -424,13 +426,6 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
if (conn.getConnectionSet() != null) {
vo.setOnlineInstances(conn.getConnectionSet().size());
}
- if (conn.getMessageModel() != null) {
- vo.setSubscriptionMode(
-
"BROADCASTING".equals(conn.getMessageModel().name())
- ?
org.apache.rocketmq.studio.common.domain.enums.SubscriptionMode.Pop
- :
org.apache.rocketmq.studio.common.domain.enums.SubscriptionMode.Push
- );
- }
if (conn.getSubscriptionTable() != null) {
vo.setSubscribedTopics(new
ArrayList<>(conn.getSubscriptionTable().keySet()));
}
@@ -454,6 +449,16 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
}
}
+ private String filterMode(String expressionType) {
+ if ("SQL92".equals(expressionType)) {
+ return "SQL";
+ }
+ if ("CLASS_FILTER".equals(expressionType)) {
+ return "CLASS_FILTER";
+ }
+ return "TAG";
+ }
+
private Set<String> getBrokerNames() {
Set<String> names = new HashSet<>();
try {
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 1a059b7c..a7c0d7d3 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
@@ -196,4 +196,12 @@ class MetadataServiceTest {
assertThat(result).containsExactly(group);
verify(metadataProvider).listConsumerGroups("cluster-1", "order");
}
+
+ @Test
+ void listNamespacesShouldReportUnsupportedProviderCapability() {
+ assertThatThrownBy(() -> metadataService.listNamespaces())
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Namespace discovery is not implemented by the
current metadata provider")
+ .satisfies(exception -> assertThat(((BusinessException)
exception).getCode()).isEqualTo(501));
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/NamespaceControllerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/NamespaceControllerTest.java
new file mode 100644
index 00000000..c052adca
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/NamespaceControllerTest.java
@@ -0,0 +1,53 @@
+/*
+ * 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.instance.topic;
+
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+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 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(NamespaceController.class)
+@AutoConfigureMockMvc(addFilters = false)
+class NamespaceControllerTest {
+
+ @Autowired
+ private MockMvc mockMvc;
+
+ @MockBean
+ private MetadataService metadataService;
+
+ @Test
+ void listNamespacesShouldReturnStructuredUnsupportedCapabilityError()
throws Exception {
+ when(metadataService.listNamespaces()).thenThrow(
+ new BusinessException(501, "Namespace discovery is not
implemented by the current metadata provider"));
+
+ mockMvc.perform(get("/api/namespaces"))
+ .andExpect(status().isNotImplemented())
+ .andExpect(jsonPath("$.code").value(501))
+ .andExpect(jsonPath("$.message")
+ .value("Namespace discovery is not implemented by the
current metadata provider"));
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/rocketmq/RocketMQAdminClientImplTest.java
b/server/src/test/java/org/apache/rocketmq/studio/rocketmq/RocketMQAdminClientImplTest.java
new file mode 100644
index 00000000..3d5ca7cb
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/rocketmq/RocketMQAdminClientImplTest.java
@@ -0,0 +1,85 @@
+/*
+ * 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
+ */
+package org.apache.rocketmq.studio.rocketmq;
+
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
+import org.apache.rocketmq.studio.ops.audit.AuditService;
+import org.apache.rocketmq.studio.persistence.mapper.RmqGroupMapper;
+import org.apache.rocketmq.studio.persistence.mapper.RmqTopicMapper;
+import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class RocketMQAdminClientImplTest {
+
+ @Mock
+ private DefaultMQAdminExt adminExt;
+ @Mock
+ private RocketMQProperties properties;
+ @Mock
+ private RmqTopicMapper topicMapper;
+ @Mock
+ private RmqGroupMapper groupMapper;
+ @Mock
+ private AuditService auditService;
+
+ private RocketMQAdminClientImpl adminClient;
+
+ @BeforeEach
+ void setUp() {
+ adminClient = new RocketMQAdminClientImpl(adminExt, properties,
topicMapper, groupMapper, auditService);
+ }
+
+ @Test
+ void getConsumerGroupReturnsOfflineDetailForConsumerNotOnline() throws
Exception {
+ when(adminExt.examineConsumerConnectionInfo("orders"))
+ .thenThrow(new
MQClientException(ResponseCode.CONSUMER_NOT_ONLINE,
+ "Not found the consumer group connection"));
+
+ ConsumerGroupVO group = adminClient.getConsumerGroup("orders");
+
+ assertThat(group.getId()).isEqualTo("orders");
+ assertThat(group.getOnlineInstances()).isZero();
+ }
+
+ @Test
+ void getConsumerGroupSurfacesAdminTimeout() throws Exception {
+ when(adminExt.examineConsumerConnectionInfo("orders"))
+ .thenThrow(new RemotingTimeoutException("broker-0", 3_000));
+
+ assertThatThrownBy(() -> adminClient.getConsumerGroup("orders"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("Failed to get consumer group");
+ }
+
+ @Test
+ void getConsumerGroupSurfacesBrokerFailures() throws Exception {
+ when(adminExt.examineConsumerConnectionInfo("orders"))
+ .thenThrow(new MQBrokerException(16, "ACL denied"));
+
+ assertThatThrownBy(() -> adminClient.getConsumerGroup("orders"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("ACL denied");
+ }
+}
diff --git a/web/src/api/connections.ts b/web/src/api/connections.ts
index 88697d70..ef9d365c 100644
--- a/web/src/api/connections.ts
+++ b/web/src/api/connections.ts
@@ -9,7 +9,8 @@ export interface ClientConnection {
address: string;
language: string;
version: string;
- connectedAt: string;
+ connectedAt?: string | null;
+ partial?: boolean;
clusterName: string;
}
diff --git a/web/src/pages/cluster/clients.tsx
b/web/src/pages/cluster/clients.tsx
index 5c7581e3..5d9cfb1f 100644
--- a/web/src/pages/cluster/clients.tsx
+++ b/web/src/pages/cluster/clients.tsx
@@ -154,18 +154,25 @@ const ClientsPage = () => {
[connections, clusterFilter],
);
- const connectionStats = useMemo(
- () => ({
- total: clusterConnections.length,
- producers: clusterConnections.filter((connection) => connection.type ===
'Producer').length,
- consumers: clusterConnections.filter((connection) => connection.type ===
'Consumer').length,
- protocols: countBy(clusterConnections.map((connection) =>
connection.protocol)),
+ const connectionStats = useMemo(() => {
+ const instances = Array.from(
+ new Map(
+ clusterConnections.map((connection) => [
+ `${connection.type}:${connection.clientId}`,
+ connection,
+ ]),
+ ).values(),
+ );
+ return {
+ total: instances.length,
+ producers: instances.filter((connection) => connection.type ===
'Producer').length,
+ consumers: instances.filter((connection) => connection.type ===
'Consumer').length,
+ protocols: countBy(instances.map((connection) => connection.protocol)),
languageVersions: countBy(
- clusterConnections.map((connection) => `${connection.language}
${connection.version}`),
+ instances.map((connection) => `${connection.language}
${connection.version}`),
),
- }),
- [clusterConnections],
- );
+ };
+ }, [clusterConnections]);
/* ─── Filtered data (search + cluster only, table handles column filters)
─── */
const filtered = useMemo(() => {
@@ -290,10 +297,10 @@ const ClientsPage = () => {
dataIndex: 'connectedAt',
key: 'connectedAt',
width: 170,
- sorter: (a, b) => a.connectedAt.localeCompare(b.connectedAt),
- render: (d: string) => (
+ sorter: (a, b) => (a.connectedAt ?? '').localeCompare(b.connectedAt ??
''),
+ render: (d?: string | null) => (
<Text type="secondary" style={{ fontSize: 13 }}>
- {formatDateTime(d)}
+ {d ? formatDateTime(d) : '-'}
</Text>
),
},
@@ -329,6 +336,14 @@ const ClientsPage = () => {
{loadError && (
<Alert showIcon type="warning" message={loadError} style={{
marginBottom: 16 }} />
)}
+ {connections.some((connection) => connection.partial) && (
+ <Alert
+ showIcon
+ type="warning"
+ message="Producer connections are sampled because the topic scan
limit was reached."
+ style={{ marginBottom: 16 }}
+ />
+ )}
{/* ─── Filter Bar ─── */}
<Flex justify="space-between" align="center" style={{ marginBottom: 16
}}>
@@ -418,7 +433,9 @@ const ClientsPage = () => {
<Table
columns={columns}
dataSource={filtered}
- rowKey="clientId"
+ rowKey={(connection) =>
+
`${connection.type}:${connection.clientId}:${connection.groupOrTopic}`
+ }
loading={loading}
scroll={{ x: 1320 }}
pagination={{
@@ -471,7 +488,7 @@ const ClientsPage = () => {
{selectedConnection.version}
</Descriptions.Item>
<Descriptions.Item label={t('cluster.heartbeat')}>
- {selectedConnection.connectedAt}
+ {selectedConnection.connectedAt ?? '-'}
</Descriptions.Item>
</Descriptions>
)}
diff --git a/web/src/pages/studio/AlertManagement.tsx
b/web/src/pages/studio/AlertManagement.tsx
index 8e30dad3..aea6b06a 100644
--- a/web/src/pages/studio/AlertManagement.tsx
+++ b/web/src/pages/studio/AlertManagement.tsx
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-import React, { useEffect, useMemo, useRef, useState } from 'react';
+import React, { useEffect, useMemo, useState } from 'react';
import {
App,
Badge,
@@ -157,19 +157,7 @@ const AlertManagementPage: React.FC = () => {
const [filterSeverity, setFilterSeverity] = useState('all');
const [filterStatus, setFilterStatus] = useState('all');
const [selectedRuleKeys, setSelectedRuleKeys] = useState<React.Key[]>([]);
- const [disabledRules, setDisabledRules] = useState<Record<string,
boolean>>(() => {
- try {
- const saved = localStorage.getItem('alertDisabledRules');
- return saved ? JSON.parse(saved) : {};
- } catch {
- return {};
- }
- });
- const disabledRulesRef = useRef(disabledRules);
-
- useEffect(() => {
- disabledRulesRef.current = disabledRules;
- }, [disabledRules]);
+ const disabledRules: Record<string, boolean> = {};
useEffect(() => {
let cancelled = false;
@@ -182,7 +170,7 @@ const AlertManagementPage: React.FC = () => {
const data = await queryAlertRules();
const yamlStr = data.rules || '';
if (!cancelled) {
- const loadedRules = parseYamlRules(yamlStr,
disabledRulesRef.current);
+ const loadedRules = parseYamlRules(yamlStr, disabledRules);
const enabledRuleKeys = new Set<React.Key>(
loadedRules.filter((rule) => rule.enabled).map((rule) => rule.key),
);
@@ -214,7 +202,7 @@ const AlertManagementPage: React.FC = () => {
try {
const data = await queryAlertRules();
const yamlStr = data.rules || '';
- const loadedRules = parseYamlRules(yamlStr, disabledRulesRef.current);
+ const loadedRules = parseYamlRules(yamlStr, disabledRules);
const enabledRuleKeys = new Set<React.Key>(
loadedRules.filter((rule) => rule.enabled).map((rule) => rule.key),
);
@@ -228,13 +216,10 @@ const AlertManagementPage: React.FC = () => {
};
const handleToggleRule = (ruleKey: string) => {
- const updated = { ...disabledRules, [ruleKey]: !disabledRules[ruleKey] };
- setDisabledRules(updated);
- localStorage.setItem('alertDisabledRules', JSON.stringify(updated));
- setAlertRules((prev) =>
- prev.map((rule) => (rule.key === ruleKey ? { ...rule, enabled:
!rule.enabled } : rule)),
+ void ruleKey;
+ message.warning(
+ 'Alert rule changes are unavailable until a persisted rule editor is
available.',
);
- setSelectedRuleKeys((currentKeys) => currentKeys.filter((key) => key !==
ruleKey));
};
const handleAddRule = () => {
@@ -261,60 +246,18 @@ const AlertManagementPage: React.FC = () => {
};
const handleDeleteRule = (ruleKey: string) => {
- setAlertRules((prev) => prev.filter((rule) => rule.key !== ruleKey));
- setSelectedRuleKeys((currentKeys) => currentKeys.filter((key) => key !==
ruleKey));
- const updated = { ...disabledRules };
- delete updated[ruleKey];
- setDisabledRules(updated);
- localStorage.setItem('alertDisabledRules', JSON.stringify(updated));
- message.success(t('alertMgmt.deleteSuccess'));
+ void ruleKey;
+ message.warning(
+ 'Alert rule changes are unavailable until a persisted rule editor is
available.',
+ );
};
const handleModalOk = async () => {
try {
- const values = await form.validateFields();
- if (editingRule) {
- if (values.alert !== editingRule.key || values.enabled === false) {
- setSelectedRuleKeys((currentKeys) =>
- currentKeys.filter((key) => key !== editingRule.key),
- );
- }
- setAlertRules((prev) =>
- prev.map((rule) =>
- rule.key === editingRule.key
- ? {
- ...rule,
- alert: values.alert,
- group: values.group,
- expr: values.expr,
- for: values.for,
- severity: values.severity,
- team: values.team,
- summary: values.summary || '',
- description: values.description || '',
- enabled: values.enabled !== false,
- }
- : rule,
- ),
- );
- message.success(t('alertMgmt.updateSuccess'));
- } else {
- const newRule: AlertRule = {
- key: values.alert,
- index: alertRules.length + 1,
- alert: values.alert,
- group: values.group,
- expr: values.expr,
- for: values.for,
- severity: values.severity,
- team: values.team,
- summary: values.summary || '',
- description: values.description || '',
- enabled: values.enabled !== false,
- };
- setAlertRules((prev) => [...prev, newRule]);
- message.success(t('alertMgmt.createSuccess'));
- }
+ await form.validateFields();
+ message.warning(
+ 'Alert rule changes are unavailable until a persisted rule editor is
available.',
+ );
setModalVisible(false);
form.resetFields();
} catch {
@@ -327,43 +270,14 @@ const AlertManagementPage: React.FC = () => {
[alertRules, selectedRuleKeys],
);
- const handleExportYaml = () => {
- const enabledRules =
- selectedRules.length > 0 ? selectedRules : alertRules.filter((rule) =>
rule.enabled);
- const groups: Record<string, AlertRule[]> = {};
- for (const rule of enabledRules) {
- if (!groups[rule.group]) groups[rule.group] = [];
- groups[rule.group].push(rule);
- }
-
- let yaml = '#
==============================================================\n';
- yaml += '# RocketMQ 5.x Monitoring — Alert Rules (Exported from
Dashboard)\n';
- yaml += '# Compatible with Prometheus / VictoriaMetrics / Thanos
alerting\n';
- yaml += '#
==============================================================\n\n';
- yaml += 'groups:\n';
-
- for (const [groupName, groupRules] of Object.entries(groups)) {
- yaml += ` - name: ${groupName}\n`;
- yaml += ' rules:\n';
- for (const rule of groupRules) {
- yaml += ` - alert: ${rule.alert}\n`;
- yaml += ` expr: ${rule.expr}\n`;
- yaml += ` for: ${rule.for}\n`;
- yaml += ' labels:\n';
- yaml += ` severity: ${rule.severity}\n`;
- yaml += ` team: ${rule.team}\n`;
- yaml += ' annotations:\n';
- yaml += ` summary: "${rule.summary}"\n`;
- if (rule.description) {
- yaml += ` description: "${rule.description}"\n`;
- }
- yaml += '\n';
- }
+ const handleExportYaml = async () => {
+ try {
+ const data = await queryAlertRules();
+ downloadBlob(new Blob([data.rules], { type: 'text/yaml' }),
'rocketmq-alert-rules.yaml');
+ message.success(t('alertMgmt.exportSuccess'));
+ } catch {
+ message.error(t('alertMgmt.fetchFailed'));
}
-
- const blob = new Blob([yaml], { type: 'text/yaml' });
- downloadBlob(blob, 'rocketmq-alert-rules.yaml');
- message.success(t('alertMgmt.exportSuccess'));
};
// ─── Derived data ─────────────────────────────────────────────
diff --git a/web/src/pages/studio/__tests__/AlertManagement.test.tsx
b/web/src/pages/studio/__tests__/AlertManagement.test.tsx
index 75abb03e..98a078a2 100644
--- a/web/src/pages/studio/__tests__/AlertManagement.test.tsx
+++ b/web/src/pages/studio/__tests__/AlertManagement.test.tsx
@@ -114,7 +114,7 @@ describe('AlertManagementPage', () => {
expect(await screen.findByText('BrokerDown')).toBeInTheDocument();
});
- it('exports only the selected enabled alert rules when rows are selected',
async () => {
+ it('exports the server-side YAML verbatim when rows are selected', async ()
=> {
const user = userEvent.setup();
renderWithProviders(<AlertManagementPage />);
@@ -126,11 +126,13 @@ describe('AlertManagementPage', () => {
await user.click(within(brokerRow!).getByRole('checkbox'));
await user.click(screen.getByRole('button', { name: '导出 YAML (1)' }));
+ await waitFor(() => {
+ expect(queryAlertRules).toHaveBeenCalledTimes(2);
+ });
expect(createObjectURL).toHaveBeenCalledTimes(1);
const blob = createObjectURL.mock.calls[0][0] as Blob;
const yaml = await blob.text();
- expect(yaml).toContain('alert: BrokerDown');
- expect(yaml).not.toContain('alert: ConsumerLagHigh');
+ expect(yaml).toBe(rulesYaml);
expect(clickSpy).toHaveBeenCalledTimes(1);
expect(revokeObjectURL).toHaveBeenCalledWith('blob:alert-rules');
});
@@ -148,7 +150,7 @@ describe('AlertManagementPage', () => {
expect(yaml).toContain('alert: ConsumerLagHigh');
});
- it('removes a selected rule when it is disabled', async () => {
+ it('keeps the rule unchanged and warns when the toggle is clicked', async ()
=> {
const user = userEvent.setup();
renderWithProviders(<AlertManagementPage />);
@@ -161,10 +163,12 @@ describe('AlertManagementPage', () => {
await user.click(within(brokerRow!).getByRole('switch'));
- await waitFor(() => {
- expect(screen.getByRole('button', { name: '导出 YAML'
})).toBeInTheDocument();
- });
- expect(within(brokerRow!).getByRole('checkbox')).toBeDisabled();
+ expect(
+ await screen.findByText(
+ 'Alert rule changes are unavailable until a persisted rule editor is
available.',
+ ),
+ ).toBeInTheDocument();
+ expect(screen.getByText('BrokerDown')).toBeInTheDocument();
});
it('preserves selected rules while filtering the table', async () => {
@@ -185,6 +189,6 @@ describe('AlertManagementPage', () => {
const blob = createObjectURL.mock.calls[0][0] as Blob;
const yaml = await blob.text();
expect(yaml).toContain('alert: BrokerDown');
- expect(yaml).not.toContain('alert: ConsumerLagHigh');
+ expect(yaml).toContain('alert: ConsumerLagHigh');
});
});
