You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by li...@apache.org on 2026/08/10 06:45:24 UTC
(rocketmq-dashboard) branch rocketmq-studio updated (269e2289 -> d58b4635)
This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a change to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
discard 269e2289 fix(ui): normalize ACL version to a string for filtering and coloring (#1259)
discard 15c1a086 fix(ui): drop misleading selected count on alert export button (#1258)
discard 9933f0b4 fix: scope consumer group diagnostics by selected instance (#1248)
discard 59f6182c fix(deploy): bootstrap local docker compose on Docker Engine (#1197)
discard c20f6ab6 fix: metrics error bodies, DLQ search and small UI fixes (#1252, #1254, #1255, #1256, #1257)
discard 958e3962 fix: instance-scoped hardening, mutation auditing and validation (#1274, #1280, #1282, #1288, #1293, #1296, #1303, #1305, #1307, #1309, #1311, #1319)
new a6eae719 fix: instance-scoped hardening, mutation auditing and validation (#1274, #1280, #1282, #1288, #1293, #1296, #1303, #1305, #1307, #1309, #1311, #1319)
new 93b90b2d fix: metrics error bodies, DLQ search and small UI fixes (#1252, #1254, #1255, #1256, #1257)
new 9dc8eeb1 fix(deploy): bootstrap local docker compose on Docker Engine (#1197)
new 06b1d933 fix: scope consumer group diagnostics by selected instance (#1248)
new 15b98d65 fix(ui): drop misleading selected count on alert export button (#1258)
new d58b4635 fix(ui): normalize ACL version to a string for filtering and coloring (#1259)
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (269e2289)
\
N -- N -- N refs/heads/rocketmq-studio (d58b4635)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 6 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
(rocketmq-dashboard) 02/06: fix: metrics error bodies, DLQ search and small UI fixes (#1252, #1254, #1255, #1256, #1257)
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
commit 93b90b2d8f43018647fd3d756e6a586d980bf6de
Author: yyqdbngt <30...@users.noreply.github.com>
AuthorDate: Mon Aug 10 13:48:29 2026 +0800
fix: metrics error bodies, DLQ search and small UI fixes (#1252, #1254, #1255, #1256, #1257)
* fix(metrics): report non-JSON error bodies with their real status
* fix(ui): make DLQ group search case-insensitive
* fix(ui): do not report form validation failure as a save failure
* fix(ui): clear topic dropdown when the instance has no topics
* fix(ui): reset audit page when changing page size
* test: supply real topic options in MessagePage history tests
#1256 intentionally empties the topic dropdown when the backend returns no topics, replacing the previous fall-through to static defaults; the query-history tests now mock listTopics with the expected topic names instead of relying on the defaults.
---------
---
.../AbstractPrometheusCompatibleMetricsSource.java | 19 ++++++++++++++++---
web/src/pages/instance/__tests__/MessagePage.test.tsx | 10 +++++++++-
web/src/pages/instance/dlq.tsx | 3 ++-
web/src/pages/instance/index.tsx | 10 ++++++++--
web/src/pages/instance/message.tsx | 6 +++---
web/src/pages/ops/alerts.tsx | 5 ++++-
web/src/pages/ops/audit.tsx | 9 +++++++--
web/src/pages/settings/index.tsx | 5 ++++-
8 files changed, 53 insertions(+), 14 deletions(-)
diff --git a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/AbstractPrometheusCompatibleMetricsSource.java b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/AbstractPrometheusCompatibleMetricsSource.java
index a0ce125d..6909c302 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/AbstractPrometheusCompatibleMetricsSource.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/AbstractPrometheusCompatibleMetricsSource.java
@@ -100,11 +100,24 @@ public abstract class AbstractPrometheusCompatibleMetricsSource implements Metri
.headers(this::applyAuthentication)
.body(form)
.exchange((request, clientResponse) -> {
- JsonNode body = objectMapper.readTree(readResponseBody(clientResponse.getBody()));
+ byte[] rawBody = readResponseBody(clientResponse.getBody());
+ int upstreamStatus = responseStatus(clientResponse.getStatusCode());
if (clientResponse.getStatusCode().isError()) {
- throw responseBodyException(body, responseStatus(clientResponse.getStatusCode()));
+ // Check the status before parsing so a non-JSON error body (e.g. a proxy
+ // HTML/plain-text page) is not misreported as a connection failure.
+ try {
+ throw responseBodyException(objectMapper.readTree(rawBody), upstreamStatus);
+ } catch (IOException nonJsonBody) {
+ throw new PrometheusException(upstreamStatus,
+ backendLabel() + " query failed (HTTP " + upstreamStatus + ")");
+ }
+ }
+ try {
+ return objectMapper.readTree(rawBody);
+ } catch (IOException malformedBody) {
+ throw new PrometheusException(HttpStatus.BAD_GATEWAY.value(),
+ backendLabel() + " returned a non-JSON response");
}
- return body;
});
return parseResponse(response);
} catch (PrometheusException exception) {
diff --git a/web/src/pages/instance/__tests__/MessagePage.test.tsx b/web/src/pages/instance/__tests__/MessagePage.test.tsx
index 36dd97a3..08882926 100644
--- a/web/src/pages/instance/__tests__/MessagePage.test.tsx
+++ b/web/src/pages/instance/__tests__/MessagePage.test.tsx
@@ -36,7 +36,15 @@ vi.mock('../../../services/instanceService', () => ({
listInstances: vi.fn().mockResolvedValue([]),
}));
vi.mock('../../../services/topicService', () => ({
- listTopics: vi.fn().mockResolvedValue([]),
+ listTopics: vi
+ .fn()
+ .mockResolvedValue([
+ { name: 'order-create' },
+ { name: 'payment-callback' },
+ { name: 'user-activity-log' },
+ { name: 'notification-push' },
+ { name: 'inventory-sync' },
+ ]),
}));
import MessagePage from '../message';
diff --git a/web/src/pages/instance/dlq.tsx b/web/src/pages/instance/dlq.tsx
index dfd574fc..3010f58c 100644
--- a/web/src/pages/instance/dlq.tsx
+++ b/web/src/pages/instance/dlq.tsx
@@ -179,7 +179,8 @@ const DLQPage = () => {
if (!search) return groups;
return groups.filter(
(g) =>
- g.groupName.includes(search) || g.dlqTopic.toLowerCase().includes(search.toLowerCase()),
+ g.groupName.toLowerCase().includes(search.toLowerCase()) ||
+ g.dlqTopic.toLowerCase().includes(search.toLowerCase()),
);
}, [groups, search]);
diff --git a/web/src/pages/instance/index.tsx b/web/src/pages/instance/index.tsx
index 0b7eefe3..90a84a6b 100644
--- a/web/src/pages/instance/index.tsx
+++ b/web/src/pages/instance/index.tsx
@@ -215,7 +215,10 @@ const InstancePage = () => {
setAddModalOpen(false);
addForm.resetFields();
setVendor(DEFAULT_VENDOR);
- } catch {
+ } catch (error) {
+ if (error && typeof error === 'object' && 'errorFields' in error) {
+ return; // validation failure; antd already shows field-level errors
+ }
message.error('添加实例失败，请稍后重试');
} finally {
setSubmitting(false);
@@ -232,7 +235,10 @@ const InstancePage = () => {
message.success(`实例「${updated.name}」备注已更新`);
setEditModalOpen(false);
editForm.resetFields();
- } catch {
+ } catch (error) {
+ if (error && typeof error === 'object' && 'errorFields' in error) {
+ return; // validation failure; antd already shows field-level errors
+ }
message.error('更新实例失败，请稍后重试');
} finally {
setSubmitting(false);
diff --git a/web/src/pages/instance/message.tsx b/web/src/pages/instance/message.tsx
index c4a7d43a..8608f967 100644
--- a/web/src/pages/instance/message.tsx
+++ b/web/src/pages/instance/message.tsx
@@ -222,9 +222,9 @@ const MessagePage = () => {
const scoped = selectedInstanceId
? nextTopics.filter((topic) => topic.instanceId === selectedInstanceId)
: nextTopics;
- if (scoped.length > 0) {
- setTopicOptions(scoped.map((topic) => topic.name));
- }
+ // Always update so an instance with no topics empties the dropdown instead of showing
+ // topics from another instance or the static defaults.
+ setTopicOptions(scoped.map((topic) => topic.name));
})
.catch(() => {
// 加载失败保持静态选项可用
diff --git a/web/src/pages/ops/alerts.tsx b/web/src/pages/ops/alerts.tsx
index 7aee76bf..62b8a890 100644
--- a/web/src/pages/ops/alerts.tsx
+++ b/web/src/pages/ops/alerts.tsx
@@ -308,7 +308,10 @@ const AlertsPage = () => {
}
setModalVisible(false);
form.resetFields();
- } catch {
+ } catch (error) {
+ if (error && typeof error === 'object' && 'errorFields' in error) {
+ return; // validation failure; antd already shows field-level errors
+ }
message.error('保存告警规则失败，请稍后重试');
} finally {
setSubmitting(false);
diff --git a/web/src/pages/ops/audit.tsx b/web/src/pages/ops/audit.tsx
index 808b88f6..68a82473 100644
--- a/web/src/pages/ops/audit.tsx
+++ b/web/src/pages/ops/audit.tsx
@@ -361,8 +361,13 @@ const AuditPage: React.FC = () => {
total,
showSizeChanger: true,
onChange: (nextPage, nextPageSize) => {
- setPage(nextPage);
- setPageSize(nextPageSize);
+ if (nextPageSize !== pageSize) {
+ // A larger page size can make the current page exceed the new total page count.
+ setPage(1);
+ setPageSize(nextPageSize);
+ } else {
+ setPage(nextPage);
+ }
},
}}
/>
diff --git a/web/src/pages/settings/index.tsx b/web/src/pages/settings/index.tsx
index be988f3a..64bb0bb0 100644
--- a/web/src/pages/settings/index.tsx
+++ b/web/src/pages/settings/index.tsx
@@ -371,7 +371,10 @@ export const DataSourceTab = () => {
message.success(editingDataSource ? '数据源已更新' : '数据源已添加');
setModalOpen(false);
dsForm.resetFields();
- } catch {
+ } catch (error) {
+ if (error && typeof error === 'object' && 'errorFields' in error) {
+ return; // validation failure; antd already shows field-level errors
+ }
message.error('保存数据源失败，请稍后重试');
} finally {
setSubmitting(false);
(rocketmq-dashboard) 03/06: fix(deploy): bootstrap local docker compose on Docker Engine (#1197)
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
commit 9dc8eeb1be2a9b43ae17009989061d31bd6f9290
Author: Lei Zhiyuan <le...@gmail.com>
AuthorDate: Mon Aug 10 14:04:23 2026 +0800
fix(deploy): bootstrap local docker compose on Docker Engine (#1197)
* fix(rocketmq): declare subnet for compose network static IPs
broker-0/broker-1/nameserver/proxy pin static IPs (10.89.2.x) but the
default network had no ipam subnet, so 'docker compose up' failed with
'user specified IP address is supported only when connecting to networks
with user configured subnets' on Docker Engine. Declare a 10.89.2.0/24
bridge subnet so the pinned addresses match.
* fix(web): use Docker embedded DNS for nginx upstream resolver
15-resolver.envsh derived the nginx resolver from the default gateway,
which is only valid under podman. On Docker Engine the gateway IP runs no
DNS server, so resolving the rocketmq-server upstream failed and every
/api/ request returned 502 ('could not be resolved', Operation timed out),
breaking login status checks. Prefer Docker's embedded DNS at 127.0.0.11
and fall back to the gateway only when it is unreachable.
---------
Co-authored-by: zhiyuanlei <zh...@tencent.com>
---
deploy/rocketmq/docker-compose.yml | 4 ++++
web/15-resolver.envsh | 19 +++++++++++++------
2 files changed, 17 insertions(+), 6 deletions(-)
diff --git a/deploy/rocketmq/docker-compose.yml b/deploy/rocketmq/docker-compose.yml
index 1393c6f8..8b83e683 100644
--- a/deploy/rocketmq/docker-compose.yml
+++ b/deploy/rocketmq/docker-compose.yml
@@ -120,6 +120,10 @@ services:
networks:
default:
name: rocketmq_default
+ driver: bridge
+ ipam:
+ config:
+ - subnet: 10.89.2.0/24
volumes:
broker-0-store:
diff --git a/web/15-resolver.envsh b/web/15-resolver.envsh
index 8f802676..ed04a8dc 100755
--- a/web/15-resolver.envsh
+++ b/web/15-resolver.envsh
@@ -1,11 +1,18 @@
#!/bin/sh
# Sourced by docker-entrypoint.sh (*.envsh), so RESOLVER is visible to the
-# later 20-envsubst-on-templates.sh step. Uses the podman network gateway
-# (the container DNS server) only — the host resolv.conf may leak public DNS
-# servers that cannot resolve container names and cause random 502s.
-RESOLVER="$(ip route | awk '/default/ {print $3; exit}')"
-if [ -z "$RESOLVER" ]; then
- RESOLVER="127.0.0.11"
+# later 20-envsubst-on-templates.sh step.
+#
+# Prefer Docker's embedded DNS (127.0.0.11): it resolves container names and
+# is the correct choice on Docker Engine. Fall back to the default gateway
+# (podman's container DNS server) only when 127.0.0.11 is unreachable.
+# The host resolv.conf may leak public DNS servers that cannot resolve
+# container names and cause random 502s, so never use it directly.
+RESOLVER="127.0.0.11"
+if ! nc -z -w2 127.0.0.11 53 >/dev/null 2>&1; then
+ GW="$(ip route | awk '/default/ {print $3; exit}')"
+ if [ -n "$GW" ]; then
+ RESOLVER="$GW"
+ fi
fi
export RESOLVER
echo "15-resolver.envsh: resolver=$RESOLVER"
(rocketmq-dashboard) 01/06: fix: instance-scoped hardening, mutation auditing and validation (#1274, #1280, #1282, #1288, #1293, #1296, #1303, #1305, #1307, #1309, #1311, #1319)
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
commit a6eae7198ed7a82b08186bf884d255e4edc796ec
Author: Aias00 <li...@apache.org>
AuthorDate: Mon Aug 10 12:57:56 2026 +0800
fix: instance-scoped hardening, mutation auditing and validation (#1274, #1280, #1282, #1288, #1293, #1296, #1303, #1305, #1307, #1309, #1311, #1319)
* fix: reject non-SEND_OK topic test messages
Signed-off-by: liuhy <li...@apache.org>
* fix: surface message trace query failures
Signed-off-by: liuhy <li...@apache.org>
* fix: require a topic for message ID queries
* fix: reject blank message trace identifiers
* fix: clear stale DLQ state when switching instances
Signed-off-by: liuhy <li...@apache.org>
* fix: handle client instance bootstrap failures
Signed-off-by: liuhy <li...@apache.org>
* fix: scope producer topics by selected instance
Signed-off-by: liuhy <li...@apache.org>
* fix: require instance IDs for client queries
* fix: ignore stale dashboard responses
Signed-off-by: liuhy <li...@apache.org>
* test: cover dashboard instance response ordering
Signed-off-by: liuhy <li...@apache.org>
* fix: initialize the H2 schema for development
* fix: reject cloud instances from remoting admin resolution
* feat: audit cloud credential mutations
* feat: audit managed instance mutations
* fix: normalize managed instance names
* feat: audit ACL mutations
* fix: clarify that ACL records are Studio-local
* feat: audit K8s certificate mutations
* fix: clarify that K8s certificate records are local
* feat: audit alert mutations
* fix: exclude disabled alert rules from export
* fix: validate runtime Topic and Group names
* fix: adapt batch to studio conventions and stabilize producer instance switch
- isolate audit write failures (try/catch + warn) in Instance/K8sCert/Acl/Alert/CloudCredential services, matching the control-plane audit isolation precedent
- replace private isBlank/JDK blank checks with Spring StringUtils in MetadataService and ClientService
- Producer page: clear stale topic/group state in the instance change handler instead of synchronously in an effect (also fixes antd Select breakage across remounts in tests)
- Producer test: scope stale-state assertions to the render container (antd keeps closed dropdown portals in document.body)
---------
Signed-off-by: liuhy <li...@apache.org>
Co-authored-by: liuhy <li...@apache.org>
---
.../cluster/broker/RuntimeAdminClientResolver.java | 12 ++-
.../studio/cluster/client/ClientService.java | 11 +-
.../studio/cluster/k8s/K8sCertService.java | 31 +++++-
.../rocketmq/studio/instance/InstanceService.java | 43 ++++++--
.../rocketmq/studio/instance/acl/AclService.java | 41 ++++++-
.../studio/instance/message/MessageService.java | 6 ++
.../studio/instance/topic/MetadataService.java | 33 +++---
.../rocketmq/studio/ops/alert/AlertService.java | 43 +++++++-
.../provider/apache/RocketMQAdminClientImpl.java | 8 ++
.../provider/apache/RocketMQMessageProvider.java | 3 +
.../credential/CloudCredentialService.java | 27 ++++-
server/src/main/resources/application-dev.yml | 4 +
server/src/main/resources/db/schema.sql | 14 ++-
.../rocketmq/studio/StudioApplicationTest.java | 20 +++-
.../broker/RuntimeAdminClientResolverTest.java | 21 ++++
.../studio/cluster/client/ClientServiceTest.java | 20 +++-
.../studio/cluster/k8s/K8sCertServiceTest.java | 18 +++-
.../studio/instance/InstanceServiceTest.java | 35 ++++++
.../studio/instance/acl/AclServiceTest.java | 20 ++++
.../instance/message/MessageServiceTest.java | 27 +++--
.../studio/instance/topic/MetadataServiceTest.java | 21 ++++
.../ops/alert/AlertServiceDefaultRulesTest.java | 8 +-
.../studio/ops/alert/AlertServiceTest.java | 70 +++++++++++-
.../apache/RocketMQAdminClientImplTest.java | 49 +++++++++
.../apache/RocketMQMessageProviderTest.java | 13 +++
.../credential/CloudCredentialServiceTest.java | 14 +++
web/src/api/producer.test.ts | 11 +-
web/src/api/producer.ts | 6 +-
web/src/i18n/translations.ts | 8 ++
.../pages/cluster/__tests__/ClientsPage.test.tsx | 28 +++++
.../pages/cluster/__tests__/K8sCertsPage.test.tsx | 8 ++
web/src/pages/cluster/certs.tsx | 9 ++
web/src/pages/cluster/clients.tsx | 40 +++++--
.../pages/home/__tests__/DashboardPage.test.tsx | 118 +++++++++++++++++++++
web/src/pages/home/dashboard.tsx | 17 ++-
web/src/pages/instance/__tests__/AclPage.test.tsx | 8 ++
web/src/pages/instance/__tests__/DLQPage.test.tsx | 46 ++++++++
.../pages/instance/__tests__/MessagePage.test.tsx | 22 +++-
web/src/pages/instance/acl.tsx | 10 ++
web/src/pages/instance/dlq.tsx | 17 ++-
web/src/pages/instance/message.tsx | 51 ++++++---
web/src/pages/studio/Producer.tsx | 22 +++-
web/src/pages/studio/__tests__/Producer.test.tsx | 64 ++++++++++-
43 files changed, 994 insertions(+), 103 deletions(-)
diff --git a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/RuntimeAdminClientResolver.java b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/RuntimeAdminClientResolver.java
index 5acead9a..2bd87db7 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/RuntimeAdminClientResolver.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/RuntimeAdminClientResolver.java
@@ -7,6 +7,7 @@
package org.apache.rocketmq.studio.cluster.broker;
import lombok.RequiredArgsConstructor;
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.instance.InstanceRepository;
import org.apache.rocketmq.studio.instance.InstanceVO;
@@ -29,7 +30,7 @@ public class RuntimeAdminClientResolver {
}
public String resolveEndpoint(String instanceId) {
- InstanceVO instance = resolveInstance(instanceId);
+ InstanceVO instance = requireApacheInstance(resolveInstance(instanceId));
if (!StringUtils.hasText(instance.getEndpoint())) {
throw new BusinessException(400, "Instance has no endpoint: " + instanceId);
}
@@ -41,9 +42,18 @@ public class RuntimeAdminClientResolver {
}
public <T> T execute(InstanceVO instance, MqAdminExtFactory.AdminAction<T> action) {
+ requireApacheInstance(instance);
if (instance == null || !StringUtils.hasText(instance.getEndpoint())) {
throw new BusinessException(400, "Instance endpoint is required");
}
return adminFactory.execute(instance.getEndpoint().trim(), null, action);
}
+
+ private InstanceVO requireApacheInstance(InstanceVO instance) {
+ if (instance != null && instance.getVendor() != null && instance.getVendor() != InstanceVendor.APACHE) {
+ throw new BusinessException(400,
+ "Runtime AdminClient only supports Apache instances: " + instance.getId());
+ }
+ return instance;
+ }
}
diff --git a/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ClientService.java b/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ClientService.java
index ed9723e4..9d04ebcf 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ClientService.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ClientService.java
@@ -18,7 +18,9 @@ package org.apache.rocketmq.studio.cluster.client;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.springframework.stereotype.Service;
+import org.springframework.util.StringUtils;
import java.util.List;
@@ -31,7 +33,14 @@ public class ClientService {
public List<ClientConnectionVO> listConnections(String instanceId, String clusterId, String type) {
log.info("Listing client connections, instanceId={}, clusterId={}, type={}", instanceId, clusterId, type);
- return clientProvider.findConnections(normalizeFilter(instanceId), normalizeFilter(clusterId), normalizeFilter(type));
+ return clientProvider.findConnections(requireInstanceId(instanceId), normalizeFilter(clusterId), normalizeFilter(type));
+ }
+
+ private String requireInstanceId(String instanceId) {
+ if (!StringUtils.hasText(instanceId)) {
+ throw new BusinessException(400, "instanceId is required");
+ }
+ return instanceId.trim();
}
private String normalizeFilter(String value) {
diff --git a/server/src/main/java/org/apache/rocketmq/studio/cluster/k8s/K8sCertService.java b/server/src/main/java/org/apache/rocketmq/studio/cluster/k8s/K8sCertService.java
index 6ca6debf..b13f9804 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/cluster/k8s/K8sCertService.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/cluster/k8s/K8sCertService.java
@@ -19,6 +19,7 @@ package org.apache.rocketmq.studio.cluster.k8s;
import org.apache.rocketmq.studio.common.domain.enums.CertStatus;
import org.apache.rocketmq.studio.common.domain.enums.CertType;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.audit.OperationAuditService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@@ -38,15 +39,17 @@ public class K8sCertService {
private static final int EXPIRING_THRESHOLD_DAYS = 30;
private final K8sCertRepository k8sCertRepository;
+ private final OperationAuditService operationAuditService;
private final Clock clock;
@Autowired
- public K8sCertService(K8sCertRepository k8sCertRepository) {
- this(k8sCertRepository, Clock.systemDefaultZone());
+ public K8sCertService(K8sCertRepository k8sCertRepository, OperationAuditService operationAuditService) {
+ this(k8sCertRepository, operationAuditService, Clock.systemDefaultZone());
}
- K8sCertService(K8sCertRepository k8sCertRepository, Clock clock) {
+ K8sCertService(K8sCertRepository k8sCertRepository, OperationAuditService operationAuditService, Clock clock) {
this.k8sCertRepository = k8sCertRepository;
+ this.operationAuditService = operationAuditService;
this.clock = clock;
}
@@ -82,6 +85,7 @@ public class K8sCertService {
cert.setUpdatedAt(now);
K8sCertVO saved = k8sCertRepository.save(cert);
+ auditCertificate("CREATE_K8S_CERTIFICATE", saved);
log.info("K8s certificate created: {} (id={})", saved.getName(), saved.getId());
return saved;
}
@@ -115,6 +119,7 @@ public class K8sCertService {
updated.setUpdatedAt(now);
K8sCertVO saved = k8sCertRepository.save(refreshExpirationState(updated, now));
+ auditCertificate("UPDATE_K8S_CERTIFICATE", saved);
log.info("K8s certificate updated: {} (id={})", saved.getName(), saved.getId());
return saved;
}
@@ -136,6 +141,7 @@ public class K8sCertService {
renewed.setUpdatedAt(now);
K8sCertVO saved = k8sCertRepository.save(renewed);
+ auditCertificate("RENEW_K8S_CERTIFICATE", saved);
log.info("K8s certificate renewed: {} (id={}), new expiry: {}", saved.getName(), saved.getId(), notAfter);
return saved;
}
@@ -146,6 +152,8 @@ public class K8sCertService {
k8sCertRepository.findById(command.getId())
.orElseThrow(() -> new BusinessException(404, "Certificate not found: " + command.getId()));
k8sCertRepository.deleteById(command.getId());
+ recordAudit("DELETE_K8S_CERTIFICATE", "K8S_CERTIFICATE", command.getId(), null,
+ null);
log.info("K8s certificate deleted: {}", command.getId());
}
@@ -192,4 +200,21 @@ public class K8sCertService {
copy.setUpdatedAt(cert.getUpdatedAt());
return copy;
}
+
+ private void auditCertificate(String operation, K8sCertVO certificate) {
+ recordAudit(operation, "K8S_CERTIFICATE", certificate.getId(), null,
+ "name=" + certificate.getName() + ", namespace=" + certificate.getNamespace()
+ + ", cluster=" + certificate.getCluster());
+ }
+
+ private void recordAudit(String operation, String resourceType, String resourceName,
+ String clusterId, String detail) {
+ try {
+ operationAuditService.record(operation, resourceType, resourceName, clusterId, detail, "SUCCESS", null);
+ } catch (Exception auditFailure) {
+ log.warn("Failed to record audit operation={} resource={}: {}", operation, resourceName,
+ auditFailure.getMessage());
+ }
+ }
+
}
diff --git a/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceService.java b/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceService.java
index 753cc464..4971f38f 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceService.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceService.java
@@ -22,6 +22,7 @@ import org.springframework.util.StringUtils;
import org.apache.rocketmq.studio.provider.credential.CloudCredentialRepository;
import org.apache.rocketmq.studio.provider.credential.CloudCredentialVO;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
+import org.apache.rocketmq.studio.audit.OperationAuditService;
import org.apache.rocketmq.studio.common.domain.enums.InstanceType;
import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
import org.apache.rocketmq.studio.common.exception.BusinessException;
@@ -45,6 +46,7 @@ public class InstanceService {
private final CloudCredentialRepository cloudCredentialRepository;
private final InstanceProviderRegistry providerRegistry;
private final MqAdminExtFactory adminFactory;
+ private final OperationAuditService operationAuditService;
public List<InstanceVO> listInstances(InstanceType type, String search) {
log.debug("Listing instances, type={}, search={}", type, search);
@@ -96,14 +98,15 @@ public class InstanceService {
instance.setId(UUID.randomUUID().toString());
instance.setCreatedAt(LocalDateTime.now());
instance.setUpdatedAt(LocalDateTime.now());
- return instanceRepository.save(instance);
+ InstanceVO saved = instanceRepository.save(instance);
+ recordAudit("CREATE_INSTANCE", "INSTANCE", saved.getId(), null,
+ instanceAuditDetail(saved));
+ return saved;
}
private void createApacheInstance(InstanceVO instance) {
instance.setVendor(InstanceVendor.APACHE);
- if (instance.getName() == null || instance.getName().isBlank()) {
- throw new BusinessException(400, "InstanceVO name is required");
- }
+ instance.setName(requireInstanceName(instance.getName()));
instance.setEndpoint(requireValidEndpoint(instance.getEndpoint()));
if (instance.getType() == null) {
throw new BusinessException(400, "InstanceVO type is required");
@@ -131,10 +134,11 @@ public class InstanceService {
}
CloudInstanceDetailVO detail = providerRegistry.catalogFor(InstanceVendor.ALIYUN)
.getCloudInstance(instance.getCredentialId(), instance.getRegionId(), instance.getCloudInstanceId());
- if (instance.getName() == null || instance.getName().isBlank()) {
+ if (!StringUtils.hasText(instance.getName())) {
instance.setName(detail.getInstanceName() != null && !detail.getInstanceName().isBlank()
? detail.getInstanceName() : detail.getInstanceId());
}
+ instance.setName(requireInstanceName(instance.getName()));
instance.setType(InstanceType.PROXY);
instance.setEndpoint(resolveEndpoint(detail));
}
@@ -176,6 +180,13 @@ public class InstanceService {
return normalized;
}
+ private String requireInstanceName(String name) {
+ if (!StringUtils.hasText(name)) {
+ throw new BusinessException(400, "InstanceVO name is required");
+ }
+ return name.trim();
+ }
+
public InstanceVO updateInstance(InstanceVO instance) {
requireInstance(instance);
log.info("Updating instance: {}", instance.getId());
@@ -194,7 +205,7 @@ public class InstanceService {
InstanceVO updated = copyOf(existing);
boolean cloudInstance = existing.getVendor() != null && existing.getVendor() != InstanceVendor.APACHE;
if (instance.getName() != null) {
- updated.setName(instance.getName());
+ updated.setName(requireInstanceName(instance.getName()));
}
if (!cloudInstance) {
if (instance.getType() != null) {
@@ -211,6 +222,8 @@ public class InstanceService {
InstanceVO saved = instanceRepository.save(updated);
releaseApacheEndpointIfUnused(existing, saved.getEndpoint());
+ recordAudit("UPDATE_INSTANCE", "INSTANCE", saved.getId(), null,
+ instanceAuditDetail(saved));
return saved;
}
@@ -235,6 +248,8 @@ public class InstanceService {
}
instanceRepository.deleteById(id);
releaseApacheEndpointIfUnused(existing, null);
+ recordAudit("DELETE_INSTANCE", "INSTANCE", id, null,
+ instanceAuditDetail(existing));
}
private void requireInstance(InstanceVO instance) {
@@ -243,6 +258,11 @@ public class InstanceService {
}
}
+ private String instanceAuditDetail(InstanceVO instance) {
+ InstanceVendor vendor = instance.getVendor() == null ? InstanceVendor.APACHE : instance.getVendor();
+ return "name=" + instance.getName() + ", vendor=" + vendor + ", type=" + instance.getType();
+ }
+
private InstanceVO copyOf(InstanceVO instance) {
InstanceVO copy = InstanceVO.builder()
.name(instance.getName())
@@ -290,4 +310,15 @@ public class InstanceService {
String normalizedEndpoint = MqAdminExtFactory.normalizeNamesrvAddr(endpoint);
return normalizedEndpoint.isEmpty() ? null : normalizedEndpoint;
}
+
+ private void recordAudit(String operation, String resourceType, String resourceName,
+ String clusterId, String detail) {
+ try {
+ operationAuditService.record(operation, resourceType, resourceName, clusterId, detail, "SUCCESS", null);
+ } catch (Exception auditFailure) {
+ log.warn("Failed to record audit operation={} resource={}: {}", operation, resourceName,
+ auditFailure.getMessage());
+ }
+ }
+
}
diff --git a/server/src/main/java/org/apache/rocketmq/studio/instance/acl/AclService.java b/server/src/main/java/org/apache/rocketmq/studio/instance/acl/AclService.java
index c3ada9d7..1916f62b 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/instance/acl/AclService.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/instance/acl/AclService.java
@@ -20,6 +20,7 @@ import org.springframework.util.StringUtils;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.common.util.CredentialUtils;
+import org.apache.rocketmq.studio.audit.OperationAuditService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
@@ -34,6 +35,7 @@ import java.util.UUID;
public class AclService {
private final AclRepository aclRepository;
+ private final OperationAuditService operationAuditService;
public List<AclRuleVO> listRules(String clusterId, String principal) {
@@ -52,7 +54,9 @@ public class AclService {
}
rule.setId(UUID.randomUUID().toString());
rule.setCreatedAt(LocalDateTime.now());
- return aclRepository.saveRule(rule);
+ AclRuleVO saved = aclRepository.saveRule(rule);
+ auditRule("CREATE_ACL_RULE", saved);
+ return saved;
}
public AclRuleVO updateRule(AclRuleVO rule) {
@@ -60,8 +64,10 @@ public class AclService {
throw new BusinessException(400, "ACL rule id is required");
}
log.info("Updating ACL rule id={}, principal={}", rule.getId(), rule.getPrincipal());
- return aclRepository.replaceRule(rule)
+ AclRuleVO saved = aclRepository.replaceRule(rule)
.orElseThrow(() -> new BusinessException(404, "ACL rule not found: " + rule.getId()));
+ auditRule("UPDATE_ACL_RULE", saved);
+ return saved;
}
public void deleteRule(String id) {
@@ -69,6 +75,7 @@ public class AclService {
if (!aclRepository.deleteRule(id)) {
throw new BusinessException(404, "ACL rule not found: " + id);
}
+ recordAudit("DELETE_ACL_RULE", "ACL_RULE", id, null, null);
}
@@ -89,7 +96,9 @@ public class AclService {
user.setAccessKey(UUID.randomUUID().toString().replace("-", ""));
user.setSecretKey(UUID.randomUUID().toString().replace("-", ""));
user.setCreatedAt(LocalDateTime.now());
- return aclRepository.saveUser(user);
+ AclUserVO saved = aclRepository.saveUser(user);
+ auditUser("CREATE_ACL_USER", saved);
+ return saved;
}
public AclUserVO updateUser(UpdateAclUserDTO user) {
@@ -108,7 +117,9 @@ public class AclService {
.clusters(user.getClusters() == null ? existing.getClusters() : user.getClusters())
.createdAt(existing.getCreatedAt())
.build();
- return maskCredentials(aclRepository.saveUser(merged));
+ AclUserVO saved = aclRepository.saveUser(merged);
+ auditUser("UPDATE_ACL_USER", saved);
+ return maskCredentials(saved);
}
public void deleteUser(String id) {
@@ -116,6 +127,7 @@ public class AclService {
if (!aclRepository.deleteUser(id)) {
throw new BusinessException(404, "ACL user not found: " + id);
}
+ recordAudit("DELETE_ACL_USER", "ACL_USER", id, null, null);
}
/**
@@ -143,4 +155,25 @@ public class AclService {
.build();
}
+ private void auditRule(String operation, AclRuleVO rule) {
+ recordAudit(operation, "ACL_RULE", rule.getId(), null,
+ "principal=" + rule.getPrincipal());
+ }
+
+ private void auditUser(String operation, AclUserVO user) {
+ recordAudit(operation, "ACL_USER", user.getId(), null,
+ "username=" + user.getUsername() + ", admin=" + user.isAdmin());
+ }
+
+
+ private void recordAudit(String operation, String resourceType, String resourceName,
+ String clusterId, String detail) {
+ try {
+ operationAuditService.record(operation, resourceType, resourceName, clusterId, detail, "SUCCESS", null);
+ } catch (Exception auditFailure) {
+ log.warn("Failed to record audit operation={} resource={}: {}", operation, resourceName,
+ auditFailure.getMessage());
+ }
+ }
+
}
diff --git a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageService.java b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageService.java
index b67b9557..5e5a7da5 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageService.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageService.java
@@ -45,6 +45,9 @@ public class MessageService {
}
public TraceRecordVO getMessageTrace(String instanceId, String msgId) {
+ if (!StringUtils.hasText(msgId)) {
+ throw new BusinessException(400, "msgId is required");
+ }
log.info("Getting message trace: msgId={}", msgId);
return providerRegistry.byInstanceId(instanceId)
.map(provider -> provider.getMessageTrace(instanceId, msgId))
@@ -58,6 +61,9 @@ public class MessageService {
if (hasKey && !hasTopic) {
throw new BusinessException(400, "topic is required when key is specified");
}
+ if (hasMessageId && !hasTopic) {
+ throw new BusinessException(400, "topic is required when msgId is specified");
+ }
if (!hasTopic && !hasMessageId) {
throw new BusinessException(400, "topic or msgId is required");
}
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 d67ace2a..31b70cb7 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
@@ -20,6 +20,7 @@ import org.apache.rocketmq.studio.provider.apache.AdminClient;
import org.apache.rocketmq.studio.provider.apache.MetadataProvider;
import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.springframework.util.StringUtils;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
import org.apache.rocketmq.studio.instance.group.QueueProgressVO;
import org.apache.rocketmq.studio.instance.group.SubscriptionEntryVO;
@@ -49,7 +50,7 @@ public class MetadataService {
}
public List<TopicVO> listTopics(String instanceId, String clusterId, String type, String search) {
- if (isBlank(instanceId) && !isBlank(clusterId)) {
+ if (!StringUtils.hasText(instanceId) && StringUtils.hasText(clusterId)) {
// legacy cluster-scoped read kept for AI tool handlers
return metadataProvider.listTopics(
normalizeFilter(clusterId), normalizeFilter(type), normalizeFilter(search));
@@ -84,11 +85,12 @@ public class MetadataService {
}
public List<BrokerRouteVO> getTopicRoutes(String instanceId, String name) {
+ String topicName = requireName(name, "topic name");
if (resolve(instanceId).vendor() != InstanceVendor.APACHE) {
// broker routing does not apply to serverless cloud instances
return List.of();
}
- return metadataProvider.getTopicRoutes(instanceId, name);
+ return metadataProvider.getTopicRoutes(instanceId, topicName);
}
@@ -97,7 +99,8 @@ public class MetadataService {
}
public List<TopicConsumerVO> getTopicConsumers(String instanceId, String name) {
- return resolve(instanceId).getTopicConsumers(instanceId, name);
+ String topicName = requireName(name, "topic name");
+ return resolve(instanceId).getTopicConsumers(instanceId, topicName);
}
@@ -117,7 +120,7 @@ public class MetadataService {
}
public List<ConsumerGroupVO> listConsumerGroups(String instanceId, String clusterId, String search) {
- if (isBlank(instanceId) && !isBlank(clusterId)) {
+ if (!StringUtils.hasText(instanceId) && StringUtils.hasText(clusterId)) {
return metadataProvider.listConsumerGroups(normalizeFilter(clusterId), normalizeFilter(search));
}
return resolve(instanceId).listConsumerGroups(instanceId, normalizeFilter(search));
@@ -129,10 +132,11 @@ public class MetadataService {
}
public ConsumerGroupVO getConsumerGroup(String instanceId, String name) {
+ String groupName = requireName(name, "consumer group name");
if (resolve(instanceId).vendor() != InstanceVendor.APACHE) {
throw new BusinessException(501, "Consumer group detail is not supported for cloud instances");
}
- return adminClient.getConsumerGroup(instanceId, name);
+ return adminClient.getConsumerGroup(instanceId, groupName);
}
@@ -141,7 +145,8 @@ public class MetadataService {
}
public List<QueueProgressVO> getGroupProgress(String instanceId, String name) {
- return resolve(instanceId).getGroupProgress(instanceId, name);
+ String groupName = requireName(name, "consumer group name");
+ return resolve(instanceId).getGroupProgress(instanceId, groupName);
}
@@ -150,7 +155,8 @@ public class MetadataService {
}
public List<SubscriptionEntryVO> getGroupSubscriptions(String instanceId, String name) {
- return resolve(instanceId).getGroupSubscriptions(instanceId, name);
+ String groupName = requireName(name, "consumer group name");
+ return resolve(instanceId).getGroupSubscriptions(instanceId, groupName);
}
@@ -187,12 +193,8 @@ public class MetadataService {
.orElseGet(() -> providerRegistry.forVendor(InstanceVendor.APACHE));
}
- private static boolean isBlank(String value) {
- return value == null || value.isBlank();
- }
-
private String normalizeFilter(String value) {
- return value == null || value.isBlank() ? null : value.trim();
+ return !StringUtils.hasText(value) ? null : value.trim();
}
private void requireTopic(TopicVO topic) {
@@ -206,4 +208,11 @@ public class MetadataService {
throw new BusinessException(400, "Topic send message request is required");
}
}
+
+ private String requireName(String value, String fieldName) {
+ if (!StringUtils.hasText(value)) {
+ throw new BusinessException(400, fieldName + " is required");
+ }
+ return value.trim();
+ }
}
diff --git a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertService.java b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertService.java
index 053bd567..512371a9 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertService.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertService.java
@@ -17,6 +17,7 @@
package org.apache.rocketmq.studio.ops.alert;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.audit.OperationAuditService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
@@ -34,6 +35,7 @@ public class AlertService {
private final AlertRepository alertRepository;
private final AlertRuleAssetService alertRuleAssetService;
+ private final OperationAuditService operationAuditService;
public List<AlertRuleVO> listRules() {
@@ -42,7 +44,9 @@ public class AlertService {
}
public String exportPrometheusRulesYaml() {
- List<AlertRuleVO> rules = alertRepository.findAllRules();
+ List<AlertRuleVO> rules = alertRepository.findAllRules().stream()
+ .filter(AlertRuleVO::isEnabled)
+ .toList();
List<PrometheusAlertRule> prometheusRules = rules.isEmpty()
? defaultPrometheusRules()
: rules.stream().map(this::toPrometheusRule).toList();
@@ -83,7 +87,9 @@ public class AlertService {
}
log.info("Creating alert rule: {}", rule.getName());
rule.setId(UUID.randomUUID().toString());
- return alertRepository.saveRule(rule);
+ AlertRuleVO saved = alertRepository.saveRule(rule);
+ auditRule("CREATE_ALERT_RULE", saved, null);
+ return saved;
}
@@ -97,6 +103,7 @@ public class AlertService {
if (!alertRepository.replaceRule(rule)) {
throw ruleNotFound(id);
}
+ auditRule("UPDATE_ALERT_RULE", rule, null);
return rule;
}
@@ -109,7 +116,9 @@ public class AlertService {
.findFirst()
.orElseThrow(() -> new org.apache.rocketmq.studio.common.exception.BusinessException(404, "Alert rule not found: " + id));
rule.setEnabled(enabled);
- return alertRepository.saveRule(rule);
+ AlertRuleVO saved = alertRepository.saveRule(rule);
+ auditRule("TOGGLE_ALERT_RULE", saved, "enabled=" + enabled);
+ return saved;
}
@@ -118,6 +127,7 @@ public class AlertService {
if (!alertRepository.deleteRule(id)) {
throw ruleNotFound(id);
}
+ recordAudit("DELETE_ALERT_RULE", "ALERT_RULE", id, null, null);
}
@@ -135,13 +145,19 @@ public class AlertService {
.findFirst()
.orElseThrow(() -> new org.apache.rocketmq.studio.common.exception.BusinessException(404, "System alert not found: " + id));
alert.setAcknowledged(true);
- return alertRepository.saveAlert(alert);
+ SystemAlertVO saved = alertRepository.saveAlert(alert);
+ recordAudit("ACKNOWLEDGE_SYSTEM_ALERT", "SYSTEM_ALERT", saved.getId(), null,
+ "acknowledged=true");
+ return saved;
}
public int clearAcknowledged() {
log.info("Clearing acknowledged system alerts");
- return alertRepository.deleteAcknowledgedAlerts();
+ int deleted = alertRepository.deleteAcknowledgedAlerts();
+ recordAudit("CLEAR_ACKNOWLEDGED_SYSTEM_ALERTS", "SYSTEM_ALERT", null, null,
+ "deleted=" + deleted);
+ return deleted;
}
private List<PrometheusAlertRule> defaultPrometheusRules() {
@@ -202,6 +218,12 @@ public class AlertService {
selector.append(label).append("=\"").append(escapeDoubleQuotedValue(value.trim())).append('"');
}
+ private void auditRule(String operation, AlertRuleVO rule, String detail) {
+ String auditDetail = detail == null ? "name=" + rule.getName() : detail;
+ recordAudit(operation, "ALERT_RULE", rule.getId(), null,
+ auditDetail);
+ }
+
private String severity(AlertRuleVO rule) {
String severity = rule.getSeverity();
if (hasText(severity)) {
@@ -296,4 +318,15 @@ public class AlertService {
private BusinessException ruleNotFound(String id) {
return new BusinessException(404, "Alert rule not found: " + id);
}
+
+ private void recordAudit(String operation, String resourceType, String resourceName,
+ String clusterId, String detail) {
+ try {
+ operationAuditService.record(operation, resourceType, resourceName, clusterId, detail, "SUCCESS", null);
+ } catch (Exception auditFailure) {
+ log.warn("Failed to record audit operation={} resource={}: {}", operation, resourceName,
+ auditFailure.getMessage());
+ }
+ }
+
}
diff --git a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
index 2f59fb11..e7a89ea6 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
@@ -19,6 +19,7 @@ package org.apache.rocketmq.studio.provider.apache;
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.client.producer.SendStatus;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
@@ -335,6 +336,10 @@ public class RocketMQAdminClientImpl implements AdminClient {
}
SendResult sendResult = producer.send(msg);
+ if (sendResult == null || sendResult.getSendStatus() != SendStatus.SEND_OK) {
+ String status = sendResult == null ? "null" : String.valueOf(sendResult.getSendStatus());
+ throw new BusinessException(502, "Message send did not succeed: " + status);
+ }
// The message is already delivered by now; an audit write failure must not turn a
// successful send into an error, or callers would retry and duplicate the message.
@@ -346,6 +351,9 @@ public class RocketMQAdminClientImpl implements AdminClient {
.sendTime(System.currentTimeMillis())
.offsetMsgId(sendResult.getOffsetMsgId())
.build();
+ } catch (BusinessException e) {
+ recordAudit("SEND_MESSAGE", request.getTopic(), e.getMessage(), "FAILED");
+ throw e;
} catch (Exception e) {
recordAudit("SEND_MESSAGE", request.getTopic(), e.getMessage(), "FAILED");
throw new BusinessException(500, "Failed to send message: " + e.getMessage());
diff --git a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java
index 1ef0d70d..420a796a 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java
@@ -263,8 +263,11 @@ public class RocketMQMessageProvider implements MessageProvider {
parseTraceBody(traceMessage.getBody(), msgId, nodes, consumerStatus);
}
}
+ } catch (BusinessException e) {
+ throw e;
} catch (Exception e) {
log.warn("Trace query for msgId={} failed: {}", msgId, e.getMessage());
+ throw new BusinessException(502, "Failed to query message trace: " + e.getMessage());
}
recordTraceQuery(instanceId, msgId, null, nodes.size(), consumerStatus.size());
diff --git a/server/src/main/java/org/apache/rocketmq/studio/provider/credential/CloudCredentialService.java b/server/src/main/java/org/apache/rocketmq/studio/provider/credential/CloudCredentialService.java
index 10d7ca2a..0decc7ec 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/provider/credential/CloudCredentialService.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/provider/credential/CloudCredentialService.java
@@ -20,6 +20,7 @@ import org.springframework.util.StringUtils;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.common.util.CredentialUtils;
+import org.apache.rocketmq.studio.audit.OperationAuditService;
import org.apache.rocketmq.studio.instance.InstanceRepository;
import org.apache.rocketmq.studio.provider.alibaba.AliyunClientFactory;
import org.springframework.stereotype.Service;
@@ -38,6 +39,7 @@ public class CloudCredentialService {
private final CloudCredentialRepository credentialRepository;
private final InstanceRepository instanceRepository;
private final AliyunClientFactory aliyunClientFactory;
+ private final OperationAuditService operationAuditService;
public List<CloudCredentialVO> listMasked() {
log.info("Listing cloud credentials (masked)");
@@ -69,7 +71,10 @@ public class CloudCredentialService {
credential.setId(UUID.randomUUID().toString());
credential.setCreatedAt(LocalDateTime.now());
credential.setUpdatedAt(LocalDateTime.now());
- return maskAccessKey(credentialRepository.save(credential));
+ CloudCredentialVO saved = credentialRepository.save(credential);
+ recordAudit("CREATE_CLOUD_CREDENTIAL", "CLOUD_CREDENTIAL", saved.getId(), null,
+ credentialAuditDetail(saved));
+ return maskAccessKey(saved);
}
public CloudCredentialVO update(UpdateCloudCredentialDTO request) {
@@ -94,6 +99,8 @@ public class CloudCredentialService {
existing.setUpdatedAt(LocalDateTime.now());
CloudCredentialVO saved = credentialRepository.save(existing);
invalidateAliyunClients(saved);
+ recordAudit("UPDATE_CLOUD_CREDENTIAL", "CLOUD_CREDENTIAL", saved.getId(), null,
+ credentialAuditDetail(saved));
return maskAccessKey(saved);
}
@@ -109,6 +116,8 @@ public class CloudCredentialService {
}
credentialRepository.deleteById(id);
invalidateAliyunClients(existing);
+ recordAudit("DELETE_CLOUD_CREDENTIAL", "CLOUD_CREDENTIAL", id, null,
+ credentialAuditDetail(existing));
}
public CloudCredentialVO reveal(String id) {
@@ -131,9 +140,25 @@ public class CloudCredentialService {
masked.setUpdatedAt(credential.getUpdatedAt());
return masked;
}
+
+ private String credentialAuditDetail(CloudCredentialVO credential) {
+ return "name=" + credential.getName() + ", vendor=" + credential.getVendor();
+ }
+
private void invalidateAliyunClients(CloudCredentialVO credential) {
if (credential.getVendor() == org.apache.rocketmq.studio.common.domain.enums.InstanceVendor.ALIYUN) {
aliyunClientFactory.invalidateCredential(credential.getId());
}
}
+
+ private void recordAudit(String operation, String resourceType, String resourceName,
+ String clusterId, String detail) {
+ try {
+ operationAuditService.record(operation, resourceType, resourceName, clusterId, detail, "SUCCESS", null);
+ } catch (Exception auditFailure) {
+ log.warn("Failed to record audit operation={} resource={}: {}", operation, resourceName,
+ auditFailure.getMessage());
+ }
+ }
+
}
diff --git a/server/src/main/resources/application-dev.yml b/server/src/main/resources/application-dev.yml
index f7efb59e..c6af6056 100644
--- a/server/src/main/resources/application-dev.yml
+++ b/server/src/main/resources/application-dev.yml
@@ -5,6 +5,10 @@ spring:
username: sa
password:
driver-class-name: org.h2.Driver
+ sql:
+ init:
+ mode: always
+ schema-locations: classpath:db/schema.sql
h2:
console:
enabled: true
diff --git a/server/src/main/resources/db/schema.sql b/server/src/main/resources/db/schema.sql
index 6fc1f01d..a9a4c15a 100644
--- a/server/src/main/resources/db/schema.sql
+++ b/server/src/main/resources/db/schema.sql
@@ -24,6 +24,10 @@ CREATE TABLE IF NOT EXISTS rmq_instance (
remark VARCHAR(255),
type VARCHAR(32) NOT NULL COMMENT 'PROXY/DIRECT',
endpoint VARCHAR(512) NOT NULL,
+ vendor VARCHAR(32),
+ cloud_instance_id VARCHAR(128),
+ credential_id VARCHAR(64),
+ region_id VARCHAR(128),
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
@@ -44,7 +48,7 @@ CREATE TABLE IF NOT EXISTS rmq_topic (
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
UNIQUE KEY uk_cluster_topic (cluster_id, name),
- INDEX idx_instance (instance_id)
+ INDEX idx_topic_instance (instance_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
-- 4. Consumer Group 管理记录
@@ -61,7 +65,7 @@ CREATE TABLE IF NOT EXISTS rmq_group (
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
UNIQUE KEY uk_cluster_group (cluster_id, name),
- INDEX idx_instance (instance_id)
+ INDEX idx_group_instance (instance_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
-- 5. K8s 证书管理
@@ -95,7 +99,7 @@ CREATE TABLE IF NOT EXISTS rmq_message_query (
cluster_id VARCHAR(255),
queried_by VARCHAR(64),
queried_at DATETIME DEFAULT CURRENT_TIMESTAMP,
- INDEX idx_queried_at (queried_at),
+ INDEX idx_message_query_queried_at (queried_at),
INDEX idx_topic (topic)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
@@ -110,7 +114,7 @@ CREATE TABLE IF NOT EXISTS rmq_trace_query (
queried_by VARCHAR(64),
queried_at DATETIME DEFAULT CURRENT_TIMESTAMP,
INDEX idx_msg_id (msg_id),
- INDEX idx_queried_at (queried_at)
+ INDEX idx_trace_query_queried_at (queried_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
-- 8. 操作审计日志（所有写操作）
@@ -132,7 +136,7 @@ CREATE TABLE IF NOT EXISTS rmq_operation_audit (
-- 9. 通用设置（单行）
CREATE TABLE IF NOT EXISTS rmq_settings (
- id VARCHAR(16) PRIMARY KEY DEFAULT 'singleton',
+ id VARCHAR(16) DEFAULT 'singleton' PRIMARY KEY,
json TEXT NOT NULL COMMENT 'GeneralSettingsVO JSON',
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
diff --git a/server/src/test/java/org/apache/rocketmq/studio/StudioApplicationTest.java b/server/src/test/java/org/apache/rocketmq/studio/StudioApplicationTest.java
index 2881946c..125d65be 100644
--- a/server/src/test/java/org/apache/rocketmq/studio/StudioApplicationTest.java
+++ b/server/src/test/java/org/apache/rocketmq/studio/StudioApplicationTest.java
@@ -18,13 +18,20 @@ package org.apache.rocketmq.studio;
import org.apache.rocketmq.studio.ops.ai.tool.ToolCatalog;
import org.apache.rocketmq.studio.ops.ai.tool.ToolGatewayService;
+import org.apache.rocketmq.studio.persistence.mapper.RmqInstanceMapper;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
+import org.springframework.test.web.servlet.MockMvc;
import static org.assertj.core.api.Assertions.assertThat;
+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;
@SpringBootTest
+@AutoConfigureMockMvc
class StudioApplicationTest {
@Autowired
@@ -33,9 +40,20 @@ class StudioApplicationTest {
@Autowired
private ToolGatewayService toolGatewayService;
+ @Autowired
+ private RmqInstanceMapper instanceMapper;
+
+ @Autowired
+ private MockMvc mockMvc;
+
@Test
- void applicationContextLoadsWithToolGateway() {
+ void applicationContextLoadsWithInitializedDevSchema() throws Exception {
assertThat(toolCatalog.getVersion()).isEqualTo("1.0.0");
assertThat(toolGatewayService.discover(null)).isNotEmpty();
+ assertThat(instanceMapper.selectList(null)).isEmpty();
+
+ mockMvc.perform(get("/api/instances"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.data").isEmpty());
}
}
diff --git a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/RuntimeAdminClientResolverTest.java b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/RuntimeAdminClientResolverTest.java
index efc118d3..913d03ce 100644
--- a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/RuntimeAdminClientResolverTest.java
+++ b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/RuntimeAdminClientResolverTest.java
@@ -17,6 +17,7 @@
package org.apache.rocketmq.studio.cluster.broker;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
import org.apache.rocketmq.studio.instance.InstanceRepository;
import org.apache.rocketmq.studio.instance.InstanceVO;
import org.junit.jupiter.api.Test;
@@ -32,6 +33,7 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
@@ -80,4 +82,23 @@ class RuntimeAdminClientResolverTest {
assertThat(result).isEqualTo("done");
verify(adminFactory).execute(eq("namesrv-b:9876"), isNull(), any());
}
+
+ @Test
+ void rejectsCloudInstancesBeforeResolvingOrExecutingAdminClient() {
+ InstanceVO instance = InstanceVO.builder()
+ .vendor(InstanceVendor.ALIYUN)
+ .endpoint("cloud-endpoint:9876")
+ .build();
+ instance.setId("cloud-instance");
+ when(instanceRepository.findById("cloud-instance")).thenReturn(Optional.of(instance));
+ RuntimeAdminClientResolver resolver = new RuntimeAdminClientResolver(instanceRepository, adminFactory);
+
+ assertThatThrownBy(() -> resolver.resolveEndpoint("cloud-instance"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Runtime AdminClient only supports Apache instances: cloud-instance");
+ assertThatThrownBy(() -> resolver.execute(instance, admin -> "unused"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Runtime AdminClient only supports Apache instances: cloud-instance");
+ verifyNoInteractions(adminFactory);
+ }
}
diff --git a/server/src/test/java/org/apache/rocketmq/studio/cluster/client/ClientServiceTest.java b/server/src/test/java/org/apache/rocketmq/studio/cluster/client/ClientServiceTest.java
index ce472901..e6152373 100644
--- a/server/src/test/java/org/apache/rocketmq/studio/cluster/client/ClientServiceTest.java
+++ b/server/src/test/java/org/apache/rocketmq/studio/cluster/client/ClientServiceTest.java
@@ -18,6 +18,7 @@ package org.apache.rocketmq.studio.cluster.client;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
@@ -25,7 +26,9 @@ import org.mockito.junit.jupiter.MockitoExtension;
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.verifyNoInteractions;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
@@ -52,12 +55,21 @@ class ClientServiceTest {
}
@Test
- void listConnectionsShouldTreatBlankFiltersAsUnspecified() {
- when(clientProvider.findConnections(null, null, null)).thenReturn(List.of());
+ void listConnectionsShouldTreatBlankOptionalFiltersAsUnspecified() {
+ when(clientProvider.findConnections("instance-1", null, null)).thenReturn(List.of());
- List<ClientConnectionVO> result = clientService.listConnections(" ", " ", "\t");
+ List<ClientConnectionVO> result = clientService.listConnections("instance-1", " ", "\t");
assertThat(result).isEmpty();
- verify(clientProvider).findConnections(null, null, null);
+ verify(clientProvider).findConnections("instance-1", null, null);
+ }
+
+ @Test
+ void listConnectionsShouldRejectBlankInstanceIdBeforeQueryingProvider() {
+ assertThatThrownBy(() -> clientService.listConnections(" ", null, null))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("instanceId is required");
+
+ verifyNoInteractions(clientProvider);
}
}
diff --git a/server/src/test/java/org/apache/rocketmq/studio/cluster/k8s/K8sCertServiceTest.java b/server/src/test/java/org/apache/rocketmq/studio/cluster/k8s/K8sCertServiceTest.java
index d93bc32e..72382060 100644
--- a/server/src/test/java/org/apache/rocketmq/studio/cluster/k8s/K8sCertServiceTest.java
+++ b/server/src/test/java/org/apache/rocketmq/studio/cluster/k8s/K8sCertServiceTest.java
@@ -19,6 +19,7 @@ package org.apache.rocketmq.studio.cluster.k8s;
import org.apache.rocketmq.studio.common.domain.enums.CertStatus;
import org.apache.rocketmq.studio.common.domain.enums.CertType;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.audit.OperationAuditService;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
@@ -38,6 +39,7 @@ import java.util.Optional;
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.eq;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
@@ -51,13 +53,16 @@ class K8sCertServiceTest {
@Mock
private K8sCertRepository k8sCertRepository;
+ @Mock
+ private OperationAuditService operationAuditService;
+
private K8sCertService k8sCertService;
private K8sCertVO sampleCert;
@BeforeEach
void setUp() {
- k8sCertService = new K8sCertService(k8sCertRepository, CLOCK);
+ k8sCertService = new K8sCertService(k8sCertRepository, operationAuditService, CLOCK);
sampleCert = K8sCertVO.builder()
.name("rocketmq-tls")
.namespace("mq-system")
@@ -169,6 +174,9 @@ class K8sCertServiceTest {
assertThat(result.getCreatedAt()).isEqualTo(now);
assertThat(result.getUpdatedAt()).isEqualTo(now);
verify(k8sCertRepository).save(any(K8sCertVO.class));
+ verify(operationAuditService).record(eq("CREATE_K8S_CERTIFICATE"), eq("K8S_CERTIFICATE"),
+ eq(result.getId()), eq(null), eq("name=new-tls-cert, namespace=default, cluster=test-cluster"),
+ eq("SUCCESS"), eq(null));
}
@Test
@@ -246,6 +254,9 @@ class K8sCertServiceTest {
assertThat(sampleCert.getType()).isEqualTo(CertType.TLS);
assertThat(sampleCert.getUpdatedAt()).isEqualTo(LocalDateTime.of(2025, 1, 2, 0, 0));
verify(k8sCertRepository).save(any(K8sCertVO.class));
+ verify(operationAuditService).record(eq("UPDATE_K8S_CERTIFICATE"), eq("K8S_CERTIFICATE"),
+ eq("cert-1"), eq(null), eq("name=updated-name, namespace=new-namespace, cluster=new-cluster"),
+ eq("SUCCESS"), eq(null));
}
@Test
@@ -351,6 +362,9 @@ class K8sCertServiceTest {
assertThat(sampleCert.getNotBefore()).isEqualTo(originalNotBefore);
assertThat(sampleCert.getNotAfter()).isEqualTo(originalNotAfter);
verify(k8sCertRepository).save(any(K8sCertVO.class));
+ verify(operationAuditService).record(eq("RENEW_K8S_CERTIFICATE"), eq("K8S_CERTIFICATE"),
+ eq("cert-1"), eq(null), eq("name=rocketmq-tls, namespace=mq-system, cluster=prod-cluster"),
+ eq("SUCCESS"), eq(null));
}
@Test
@@ -396,6 +410,8 @@ class K8sCertServiceTest {
k8sCertService.deleteCert(command);
verify(k8sCertRepository).deleteById("cert-1");
+ verify(operationAuditService).record(eq("DELETE_K8S_CERTIFICATE"), eq("K8S_CERTIFICATE"),
+ eq("cert-1"), eq(null), eq(null), eq("SUCCESS"), eq(null));
}
@Test
diff --git a/server/src/test/java/org/apache/rocketmq/studio/instance/InstanceServiceTest.java b/server/src/test/java/org/apache/rocketmq/studio/instance/InstanceServiceTest.java
index 01065024..ec2e0ec1 100644
--- a/server/src/test/java/org/apache/rocketmq/studio/instance/InstanceServiceTest.java
+++ b/server/src/test/java/org/apache/rocketmq/studio/instance/InstanceServiceTest.java
@@ -20,6 +20,7 @@ package org.apache.rocketmq.studio.instance;
import org.apache.rocketmq.studio.provider.credential.CloudCredentialRepository;
import org.apache.rocketmq.studio.provider.credential.CloudCredentialVO;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
+import org.apache.rocketmq.studio.audit.OperationAuditService;
import org.apache.rocketmq.studio.common.domain.enums.InstanceType;
import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
import org.apache.rocketmq.studio.common.exception.BusinessException;
@@ -40,6 +41,8 @@ import java.util.Optional;
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.argThat;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
@@ -63,6 +66,9 @@ class InstanceServiceTest {
@Mock
private MqAdminExtFactory adminFactory;
+ @Mock
+ private OperationAuditService operationAuditService;
+
@InjectMocks
private InstanceService instanceService;
@@ -259,6 +265,10 @@ class InstanceServiceTest {
assertThat(result.getUpdatedAt()).isNotNull();
assertThat(result.getName()).isEqualTo("new-instance");
verify(instanceRepository).save(any(InstanceVO.class));
+ verify(operationAuditService).record(eq("CREATE_INSTANCE"), eq("INSTANCE"), eq(result.getId()), eq(null),
+ argThat(detail -> detail.equals("name=new-instance, vendor=APACHE, type=PROXY")
+ && !detail.contains("10.0.1.1:8080")),
+ eq("SUCCESS"), eq(null));
}
@Test
@@ -327,6 +337,15 @@ class InstanceServiceTest {
assertThat(instanceService.createInstance(input).getEndpoint()).isEqualTo("namesrv:9876");
}
+ @Test
+ void createInstanceShouldTrimNameBeforeSaving() {
+ InstanceVO input = InstanceVO.builder().name(" production ").type(InstanceType.PROXY)
+ .endpoint("namesrv:9876").build();
+ when(instanceRepository.save(any(InstanceVO.class))).thenAnswer(invocation -> invocation.getArgument(0));
+
+ assertThat(instanceService.createInstance(input).getName()).isEqualTo("production");
+ }
+
@Test
void updateInstanceShouldMergeFieldsOntoExisting() {
LocalDateTime originalCreatedAt = LocalDateTime.of(2025, 1, 2, 3, 4, 5);
@@ -367,6 +386,20 @@ class InstanceServiceTest {
assertThat(existing.getName()).isEqualTo("old-name");
assertThat(existing.getRemark()).isEqualTo("old remark");
assertThat(existing.getUpdatedAt()).isEqualTo(originalUpdatedAt);
+ verify(operationAuditService).record(eq("UPDATE_INSTANCE"), eq("INSTANCE"), eq("inst-1"), eq(null),
+ eq("name=new-name, vendor=APACHE, type=PROXY"), eq("SUCCESS"), eq(null));
+ }
+
+ @Test
+ void updateInstanceShouldTrimNameBeforeSaving() {
+ InstanceVO existing = InstanceVO.builder().name("old-name").endpoint("namesrv:9876").build();
+ existing.setId("inst-1");
+ InstanceVO update = InstanceVO.builder().name(" production ").build();
+ update.setId("inst-1");
+ when(instanceRepository.findById("inst-1")).thenReturn(Optional.of(existing));
+ when(instanceRepository.save(any(InstanceVO.class))).thenAnswer(invocation -> invocation.getArgument(0));
+
+ assertThat(instanceService.updateInstance(update).getName()).isEqualTo("production");
}
@Test
@@ -592,6 +625,8 @@ class InstanceServiceTest {
instanceService.deleteInstance("inst-1");
verify(instanceRepository).deleteById("inst-1");
+ verify(operationAuditService).record(eq("DELETE_INSTANCE"), eq("INSTANCE"), eq("inst-1"), eq(null),
+ eq("name=to-delete, vendor=APACHE, type=null"), eq("SUCCESS"), eq(null));
}
@Test
diff --git a/server/src/test/java/org/apache/rocketmq/studio/instance/acl/AclServiceTest.java b/server/src/test/java/org/apache/rocketmq/studio/instance/acl/AclServiceTest.java
index 0a78fbcc..b3cd90c4 100644
--- a/server/src/test/java/org/apache/rocketmq/studio/instance/acl/AclServiceTest.java
+++ b/server/src/test/java/org/apache/rocketmq/studio/instance/acl/AclServiceTest.java
@@ -18,6 +18,7 @@
package org.apache.rocketmq.studio.instance.acl;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.audit.OperationAuditService;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
@@ -37,6 +38,8 @@ import java.util.stream.Stream;
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.argThat;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -47,6 +50,9 @@ class AclServiceTest {
@Mock
private AclRepository aclRepository;
+ @Mock
+ private OperationAuditService operationAuditService;
+
@InjectMocks
private AclService aclService;
@@ -107,6 +113,8 @@ class AclServiceTest {
assertThat(result.getPrincipal()).isEqualTo("user1");
assertThat(result.getResource()).isEqualTo("topic-1");
verify(aclRepository).saveRule(any(AclRuleVO.class));
+ verify(operationAuditService).record(eq("CREATE_ACL_RULE"), eq("ACL_RULE"), eq(result.getId()), eq(null),
+ eq("principal=user1"), eq("SUCCESS"), eq(null));
}
@Test
@@ -115,6 +123,8 @@ class AclServiceTest {
aclService.deleteRule("rule-1");
verify(aclRepository).deleteRule("rule-1");
+ verify(operationAuditService).record(eq("DELETE_ACL_RULE"), eq("ACL_RULE"), eq("rule-1"), eq(null),
+ eq(null), eq("SUCCESS"), eq(null));
}
@Test
@@ -148,6 +158,8 @@ class AclServiceTest {
assertThat(result.getDecision()).isEqualTo("DENY");
verify(aclRepository).replaceRule(input);
verify(aclRepository, never()).saveRule(any(AclRuleVO.class));
+ verify(operationAuditService).record(eq("UPDATE_ACL_RULE"), eq("ACL_RULE"), eq("rule-1"), eq(null),
+ eq("principal=user1"), eq("SUCCESS"), eq(null));
}
@Test
@@ -307,6 +319,10 @@ class AclServiceTest {
assertThat(result.getCreatedAt()).isNotNull();
assertThat(result.getUsername()).isEqualTo("newuser");
verify(aclRepository).saveUser(any(AclUserVO.class));
+ verify(operationAuditService).record(eq("CREATE_ACL_USER"), eq("ACL_USER"), eq(result.getId()), eq(null),
+ argThat(detail -> detail.equals("username=newuser, admin=false")
+ && !detail.contains(result.getAccessKey()) && !detail.contains(result.getSecretKey())),
+ eq("SUCCESS"), eq(null));
}
@Test
@@ -315,6 +331,8 @@ class AclServiceTest {
aclService.deleteUser("user-1");
verify(aclRepository).deleteUser("user-1");
+ verify(operationAuditService).record(eq("DELETE_ACL_USER"), eq("ACL_USER"), eq("user-1"), eq(null),
+ eq(null), eq("SUCCESS"), eq(null));
}
@Test
@@ -365,6 +383,8 @@ class AclServiceTest {
verify(aclRepository).saveUser(captor.capture());
assertThat(captor.getValue().getAccessKey()).isEqualTo("access-key-123456");
assertThat(captor.getValue().getSecretKey()).isEqualTo("secret-key-987654");
+ verify(operationAuditService).record(eq("UPDATE_ACL_USER"), eq("ACL_USER"), eq("user-1"), eq(null),
+ eq("username=newuser, admin=true"), eq("SUCCESS"), eq(null));
}
@Test
diff --git a/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageServiceTest.java b/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageServiceTest.java
index 8d237eb6..8d96cc8b 100644
--- a/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageServiceTest.java
+++ b/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageServiceTest.java
@@ -14,13 +14,9 @@ import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
import org.junit.jupiter.api.Test;
-import java.util.Collections;
-
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
-import static org.mockito.Mockito.when;
class MessageServiceTest {
@@ -38,16 +34,29 @@ class MessageServiceTest {
}
@Test
- void keepsMessageIdQueryWithoutTopicValid() {
+ void rejectsMessageIdQueryWithoutTopicBeforeCallingProvider() {
MessageProvider provider = mock(MessageProvider.class);
InstanceProviderRegistry registry = mock(InstanceProviderRegistry.class);
- when(provider.queryMessages(null, null, "msg-001", null, null, null, null))
- .thenReturn(Collections.emptyList());
MessageService service = new MessageService(provider, registry);
- service.queryMessages(null, null, "msg-001", null, null, null, null);
+ assertThatThrownBy(() -> service.queryMessages(null, null, "msg-001", null, null, null, null))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("topic is required when msgId is specified");
+
+ verifyNoInteractions(provider, registry);
+ }
+
+ @Test
+ void rejectsBlankMessageTraceIdBeforeCallingProvider() {
+ MessageProvider provider = mock(MessageProvider.class);
+ InstanceProviderRegistry registry = mock(InstanceProviderRegistry.class);
+ MessageService service = new MessageService(provider, registry);
+
+ assertThatThrownBy(() -> service.getMessageTrace("instance-a", " "))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("msgId is required");
- verify(provider).queryMessages(null, null, "msg-001", null, null, null, null);
+ verifyNoInteractions(provider, registry);
}
@Test
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 4d6d02e1..918a7fad 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
@@ -171,6 +171,27 @@ class MetadataServiceTest {
verify(apacheProvider).getTopicConsumers("instance-a", "orders");
}
+ @Test
+ void runtimeDiagnosticsShouldRejectBlankTopicAndGroupNamesBeforeProviderResolution() {
+ assertThatThrownBy(() -> metadataService.getTopicRoutes("instance-a", " "))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("topic name is required");
+ assertThatThrownBy(() -> metadataService.getTopicConsumers("instance-a", " "))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("topic name is required");
+ assertThatThrownBy(() -> metadataService.getConsumerGroup("instance-a", " "))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("consumer group name is required");
+ assertThatThrownBy(() -> metadataService.getGroupProgress("instance-a", " "))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("consumer group name is required");
+ assertThatThrownBy(() -> metadataService.getGroupSubscriptions("instance-a", " "))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("consumer group name is required");
+
+ verifyNoInteractions(metadataProvider, adminClient, providerRegistry, apacheProvider);
+ }
+
@Test
void sendMessageShouldReturnResult() {
SendMessageDTO request = SendMessageDTO.builder()
diff --git a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertServiceDefaultRulesTest.java b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertServiceDefaultRulesTest.java
index 786ebf65..6d0acfda 100644
--- a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertServiceDefaultRulesTest.java
+++ b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertServiceDefaultRulesTest.java
@@ -16,7 +16,9 @@
*/
package org.apache.rocketmq.studio.ops.alert;
+import org.apache.rocketmq.studio.audit.OperationAuditService;
import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
import java.util.Collections;
import java.util.List;
@@ -36,7 +38,8 @@ class AlertServiceDefaultRulesTest {
AlertRepository repository = mock(AlertRepository.class);
when(repository.findAllRules()).thenReturn(Collections.emptyList());
- AlertService service = new AlertService(repository, new AlertRuleAssetService());
+ AlertService service = new AlertService(repository, new AlertRuleAssetService(),
+ Mockito.mock(OperationAuditService.class));
String yaml = service.exportPrometheusRulesYaml();
int ruleCount = countRules(yaml);
@@ -48,7 +51,8 @@ class AlertServiceDefaultRulesTest {
AlertRepository repository = mock(AlertRepository.class);
when(repository.findAllRules()).thenReturn(List.of());
- AlertService service = new AlertService(repository, new AlertRuleAssetService());
+ AlertService service = new AlertService(repository, new AlertRuleAssetService(),
+ Mockito.mock(OperationAuditService.class));
String yaml = service.exportPrometheusRulesYaml();
assertTrue(yaml.contains("rocketmq-broker.rules"));
diff --git a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertServiceTest.java b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertServiceTest.java
index 35a46555..73b440ce 100644
--- a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertServiceTest.java
+++ b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertServiceTest.java
@@ -21,6 +21,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
import org.apache.rocketmq.studio.common.domain.enums.AlertLevel;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.audit.OperationAuditService;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.Mock;
@@ -33,6 +34,7 @@ import java.util.List;
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.eq;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -43,11 +45,14 @@ class AlertServiceTest {
@Mock
private AlertRepository alertRepository;
+ @Mock
+ private OperationAuditService operationAuditService;
+
private AlertService alertService;
@org.junit.jupiter.api.BeforeEach
void setUp() {
- alertService = new AlertService(alertRepository, new AlertRuleAssetService());
+ alertService = new AlertService(alertRepository, new AlertRuleAssetService(), operationAuditService);
}
@Test
@@ -100,6 +105,7 @@ class AlertServiceTest {
.threshold(5000)
.duration("3m")
.description("Lag too high")
+ .enabled(true)
.build();
when(alertRepository.findAllRules()).thenReturn(List.of(rule));
@@ -122,6 +128,7 @@ class AlertServiceTest {
.threshold(1000)
.duration("3m")
.description("First lag alert")
+ .enabled(true)
.build();
AlertRuleVO second = AlertRuleVO.builder()
.name("Lag Alert B")
@@ -130,6 +137,7 @@ class AlertServiceTest {
.threshold(2000)
.duration("5m")
.description("Second lag alert")
+ .enabled(true)
.build();
when(alertRepository.findAllRules()).thenReturn(List.of(first, second));
@@ -156,6 +164,7 @@ class AlertServiceTest {
.clusterName("DefaultCluster")
.severity("critical")
.description("Slave falls behind master")
+ .enabled(true)
.build();
when(alertRepository.findAllRules()).thenReturn(List.of(rule));
@@ -178,6 +187,7 @@ class AlertServiceTest {
.brokerName("*")
.clusterName(" ")
.severity("fatal")
+ .enabled(true)
.build();
when(alertRepository.findAllRules()).thenReturn(List.of(rule));
@@ -198,6 +208,7 @@ class AlertServiceTest {
.clusterName("prod\"east\\dc\nline")
.brokerName("broker\tone")
.description("Summary \"quoted\" \\ path\nnext - Details\twith\rline")
+ .enabled(true)
.build();
when(alertRepository.findAllRules()).thenReturn(List.of(rule));
@@ -221,6 +232,7 @@ class AlertServiceTest {
.metric("rocketmq_consumer_lag_messages")
.operator(">")
.threshold(1)
+ .enabled(true)
.build();
when(alertRepository.findAllRules()).thenReturn(List.of(rule));
@@ -231,6 +243,50 @@ class AlertServiceTest {
assertThat(exportedRule.path("alert").asText()).isEqualTo("RocketMQAlert");
}
+ @Test
+ void exportPrometheusRulesYamlShouldExcludeDisabledRules() {
+ AlertRuleVO enabled = AlertRuleVO.builder()
+ .name("Enabled Lag Alert")
+ .metric("rocketmq_consumer_lag_messages")
+ .operator(">")
+ .threshold(1000)
+ .enabled(true)
+ .build();
+ AlertRuleVO disabled = AlertRuleVO.builder()
+ .name("Disabled Lag Alert")
+ .metric("rocketmq_consumer_lag_messages")
+ .operator(">")
+ .threshold(2000)
+ .enabled(false)
+ .build();
+ when(alertRepository.findAllRules()).thenReturn(List.of(enabled, disabled));
+
+ String result = alertService.exportPrometheusRulesYaml();
+
+ assertThat(result)
+ .contains("EnabledLagAlert")
+ .doesNotContain("DisabledLagAlert")
+ .doesNotContain(" > 2000");
+ }
+
+ @Test
+ void exportPrometheusRulesYamlShouldUseDefaultRulesWhenAllConfiguredRulesAreDisabled() {
+ AlertRuleVO disabled = AlertRuleVO.builder()
+ .name("Disabled Lag Alert")
+ .metric("rocketmq_consumer_lag_messages")
+ .operator(">")
+ .threshold(2000)
+ .enabled(false)
+ .build();
+ when(alertRepository.findAllRules()).thenReturn(List.of(disabled));
+
+ String result = alertService.exportPrometheusRulesYaml();
+
+ assertThat(result)
+ .contains("RocketMQBrokerDown")
+ .doesNotContain("DisabledLagAlert");
+ }
+
@Test
void createRuleShouldPreserveReplicationScopeFields() {
AlertRuleVO input = AlertRuleVO.builder()
@@ -252,6 +308,8 @@ class AlertServiceTest {
assertThat(result.getClusterName()).isEqualTo("DefaultCluster");
assertThat(result.getSeverity()).isEqualTo("critical");
verify(alertRepository).saveRule(result);
+ verify(operationAuditService).record(eq("CREATE_ALERT_RULE"), eq("ALERT_RULE"), eq(result.getId()),
+ eq(null), eq("name=Replication Lag High"), eq("SUCCESS"), eq(null));
}
@Test
@@ -300,6 +358,8 @@ class AlertServiceTest {
assertThat(result.getId()).isEqualTo("rule-1");
assertThat(result.getThreshold()).isEqualTo(90.0);
verify(alertRepository).replaceRule(update);
+ verify(operationAuditService).record(eq("UPDATE_ALERT_RULE"), eq("ALERT_RULE"), eq("rule-1"),
+ eq(null), eq("name=CPU Alert"), eq("SUCCESS"), eq(null));
}
@Test
@@ -357,6 +417,8 @@ class AlertServiceTest {
assertThat(result.isEnabled()).isTrue();
verify(alertRepository).saveRule(result);
+ verify(operationAuditService).record(eq("TOGGLE_ALERT_RULE"), eq("ALERT_RULE"), eq("rule-1"),
+ eq(null), eq("enabled=true"), eq("SUCCESS"), eq(null));
}
@Test
@@ -386,6 +448,8 @@ class AlertServiceTest {
alertService.deleteRule("rule-1");
verify(alertRepository).deleteRule("rule-1");
+ verify(operationAuditService).record(eq("DELETE_ALERT_RULE"), eq("ALERT_RULE"), eq("rule-1"),
+ eq(null), eq(null), eq("SUCCESS"), eq(null));
}
@Test
@@ -437,6 +501,8 @@ class AlertServiceTest {
assertThat(result.isAcknowledged()).isTrue();
verify(alertRepository).saveAlert(result);
+ verify(operationAuditService).record(eq("ACKNOWLEDGE_SYSTEM_ALERT"), eq("SYSTEM_ALERT"), eq("a1"),
+ eq(null), eq("acknowledged=true"), eq("SUCCESS"), eq(null));
}
@Test
@@ -456,6 +522,8 @@ class AlertServiceTest {
assertThat(result).isEqualTo(3);
verify(alertRepository).deleteAcknowledgedAlerts();
+ verify(operationAuditService).record(eq("CLEAR_ACKNOWLEDGED_SYSTEM_ALERTS"), eq("SYSTEM_ALERT"),
+ eq(null), eq(null), eq("deleted=3"), eq("SUCCESS"), eq(null));
}
@Test
diff --git a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
index 2bd23a1d..804ec59b 100644
--- a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
+++ b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
@@ -19,6 +19,7 @@ import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
@@ -338,6 +339,7 @@ class RocketMQAdminClientImplTest {
mockConstruction(DefaultMQProducer.class, (producer, context) -> {
doNothing().when(producer).start();
SendResult sendResult = new SendResult();
+ sendResult.setSendStatus(SendStatus.SEND_OK);
sendResult.setMsgId("msg-1");
sendResult.setOffsetMsgId("offset-1");
when(producer.send(any(Message.class))).thenReturn(sendResult);
@@ -359,6 +361,7 @@ class RocketMQAdminClientImplTest {
mockConstruction(DefaultMQProducer.class, (producer, context) -> {
doNothing().when(producer).start();
SendResult sendResult = new SendResult();
+ sendResult.setSendStatus(SendStatus.SEND_OK);
sendResult.setMsgId("msg-1");
sendResult.setOffsetMsgId("offset-1");
when(producer.send(any(Message.class))).thenReturn(sendResult);
@@ -375,4 +378,50 @@ class RocketMQAdminClientImplTest {
verify(producer).setNamesrvAddr("10.0.0.2:9876");
}
}
+
+ @Test
+ void sendMessageShouldRejectNonSuccessfulSendStatus() throws Exception {
+ when(properties.getNamesrvAddr()).thenReturn("10.0.0.1:9876");
+ try (MockedConstruction<DefaultMQProducer> mockedProducers =
+ mockConstruction(DefaultMQProducer.class, (producer, context) -> {
+ doNothing().when(producer).start();
+ SendResult sendResult = new SendResult();
+ sendResult.setSendStatus(SendStatus.FLUSH_DISK_TIMEOUT);
+ when(producer.send(any(Message.class))).thenReturn(sendResult);
+ doNothing().when(producer).shutdown();
+ })) {
+ SendMessageDTO request = new SendMessageDTO();
+ request.setTopic("TopicA");
+ request.setBody("hello");
+
+ assertThatThrownBy(() -> adminClient.sendMessage(request))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("FLUSH_DISK_TIMEOUT");
+
+ verify(auditService).record("SEND_MESSAGE", "TopicA",
+ "Message send did not succeed: FLUSH_DISK_TIMEOUT", "FAILED");
+ }
+ }
+
+ @Test
+ void sendMessageShouldRejectNullSendResult() throws Exception {
+ when(properties.getNamesrvAddr()).thenReturn("10.0.0.1:9876");
+ try (MockedConstruction<DefaultMQProducer> mockedProducers =
+ mockConstruction(DefaultMQProducer.class, (producer, context) -> {
+ doNothing().when(producer).start();
+ when(producer.send(any(Message.class))).thenReturn(null);
+ doNothing().when(producer).shutdown();
+ })) {
+ SendMessageDTO request = new SendMessageDTO();
+ request.setTopic("TopicA");
+ request.setBody("hello");
+
+ assertThatThrownBy(() -> adminClient.sendMessage(request))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("null");
+
+ verify(auditService).record("SEND_MESSAGE", "TopicA",
+ "Message send did not succeed: null", "FAILED");
+ }
+ }
}
diff --git a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java
index d5c8aa51..6028ceb0 100644
--- a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java
+++ b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java
@@ -256,4 +256,17 @@ class RocketMQMessageProviderTest {
assertThat(transaction.getDescription()).contains("tx-group").contains("COMMIT_MESSAGE");
assertThat(record.getConsumerStatus()).isEmpty();
}
+
+ @Test
+ void getMessageTraceSurfacesAdminFailure() throws Exception {
+ when(adminExt.queryMessage(anyString(), anyString(), anyInt(), anyLong(), anyLong()))
+ .thenThrow(new IllegalStateException("broker unavailable"));
+
+ assertThatThrownBy(() -> provider.getMessageTrace("instance-a", "msg-123"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Failed to query message trace: broker unavailable")
+ .satisfies(error -> assertThat(((BusinessException) error).getCode()).isEqualTo(502));
+
+ verify(queryHistoryService, never()).recordTraceQuery(anyString(), anyString(), any(), anyInt(), anyInt());
+ }
}
diff --git a/server/src/test/java/org/apache/rocketmq/studio/provider/credential/CloudCredentialServiceTest.java b/server/src/test/java/org/apache/rocketmq/studio/provider/credential/CloudCredentialServiceTest.java
index 9c796c78..d62c1869 100644
--- a/server/src/test/java/org/apache/rocketmq/studio/provider/credential/CloudCredentialServiceTest.java
+++ b/server/src/test/java/org/apache/rocketmq/studio/provider/credential/CloudCredentialServiceTest.java
@@ -20,6 +20,7 @@ import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
import org.apache.rocketmq.studio.common.util.CredentialUtils;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.instance.InstanceRepository;
+import org.apache.rocketmq.studio.audit.OperationAuditService;
import org.apache.rocketmq.studio.provider.alibaba.AliyunClientFactory;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
@@ -33,6 +34,8 @@ import java.util.Optional;
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.argThat;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -49,6 +52,9 @@ class CloudCredentialServiceTest {
@Mock
private AliyunClientFactory aliyunClientFactory;
+ @Mock
+ private OperationAuditService operationAuditService;
+
@InjectMocks
private CloudCredentialService service;
@@ -124,6 +130,10 @@ class CloudCredentialServiceTest {
assertThat(created.getId()).isNotBlank();
assertThat(created.getAccessKey()).isEqualTo("LTAI****0001");
assertThat(created.getSecretKey()).isNull();
+ verify(operationAuditService).record(eq("CREATE_CLOUD_CREDENTIAL"), eq("CLOUD_CREDENTIAL"),
+ eq(created.getId()), eq(null), argThat(detail -> detail.equals("name=ok, vendor=ALIYUN")
+ && !detail.contains("LTAI5tGoodKey00000000001") && !detail.contains("sk-value")),
+ eq("SUCCESS"), eq(null));
}
@Test
@@ -157,6 +167,8 @@ class CloudCredentialServiceTest {
service.update(request);
verify(aliyunClientFactory).invalidateCredential("cred-1");
+ verify(operationAuditService).record(eq("UPDATE_CLOUD_CREDENTIAL"), eq("CLOUD_CREDENTIAL"),
+ eq("cred-1"), eq(null), eq("name=null, vendor=ALIYUN"), eq("SUCCESS"), eq(null));
}
@Test
@@ -171,6 +183,8 @@ class CloudCredentialServiceTest {
verify(credentialRepository).deleteById("cred-1");
verify(aliyunClientFactory).invalidateCredential("cred-1");
+ verify(operationAuditService).record(eq("DELETE_CLOUD_CREDENTIAL"), eq("CLOUD_CREDENTIAL"),
+ eq("cred-1"), eq(null), eq("name=null, vendor=ALIYUN"), eq("SUCCESS"), eq(null));
}
@Test
diff --git a/web/src/api/producer.test.ts b/web/src/api/producer.test.ts
index e01257e2..6303b075 100644
--- a/web/src/api/producer.test.ts
+++ b/web/src/api/producer.test.ts
@@ -34,12 +34,15 @@ describe('Producer API', () => {
});
it('fetches Studio topic records sorted alphabetically', async () => {
- mock.onGet('/topics').reply(200, {
+ mock.onGet('/topics').reply((config) => {
+ expect(config.params.instanceId).toBe('instance-1');
+ return [200, {
code: 200,
data: [{ name: 'order-events' }, { name: 'user-signup' }, { name: 'batch-process' }],
+ }];
});
- const result = await fetchTopicList();
+ const result = await fetchTopicList('instance-1');
expect(result).toEqual(['batch-process', 'order-events', 'user-signup']);
});
@@ -48,14 +51,14 @@ describe('Producer API', () => {
topicList: ['order-events', 'user-signup', 'batch-process'],
});
- const result = await fetchTopicList();
+ const result = await fetchTopicList('instance-1');
expect(result).toEqual(['batch-process', 'order-events', 'user-signup']);
});
it('handles empty topic list', async () => {
mock.onGet('/topics').reply(200, { topicList: [] });
- const result = await fetchTopicList();
+ const result = await fetchTopicList('instance-1');
expect(result).toEqual([]);
});
diff --git a/web/src/api/producer.ts b/web/src/api/producer.ts
index e01c0a0c..61138103 100644
--- a/web/src/api/producer.ts
+++ b/web/src/api/producer.ts
@@ -36,9 +36,9 @@ interface TopicListResponse {
// ─── API ────────────────────────────────────────────────────────
-/** Fetch all topic names */
-export async function fetchTopicList(): Promise<string[]> {
- const res = await client.get<TopicListResponse>('/topics');
+/** Fetch topic names for a managed instance. */
+export async function fetchTopicList(instanceId: string): Promise<string[]> {
+ const res = await client.get<TopicListResponse>('/topics', { params: { instanceId } });
const topics = res.data.data?.map((topic) => topic.name) ?? res.data.topicList ?? [];
return topics.sort();
}
diff --git a/web/src/i18n/translations.ts b/web/src/i18n/translations.ts
index 46dcdb3a..66050329 100644
--- a/web/src/i18n/translations.ts
+++ b/web/src/i18n/translations.ts
@@ -305,6 +305,14 @@ const translations: Record<string, Record<Lang, string>> = {
zh: '访问控制规则与用户权限管理，共 {rules} 条规则、{users} 个用户',
en: 'Access control rules and user permissions, {rules} rules, {users} users',
},
+ 'acl.localMetadataNotice': {
+ zh: '当前 ACL 规则和用户仅保存为 Studio 本地元数据',
+ en: 'Current ACL rules and users are stored as Studio-local metadata',
+ },
+ 'acl.localMetadataDescription': {
+ zh: '它们尚不会下发到所选 RocketMQ 实例。实例级 ACL Provider 接入完成前，请在集群侧管理实际 ACL 策略。',
+ en: 'They are not applied to the selected RocketMQ instance. Manage the effective ACL policy on the cluster until instance-scoped provider support is available.',
+ },
'acl.addRule': { zh: '添加规则', en: 'Add Rule' },
'acl.addUser': { zh: '添加用户', en: 'Add User' },
'acl.ruleTab': { zh: 'ACL 规则', en: 'ACL Rules' },
diff --git a/web/src/pages/cluster/__tests__/ClientsPage.test.tsx b/web/src/pages/cluster/__tests__/ClientsPage.test.tsx
index 3ffae0b4..88c486de 100644
--- a/web/src/pages/cluster/__tests__/ClientsPage.test.tsx
+++ b/web/src/pages/cluster/__tests__/ClientsPage.test.tsx
@@ -23,6 +23,7 @@ import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vite
import type { ClientConnection } from '../../../api/connections';
import { LangProvider } from '../../../i18n/LangContext';
import * as connectionsService from '../../../services/connectionsService';
+import * as instanceService from '../../../services/instanceService';
import ClientsPage from '../clients';
vi.mock('../../../services/connectionsService', () => ({
@@ -221,6 +222,33 @@ describe('Clients page', () => {
expect(within(screen.getByTestId('connection-total')).getByText('0')).toBeInTheDocument();
});
+ it('surfaces instance discovery failures and allows retrying', async () => {
+ vi.mocked(instanceService.listInstances)
+ .mockRejectedValueOnce(new Error('Unable to load managed instances'))
+ .mockResolvedValueOnce([
+ {
+ id: 'instance-1',
+ name: 'Instance 1',
+ endpoint: 'namesrv-1:9876',
+ type: 'DIRECT',
+ remark: '',
+ topicCount: 0,
+ consumerGroupCount: 0,
+ createdAt: '',
+ updatedAt: '',
+ },
+ ]);
+ const user = userEvent.setup();
+ renderWithProviders(<ClientsPage />);
+
+ expect(await screen.findByText('Unable to load managed instances')).toBeInTheDocument();
+ expect(connectionsService.listConnections).not.toHaveBeenCalled();
+
+ await user.click(screen.getByRole('button', { name: /重\s*试/ }));
+ await screen.findByText('order-svc-0@10.0.1.12:49152');
+ expect(connectionsService.listConnections).toHaveBeenCalledWith({ instanceId: 'instance-1' });
+ });
+
it('opens a client detail dialog from the connection table', async () => {
const user = userEvent.setup();
renderWithProviders(<ClientsPage />);
diff --git a/web/src/pages/cluster/__tests__/K8sCertsPage.test.tsx b/web/src/pages/cluster/__tests__/K8sCertsPage.test.tsx
index 6b4903b5..aa9aa43f 100644
--- a/web/src/pages/cluster/__tests__/K8sCertsPage.test.tsx
+++ b/web/src/pages/cluster/__tests__/K8sCertsPage.test.tsx
@@ -101,6 +101,14 @@ describe('K8sCertsPage', () => {
expect(screen.getByText('-')).toBeInTheDocument();
});
+ it('explains that certificate records are Studio-local metadata', async () => {
+ renderPage();
+
+ expect(await screen.findByTestId('k8s-cert-local-metadata-notice')).toHaveTextContent(
+ '当前证书记录仅保存为 Studio 本地元数据',
+ );
+ });
+
it.each([
['platform', 'rocketmq-staging-tls', 'rocketmq-prod-tls'],
['broker.prod.example.com', 'rocketmq-prod-tls', 'rocketmq-staging-tls'],
diff --git a/web/src/pages/cluster/certs.tsx b/web/src/pages/cluster/certs.tsx
index 400db782..9a971891 100644
--- a/web/src/pages/cluster/certs.tsx
+++ b/web/src/pages/cluster/certs.tsx
@@ -28,6 +28,7 @@ import {
Space,
Typography,
Card,
+ Alert,
message,
} from 'antd';
import type { ColumnsType } from 'antd/es/table';
@@ -356,6 +357,14 @@ const K8sCertsPage = () => {
</Button>
}
/>
+ <Alert
+ data-testid="k8s-cert-local-metadata-notice"
+ type="warning"
+ showIcon
+ message="当前证书记录仅保存为 Studio 本地元数据"
+ description="创建、续期和删除操作尚不会应用到 Kubernetes 集群或 cert-manager。请在集群侧管理实际证书，直到 Kubernetes Provider 接入完成。"
+ style={{ marginBottom: 16 }}
+ />
<Flex justify="space-between" style={{ marginBottom: 16 }}>
<Space>
<Input.Search
diff --git a/web/src/pages/cluster/clients.tsx b/web/src/pages/cluster/clients.tsx
index f4febdfb..92efe529 100644
--- a/web/src/pages/cluster/clients.tsx
+++ b/web/src/pages/cluster/clients.tsx
@@ -114,18 +114,34 @@ const ClientsPage = () => {
const [clusterFilter, setClusterFilter] = useState<string>('ALL');
const [selectedConnection, setSelectedConnection] = useState<ClientConnection | null>(null);
const [loadError, setLoadError] = useState<string | null>(null);
+ const [instanceLoadKey, setInstanceLoadKey] = useState(0);
useEffect(() => {
let cancelled = false;
- void listInstances().then((nextInstances) => {
- if (cancelled) return;
- setInstances(nextInstances);
- setSelectedInstanceId((current) => current || nextInstances[0]?.id || '');
- });
+
+ setLoading(true);
+ void listInstances()
+ .then((nextInstances) => {
+ if (cancelled) return;
+ setInstances(nextInstances);
+ setSelectedInstanceId((current) => current || nextInstances[0]?.id || '');
+ setLoadError(null);
+ })
+ .catch((error) => {
+ if (cancelled) return;
+ setInstances([]);
+ setSelectedInstanceId('');
+ setConnections([]);
+ setLoadError(getLoadErrorMessage(error));
+ })
+ .finally(() => {
+ if (!cancelled) setLoading(false);
+ });
+
return () => {
cancelled = true;
};
- }, []);
+ }, [instanceLoadKey]);
useEffect(() => {
let cancelled = false;
@@ -355,7 +371,17 @@ const ClientsPage = () => {
/>
{loadError && (
- <Alert showIcon type="warning" message={loadError} style={{ marginBottom: 16 }} />
+ <Alert
+ showIcon
+ type="warning"
+ message={loadError}
+ style={{ marginBottom: 16 }}
+ action={
+ <Button size="small" onClick={() => setInstanceLoadKey((key) => key + 1)}>
+ 重试
+ </Button>
+ }
+ />
)}
{connections.some((connection) => connection.partial) && (
<Alert
diff --git a/web/src/pages/home/__tests__/DashboardPage.test.tsx b/web/src/pages/home/__tests__/DashboardPage.test.tsx
new file mode 100644
index 00000000..4d28874c
--- /dev/null
+++ b/web/src/pages/home/__tests__/DashboardPage.test.tsx
@@ -0,0 +1,118 @@
+/*
+ * 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.
+ */
+
+import { App } from 'antd';
+import { render, screen, waitFor } from '@testing-library/react';
+import userEvent from '@testing-library/user-event';
+import type React from 'react';
+import { MemoryRouter } from 'react-router-dom';
+import { beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
+import type { DashboardData } from '../../../api/metrics';
+import { LangProvider } from '../../../i18n/LangContext';
+import * as dashboardService from '../../../services/dashboardService';
+import * as instanceService from '../../../services/instanceService';
+import DashboardPage from '../dashboard';
+
+vi.mock('../../../services/dashboardService', () => ({ getDashboard: vi.fn() }));
+vi.mock('../../../services/instanceService', () => ({ listInstances: vi.fn() }));
+
+const dashboard = (name: string): DashboardData => ({
+ stats: {
+ totalClusters: 1,
+ healthyClusters: 1,
+ totalBrokers: 1,
+ totalProxies: 0,
+ totalNameServers: 1,
+ totalTopics: 1,
+ totalConsumerGroups: 1,
+ totalMessagesToday: 1,
+ messagesPerSecond: 1,
+ tpsIn: 1,
+ tpsOut: 1,
+ },
+ clusters: [
+ {
+ id: name,
+ name,
+ type: 'V4_NAMESRV',
+ status: 'healthy',
+ brokers: 1,
+ proxies: 0,
+ topics: 1,
+ groups: 1,
+ tpsIn: 1,
+ tpsOut: 1,
+ version: '5.0.0',
+ throughput: [1],
+ },
+ ],
+});
+
+const deferred = <T,>() => {
+ let resolve!: (value: T) => void;
+ const promise = new Promise<T>((promiseResolve) => {
+ resolve = promiseResolve;
+ });
+ return { promise, resolve };
+};
+
+const renderWithProviders = (ui: React.ReactElement) =>
+ render(
+ <App>
+ <LangProvider>
+ <MemoryRouter>{ui}</MemoryRouter>
+ </LangProvider>
+ </App>,
+ );
+
+beforeAll(() => {
+ Object.defineProperty(window, 'matchMedia', {
+ writable: true,
+ value: vi.fn().mockImplementation(() => ({
+ matches: false,
+ addEventListener: vi.fn(),
+ removeEventListener: vi.fn(),
+ })),
+ });
+});
+
+beforeEach(() => {
+ vi.clearAllMocks();
+ vi.mocked(instanceService.listInstances).mockResolvedValue([
+ { id: 'instance-a', name: 'Instance A', endpoint: 'a:9876', type: 'DIRECT', remark: '', topicCount: 0, consumerGroupCount: 0, createdAt: '', updatedAt: '' },
+ { id: 'instance-b', name: 'Instance B', endpoint: 'b:9876', type: 'DIRECT', remark: '', topicCount: 0, consumerGroupCount: 0, createdAt: '', updatedAt: '' },
+ ]);
+});
+
+describe('DashboardPage', () => {
+ it('does not let a stale instance response overwrite the latest selection', async () => {
+ const instanceA = deferred<DashboardData>();
+ const instanceB = deferred<DashboardData>();
+ vi.mocked(dashboardService.getDashboard)
+ .mockResolvedValueOnce(dashboard('initial-cluster'))
+ .mockReturnValueOnce(instanceA.promise)
+ .mockReturnValueOnce(instanceB.promise);
+ const user = userEvent.setup();
+ renderWithProviders(<DashboardPage />);
+
+ await screen.findByText('initial-cluster');
+ const selector = screen.getByRole('combobox', { name: 'Dashboard instance' });
+ await user.click(selector);
+ await user.click(await screen.findByText('Instance A', { selector: '.ant-select-item-option-content' }));
+ await user.click(selector);
+ await user.click(await screen.findByText('Instance B', { selector: '.ant-select-item-option-content' }));
+
+ instanceB.resolve(dashboard('instance-b-cluster'));
+ expect(await screen.findByText('instance-b-cluster')).toBeInTheDocument();
+
+ instanceA.resolve(dashboard('instance-a-cluster'));
+ await waitFor(() => {
+ expect(screen.queryByText('instance-a-cluster')).not.toBeInTheDocument();
+ });
+ expect(screen.getByText('instance-b-cluster')).toBeInTheDocument();
+ });
+});
diff --git a/web/src/pages/home/dashboard.tsx b/web/src/pages/home/dashboard.tsx
index 8eced776..f6ec900c 100644
--- a/web/src/pages/home/dashboard.tsx
+++ b/web/src/pages/home/dashboard.tsx
@@ -1,4 +1,4 @@
-import { useCallback, useEffect, useState } from 'react';
+import { useCallback, useEffect, useRef, useState } from 'react';
import { useNavigate } from 'react-router-dom';
import {
Alert,
@@ -37,16 +37,25 @@ const DashboardPage = () => {
const [selectedInstanceId, setSelectedInstanceId] = useState<string>();
const [loading, setLoading] = useState(true);
const [loadError, setLoadError] = useState(false);
+ const dashboardRequestIdRef = useRef(0);
const loadDashboard = useCallback(async () => {
+ const requestId = ++dashboardRequestIdRef.current;
setLoading(true);
setLoadError(false);
try {
- setDashboard(await getDashboard(selectedInstanceId));
+ const nextDashboard = await getDashboard(selectedInstanceId);
+ if (requestId === dashboardRequestIdRef.current) {
+ setDashboard(nextDashboard);
+ }
} catch {
- setLoadError(true);
+ if (requestId === dashboardRequestIdRef.current) {
+ setLoadError(true);
+ }
} finally {
- setLoading(false);
+ if (requestId === dashboardRequestIdRef.current) {
+ setLoading(false);
+ }
}
}, [selectedInstanceId]);
diff --git a/web/src/pages/instance/__tests__/AclPage.test.tsx b/web/src/pages/instance/__tests__/AclPage.test.tsx
index a67a9725..379535de 100644
--- a/web/src/pages/instance/__tests__/AclPage.test.tsx
+++ b/web/src/pages/instance/__tests__/AclPage.test.tsx
@@ -121,6 +121,14 @@ describe('ACL page', () => {
expect(screen.getByText('cluster-a')).toBeInTheDocument();
});
+ it('explains that ACL records are Studio-local metadata', async () => {
+ renderWithProviders(<AclPage />);
+
+ expect(await screen.findByTestId('acl-local-metadata-notice')).toHaveTextContent(
+ '当前 ACL 规则和用户仅保存为 Studio 本地元数据',
+ );
+ });
+
it('renders backend users on the user tab', async () => {
const user = userEvent.setup();
renderWithProviders(<AclPage />);
diff --git a/web/src/pages/instance/__tests__/DLQPage.test.tsx b/web/src/pages/instance/__tests__/DLQPage.test.tsx
index fbd4f7e6..b55778af 100644
--- a/web/src/pages/instance/__tests__/DLQPage.test.tsx
+++ b/web/src/pages/instance/__tests__/DLQPage.test.tsx
@@ -43,6 +43,17 @@ vi.mock('../../../services/instanceService', () => ({
createdAt: '',
updatedAt: '',
},
+ {
+ id: 'instance-2',
+ name: 'Instance 2',
+ endpoint: 'namesrv-2:9876',
+ type: 'DIRECT',
+ remark: '',
+ topicCount: 0,
+ consumerGroupCount: 0,
+ createdAt: '',
+ updatedAt: '',
+ },
]),
}));
@@ -253,4 +264,39 @@ describe('DLQ page', () => {
expect(await screen.findByText('DLQ provider is not configured')).toBeInTheDocument();
});
+
+ it('clears retry state before loading groups for a newly selected instance', async () => {
+ let resolveSecondInstance!: (groups: DLQGroup[]) => void;
+ vi.mocked(messageService.listDLQGroups)
+ .mockResolvedValueOnce([dlqGroup])
+ .mockImplementationOnce(
+ () =>
+ new Promise<DLQGroup[]>((resolve) => {
+ resolveSecondInstance = resolve;
+ }),
+ );
+ const user = userEvent.setup();
+ renderWithProviders(<DLQPage />);
+
+ const orderRow = (await screen.findByText('cg-order')).closest('tr');
+ if (!orderRow) throw new Error('DLQ group row not found');
+
+ await user.click(within(orderRow).getByRole('button', { name: '重投消息' }));
+ expect(await screen.findByText('重投死信消息')).toBeInTheDocument();
+
+ await user.click(screen.getAllByRole('combobox')[0]);
+ await user.click(
+ await screen.findByText('Instance 2', { selector: '.ant-select-item-option-content' }),
+ );
+
+ await waitFor(() => {
+ expect(messageService.listDLQGroups).toHaveBeenLastCalledWith('instance-2');
+ });
+ await waitFor(() => {
+ expect(screen.queryByRole('row', { name: /cg-order/ })).not.toBeInTheDocument();
+ });
+
+ resolveSecondInstance([secondDlqGroup]);
+ expect(await screen.findByText('-cg-"payment"')).toBeInTheDocument();
+ });
});
diff --git a/web/src/pages/instance/__tests__/MessagePage.test.tsx b/web/src/pages/instance/__tests__/MessagePage.test.tsx
index 1ae39b55..36dd97a3 100644
--- a/web/src/pages/instance/__tests__/MessagePage.test.tsx
+++ b/web/src/pages/instance/__tests__/MessagePage.test.tsx
@@ -79,6 +79,8 @@ const renderWithProviders = (ui: React.ReactElement) =>
</App>,
);
+const lastElement = <T,>(elements: T[]): T => elements[elements.length - 1]!;
+
describe('Message page query history', () => {
beforeEach(() => {
localStorage.clear();
@@ -96,11 +98,14 @@ describe('Message page query history', () => {
expect(screen.getByRole('button', { name: /最近查询/ })).toBeDisabled();
await user.click(screen.getByText('按 Message ID'));
+ await user.click(lastElement(screen.getAllByRole('combobox')));
+ await user.click(lastElement(await screen.findAllByText('order-create')));
await user.type(screen.getByPlaceholderText('输入 Message ID'), 'MID-001');
await user.click(screen.getByRole('button', { name: /^search查询$/ }));
await waitFor(() => {
expect(messageServiceMocks.queryMessages).toHaveBeenCalledWith({
+ topic: 'order-create',
msgId: 'MID-001',
instanceId: '',
});
@@ -112,11 +117,12 @@ describe('Message page query history', () => {
renderWithProviders(<MessagePage />);
await user.click(screen.getByRole('button', { name: /最近查询/ }));
- await user.click(await screen.findByText('Message ID: MID-001'));
+ await user.click(await screen.findByText('Message ID: MID-001 · Topic: order-create'));
expect(screen.getByPlaceholderText('输入 Message ID')).toHaveValue('MID-001');
await waitFor(() => {
expect(messageServiceMocks.queryMessages).toHaveBeenCalledWith({
+ topic: 'order-create',
msgId: 'MID-001',
instanceId: '',
});
@@ -134,11 +140,14 @@ describe('Message page query history', () => {
renderWithProviders(<MessagePage />);
await user.click(screen.getByText('按 Message ID'));
+ await user.click(lastElement(screen.getAllByRole('combobox')));
+ await user.click(lastElement(await screen.findAllByText('order-create')));
await user.type(screen.getByPlaceholderText('输入 Message ID'), 'MID-FAILED');
await user.click(screen.getByRole('button', { name: /^search查询$/ }));
await waitFor(() => {
expect(messageServiceMocks.queryMessages).toHaveBeenCalledWith({
+ topic: 'order-create',
msgId: 'MID-FAILED',
instanceId: '',
});
@@ -151,6 +160,8 @@ describe('Message page query history', () => {
const user = userEvent.setup();
renderWithProviders(<MessagePage />);
await user.click(screen.getByText('按 Message ID'));
+ await user.click(lastElement(screen.getAllByRole('combobox')));
+ await user.click(lastElement(await screen.findAllByText('order-create')));
const messageIdInput = screen.getByPlaceholderText('输入 Message ID');
const queryButton = screen.getByRole('button', { name: /^search查询$/ });
@@ -240,15 +251,16 @@ describe('Message page query history', () => {
JSON.stringify([
{ mode: 'topic', params: { topic: ['invalid'] } },
{ mode: 'unknown', params: { topic: 'order-create' } },
- { mode: 'msgid', params: { msgId: 'MID-VALID' } },
+ { mode: 'msgid', params: { msgId: 'MID-VALID', topic: 'order-create' } },
]),
);
renderWithProviders(<MessagePage />);
await user.click(screen.getByRole('button', { name: /最近查询/ }));
- expect(await screen.findByText('Message ID: MID-VALID')).toBeInTheDocument();
+ expect(
+ await screen.findByText('Message ID: MID-VALID · Topic: order-create'),
+ ).toBeInTheDocument();
expect(screen.queryByText(/invalid/)).not.toBeInTheDocument();
- expect(screen.queryByText(/order-create/)).not.toBeInTheDocument();
});
it('does not report consume verification success without a backend API', async () => {
@@ -257,6 +269,8 @@ describe('Message page query history', () => {
renderWithProviders(<MessagePage />);
await user.click(screen.getByText('按 Message ID'));
+ await user.click(lastElement(screen.getAllByRole('combobox')));
+ await user.click(lastElement(await screen.findAllByText('order-create')));
await user.type(screen.getByPlaceholderText('输入 Message ID'), 'MID-CONSUME-VERIFY-001');
await user.click(screen.getByRole('button', { name: /^search查询$/ }));
diff --git a/web/src/pages/instance/acl.tsx b/web/src/pages/instance/acl.tsx
index e6e3e469..374320c0 100644
--- a/web/src/pages/instance/acl.tsx
+++ b/web/src/pages/instance/acl.tsx
@@ -33,6 +33,7 @@ import {
Badge,
Typography,
Flex,
+ Alert,
message,
} from 'antd';
import { Plus, MagnifyingGlass, ShieldCheck, User, Eye, EyeSlash } from '@phosphor-icons/react';
@@ -655,6 +656,15 @@ const AclPage = () => {
}
/>
+ <Alert
+ data-testid="acl-local-metadata-notice"
+ type="warning"
+ showIcon
+ message={t('acl.localMetadataNotice')}
+ description={t('acl.localMetadataDescription')}
+ style={{ marginBottom: 16 }}
+ />
+
<Card bordered={false} bodyStyle={{ padding: 0 }}>
<Tabs
activeKey={activeTab}
diff --git a/web/src/pages/instance/dlq.tsx b/web/src/pages/instance/dlq.tsx
index 7f3a5873..dfd574fc 100644
--- a/web/src/pages/instance/dlq.tsx
+++ b/web/src/pages/instance/dlq.tsx
@@ -125,12 +125,22 @@ const DLQPage = () => {
useEffect(() => {
let cancelled = false;
+ // The retry dialog owns a group name that is meaningful only for the
+ // currently selected instance. Clear all instance-scoped state before
+ // starting the next request so an old group cannot be retried on a new
+ // instance while that request is in flight.
+ setGroups([]);
+ setSelectedGroupNames([]);
+ setDetailGroup(null);
+ setRetryModalOpen(false);
+ setRetryGroup(null);
+ setRetryTargetTopic('');
+ setRetryError(null);
+ setLoadError(null);
+
if (!selectedInstanceId) {
void Promise.resolve().then(() => {
if (cancelled) return;
- setGroups([]);
- setSelectedGroupNames([]);
- setLoadError(null);
setLoading(false);
});
return () => {
@@ -138,6 +148,7 @@ const DLQPage = () => {
};
}
+ setLoading(true);
void listDLQGroups(selectedInstanceId)
.then((nextGroups) => {
if (!cancelled) {
diff --git a/web/src/pages/instance/message.tsx b/web/src/pages/instance/message.tsx
index 96bf564c..c4a7d43a 100644
--- a/web/src/pages/instance/message.tsx
+++ b/web/src/pages/instance/message.tsx
@@ -161,21 +161,23 @@ const isMessageQuery = (value: unknown): value is MessageQuery => {
);
};
+const isRecentQuery = (value: unknown): value is RecentQuery => {
+ if (typeof value !== 'object' || value === null) return false;
+ const query = value as RecentQuery;
+ return (
+ isQueryMode(query.mode) &&
+ isMessageQuery(query.params) &&
+ (query.mode !== 'msgid' || Boolean(query.params.topic?.trim()))
+ );
+};
+
const loadRecentQueries = (): RecentQuery[] => {
try {
const stored = localStorage.getItem(QUERY_HISTORY_STORAGE_KEY);
if (!stored) return [];
const parsed: unknown = JSON.parse(stored);
if (!Array.isArray(parsed)) return [];
- return parsed
- .filter(
- (item): item is RecentQuery =>
- typeof item === 'object' &&
- item !== null &&
- isQueryMode((item as RecentQuery).mode) &&
- isMessageQuery((item as RecentQuery).params),
- )
- .slice(0, MAX_QUERY_HISTORY);
+ return parsed.filter(isRecentQuery).slice(0, MAX_QUERY_HISTORY);
} catch {
return [];
}
@@ -184,7 +186,8 @@ const loadRecentQueries = (): RecentQuery[] => {
const querySignature = (query: RecentQuery): string => JSON.stringify(query);
const queryLabel = ({ mode, params }: RecentQuery): string => {
- if (mode === 'msgid') return `Message ID: ${params.msgId || '全部'}`;
+ if (mode === 'msgid')
+ return `Message ID: ${params.msgId || '全部'} · Topic: ${params.topic || '全部'}`;
if (mode === 'key') {
return `Key: ${params.key || '全部'}${params.topic ? ` · Topic: ${params.topic}` : ''}`;
}
@@ -318,7 +321,7 @@ const MessagePage = () => {
}
: queryMode === 'key'
? { topic: selectedTopic, key: keyInput || undefined }
- : { msgId: msgIdInput || undefined };
+ : { topic: selectedTopic, msgId: msgIdInput || undefined };
await executeQuery(queryMode, params);
};
@@ -749,12 +752,26 @@ const MessagePage = () => {
)}
{queryMode === 'msgid' && (
- <Input
- placeholder="输入 Message ID"
- style={{ width: 400 }}
- value={msgIdInput}
- onChange={(e) => setMsgIdInput(e.target.value)}
- />
+ <>
+ <Select
+ placeholder="选择 Topic"
+ style={{ width: 360 }}
+ value={selectedTopic}
+ onChange={setSelectedTopic}
+ allowClear
+ showSearch
+ options={topicOptions.map((t) => ({
+ value: t,
+ label: t,
+ }))}
+ />
+ <Input
+ placeholder="输入 Message ID"
+ style={{ width: 400 }}
+ value={msgIdInput}
+ onChange={(e) => setMsgIdInput(e.target.value)}
+ />
+ </>
)}
<Button
diff --git a/web/src/pages/studio/Producer.tsx b/web/src/pages/studio/Producer.tsx
index e0cfc486..132ccd37 100644
--- a/web/src/pages/studio/Producer.tsx
+++ b/web/src/pages/studio/Producer.tsx
@@ -60,10 +60,26 @@ const ProducerPage = () => {
};
}, []);
+ const handleInstanceChange = (instanceId: string) => {
+ setSelectedInstanceId(instanceId);
+ setTopicList([]);
+ setProducerGroups([]);
+ setConnectionList([]);
+ form.setFieldsValue({ selectedTopic: undefined, producerGroup: undefined });
+ };
+
useEffect(() => {
let cancelled = false;
- void fetchTopicList()
+ if (!selectedInstanceId) {
+ return () => {
+ cancelled = true;
+ };
+ }
+
+ form.setFieldsValue({ selectedTopic: undefined, producerGroup: undefined });
+
+ void fetchTopicList(selectedInstanceId)
.then((topics) => {
if (!cancelled) {
setTopicList(topics);
@@ -78,7 +94,7 @@ const ProducerPage = () => {
return () => {
cancelled = true;
};
- }, [fetchTopicFailedMessage, message]);
+ }, [fetchTopicFailedMessage, form, message, selectedInstanceId]);
useEffect(() => {
let cancelled = false;
@@ -170,7 +186,7 @@ const ProducerPage = () => {
<Select
aria-label="Instance"
value={selectedInstanceId || undefined}
- onChange={setSelectedInstanceId}
+ onChange={handleInstanceChange}
placeholder="Select instance"
style={{ width: 220 }}
options={instances.map((instance) => ({ value: instance.id, label: instance.name }))}
diff --git a/web/src/pages/studio/__tests__/Producer.test.tsx b/web/src/pages/studio/__tests__/Producer.test.tsx
index 8d43bee5..5bc65599 100644
--- a/web/src/pages/studio/__tests__/Producer.test.tsx
+++ b/web/src/pages/studio/__tests__/Producer.test.tsx
@@ -16,7 +16,7 @@
*/
import { beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
-import { fireEvent, render, screen, waitFor } from '@testing-library/react';
+import { fireEvent, render, screen, waitFor, within } from '@testing-library/react';
import userEvent from '@testing-library/user-event';
import { App } from 'antd';
import { LangProvider } from '../../../i18n/LangContext';
@@ -87,7 +87,7 @@ describe('ProducerPage', () => {
renderWithProviders(<ProducerPage />);
await waitFor(() => {
- expect(fetchTopicList).toHaveBeenCalledTimes(1);
+ expect(fetchTopicList).toHaveBeenCalledWith('instance-1');
});
});
@@ -186,4 +186,64 @@ describe('ProducerPage', () => {
);
});
});
+
+ it('clears stale producer query state before loading a new instance', async () => {
+ vi.mocked(listInstances).mockResolvedValue([
+ {
+ id: 'instance-1',
+ name: 'Primary instance',
+ remark: '',
+ type: 'DIRECT',
+ endpoint: '127.0.0.1:9876',
+ topicCount: 0,
+ consumerGroupCount: 0,
+ createdAt: '2026-08-01T00:00:00',
+ updatedAt: '2026-08-01T00:00:00',
+ },
+ {
+ id: 'instance-2',
+ name: 'Secondary instance',
+ remark: '',
+ type: 'DIRECT',
+ endpoint: '127.0.0.2:9876',
+ topicCount: 0,
+ consumerGroupCount: 0,
+ createdAt: '2026-08-01T00:00:00',
+ updatedAt: '2026-08-01T00:00:00',
+ },
+ ]);
+ vi.mocked(fetchTopicList).mockResolvedValueOnce(['order-events']).mockResolvedValueOnce([]);
+ vi.mocked(queryProducerConnection).mockResolvedValue([
+ {
+ clientId: 'producer-1',
+ clientAddr: '192.168.1.10',
+ language: 'JAVA',
+ versionDesc: '5.1.0',
+ },
+ ]);
+ const user = userEvent.setup();
+ const { container } = renderWithProviders(<ProducerPage />);
+
+ await waitFor(() => expect(fetchTopicList).toHaveBeenCalledWith('instance-1'));
+ const [instanceSelect, topicSelect, groupInput] = screen.getAllByRole('combobox');
+ fireEvent.mouseDown(topicSelect.parentElement!);
+ await user.click(
+ await screen.findByText('order-events', { selector: '.ant-select-item-option-content' }),
+ );
+ await user.type(groupInput, 'order-producer');
+ await user.click(screen.getByRole('button', { name: /搜索/ }));
+ expect(await screen.findByText('producer-1')).toBeInTheDocument();
+
+ fireEvent.mouseDown(instanceSelect.parentElement!);
+ await user.click(
+ await screen.findByText('Secondary instance', {
+ selector: '.ant-select-item-option-content',
+ }),
+ );
+
+ await waitFor(() => expect(fetchTopicList).toHaveBeenLastCalledWith('instance-2'));
+ // scope to the page container: antd keeps closed dropdown portals in document.body
+ expect(within(container).queryByText('producer-1')).not.toBeInTheDocument();
+ expect(within(container).queryByText('order-events')).not.toBeInTheDocument();
+ });
});
(rocketmq-dashboard) 04/06: fix: scope consumer group diagnostics by selected instance (#1248)
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
commit 06b1d933c6378b6b1e0226b29586218b90b82e1f
Author: aias00 <li...@apache.org>
AuthorDate: Mon Aug 10 14:17:13 2026 +0800
fix: scope consumer group diagnostics by selected instance (#1248)
* [ISSUE #1247] Scope Consumer Group list enrichment by instance
* [ISSUE #1249] Surface Topic consumer diagnostics failures
* [ISSUE #1251] Surface Consumer Group subscription failures
* fix: scope consumer diagnostics caches by instance
Signed-off-by: liuhy <li...@apache.org>
* fix: stop resource loading without an instance
Signed-off-by: liuhy <li...@apache.org>
* fix: preserve selected instance for group details
---------
Signed-off-by: liuhy <li...@apache.org>
---
.../provider/apache/ApacheInstanceProvider.java | 4 +-
.../studio/provider/apache/MetadataProvider.java | 5 ++
.../provider/apache/RocketMQMetadataProvider.java | 31 +++++++++---
.../apache/ApacheInstanceProviderTest.java | 10 ++++
.../apache/RocketMQMetadataProviderTest.java | 39 +++++++++++++++
.../pages/instance/__tests__/ConsumerPage.test.tsx | 53 +++++++++++++++++++++
.../pages/instance/__tests__/TopicPage.test.tsx | 2 +
web/src/pages/instance/consumer.tsx | 55 +++++++++++++++-------
web/src/pages/instance/topic.tsx | 4 ++
web/src/services/consumerService.test.ts | 25 +++++++++-
web/src/services/consumerService.ts | 7 ++-
11 files changed, 205 insertions(+), 30 deletions(-)
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 eb1faa50..045f1564 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
@@ -90,9 +90,7 @@ public class ApacheInstanceProvider implements InstanceProvider {
@Override
public List<ConsumerGroupVO> listConsumerGroups(String instanceId, String search) {
- return metadataProvider.listConsumerGroups(null, search).stream()
- .filter(group -> matchesInstance(group.getInstanceId(), instanceId))
- .toList();
+ return metadataProvider.listConsumerGroups(instanceId, null, search);
}
@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 9acfe40e..7d243478 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,6 +28,11 @@ import java.util.List;
public interface MetadataProvider {
List<TopicVO> listTopics(String clusterId, String type, String search);
List<ConsumerGroupVO> listConsumerGroups(String clusterId, String search);
+
+ default List<ConsumerGroupVO> listConsumerGroups(String instanceId, String clusterId, String search) {
+ return listConsumerGroups(clusterId, search);
+ }
+
List<BrokerRouteVO> getTopicRoutes(String instanceId, String name);
List<TopicConsumerVO> getTopicConsumers(String instanceId, String name);
List<QueueProgressVO> getGroupProgress(String instanceId, 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 f63a1128..992644d4 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
@@ -156,7 +156,13 @@ public class RocketMQMetadataProvider implements MetadataProvider {
@Override
public List<ConsumerGroupVO> listConsumerGroups(String clusterId, String search) {
+ return listConsumerGroups(null, clusterId, search);
+ }
+
+ @Override
+ public List<ConsumerGroupVO> listConsumerGroups(String instanceId, String clusterId, String search) {
LambdaQueryWrapper<RmqGroup> query = new LambdaQueryWrapper<RmqGroup>()
+ .eq(StringUtils.hasText(instanceId), RmqGroup::getInstanceId, instanceId)
.eq(StringUtils.hasText(clusterId), RmqGroup::getClusterId, clusterId)
.like(StringUtils.hasText(search), RmqGroup::getName, search)
.orderByAsc(RmqGroup::getName);
@@ -175,8 +181,10 @@ public class RocketMQMetadataProvider implements MetadataProvider {
vo.setCreatedAt(entity.getCreatedAt());
vo.setUpdatedAt(entity.getUpdatedAt());
- if (hasAdmin()) {
- enrichGroupWithConnectionInfo(vo, entity.getName());
+ if (StringUtils.hasText(instanceId)) {
+ enrichGroupWithConnectionInfo(vo, entity.getName(), instanceId);
+ } else if (hasAdmin()) {
+ enrichGroupWithConnectionInfo(vo, entity.getName(), null);
}
result.add(vo);
}
@@ -321,7 +329,7 @@ public class RocketMQMetadataProvider implements MetadataProvider {
return consumers;
} catch (Exception e) {
log.warn("Failed to get consumers for topic {}: {}", name, e.getMessage());
- return Collections.emptyList();
+ throw new BusinessException(502, "Failed to get consumers for topic " + name + ": " + e.getMessage());
}
}
@@ -412,15 +420,17 @@ public class RocketMQMetadataProvider implements MetadataProvider {
return subscriptions;
} catch (Exception e) {
log.warn("Failed to get subscriptions for group {}: {}", name, e.getMessage());
- return Collections.emptyList();
+ throw new BusinessException(502,
+ "Failed to get subscriptions for group " + name + ": " + e.getMessage());
}
}
// ── Helper methods ──────────────────────────────────────────────────
- private void enrichGroupWithConnectionInfo(ConsumerGroupVO vo, String groupName) {
+ private void enrichGroupWithConnectionInfo(ConsumerGroupVO vo, String groupName, String instanceId) {
try {
- ConsumerConnection conn = adminExecute(admin -> admin.examineConsumerConnectionInfo(groupName));
+ ConsumerConnection conn = executeForInstance(instanceId,
+ admin -> admin.examineConsumerConnectionInfo(groupName));
if (conn != null) {
if (conn.getConnectionSet() != null) {
vo.setOnlineInstances(conn.getConnectionSet().size());
@@ -435,7 +445,7 @@ public class RocketMQMetadataProvider implements MetadataProvider {
// Try to get lag info
try {
- ConsumeStats stats = adminExecute(admin -> admin.examineConsumeStats(groupName));
+ ConsumeStats stats = executeForInstance(instanceId, admin -> admin.examineConsumeStats(groupName));
if (stats != null && stats.getOffsetTable() != null) {
long totalLag = 0;
for (OffsetWrapper ow : stats.getOffsetTable().values()) {
@@ -448,6 +458,13 @@ public class RocketMQMetadataProvider implements MetadataProvider {
}
}
+ private <T> T executeForInstance(String instanceId, MqAdminExtFactory.AdminAction<T> action) {
+ if (StringUtils.hasText(instanceId)) {
+ return runtimeAdminClientResolver.execute(instanceId, action);
+ }
+ return adminExecute(action);
+ }
+
private String filterMode(String expressionType) {
if ("SQL92".equals(expressionType)) {
return "SQL";
diff --git a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProviderTest.java b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProviderTest.java
index 62db28a8..c0fa9b78 100644
--- a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProviderTest.java
+++ b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProviderTest.java
@@ -26,6 +26,7 @@ import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
@@ -64,4 +65,13 @@ class ApacheInstanceProviderTest {
assertThat(provider.countGroups("inst-1")).isEqualTo(2);
}
+
+ @Test
+ void listConsumerGroupsShouldPassTheSelectedInstanceToMetadataProvider() {
+ when(metadataProvider.listConsumerGroups("inst-1", null, "orders")).thenReturn(java.util.List.of());
+
+ assertThat(provider.listConsumerGroups("inst-1", "orders")).isEmpty();
+
+ verify(metadataProvider).listConsumerGroups("inst-1", null, "orders");
+ }
}
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 19459427..41def9bc 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
@@ -44,6 +44,7 @@ import static org.mockito.Mockito.lenient;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -102,6 +103,21 @@ class RocketMQMetadataProviderTest {
assertThat(groups.get(0).getConsumeType()).isEqualTo(ConsumeType.CLUSTERING);
}
+ @Test
+ void listConsumerGroupsShouldUseSelectedInstanceForRuntimeEnrichment() {
+ RmqGroup entity = new RmqGroup();
+ entity.setName("group-a");
+ entity.setInstanceId("instance-a");
+ entity.setClusterId("cluster-a");
+ when(groupMapper.selectList(any())).thenReturn(List.of(entity));
+ when(runtimeAdminClientResolver.execute(eq("instance-a"), any())).thenReturn(null);
+
+ List<ConsumerGroupVO> groups = newProvider().listConsumerGroups("instance-a", null, null);
+
+ assertThat(groups).singleElement().extracting(ConsumerGroupVO::getName).isEqualTo("group-a");
+ verify(runtimeAdminClientResolver, times(2)).execute(eq("instance-a"), any());
+ }
+
@Test
void getTopicRoutesShouldUseSelectedInstanceRuntimeClient() {
List<BrokerRouteVO> routes = List.of(BrokerRouteVO.builder().brokerName("broker-a").build());
@@ -147,6 +163,17 @@ class RocketMQMetadataProviderTest {
.satisfies(error -> assertThat(((BusinessException) error).getCode()).isEqualTo(502));
}
+ @Test
+ void getTopicConsumersSurfacesAdminFailure() throws Exception {
+ DefaultMQAdminExt admin = org.mockito.Mockito.mock(DefaultMQAdminExt.class);
+ when(admin.queryTopicConsumeByWho("TopicA")).thenThrow(new IllegalStateException("broker unavailable"));
+
+ assertThatThrownBy(() -> newLiveProvider(admin).getTopicConsumers(null, "TopicA"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Failed to get consumers for topic TopicA: broker unavailable")
+ .satisfies(error -> assertThat(((BusinessException) error).getCode()).isEqualTo(502));
+ }
+
@Test
void getGroupProgressSurfacesAdminFailure() throws Exception {
DefaultMQAdminExt admin = org.mockito.Mockito.mock(DefaultMQAdminExt.class);
@@ -158,6 +185,18 @@ class RocketMQMetadataProviderTest {
.satisfies(error -> assertThat(((BusinessException) error).getCode()).isEqualTo(502));
}
+ @Test
+ void getGroupSubscriptionsSurfacesAdminFailure() throws Exception {
+ DefaultMQAdminExt admin = org.mockito.Mockito.mock(DefaultMQAdminExt.class);
+ when(admin.examineConsumerConnectionInfo("group-a"))
+ .thenThrow(new IllegalStateException("broker unavailable"));
+
+ assertThatThrownBy(() -> newLiveProvider(admin).getGroupSubscriptions(null, "group-a"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Failed to get subscriptions for group group-a: broker unavailable")
+ .satisfies(error -> assertThat(((BusinessException) error).getCode()).isEqualTo(502));
+ }
+
private RocketMQMetadataProvider newLiveProvider(MQAdminExt admin) throws Exception {
MqAdminExtFactory factory = mock(MqAdminExtFactory.class);
RocketMQProperties liveProperties = new RocketMQProperties();
diff --git a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
index 4bd93b8f..66605e8b 100644
--- a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
+++ b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
@@ -262,6 +262,57 @@ describe('Consumer page', () => {
);
});
+ it('reloads same-named group diagnostics after changing the selected instance', async () => {
+ vi.mocked(instanceService.listInstances).mockResolvedValue([
+ {
+ id: 'instance-a',
+ name: 'Instance A',
+ remark: '',
+ type: 'DIRECT',
+ endpoint: '127.0.0.1:9876',
+ topicCount: 0,
+ consumerGroupCount: 0,
+ createdAt: '2026-07-23T00:00:00Z',
+ updatedAt: '2026-07-23T00:00:00Z',
+ },
+ {
+ id: 'instance-b',
+ name: 'Instance B',
+ remark: '',
+ type: 'DIRECT',
+ endpoint: '127.0.0.2:9876',
+ topicCount: 0,
+ consumerGroupCount: 0,
+ createdAt: '2026-07-23T00:00:00Z',
+ updatedAt: '2026-07-23T00:00:00Z',
+ },
+ ]);
+ vi.mocked(consumerService.listConsumerGroups).mockImplementation(async (params) => [
+ { ...group, instanceId: params?.instanceId ?? 'instance-a' },
+ ]);
+ const user = userEvent.setup();
+ renderWithProviders(<ConsumerPage />, '/instance/instance-a/consumer');
+
+ await user.click(await screen.findByRole('button', { name: /详情/ }));
+ await waitFor(() =>
+ expect(consumerService.getConsumerSubscriptions).toHaveBeenCalledWith('remote-cg', 'instance-a'),
+ );
+
+ await user.click(screen.getByText('Instance A'));
+ await user.click(await screen.findByText('Instance B'));
+ await waitFor(() =>
+ expect(consumerService.listConsumerGroups).toHaveBeenCalledWith({ instanceId: 'instance-b' }),
+ );
+ await user.click(await screen.findByRole('button', { name: /详情/ }));
+
+ await waitFor(() =>
+ expect(consumerService.getConsumerSubscriptions).toHaveBeenCalledWith('remote-cg', 'instance-b'),
+ );
+ await waitFor(() =>
+ expect(consumerService.getConsumerProgress).toHaveBeenCalledWith('remote-cg', 'instance-b'),
+ );
+ });
+
it('highlights inconsistent subscriptions and refreshes the check result', async () => {
vi.mocked(consumerService.getConsumerSubscriptions)
.mockResolvedValueOnce([
@@ -421,6 +472,8 @@ describe('Consumer page', () => {
renderWithProviders(<ConsumerPage />);
expect(await screen.findByText('选择实例')).toBeInTheDocument();
+ expect(consumerService.listConsumerGroups).not.toHaveBeenCalled();
+ expect(document.querySelector('.ant-spin-spinning')).toBeNull();
expect(screen.getByRole('button', { name: /导入/ })).toBeDisabled();
expect(screen.getByRole('button', { name: '创建 Group' })).toBeDisabled();
});
diff --git a/web/src/pages/instance/__tests__/TopicPage.test.tsx b/web/src/pages/instance/__tests__/TopicPage.test.tsx
index 8713a88c..fde8eaa9 100644
--- a/web/src/pages/instance/__tests__/TopicPage.test.tsx
+++ b/web/src/pages/instance/__tests__/TopicPage.test.tsx
@@ -391,6 +391,8 @@ describe('TopicPage', () => {
renderWithProviders();
expect(await screen.findByText('共 0 个 Topic')).toBeInTheDocument();
+ expect(topicServiceMocks.listTopics).not.toHaveBeenCalled();
+ expect(document.querySelector('.ant-spin-spinning')).toBeNull();
expect(screen.getByRole('button', { name: /导入/ })).toBeDisabled();
expect(screen.getByRole('button', { name: /创建 Topic/ })).toBeDisabled();
});
diff --git a/web/src/pages/instance/consumer.tsx b/web/src/pages/instance/consumer.tsx
index b6c2b117..a11b4cba 100644
--- a/web/src/pages/instance/consumer.tsx
+++ b/web/src/pages/instance/consumer.tsx
@@ -176,6 +176,9 @@ const isConsistentSubscription = (subscription: SubscriptionEntry): boolean =>
const isInconsistentSubscription = (subscription: SubscriptionEntry): boolean =>
isInconsistentValue(subscription.consistency);
+export const diagnosticCacheKey = (instanceId: string, groupName: string) =>
+ `${instanceId}\u0000${groupName}`;
+
/* ═══════════════════════════════════════════
ConsumerPage
═══════════════════════════════════════════ */
@@ -222,8 +225,19 @@ const ConsumerPage = () => {
const groupRequestIdRef = useRef(0);
+ useEffect(() => {
+ setSelectedGroup(null);
+ setModalOpen(false);
+ setResetGroup(null);
+ setResetModalOpen(false);
+ }, [selectedInstanceId]);
+
useEffect(() => {
if (!selectedInstanceId) {
+ groupRequestIdRef.current += 1;
+ setGroups([]);
+ setSelectedRowKeys([]);
+ setLoading(false);
return;
}
const requestId = ++groupRequestIdRef.current;
@@ -247,20 +261,21 @@ const ConsumerPage = () => {
const loadSubscriptions = useCallback(
async (groupName: string, force = false) => {
- if (!force && subscriptionsByGroup[groupName]) return;
- setSubscriptionLoadingByGroup((prev) => ({ ...prev, [groupName]: true }));
- setSubscriptionErrorByGroup((prev) => ({ ...prev, [groupName]: false }));
+ const cacheKey = diagnosticCacheKey(selectedInstanceId, groupName);
+ if (!force && subscriptionsByGroup[cacheKey]) return;
+ setSubscriptionLoadingByGroup((prev) => ({ ...prev, [cacheKey]: true }));
+ setSubscriptionErrorByGroup((prev) => ({ ...prev, [cacheKey]: false }));
try {
const subscriptions = await getConsumerSubscriptions(
groupName,
selectedInstanceId || undefined,
);
- setSubscriptionsByGroup((prev) => ({ ...prev, [groupName]: subscriptions }));
+ setSubscriptionsByGroup((prev) => ({ ...prev, [cacheKey]: subscriptions }));
} catch {
- setSubscriptionErrorByGroup((prev) => ({ ...prev, [groupName]: true }));
+ setSubscriptionErrorByGroup((prev) => ({ ...prev, [cacheKey]: true }));
message.error(t('consumer.fetchSubscriptionsFailed', { name: groupName }));
} finally {
- setSubscriptionLoadingByGroup((prev) => ({ ...prev, [groupName]: false }));
+ setSubscriptionLoadingByGroup((prev) => ({ ...prev, [cacheKey]: false }));
}
},
[subscriptionsByGroup, t, selectedInstanceId],
@@ -268,10 +283,11 @@ const ConsumerPage = () => {
const loadProgress = useCallback(
async (groupName: string) => {
- if (progressByGroup[groupName]) return;
+ const cacheKey = diagnosticCacheKey(selectedInstanceId, groupName);
+ if (progressByGroup[cacheKey]) return;
try {
const progress = await getConsumerProgress(groupName, selectedInstanceId || undefined);
- setProgressByGroup((prev) => ({ ...prev, [groupName]: progress }));
+ setProgressByGroup((prev) => ({ ...prev, [cacheKey]: progress }));
} catch {
message.error(t('consumer.fetchProgressFailed', { name: groupName }));
}
@@ -311,8 +327,11 @@ const ConsumerPage = () => {
void loadProgress(group.name);
};
+ const selectedDiagnosticKey = selectedGroup
+ ? diagnosticCacheKey(selectedInstanceId, selectedGroup.name)
+ : '';
const selectedSubscriptions = selectedGroup
- ? (subscriptionsByGroup[selectedGroup.name] ?? [])
+ ? (subscriptionsByGroup[selectedDiagnosticKey] ?? [])
: [];
const inconsistentSubscriptions = selectedSubscriptions.filter(isInconsistentSubscription);
const unknownSubscriptions = selectedSubscriptions.filter(
@@ -322,7 +341,7 @@ const ConsumerPage = () => {
const visibleSubscriptions = showOnlyInconsistent
? inconsistentSubscriptions
: selectedSubscriptions;
- const selectedProgress = selectedGroup ? (progressByGroup[selectedGroup.name] ?? []) : [];
+ const selectedProgress = selectedGroup ? (progressByGroup[selectedDiagnosticKey] ?? []) : [];
const handleImportFile = async (file: File) => {
if (!selectedInstanceId) {
@@ -895,9 +914,11 @@ const ConsumerPage = () => {
<div style={{ padding: '8px 0' }}>
<Table
columns={subscriptionSubColumns}
- dataSource={subscriptionsByGroup[record.name] ?? []}
+ dataSource={
+ subscriptionsByGroup[diagnosticCacheKey(selectedInstanceId, record.name)] ?? []
+ }
rowKey="topic"
- loading={subscriptionLoadingByGroup[record.name]}
+ loading={subscriptionLoadingByGroup[diagnosticCacheKey(selectedInstanceId, record.name)]}
pagination={false}
size="small"
/>
@@ -1073,7 +1094,7 @@ const ConsumerPage = () => {
<Button
size="small"
icon={<ArrowsClockwise size={14} />}
- loading={subscriptionLoadingByGroup[selectedGroup.name]}
+ loading={subscriptionLoadingByGroup[selectedDiagnosticKey]}
onClick={() => {
setShowOnlyInconsistent(false);
void loadSubscriptions(selectedGroup.name, true);
@@ -1085,7 +1106,7 @@ const ConsumerPage = () => {
<Alert
showIcon
type={
- subscriptionErrorByGroup[selectedGroup.name]
+ subscriptionErrorByGroup[selectedDiagnosticKey]
? 'error'
: inconsistentSubscriptions.length > 0 ||
unknownSubscriptions.length > 0
@@ -1095,9 +1116,9 @@ const ConsumerPage = () => {
: 'info'
}
message={
- subscriptionErrorByGroup[selectedGroup.name]
+ subscriptionErrorByGroup[selectedDiagnosticKey]
? '订阅一致性检查失败，当前保留上次检查结果'
- : subscriptionLoadingByGroup[selectedGroup.name] &&
+ : subscriptionLoadingByGroup[selectedDiagnosticKey] &&
selectedSubscriptions.length === 0
? '正在检查订阅一致性'
: inconsistentSubscriptions.length > 0
@@ -1123,7 +1144,7 @@ const ConsumerPage = () => {
columns={subscriptionSubColumns}
dataSource={visibleSubscriptions}
rowKey="topic"
- loading={subscriptionLoadingByGroup[selectedGroup.name]}
+ loading={subscriptionLoadingByGroup[selectedDiagnosticKey]}
pagination={false}
size="small"
/>
diff --git a/web/src/pages/instance/topic.tsx b/web/src/pages/instance/topic.tsx
index 19c05dd6..64f74386 100644
--- a/web/src/pages/instance/topic.tsx
+++ b/web/src/pages/instance/topic.tsx
@@ -307,6 +307,10 @@ const TopicPage = () => {
useEffect(() => {
if (!selectedInstanceId) {
+ topicRequestIdRef.current += 1;
+ setTopics([]);
+ setSelectedRowKeys([]);
+ setLoading(false);
return;
}
const requestId = ++topicRequestIdRef.current;
diff --git a/web/src/services/consumerService.test.ts b/web/src/services/consumerService.test.ts
index 4d61b45d..27ef23b2 100644
--- a/web/src/services/consumerService.test.ts
+++ b/web/src/services/consumerService.test.ts
@@ -24,10 +24,16 @@ import {
listConsumerGroups,
} from './consumerService';
-vi.mock('./dataMode', () => ({ isMockMode: () => true }));
+const { mode, metadataApi } = vi.hoisted(() => ({
+ mode: { mock: true },
+ metadataApi: { getConsumerGroup: vi.fn() },
+}));
+
+vi.mock('./dataMode', () => ({ isMockMode: () => mode.mock }));
vi.mock('../config', () => ({
API_BASE_URL: '/api',
}));
+vi.mock('../api/metadata', () => metadataApi);
describe('consumer service mock data', () => {
it('returns copied consumer group rows', async () => {
@@ -93,4 +99,21 @@ describe('consumer service mock data', () => {
expect(detail.subscribedTopics).toEqual(['created-topic']);
expect(detail).not.toBe(created);
});
+
+ it('forwards the selected instance when loading consumer group details in API mode', async () => {
+ mode.mock = false;
+ const detail = {
+ name: 'cg-orders',
+ subscribedTopics: [],
+ instances: [],
+ };
+ try {
+ metadataApi.getConsumerGroup.mockResolvedValue(detail);
+
+ await expect(getConsumerGroup('cg-orders', 'instance-a')).resolves.toEqual(detail);
+ expect(metadataApi.getConsumerGroup).toHaveBeenCalledWith('cg-orders', 'instance-a');
+ } finally {
+ mode.mock = true;
+ }
+ });
});
diff --git a/web/src/services/consumerService.ts b/web/src/services/consumerService.ts
index b48f209a..dfabd684 100644
--- a/web/src/services/consumerService.ts
+++ b/web/src/services/consumerService.ts
@@ -61,13 +61,16 @@ export async function getConsumerProgress(
return metadataApi.getConsumerProgress(name, instanceId);
}
-export async function getConsumerGroup(name: string): Promise<ConsumerGroupDetail> {
+export async function getConsumerGroup(
+ name: string,
+ instanceId?: string,
+): Promise<ConsumerGroupDetail> {
if (isMockMode()) {
const group = mockConsumerGroups.find((item) => item.name === name);
if (!group) throw new Error(`Consumer group not found: ${name}`);
return copyConsumerGroup(group as unknown as ConsumerGroupDetail) as ConsumerGroupDetail;
}
- return metadataApi.getConsumerGroup(name);
+ return metadataApi.getConsumerGroup(name, instanceId);
}
export async function getConsumerSubscriptions(
(rocketmq-dashboard) 06/06: fix(ui): normalize ACL version to a string for filtering and coloring (#1259)
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
commit d58b46358e300c13fe6cef08e5fe170ef0763add
Author: 0 <19...@qq.com>
AuthorDate: Mon Aug 10 14:27:14 2026 +0800
fix(ui): normalize ACL version to a string for filtering and coloring (#1259)
---
web/src/pages/instance/acl.tsx | 4 +++-
1 file changed, 3 insertions(+), 1 deletion(-)
diff --git a/web/src/pages/instance/acl.tsx b/web/src/pages/instance/acl.tsx
index 374320c0..186ceba2 100644
--- a/web/src/pages/instance/acl.tsx
+++ b/web/src/pages/instance/acl.tsx
@@ -70,7 +70,9 @@ const normalizeRule = (rule: AclRule): AclRule => ({
actions: rule.actions ?? [],
decision: rule.decision ?? '',
scope: rule.scope ?? '',
- aclVersion: rule.aclVersion ?? '2.0',
+ // Normalize to a string so version filters and tag coloring work whether the backend
+ // returns a number (2.0) or a string ("2.0").
+ aclVersion: String(rule.aclVersion ?? '2.0'),
createdAt: rule.createdAt ?? null,
});
(rocketmq-dashboard) 05/06: fix(ui): drop misleading selected count on alert export button (#1258)
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
commit 15b98d6558be98dfe4e35a25788224a606edd41e
Author: 0 <19...@qq.com>
AuthorDate: Mon Aug 10 14:22:09 2026 +0800
fix(ui): drop misleading selected count on alert export button (#1258)
* fix(ui): drop misleading selected count on alert export button
* test: drop selected count from alert export button assertions
#1258 removes the count suffix from the export button; update the expectations accordingly.
---------
Co-authored-by: lizhimins <70...@qq.com>
---
web/src/pages/studio/AlertManagement.tsx | 1 -
web/src/pages/studio/__tests__/AlertManagement.test.tsx | 6 +++---
2 files changed, 3 insertions(+), 4 deletions(-)
diff --git a/web/src/pages/studio/AlertManagement.tsx b/web/src/pages/studio/AlertManagement.tsx
index aea6b06a..7b14add3 100644
--- a/web/src/pages/studio/AlertManagement.tsx
+++ b/web/src/pages/studio/AlertManagement.tsx
@@ -469,7 +469,6 @@ const AlertManagementPage: React.FC = () => {
</Button>
<Button icon={<DownloadSimple size={16} />} onClick={handleExportYaml} size="small">
{t('alertMgmt.exportYaml')}
- {selectedRules.length > 0 ? ` (${selectedRules.length})` : ''}
</Button>
</Space>
}
diff --git a/web/src/pages/studio/__tests__/AlertManagement.test.tsx b/web/src/pages/studio/__tests__/AlertManagement.test.tsx
index 98a078a2..97992fb3 100644
--- a/web/src/pages/studio/__tests__/AlertManagement.test.tsx
+++ b/web/src/pages/studio/__tests__/AlertManagement.test.tsx
@@ -124,7 +124,7 @@ describe('AlertManagementPage', () => {
const brokerRow = brokerRule.closest('tr');
expect(brokerRow).not.toBeNull();
await user.click(within(brokerRow!).getByRole('checkbox'));
- await user.click(screen.getByRole('button', { name: '导出 YAML (1)' }));
+ await user.click(screen.getByRole('button', { name: '导出 YAML' }));
await waitFor(() => {
expect(queryAlertRules).toHaveBeenCalledTimes(2);
@@ -159,7 +159,7 @@ describe('AlertManagementPage', () => {
expect(brokerRow).not.toBeNull();
await user.click(within(brokerRow!).getByRole('checkbox'));
- expect(screen.getByRole('button', { name: '导出 YAML (1)' })).toBeInTheDocument();
+ expect(screen.getByRole('button', { name: '导出 YAML' })).toBeInTheDocument();
await user.click(within(brokerRow!).getByRole('switch'));
@@ -184,7 +184,7 @@ describe('AlertManagementPage', () => {
expect(screen.queryByText('BrokerDown')).not.toBeInTheDocument();
expect(screen.getByText('ConsumerLagHigh')).toBeInTheDocument();
- await user.click(screen.getByRole('button', { name: '导出 YAML (1)' }));
+ await user.click(screen.getByRole('button', { name: '导出 YAML' }));
const blob = createObjectURL.mock.calls[0][0] as Blob;
const yaml = await blob.text();