This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new ace3fb91 fix: instance-scoped hardening, mutation auditing and
validation (#1274, #1280, #1282, #1288, #1293, #1296, #1303, #1305, #1307,
#1309, #1311, #1319)
ace3fb91 is described below
commit ace3fb9100631e4e2fb49d2292f4a94b15f99806
Author: lizhimins <[email protected]>
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 <[email protected]>
* fix: surface message trace query failures
Signed-off-by: liuhy <[email protected]>
* 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 <[email protected]>
* fix: handle client instance bootstrap failures
Signed-off-by: liuhy <[email protected]>
* fix: scope producer topics by selected instance
Signed-off-by: liuhy <[email protected]>
* fix: require instance IDs for client queries
* fix: ignore stale dashboard responses
Signed-off-by: liuhy <[email protected]>
* test: cover dashboard instance response ordering
Signed-off-by: liuhy <[email protected]>
* 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 <[email protected]>
Co-authored-by: liuhy <[email protected]>
---
.../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('[email protected]: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();
+ });
});
