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 301008d62 fix: auth concurrency, history columns, metrics histograms,
resource name rules, registry probes, LiteTopic quotas (#2516)
301008d62 is described below
commit 301008d62a1c6cb7c6793dffea6bafd5b07ac919
Author: btlqql <[email protected]>
AuthorDate: Sat Aug 22 16:44:01 2026 +0800
fix: auth concurrency, history columns, metrics histograms, resource name
rules, registry probes, LiteTopic quotas (#2516)
* fix(auth): serialize concurrent administrator disables
The check-then-update in setUserEnabled let two concurrent disables of
different administrators both observe an enabled-admin count of 2 and disable
everyone, leaving the console without an administrator.
Lock the enabled administrator rows (SELECT ... FOR UPDATE) inside a
transaction before deciding, so concurrent disables serialize: the winner
disables its target, the loser re-reads a single remaining administrator and
fails with a 409. The stale-read case where the target was already disabled
concurrently no longer trips the last-admin guard.
Fixes #2484
* fix(history): align query-history owner columns with username limits
Usernames may be up to 128 characters, but the queried_by columns of
rmq_instance_message and rmq_instance_trace were VARCHAR(64). Queries made by
users with 65-128 character names failed the history insert, and MessageService
swallowed the exception, so the business query succeeded while the history row
was silently lost.
Widen both columns to VARCHAR(128) and append idempotent MODIFY upgrades so
pre-existing databases are upgraded when the DDL is re-applied. The
rmq_operation_audit.operator column stores the same username and has the
identical gap, so it is widened to VARCHAR(128) in the same way. The service
layer keeps recording the full username.
Fixes #2491
* fix(metrics): render native histogram samples in Metrics Explorer
The chart only read series.values, so a query that returned native
histograms with no scalar samples was plotted as the empty state even though
data existed.
Derive a trend value per histogram - the observed sum in the metric's unit,
falling back to the observation count when the sum is blank or non-finite - and
mark such series with a histogram tag in the legend. Scalar samples still take
precedence when both are present. The tag's tooltip is localized into the zh/en
copy instead of hardcoded English.
Fixes #2492
* fix(csv): align resource name validation with RocketMQ rules
The CSV importer accepted / and * in topic names while rejecting the % and
| characters RocketMQ actually supports, forced consumer group names to start
with a letter, and applied no length limit at all.
Introduce a shared validateResourceName used by both the topic and group
CSV imports: the RocketMQ character set [%|a-zA-Z0-9_-] with the Topic cap of
127 and the group cap of 120. The inline create forms reference the same
pattern and caps so the UI and the importer agree. The unreferenced translation
key topic.topicNameRule is removed, since the create forms use inline rule
messages instead.
Fixes #2493
* fix(cluster): bound timed-out NameServer registry probes
listRegistryClusters fanned probes out onto an unbounded cached thread
pool. CompletableFuture.orTimeout only completed the wrapper future, so a probe
blocked on an unreachable NameServer kept its thread, and every refresh
accumulated more blocked threads.
Replace the pool with a bounded, closeable RegistryProbeRunner: a
fixed-size worker pool with a bounded queue, explicit cancellation (worker
interrupt) when the probe deadline is reached, and a saturated queue that
degrades the offending entry to unavailable instead of throwing. The runner is
closed on context shutdown. The admin clients it drives already carry a 5s RPC
timeout (MqAdminExtFactory), so released workers do not pile up behind hung
calls.
Fixes #2472
* fix(topic): normalize unavailable LiteTopic quotas
A quota with a zero or negative maxTopicCount/maxSessionCount represents an
unavailable feature, not an exhausted one: isQuotaExceeded reported such quotas
as exceeded (current >= 0 max), and getRemainingQuota only guarded against null
maxes.
Treat non-positive limits the same as unset limits everywhere: usage rates
report 0, isQuotaExceeded requires a positive max, and getRemainingQuota
returns 0. The console then shows "unavailable" instead of "quota exceeded" for
LiteTopic plans that are not enabled.
---
.../apache/rocketmq/studio/auth/AuthService.java | 22 ++--
.../studio/cluster/broker/ClusterService.java | 49 ++++----
.../studio/cluster/broker/RegistryProbeRunner.java | 139 +++++++++++++++++++++
.../rocketmq/studio/model/LiteTopicQuota.java | 8 +-
server/src/main/resources/db/schema.sql | 15 ++-
.../AuthServiceConcurrencyIntegrationTest.java | 124 ++++++++++++++++++
.../studio/auth/AuthServiceDatabaseTest.java | 30 ++++-
.../cluster/broker/ClusterServiceRegistryTest.java | 72 +++++++++++
.../QueryHistoryServiceIntegrationTest.java | 73 +++++++++++
.../rocketmq/studio/model/LiteTopicQuotaTest.java | 24 +++-
web/src/components/MetricsExplorer.tsx | 84 +++++++++++--
.../components/__tests__/MetricsExplorer.test.tsx | 76 ++++++++++-
web/src/i18n/translations.ts | 4 -
web/src/pages/instance/consumer.tsx | 10 +-
web/src/pages/instance/topic.tsx | 10 +-
web/src/utils/resourceCsvImport.test.ts | 42 +++++++
web/src/utils/resourceCsvImport.ts | 38 ++++--
17 files changed, 745 insertions(+), 75 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/auth/AuthService.java
b/server/src/main/java/org/apache/rocketmq/studio/auth/AuthService.java
index d5a67f4ba..21ae0d07a 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/auth/AuthService.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/auth/AuthService.java
@@ -31,6 +31,7 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
@@ -184,11 +185,22 @@ public class AuthService {
return user;
}
+ @Transactional
public RmqStudioUser setUserEnabled(Long userId, boolean enabled) {
requireDatabaseBacked();
RmqStudioUser user = getUser(userId);
- if (!enabled && Boolean.TRUE.equals(user.getAdmin()) &&
enabledAdminCount() <= 1) {
- throw new BusinessException(409, "The last enabled administrator
cannot be disabled");
+ if (!enabled && Boolean.TRUE.equals(user.getAdmin())) {
+ // Lock the enabled administrator rows so concurrent disables
serialize. A plain
+ // count would let two requests both observe a count of 2 and
disable everyone.
+ List<RmqStudioUser> enabledAdmins = userMapper.selectList(new
QueryWrapper<RmqStudioUser>()
+ .eq("admin", true)
+ .eq("enabled", true)
+ .last("FOR UPDATE"));
+ boolean targetStillEnabled = enabledAdmins.stream()
+ .anyMatch(admin -> userId.equals(admin.getId()));
+ if (targetStillEnabled && enabledAdmins.size() <= 1) {
+ throw new BusinessException(409, "The last enabled
administrator cannot be disabled");
+ }
}
userMapper.updateById(userWithEnabled(user, enabled));
if (!enabled) {
@@ -339,12 +351,6 @@ public class AuthService {
return update;
}
- private long enabledAdminCount() {
- return userMapper.selectCount(new QueryWrapper<RmqStudioUser>()
- .eq("admin", true)
- .eq("enabled", true));
- }
-
private void revokeUserSessions(Long userId) {
sessionMapper.update(null, new UpdateWrapper<RmqStudioSession>()
.eq("user_id", userId)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterService.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterService.java
index a527b24f6..e7566deb0 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterService.java
@@ -35,6 +35,7 @@ import
org.apache.rocketmq.studio.common.domain.enums.FlushDiskType;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.ops.audit.AuditService;
import org.apache.rocketmq.studio.provider.apache.RocketMQBrokerConfigService;
+import jakarta.annotation.PreDestroy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
@@ -42,10 +43,6 @@ import org.springframework.stereotype.Service;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
@Slf4j
@Service
@@ -53,12 +50,8 @@ import java.util.concurrent.TimeUnit;
public class ClusterService {
private static final long REGISTRY_PROBE_TIMEOUT_SECONDS = 15;
-
- private static final ExecutorService REGISTRY_PROBE_EXECUTOR =
Executors.newCachedThreadPool(runnable -> {
- Thread thread = new Thread(runnable, "nameserver-registry-probe");
- thread.setDaemon(true);
- return thread;
- });
+ private static final int REGISTRY_PROBE_MAX_CONCURRENCY = 8;
+ private static final int REGISTRY_PROBE_QUEUE_CAPACITY = 32;
private final ClusterRepository clusterRepository;
private final ClusterProvider clusterProvider;
@@ -66,6 +59,20 @@ public class ClusterService {
private final AuditService auditService;
private final NameserverRegistryService registryService;
+ // Bounded so blocked probes cannot accumulate threads; replaceable in
unit tests.
+ private RegistryProbeRunner registryProbeRunner = new RegistryProbeRunner(
+ REGISTRY_PROBE_MAX_CONCURRENCY, REGISTRY_PROBE_QUEUE_CAPACITY,
+ REGISTRY_PROBE_TIMEOUT_SECONDS * 1000L);
+
+ void setRegistryProbeRunner(RegistryProbeRunner registryProbeRunner) {
+ this.registryProbeRunner = registryProbeRunner;
+ }
+
+ @PreDestroy
+ void closeRegistryProbeRunner() {
+ registryProbeRunner.close();
+ }
+
public List<ClusterVO> listClusters() {
log.info("Listing all clusters");
List<ClusterVO> discovered = clusterProvider.discoverClusters();
@@ -82,25 +89,13 @@ public class ClusterService {
* logged and skipped without affecting the other entries.
*/
public List<ClusterVO> listRegistryClusters() {
- List<NameserverRegistryVO> entries = registryService.list();
- if (entries.isEmpty()) {
- return List.of();
- }
- List<CompletableFuture<List<ClusterVO>>> futures = entries.stream()
+ List<NameserverRegistryVO> probeable = registryService.list().stream()
.filter(entry -> entry.getNamesrvAddr() != null &&
!entry.getNamesrvAddr().isBlank())
- .map(entry -> CompletableFuture
- .supplyAsync(() -> probeRegistryEntry(entry),
REGISTRY_PROBE_EXECUTOR)
- .orTimeout(REGISTRY_PROBE_TIMEOUT_SECONDS,
TimeUnit.SECONDS)
- .exceptionally(ex -> {
- log.warn("NameServer registry probe timed out or
failed for {} ({}): {}",
- entry.getName(), entry.getNamesrvAddr(),
ex.getMessage());
- return List.of();
- }))
- .toList();
- return futures.stream()
- .map(CompletableFuture::join)
- .flatMap(List::stream)
.toList();
+ if (probeable.isEmpty()) {
+ return List.of();
+ }
+ return registryProbeRunner.probeAll(probeable,
this::probeRegistryEntry);
}
private List<ClusterVO> probeRegistryEntry(NameserverRegistryVO entry) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/RegistryProbeRunner.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/RegistryProbeRunner.java
new file mode 100644
index 000000000..16be981a6
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/RegistryProbeRunner.java
@@ -0,0 +1,139 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.cluster.broker;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.studio.cluster.nameserver.NameserverRegistryVO;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * Bounded, closeable executor that runs NameServer registry probes
concurrently.
+ *
+ * <p>Replaces an unbounded cached thread pool so that probes blocked on an
unreachable
+ * NameServer cannot accumulate threads without limit. Each probe is cancelled
(its worker
+ * interrupted) when the deadline is reached so the pool thread is released,
and a saturated
+ * queue degrades the offending entry to unavailable instead of growing the
pool.</p>
+ */
+@Slf4j
+class RegistryProbeRunner implements AutoCloseable {
+
+ private static final long KEEP_ALIVE_SECONDS = 60L;
+
+ private final ThreadPoolExecutor executor;
+ private final long timeoutMillis;
+ private final int maxConcurrency;
+
+ RegistryProbeRunner(int maxConcurrency, int queueCapacity, long
timeoutMillis) {
+ this.maxConcurrency = maxConcurrency;
+ this.timeoutMillis = timeoutMillis;
+ this.executor = new ThreadPoolExecutor(
+ maxConcurrency,
+ maxConcurrency,
+ KEEP_ALIVE_SECONDS,
+ TimeUnit.SECONDS,
+ new LinkedBlockingQueue<>(queueCapacity),
+ runnable -> {
+ Thread thread = new Thread(runnable,
"nameserver-registry-probe");
+ thread.setDaemon(true);
+ return thread;
+ },
+ new ThreadPoolExecutor.AbortPolicy());
+ // Let idle workers (including core) time out so a quiet registry
holds no threads.
+ this.executor.allowCoreThreadTimeOut(true);
+ }
+
+ /** Probes a single registry entry; may block on the underlying admin
client. */
+ interface ProbeFunction {
+ List<ClusterVO> probe(NameserverRegistryVO entry) throws Exception;
+ }
+
+ /**
+ * Probes every entry concurrently, bounded by the pool. A rejected
(saturated), timed-out
+ * or failing entry contributes an empty result without affecting the
others.
+ */
+ List<ClusterVO> probeAll(List<NameserverRegistryVO> entries, ProbeFunction
function) {
+ return entries.stream()
+ .map(entry -> probeOne(entry, function))
+ .map(CompletableFuture::join)
+ .flatMap(List::stream)
+ .toList();
+ }
+
+ private CompletableFuture<List<ClusterVO>> probeOne(NameserverRegistryVO
entry, ProbeFunction function) {
+ CompletableFuture<List<ClusterVO>> result = new CompletableFuture<>();
+ Future<?> task;
+ try {
+ task = executor.submit(() -> {
+ try {
+ result.complete(function.probe(entry));
+ } catch (Exception exception) {
+ result.completeExceptionally(exception);
+ }
+ });
+ } catch (RejectedExecutionException rejected) {
+ // Queue saturated: degrade this entry to unavailable instead of
growing the pool.
+ log.warn("NameServer registry probe rejected for {} ({}): executor
saturated",
+ entry.getName(), entry.getNamesrvAddr());
+ return CompletableFuture.completedFuture(List.of());
+ }
+ return result
+ .orTimeout(timeoutMillis, TimeUnit.MILLISECONDS)
+ .whenComplete((value, error) -> {
+ if (error instanceof TimeoutException) {
+ // Interrupt the worker so the blocked probe releases
its pool thread.
+ task.cancel(true);
+ }
+ })
+ .exceptionally(error -> {
+ log.warn("NameServer registry probe failed for {} ({}):
{}",
+ entry.getName(), entry.getNamesrvAddr(),
rootMessage(error));
+ return List.of();
+ });
+ }
+
+ int activeCount() {
+ return executor.getActiveCount();
+ }
+
+ int poolSize() {
+ return executor.getPoolSize();
+ }
+
+ @Override
+ public void close() {
+ executor.shutdownNow();
+ }
+
+ private static String rootMessage(Throwable error) {
+ Throwable current = error;
+ while (current.getCause() != null) {
+ current = current.getCause();
+ }
+ String message = current.getMessage();
+ return message == null || message.isBlank()
+ ? current.getClass().getSimpleName()
+ : message;
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/model/LiteTopicQuota.java
b/server/src/main/java/org/apache/rocketmq/studio/model/LiteTopicQuota.java
index 107e07eda..40a1f25a4 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/model/LiteTopicQuota.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/model/LiteTopicQuota.java
@@ -38,14 +38,14 @@ public class LiteTopicQuota {
private Double maxCreationRate;
public double getUsageRate() {
- if (maxTopicCount == null || maxTopicCount == 0 || currentTopicCount
== null) {
+ if (maxTopicCount == null || maxTopicCount <= 0 || currentTopicCount
== null) {
return 0.0;
}
return (double) currentTopicCount / maxTopicCount;
}
public double getSessionUsageRate() {
- if (maxSessionCount == null || maxSessionCount == 0 ||
currentSessionCount == null) {
+ if (maxSessionCount == null || maxSessionCount <= 0 ||
currentSessionCount == null) {
return 0.0;
}
return (double) currentSessionCount / maxSessionCount;
@@ -57,12 +57,12 @@ public class LiteTopicQuota {
public boolean isQuotaExceeded() {
// Guard against unset fields, mirroring getUsageRate; an unconfigured
max is not exceeded.
- return maxTopicCount != null && currentTopicCount != null
+ return maxTopicCount != null && maxTopicCount > 0 && currentTopicCount
!= null
&& currentTopicCount >= maxTopicCount;
}
public Integer getRemainingQuota() {
- if (maxTopicCount == null) {
+ if (maxTopicCount == null || maxTopicCount <= 0) {
return 0;
}
return Math.max(0, maxTopicCount - (currentTopicCount == null ? 0 :
currentTopicCount));
diff --git a/server/src/main/resources/db/schema.sql
b/server/src/main/resources/db/schema.sql
index 8f2a67eed..9e58a7d78 100644
--- a/server/src/main/resources/db/schema.sql
+++ b/server/src/main/resources/db/schema.sql
@@ -145,7 +145,7 @@ CREATE TABLE IF NOT EXISTS rmq_instance_message (
end_time BIGINT,
result_count INT DEFAULT 0,
cluster_id VARCHAR(255),
- queried_by VARCHAR(64),
+ queried_by VARCHAR(128),
PRIMARY KEY (`id`),
INDEX idx_message_query_gmt_create (gmt_create),
INDEX idx_topic (topic)
@@ -161,7 +161,7 @@ CREATE TABLE IF NOT EXISTS rmq_instance_trace (
node_count INT DEFAULT 0,
consumer_count INT DEFAULT 0,
cluster_id VARCHAR(255),
- queried_by VARCHAR(64),
+ queried_by VARCHAR(128),
PRIMARY KEY (`id`),
INDEX idx_msg_id (msg_id),
INDEX idx_trace_query_gmt_create (gmt_create)
@@ -179,7 +179,7 @@ CREATE TABLE IF NOT EXISTS rmq_operation_audit (
detail TEXT COMMENT 'JSON: 操作详情/变更内容',
result VARCHAR(16) DEFAULT 'SUCCESS' COMMENT 'SUCCESS/FAILED',
error_message TEXT,
- operator VARCHAR(64),
+ operator VARCHAR(128),
PRIMARY KEY (`id`),
INDEX idx_gmt_create (gmt_create),
INDEX idx_resource (resource_type, resource_name),
@@ -288,3 +288,12 @@ CREATE TABLE IF NOT EXISTS rmq_cloud_credential (
PRIMARY KEY (`id`),
UNIQUE KEY uk_vendor_access_key (vendor, access_key)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+-- Idempotent upgrades for databases created before the corresponding CREATE
statements
+-- were widened. Safe to re-run: on fresh databases the columns already match,
and on
+-- existing databases the MODIFY below only grows the column width. Usernames
may be up
+-- to 128 characters (see AuthService), so columns that store the acting
username — the
+-- query-history owner columns and the audit operator — must match.
+ALTER TABLE rmq_instance_message MODIFY queried_by VARCHAR(128);
+ALTER TABLE rmq_instance_trace MODIFY queried_by VARCHAR(128);
+ALTER TABLE rmq_operation_audit MODIFY operator VARCHAR(128);
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/auth/AuthServiceConcurrencyIntegrationTest.java
b/server/src/test/java/org/apache/rocketmq/studio/auth/AuthServiceConcurrencyIntegrationTest.java
new file mode 100644
index 000000000..60a70138b
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/auth/AuthServiceConcurrencyIntegrationTest.java
@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.auth;
+
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.persistence.entity.RmqStudioUser;
+import org.apache.rocketmq.studio.persistence.mapper.RmqStudioUserMapper;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@SpringBootTest(properties = "studio.auth.login-required=true")
+class AuthServiceConcurrencyIntegrationTest {
+
+ @Autowired
+ private AuthService authService;
+
+ @Autowired
+ private RmqStudioUserMapper userMapper;
+
+ @Autowired
+ private PasswordHasher passwordHasher;
+
+ @Test
+ void concurrentDisablesKeepAtLeastOneEnabledAdministratorTest() throws
Exception {
+ // The in-memory dev database is shared across test classes in one
JVM, so park any
+ // pre-existing enabled administrators and restore them afterwards to
keep the race
+ // deterministic.
+ List<RmqStudioUser> parked = userMapper.selectList(new
QueryWrapper<RmqStudioUser>()
+ .eq("admin", true)
+ .eq("enabled", true));
+ RmqStudioUser first = adminUser("race-admin-one");
+ RmqStudioUser second = adminUser("race-admin-two");
+ userMapper.insert(first);
+ userMapper.insert(second);
+ for (RmqStudioUser parkedUser : parked) {
+ userMapper.update(null, new UpdateWrapper<RmqStudioUser>()
+ .eq("id", parkedUser.getId())
+ .set("enabled", false));
+ }
+ try {
+ int[] outcomes = new int[2];
+ Exception[] threadErrors = new Exception[2];
+ CyclicBarrier barrier = new CyclicBarrier(2);
+ Thread firstThread = disableThread(first.getId(), barrier,
outcomes, threadErrors, 0);
+ Thread secondThread = disableThread(second.getId(), barrier,
outcomes, threadErrors, 1);
+ firstThread.start();
+ secondThread.start();
+ firstThread.join(TimeUnit.SECONDS.toMillis(30));
+ secondThread.join(TimeUnit.SECONDS.toMillis(30));
+
+ assertThat(firstThread.isAlive()).isFalse();
+ assertThat(secondThread.isAlive()).isFalse();
+ assertThat(threadErrors[0]).isNull();
+ assertThat(threadErrors[1]).isNull();
+ // Exactly one disable may win the race; the other must fail with
a conflict.
+ assertThat(outcomes[0] + outcomes[1]).isEqualTo(1);
+ long enabledAdmins = userMapper.selectCount(new
QueryWrapper<RmqStudioUser>()
+ .eq("admin", true)
+ .eq("enabled", true));
+ assertThat(enabledAdmins).isEqualTo(1);
+ } finally {
+ userMapper.deleteById(first.getId());
+ userMapper.deleteById(second.getId());
+ for (RmqStudioUser parkedUser : parked) {
+ userMapper.update(null, new UpdateWrapper<RmqStudioUser>()
+ .eq("id", parkedUser.getId())
+ .set("enabled", true));
+ }
+ }
+ }
+
+ private Thread disableThread(Long userId, CyclicBarrier barrier, int[]
outcomes,
+ Exception[] threadErrors, int slot) {
+ return new Thread(() -> {
+ try {
+ barrier.await();
+ try {
+ authService.setUserEnabled(userId, false);
+ outcomes[slot] = 1;
+ } catch (BusinessException exception) {
+ if (exception.getCode() != 409) {
+ throw exception;
+ }
+ }
+ } catch (Exception exception) {
+ threadErrors[slot] = exception;
+ }
+ }, "auth-disable-" + userId);
+ }
+
+ private RmqStudioUser adminUser(String username) {
+ RmqStudioUser user = new RmqStudioUser();
+ user.setUsername(username);
+ user.setPasswordHash(passwordHasher.hash("password-1"));
+ user.setAdmin(true);
+ user.setEnabled(true);
+ user.setPasswordChangedAt(LocalDateTime.now());
+ return user;
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/auth/AuthServiceDatabaseTest.java
b/server/src/test/java/org/apache/rocketmq/studio/auth/AuthServiceDatabaseTest.java
index 9ef733e6a..394a410f9 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/auth/AuthServiceDatabaseTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/auth/AuthServiceDatabaseTest.java
@@ -100,13 +100,41 @@ class AuthServiceDatabaseTest {
void disablingLastEnabledAdministratorIsRejected() {
RmqStudioUser user = user(1L, "admin", true, true, "password-1");
when(userMapper.selectById(1L)).thenReturn(user);
- when(userMapper.selectCount(any(Wrapper.class))).thenReturn(1L);
+
when(userMapper.selectList(any(Wrapper.class))).thenReturn(List.of(user));
assertThatThrownBy(() -> authService.setUserEnabled(1L, false))
.isInstanceOf(BusinessException.class)
.hasMessage("The last enabled administrator cannot be
disabled");
}
+ @Test
+ void disablingAnAdministratorKeepsTheOtherEnabledAdministratorTest() {
+ RmqStudioUser target = user(1L, "admin-one", true, true, "password-1");
+ RmqStudioUser other = user(2L, "admin-two", true, true, "password-1");
+ when(userMapper.selectById(1L)).thenReturn(target);
+
when(userMapper.selectList(any(Wrapper.class))).thenReturn(List.of(target,
other));
+
+ RmqStudioUser result = authService.setUserEnabled(1L, false);
+
+ assertThat(result.getEnabled()).isFalse();
+ verify(userMapper).updateById(any(RmqStudioUser.class));
+ }
+
+ @Test
+ void
disablingAnAlreadyDisabledAdministratorIsNotRejectedByTheOtherAdminTest() {
+ // A concurrent request disabled the target first, so it no longer
appears among the
+ // enabled administrators; the stale read must not be mistaken for
"last admin".
+ RmqStudioUser target = user(1L, "admin-one", true, true, "password-1");
+ RmqStudioUser other = user(2L, "admin-two", true, true, "password-1");
+ when(userMapper.selectById(1L)).thenReturn(target);
+
when(userMapper.selectList(any(Wrapper.class))).thenReturn(List.of(other));
+
+ RmqStudioUser result = authService.setUserEnabled(1L, false);
+
+ assertThat(result.getEnabled()).isFalse();
+ verify(userMapper).updateById(any(RmqStudioUser.class));
+ }
+
@Test
void databaseAuthenticationThrottlesLastSeenWrites() {
RmqStudioUser user = user(1L, "operator", false, true, "password-1");
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterServiceRegistryTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterServiceRegistryTest.java
index 2dcd3781d..45f60db62 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterServiceRegistryTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterServiceRegistryTest.java
@@ -28,8 +28,12 @@ import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
@@ -94,4 +98,72 @@ class ClusterServiceRegistryTest {
assertThat(clusterService.listRegistryClusters()).isEmpty();
}
+
+ @Test
+ void listRegistryClustersShouldBoundProbeThreadsWhenProbesBlockTest()
throws Exception {
+ int maxConcurrency = 4;
+ when(registryService.list()).thenReturn(List.of(
+
NameserverRegistryVO.builder().id(1L).name("ns-1").namesrvAddr("ns-1:9876").build(),
+
NameserverRegistryVO.builder().id(2L).name("ns-2").namesrvAddr("ns-2:9876").build(),
+
NameserverRegistryVO.builder().id(3L).name("ns-3").namesrvAddr("ns-3:9876").build(),
+
NameserverRegistryVO.builder().id(4L).name("ns-4").namesrvAddr("ns-4:9876").build()));
+
+ CountDownLatch block = new CountDownLatch(1);
+ AtomicInteger probes = new AtomicInteger();
+
when(clusterProvider.discoverClustersAt(anyString())).thenAnswer(invocation -> {
+ probes.incrementAndGet();
+ try {
+ block.await();
+ } catch (InterruptedException exception) {
+ Thread.currentThread().interrupt();
+ }
+ return List.of();
+ });
+
+ try (RegistryProbeRunner runner = new
RegistryProbeRunner(maxConcurrency, 8, 150)) {
+ clusterService.setRegistryProbeRunner(runner);
+
+ // Two refresh rounds with every probe blocked: a cached thread
pool would
+ // accumulate a fresh blocked thread per entry per round.
+ assertThat(clusterService.listRegistryClusters()).isEmpty();
+ assertThat(clusterService.listRegistryClusters()).isEmpty();
+
+ // The pool never grows past its bound and both rounds actually
ran every probe.
+ assertThat(runner.poolSize()).isLessThanOrEqualTo(maxConcurrency);
+
assertThat(runner.activeCount()).isLessThanOrEqualTo(maxConcurrency);
+ assertThat(probes).hasValue(8);
+ } finally {
+ block.countDown();
+ }
+ }
+
+ @Test
+ void
listRegistryClustersShouldDegradeSaturatedProbeQueueWithoutThrowingTest()
throws Exception {
+ // One worker plus a one-slot queue: the third concurrent probe must
be rejected
+ // and degraded to unavailable rather than growing the pool or failing
the request.
+ when(registryService.list()).thenReturn(List.of(
+
NameserverRegistryVO.builder().id(1L).name("ns-1").namesrvAddr("ns-1:9876").build(),
+
NameserverRegistryVO.builder().id(2L).name("ns-2").namesrvAddr("ns-2:9876").build(),
+
NameserverRegistryVO.builder().id(3L).name("ns-3").namesrvAddr("ns-3:9876").build()));
+
+ CountDownLatch block = new CountDownLatch(1);
+
when(clusterProvider.discoverClustersAt(anyString())).thenAnswer(invocation -> {
+ try {
+ block.await();
+ } catch (InterruptedException exception) {
+ Thread.currentThread().interrupt();
+ }
+ return List.of();
+ });
+
+ try (RegistryProbeRunner runner = new RegistryProbeRunner(1, 1, 150)) {
+ clusterService.setRegistryProbeRunner(runner);
+
+ assertThatCode(() ->
assertThat(clusterService.listRegistryClusters()).isEmpty())
+ .doesNotThrowAnyException();
+ assertThat(runner.poolSize()).isLessThanOrEqualTo(1);
+ } finally {
+ block.countDown();
+ }
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryServiceIntegrationTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryServiceIntegrationTest.java
new file mode 100644
index 000000000..31c9d6537
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryServiceIntegrationTest.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.instance.message;
+
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import org.apache.rocketmq.studio.auth.AuthenticatedUserContext;
+import org.apache.rocketmq.studio.common.domain.PageResult;
+import org.apache.rocketmq.studio.persistence.entity.RmqMessageQuery;
+import org.apache.rocketmq.studio.persistence.entity.RmqTraceQuery;
+import org.apache.rocketmq.studio.persistence.mapper.RmqMessageQueryMapper;
+import org.apache.rocketmq.studio.persistence.mapper.RmqTraceQueryMapper;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@SpringBootTest(properties = "studio.auth.login-required=true")
+class QueryHistoryServiceIntegrationTest {
+
+ @Autowired
+ private QueryHistoryService queryHistoryService;
+
+ @Autowired
+ private RmqMessageQueryMapper messageQueryMapper;
+
+ @Autowired
+ private RmqTraceQueryMapper traceQueryMapper;
+
+ @Test
+ void queryHistoryKeepsFullLengthUsernamesTest() {
+ // Usernames may be up to 128 characters; the query-history owner
columns must
+ // store the full value instead of rejecting or truncating it.
+ String longUsername = "long".repeat(32);
+ assertThat(longUsername).hasSize(128);
+ try {
+ AuthenticatedUserContext.setUser(longUsername, true);
+ queryHistoryService.recordMessageQuery("qh-cluster", "TOPIC",
"qh-topic",
+ null, null, null, null, null, 3);
+ queryHistoryService.recordTraceQuery("qh-cluster", "qh-msg-id",
"qh-topic", 2, 1);
+
+ PageResult<MessageQueryHistoryVO> messageHistory =
+ queryHistoryService.listMessageQueries("qh-cluster", null,
null, 1, 20);
+ assertThat(messageHistory.getItems()).anySatisfy(item ->
+ assertThat(item.getQueriedBy()).isEqualTo(longUsername));
+
+ PageResult<TraceQueryHistoryVO> traceHistory =
+ queryHistoryService.listTraceQueries("qh-cluster", null,
1, 20);
+ assertThat(traceHistory.getItems()).anySatisfy(item ->
+ assertThat(item.getQueriedBy()).isEqualTo(longUsername));
+ } finally {
+ AuthenticatedUserContext.clear();
+ messageQueryMapper.delete(new QueryWrapper<RmqMessageQuery>()
+ .eq("queried_by", longUsername));
+ traceQueryMapper.delete(new QueryWrapper<RmqTraceQuery>()
+ .eq("queried_by", longUsername));
+ }
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/model/LiteTopicQuotaTest.java
b/server/src/test/java/org/apache/rocketmq/studio/model/LiteTopicQuotaTest.java
index 3825f91d4..b585fdc6b 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/model/LiteTopicQuotaTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/model/LiteTopicQuotaTest.java
@@ -25,12 +25,34 @@ class LiteTopicQuotaTest {
}
@Test
- void quotaShouldBeExceededWhenCurrentReachesMax() {
+ void nonPositiveLimitsShouldRemainUnavailableInsteadOfExceeded() {
+ LiteTopicQuota quota = new LiteTopicQuota();
+ quota.setMaxTopicCount(0);
+ quota.setCurrentTopicCount(1);
+ quota.setMaxSessionCount(-1);
+ quota.setCurrentSessionCount(1);
+
+ assertThat(quota.getUsageRate()).isZero();
+ assertThat(quota.getSessionUsageRate()).isZero();
+ assertThat(quota.isQuotaExceeded()).isFalse();
+ assertThat(quota.getRemainingQuota()).isZero();
+ }
+
+ @Test
+ void quotaShouldBeExceededOnlyAtOrAbovePositiveMax() {
LiteTopicQuota quota = new LiteTopicQuota();
quota.setMaxTopicCount(10);
+
+ quota.setCurrentTopicCount(9);
+ assertThat(quota.isQuotaExceeded()).isFalse();
+ assertThat(quota.getRemainingQuota()).isEqualTo(1);
+
quota.setCurrentTopicCount(10);
+ assertThat(quota.isQuotaExceeded()).isTrue();
+ quota.setCurrentTopicCount(11);
assertThat(quota.isQuotaExceeded()).isTrue();
+ assertThat(quota.getRemainingQuota()).isZero();
}
@Test
diff --git a/web/src/components/MetricsExplorer.tsx
b/web/src/components/MetricsExplorer.tsx
index 0c38fdfd1..d4df753ab 100644
--- a/web/src/components/MetricsExplorer.tsx
+++ b/web/src/components/MetricsExplorer.tsx
@@ -57,13 +57,49 @@ interface NumericSample {
value: number;
}
-const toNumericSamples = (series: MetricSeries): NumericSample[] =>
- series.values
- .map((sample, index) => ({ timestamp: sample.timestamp, value:
Number(sample.value), index }))
+const sortAndStrip = (
+ samples: { timestamp: number; value: number; index: number }[],
+): NumericSample[] =>
+ samples
.filter((sample) => Number.isFinite(sample.timestamp) &&
Number.isFinite(sample.value))
.sort((left, right) => left.timestamp - right.timestamp || left.index -
right.index)
.map(({ timestamp, value }) => ({ timestamp, value }));
+const toScalarSamples = (series: MetricSeries): NumericSample[] =>
+ sortAndStrip(
+ series.values.map((sample, index) => ({
+ timestamp: sample.timestamp,
+ value: Number(sample.value),
+ index,
+ })),
+ );
+
+// Native histograms carry no scalar samples. To avoid plotting them as "no
data", derive a
+// trend value per histogram: the observed sum (in the metric's unit), falling
back to the
+// observation count when the sum is absent or non-finite.
+const toHistogramSamples = (series: MetricSeries): NumericSample[] =>
+ sortAndStrip(
+ series.histograms.map((sample, index) => {
+ // An empty string parses to 0, so treat a blank field as missing rather
than zero.
+ const sumText = sample.histogram.sum?.trim();
+ const countText = sample.histogram.count?.trim();
+ const sum = sumText ? Number(sumText) : Number.NaN;
+ const count = countText ? Number(countText) : Number.NaN;
+ const value = Number.isFinite(sum) ? sum : count;
+ return { timestamp: sample.timestamp, value, index };
+ }),
+ );
+
+const toNumericSamples = (
+ series: MetricSeries,
+): { samples: NumericSample[]; fromHistogram: boolean } => {
+ const scalar = toScalarSamples(series);
+ if (scalar.length > 0) {
+ return { samples: scalar, fromHistogram: false };
+ }
+ return { samples: toHistogramSamples(series), fromHistogram: true };
+};
+
const seriesLabel = (series: MetricSeries, fallback: string) => {
const labels = Object.entries(series.labels)
.filter(([key]) => key !== '__name__')
@@ -80,15 +116,28 @@ interface MetricChartProps {
metric: MetricMapping;
locale: string;
noSamples: string;
+ histogramLabel: string;
+ histogramTooltip: string;
}
-const MetricChart = ({ data, metric, locale, noSamples }: MetricChartProps) =>
{
+const MetricChart = ({
+ data,
+ metric,
+ locale,
+ noSamples,
+ histogramLabel,
+ histogramTooltip,
+}: MetricChartProps) => {
const chartSeries = data.series
- .map((series, index) => ({
- color: SERIES_COLORS[index % SERIES_COLORS.length],
- label: seriesLabel(series, metric.name),
- samples: toNumericSamples(series),
- }))
+ .map((series, index) => {
+ const { samples, fromHistogram } = toNumericSamples(series);
+ return {
+ color: SERIES_COLORS[index % SERIES_COLORS.length],
+ label: seriesLabel(series, metric.name),
+ samples,
+ fromHistogram,
+ };
+ })
.filter((series) => series.samples.length > 0);
if (chartSeries.length === 0) {
@@ -205,6 +254,13 @@ const MetricChart = ({ data, metric, locale, noSamples }:
MetricChartProps) => {
<Text ellipsis={{ tooltip: series.label }} style={{ maxWidth:
220 }}>
{series.label}
</Text>
+ {series.fromHistogram ? (
+ <Tooltip title={histogramTooltip}>
+ <Tag color="purple" style={{ marginInlineEnd: 0 }}>
+ {histogramLabel}
+ </Tag>
+ </Tooltip>
+ ) : null}
<Text strong>
{formatMetricValue(latest.value)} {metric.unit}
</Text>
@@ -272,7 +328,9 @@ const MetricsExplorer = ({ instanceId }:
MetricsExplorerProps) => {
profileError: '指标模板加载失败',
queryError: queryErrorFallback,
noProfiles: '暂无指标模板',
- noSamples: '暂无标量数据',
+ noSamples: '暂无数据',
+ histogram: '直方图',
+ histogramTooltip: '无标量样本，趋势由直方图观测值推导',
defaultDataSource: '默认数据源',
authTitle: '数据源认证',
authDescription: '凭据仅用于当前数据源，离开该数据源后会被清除。',
@@ -292,7 +350,9 @@ const MetricsExplorer = ({ instanceId }:
MetricsExplorerProps) => {
profileError: 'Failed to load metric profiles',
queryError: queryErrorFallback,
noProfiles: 'No metric profiles',
- noSamples: 'No scalar samples',
+ noSamples: 'No samples',
+ histogram: 'Histogram',
+ histogramTooltip: 'No scalar samples; trend derived from histogram
observations',
defaultDataSource: 'Default source',
authTitle: 'Data source authentication',
authDescription:
@@ -607,6 +667,8 @@ const MetricsExplorer = ({ instanceId }:
MetricsExplorerProps) => {
metric={selectedMetric}
locale={lang === 'zh' ? 'zh-CN' : 'en-US'}
noSamples={copy.noSamples}
+ histogramLabel={copy.histogram}
+ histogramTooltip={copy.histogramTooltip}
/>
</>
) : null}
diff --git a/web/src/components/__tests__/MetricsExplorer.test.tsx
b/web/src/components/__tests__/MetricsExplorer.test.tsx
index aee5234fc..b14a3f140 100644
--- a/web/src/components/__tests__/MetricsExplorer.test.tsx
+++ b/web/src/components/__tests__/MetricsExplorer.test.tsx
@@ -84,6 +84,27 @@ const metricData = {
warnings: [],
};
+const histogramOnlyData = {
+ resultType: 'matrix',
+ series: [
+ {
+ labels: { cluster: 'prod', node_id: 'broker-a' },
+ values: [],
+ histograms: [
+ {
+ timestamp: 1_799_996_400,
+ histogram: { count: '10', sum: '250', buckets: [] },
+ },
+ {
+ timestamp: 1_800_000_000,
+ histogram: { count: '20', sum: '600', buckets: [] },
+ },
+ ],
+ },
+ ],
+ warnings: [],
+};
+
beforeAll(() => {
Object.defineProperty(window, 'matchMedia', {
writable: true,
@@ -257,12 +278,63 @@ describe('MetricsExplorer', () => {
expect(await screen.findByText('Prometheus base URL is not
configured')).toBeInTheDocument();
});
- it('shows an empty state when Prometheus returns no scalar samples', async
() => {
+ it('shows an empty state when Prometheus returns no samples at all', async
() => {
vi.mocked(queryMetrics).mockResolvedValue({ ...metricData, series: [] });
renderWithProviders(<MetricsExplorer />);
- expect(await screen.findByText('暂无标量数据')).toBeInTheDocument();
+ expect(await screen.findByText('暂无数据')).toBeInTheDocument();
+ });
+
+ it('renders histogram-only series from observed sums instead of an empty
state', async () => {
+ vi.mocked(queryMetrics).mockResolvedValue(histogramOnlyData);
+
+ renderWithProviders(<MetricsExplorer />);
+
+ expect(
+ await screen.findByRole('img', { name: 'Message In TPS time series' }),
+ ).toBeInTheDocument();
+ expect(screen.getByText('600 messages/s')).toBeInTheDocument();
+ expect(screen.getByText('直方图')).toBeInTheDocument();
+ expect(screen.queryByText('暂无数据')).not.toBeInTheDocument();
+ });
+
+ it('falls back to the observation count when the histogram sum is missing',
async () => {
+ vi.mocked(queryMetrics).mockResolvedValue({
+ ...histogramOnlyData,
+ series: [
+ {
+ ...histogramOnlyData.series[0],
+ histograms: [
+ { timestamp: 1_800_000_000, histogram: { count: '20', sum: '',
buckets: [] } },
+ ],
+ },
+ ],
+ });
+
+ renderWithProviders(<MetricsExplorer />);
+
+ expect(await screen.findByText('20 messages/s')).toBeInTheDocument();
+ });
+
+ it('prefers scalar samples when a series has both values and histograms',
async () => {
+ vi.mocked(queryMetrics).mockResolvedValue({
+ ...metricData,
+ series: [
+ {
+ ...metricData.series[0],
+ histograms: [
+ { timestamp: 1_800_000_000, histogram: { count: '99', sum: '999',
buckets: [] } },
+ ],
+ },
+ ],
+ });
+
+ renderWithProviders(<MetricsExplorer />);
+
+ expect(await screen.findByText('42 messages/s')).toBeInTheDocument();
+ expect(screen.queryByText('999 messages/s')).not.toBeInTheDocument();
+ expect(screen.queryByText('直方图')).not.toBeInTheDocument();
});
it('queries the selected data source through the datasource endpoint', async
() => {
diff --git a/web/src/i18n/translations.ts b/web/src/i18n/translations.ts
index 648e2ae88..171c08a28 100644
--- a/web/src/i18n/translations.ts
+++ b/web/src/i18n/translations.ts
@@ -753,10 +753,6 @@ const translations: Record<string, Record<Lang, string>> =
{
en: 'Topic "{name}" created successfully',
},
'topic.topicNamePlaceholder': { zh: '请输入 Topic 名称', en: 'Enter topic name' },
- 'topic.topicNameRule': {
- zh: '仅支持字母、数字、下划线、中划线、斜杠和星号',
- en: 'Only letters, numbers, underscore, hyphen, slash and asterisk',
- },
'topic.topicNameRequired': { zh: '请输入 Topic 名称', en: 'Please enter topic
name' },
'topic.queueExtra': { zh: '每个 Broker 节点 8 个队列', en: '8 queues per Broker
node' },
'topic.remarkPlaceholder': {
diff --git a/web/src/pages/instance/consumer.tsx
b/web/src/pages/instance/consumer.tsx
index cff253685..f8bc9d8ca 100644
--- a/web/src/pages/instance/consumer.tsx
+++ b/web/src/pages/instance/consumer.tsx
@@ -95,6 +95,8 @@ import {
import { useInstanceFilter } from '../../hooks/useInstanceFilter';
import {
parseCsvTable,
+ RESOURCE_NAME_MAX_LENGTH,
+ RESOURCE_NAME_PATTERN,
validateConsumerGroupCsvImport,
type ResourceImportRow,
} from '../../utils/resourceCsvImport';
@@ -1743,8 +1745,12 @@ const ConsumerPageContent = ({
rules={[
{ required: true, message: '请输入 Group 名称' },
{
- pattern: /^[a-zA-Z][a-zA-Z0-9_-]*$/,
- message: '名称以字母开头，仅包含字母、数字、下划线和短横线',
+ pattern: RESOURCE_NAME_PATTERN,
+ message: '仅支持字母、数字、下划线、短横线、% 和 |',
+ },
+ {
+ max: RESOURCE_NAME_MAX_LENGTH.group,
+ message: `名称不能超过 ${RESOURCE_NAME_MAX_LENGTH.group} 个字符`,
},
]}
>
diff --git a/web/src/pages/instance/topic.tsx b/web/src/pages/instance/topic.tsx
index c2bc59e59..da01ce078 100644
--- a/web/src/pages/instance/topic.tsx
+++ b/web/src/pages/instance/topic.tsx
@@ -74,6 +74,8 @@ import { useInstanceFilter } from
'../../hooks/useInstanceFilter';
import type { Instance } from '../../api/instance';
import {
parseCsvTable,
+ RESOURCE_NAME_MAX_LENGTH,
+ RESOURCE_NAME_PATTERN,
validateTopicCsvImport,
type ResourceImportRow,
} from '../../utils/resourceCsvImport';
@@ -1329,8 +1331,12 @@ const TopicPage = () => {
rules={[
{ required: true, message: '请输入 Topic 名称' },
{
- pattern: /^[a-zA-Z0-9_\-/*]+$/,
- message: '仅支持字母、数字、下划线、中划线、斜杠和星号',
+ pattern: RESOURCE_NAME_PATTERN,
+ message: '仅支持字母、数字、下划线、短横线、% 和 |',
+ },
+ {
+ max: RESOURCE_NAME_MAX_LENGTH.topic,
+ message: `名称不能超过 ${RESOURCE_NAME_MAX_LENGTH.topic} 个字符`,
},
]}
>
diff --git a/web/src/utils/resourceCsvImport.test.ts
b/web/src/utils/resourceCsvImport.test.ts
index 4621e38df..979f38fb9 100644
--- a/web/src/utils/resourceCsvImport.test.ts
+++ b/web/src/utils/resourceCsvImport.test.ts
@@ -20,6 +20,7 @@ import {
parseCsvTable,
RESOURCE_IMPORT_ROW_LIMIT,
validateConsumerGroupCsvImport,
+ validateResourceName,
validateTopicCsvImport,
} from './resourceCsvImport';
@@ -125,4 +126,45 @@ describe('resourceCsvImport', () => {
instanceId: 'instance-2',
});
});
+
+ it('aligns topic and group names with the RocketMQ validators', () => {
+ const topicStatus = (name: string) =>
+ validateTopicCsvImport(parseCsvTable(['"Name"',
`"${name}"`].join('\n'))).rows[0].status;
+ const groupStatus = (name: string) =>
+ validateConsumerGroupCsvImport(parseCsvTable(['"Name"',
`"${name}"`].join('\n'))).rows[0]
+ .status;
+
+ // RocketMQ accepts % and | in both topic and group names
+ expect(topicStatus('100%topic')).toBe('pending');
+ expect(topicStatus('topic|pipe')).toBe('pending');
+ expect(groupStatus('cg|pipe')).toBe('pending');
+ // Groups may start with a digit (no leading-letter rule)
+ expect(groupStatus('1cg')).toBe('pending');
+ // / and * are not part of the RocketMQ name character set
+ expect(topicStatus('topic/with-slash')).toBe('invalid');
+ expect(topicStatus('topic*star')).toBe('invalid');
+ });
+
+ it('applies the RocketMQ length caps to imported names', () => {
+ const topicStatus = (name: string) =>
+ validateTopicCsvImport(parseCsvTable(['"Name"',
`"${name}"`].join('\n'))).rows[0].status;
+ const groupStatus = (name: string) =>
+ validateConsumerGroupCsvImport(parseCsvTable(['"Name"',
`"${name}"`].join('\n'))).rows[0]
+ .status;
+
+ expect(topicStatus('t'.repeat(127))).toBe('pending');
+ expect(topicStatus('t'.repeat(128))).toBe('invalid');
+ expect(groupStatus('g'.repeat(120))).toBe('pending');
+ expect(groupStatus('g'.repeat(121))).toBe('invalid');
+ });
+
+ it('reports the RocketMQ-oriented name error messages', () => {
+ expect(validateResourceName('', 'topic')).toBe('Name 不能为空');
+ expect(validateResourceName('a'.repeat(128), 'topic')).toBe('Name 长度不能超过
127 个字符');
+ expect(validateResourceName('a'.repeat(121), 'group')).toBe('Name 长度不能超过
120 个字符');
+ expect(validateResourceName('bad/name', 'topic')).toBe(
+ 'Name 仅支持字母、数字、下划线、短横线、% 和 |',
+ );
+ expect(validateResourceName('ok-name|100%', 'group')).toBeNull();
+ });
});
diff --git a/web/src/utils/resourceCsvImport.ts
b/web/src/utils/resourceCsvImport.ts
index 19a6d10f7..bda78d46a 100644
--- a/web/src/utils/resourceCsvImport.ts
+++ b/web/src/utils/resourceCsvImport.ts
@@ -44,8 +44,28 @@ interface ParsedCsvRow {
}
const FORMULA_SAFE_PREFIX_PATTERN = /^'(?=[=+\-@\t\r\n])/;
-const TOPIC_NAME_PATTERN = /^[a-zA-Z0-9_\-/*]+$/;
-const GROUP_NAME_PATTERN = /^[a-zA-Z][a-zA-Z0-9_-]*$/;
+
+// Aligned with RocketMQ's TopicValidator/GroupValidator: a shared character
set (letters,
+// digits, underscore, hyphen, % and |) with per-kind length caps. Topics cap
at 127 and
+// consumer groups at 120; both may start with a digit or symbol, so no
leading-letter rule.
+export const RESOURCE_NAME_PATTERN = /^[%|a-zA-Z0-9_-]+$/;
+export const RESOURCE_NAME_MAX_LENGTH = { topic: 127, group: 120 } as const;
+
+export type ResourceNameKind = keyof typeof RESOURCE_NAME_MAX_LENGTH;
+
+export const validateResourceName = (name: string, kind: ResourceNameKind):
string | null => {
+ if (!name) {
+ return 'Name 不能为空';
+ }
+ const maxLength = RESOURCE_NAME_MAX_LENGTH[kind];
+ if (name.length > maxLength) {
+ return `Name 长度不能超过 ${maxLength} 个字符`;
+ }
+ if (!RESOURCE_NAME_PATTERN.test(name)) {
+ return 'Name 仅支持字母、数字、下划线、短横线、% 和 |';
+ }
+ return null;
+};
const TOPIC_TYPES = new Set(['NORMAL', 'FIFO', 'DELAY', 'TRANSACTION',
'LITE']);
const TOPIC_PERMISSIONS = new Set(['RW', 'RO', 'WO']);
@@ -260,10 +280,9 @@ export const validateTopicCsvImport = (
const duplicateMessage = duplicateMessages.get(record.lineNumber);
if (duplicateMessage) rowErrors.push(duplicateMessage);
- if (!name) {
- rowErrors.push('Name 不能为空');
- } else if (!TOPIC_NAME_PATTERN.test(name)) {
- rowErrors.push('Name 仅支持字母、数字、下划线、中划线、斜杠和星号');
+ const nameError = validateResourceName(name, 'topic');
+ if (nameError) {
+ rowErrors.push(nameError);
}
if (!TOPIC_TYPES.has(type)) {
rowErrors.push(`Type 不支持：${type}`);
@@ -319,10 +338,9 @@ export const validateConsumerGroupCsvImport = (
const duplicateMessage = duplicateMessages.get(record.lineNumber);
if (duplicateMessage) rowErrors.push(duplicateMessage);
- if (!name) {
- rowErrors.push('Name 不能为空');
- } else if (!GROUP_NAME_PATTERN.test(name)) {
- rowErrors.push('Name 需以字母开头，仅包含字母、数字、下划线和短横线');
+ const nameError = validateResourceName(name, 'group');
+ if (nameError) {
+ rowErrors.push(nameError);
}
if (!GROUP_SUBSCRIPTION_MODES.has(subscriptionMode)) {
rowErrors.push(`Subscription Mode 不支持：${subscriptionMode}`);
