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 25c895f3 feat: add MQAdminExt-backed cluster provider and connection
test (#694)
25c895f3 is described below
commit 25c895f3f287e68c90200631301010e43c42134b
Author: zhaohai <[email protected]>
AuthorDate: Mon Aug 3 10:54:58 2026 +0800
feat: add MQAdminExt-backed cluster provider and connection test (#694)
---
server/pom.xml | 16 +++
.../cluster/broker/ClusterConnectionService.java | 65 ++++++++++
.../studio/cluster/broker/ClusterController.java | 6 +
.../studio/cluster/broker/ClusterProbeResult.java | 58 +++++++++
.../studio/cluster/broker/ClusterProviderStub.java | 87 -------------
.../studio/cluster/broker/MqAdminExtFactory.java | 141 +++++++++++++++++++++
.../studio/cluster/broker/MqAdminProperties.java | 38 ++++++
.../studio/cluster/broker/RealClusterProvider.java | 136 ++++++++++++++++++++
.../studio/cluster/broker/TestConnectionDTO.java | 37 ++++++
.../broker/ClusterConnectionServiceTest.java | 79 ++++++++++++
.../cluster/broker/ClusterControllerTest.java | 45 +++++++
.../cluster/broker/ClusterProviderStubTest.java | 48 -------
.../cluster/broker/MqAdminExtFactoryTest.java | 95 ++++++++++++++
.../cluster/broker/RealClusterProviderTest.java | 116 +++++++++++++++++
web/src/api/cluster.ts | 17 +++
web/src/i18n/translations.ts | 16 ++-
.../pages/cluster/__tests__/ClusterPage.test.tsx | 2 +
web/src/pages/cluster/index.tsx | 100 +++++++++++++--
web/src/services/clusterService.ts | 20 ++-
19 files changed, 972 insertions(+), 150 deletions(-)
diff --git a/server/pom.xml b/server/pom.xml
index 6450dcc0..1f2dccf3 100644
--- a/server/pom.xml
+++ b/server/pom.xml
@@ -19,6 +19,7 @@
<properties>
<java.version>21</java.version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ <rocketmq.version>5.3.3</rocketmq.version>
</properties>
<dependencies>
@@ -65,6 +66,21 @@
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-tools</artifactId>
+ <version>${rocketmq.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>logback-classic</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>logback-core</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
</dependencies>
<build>
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterConnectionService.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterConnectionService.java
new file mode 100644
index 00000000..8b0e9b6f
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterConnectionService.java
@@ -0,0 +1,65 @@
+/*
+ * 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.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+
+/**
+ * Verifies live connectivity to a RocketMQ NameServer and summarises the
topology it reports.
+ *
+ * <p>Powers the "test connection" action in the cluster UI: it opens a real
admin connection
+ * through {@link RealClusterProvider}, so a failure surfaces as a {@code
BusinessException} with the
+ * underlying cause instead of a fabricated success.
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class ClusterConnectionService {
+
+ private final RealClusterProvider clusterProvider;
+
+ /**
+ * Probes the NameServer described by the command.
+ *
+ * @param command the connection request holding the NameServer address
+ * @return a populated {@link ClusterProbeResult} on success
+ */
+ public ClusterProbeResult testConnection(TestConnectionDTO command) {
+ String namesrvAddr = command.getNamesrvAddr().trim();
+ log.info("Testing connection to NameServer {}", namesrvAddr);
+ long start = System.currentTimeMillis();
+ ClusterVO cluster = clusterProvider.describeCluster(namesrvAddr);
+ long elapsed = System.currentTimeMillis() - start;
+
+ List<String> brokerNames = cluster.getBrokers() == null ? List.of()
+ :
cluster.getBrokers().stream().map(BrokerVO::getName).toList();
+
+ return ClusterProbeResult.builder()
+ .connected(true)
+ .namesrvAddr(namesrvAddr)
+ .clusterName(cluster.getName())
+ .brokerCount(brokerNames.size())
+ .brokerNames(brokerNames)
+ .elapsedMillis(elapsed)
+ .message("Connected to " + brokerNames.size() + " broker(s) in
" + elapsed + "ms")
+ .build();
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterController.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterController.java
index 1cd53c2c..3ba3161e 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterController.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterController.java
@@ -37,12 +37,18 @@ import java.util.Map;
public class ClusterController {
private final ClusterService clusterService;
+ private final ClusterConnectionService clusterConnectionService;
@GetMapping
public Result<List<ClusterVO>> listClusters() {
return Result.ok(clusterService.listClusters());
}
+ @PostMapping("/test-connection")
+ public Result<ClusterProbeResult> testConnection(@Valid @RequestBody
TestConnectionDTO command) {
+ return Result.ok(clusterConnectionService.testConnection(command));
+ }
+
@GetMapping("/{id}")
public Result<ClusterVO> getCluster(@PathVariable String id) {
return Result.ok(clusterService.getCluster(id));
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterProbeResult.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterProbeResult.java
new file mode 100644
index 00000000..5e0dd4ad
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterProbeResult.java
@@ -0,0 +1,58 @@
+/*
+ * 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.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.List;
+
+/**
+ * Outcome of a live connectivity probe against a RocketMQ NameServer.
+ *
+ * <p>Returned by the {@code POST /api/clusters/test-connection} endpoint so
the UI can confirm a
+ * NameServer is reachable and preview the topology it exposes before the
cluster is registered.
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class ClusterProbeResult {
+
+ /** Whether the NameServer accepted the connection and returned cluster
info. */
+ private boolean connected;
+
+ /** The probed NameServer address list. */
+ private String namesrvAddr;
+
+ /** First cluster name reported by the NameServer, if any. */
+ private String clusterName;
+
+ /** Number of brokers registered with the cluster. */
+ private int brokerCount;
+
+ /** Names of the registered brokers. */
+ private List<String> brokerNames;
+
+ /** Round-trip time of the probe in milliseconds. */
+ private long elapsedMillis;
+
+ /** Human-readable outcome message. */
+ private String message;
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterProviderStub.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterProviderStub.java
deleted file mode 100644
index 10e7271e..00000000
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterProviderStub.java
+++ /dev/null
@@ -1,87 +0,0 @@
-/*
- * 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 org.apache.rocketmq.studio.cluster.nameserver.NameServerVO;
-import org.apache.rocketmq.studio.cluster.proxy.ProxyVO;
-
-import org.apache.rocketmq.studio.common.domain.enums.BrokerStatus;
-import org.apache.rocketmq.studio.common.domain.enums.ClusterStatus;
-import org.springframework.stereotype.Component;
-
-import java.util.List;
-
-@Component
-public class ClusterProviderStub implements ClusterProvider {
-
- @Override
- public List<ClusterVO> discoverClusters() {
- return List.of(buildSampleCluster("cluster-001"));
- }
-
- @Override
- public ClusterVO refreshClusterDetail(String clusterId) {
- return buildSampleCluster(clusterId);
- }
-
- private ClusterVO buildSampleCluster(String clusterId) {
- ClusterVO cluster = ClusterVO.builder()
- .name("rmq-cluster-01")
- .brokers(List.of(
- BrokerVO.builder()
- .name("broker-a")
- .addr("10.0.0.1:10911")
- .version("5.2.0")
- .status(BrokerStatus.running)
- .diskUsage(45.2)
- .tpsIn(1200)
- .tpsOut(800)
- .build(),
- BrokerVO.builder()
- .name("broker-b")
- .addr("10.0.0.2:10911")
- .version("5.2.0")
- .status(BrokerStatus.running)
- .diskUsage(38.7)
- .tpsIn(980)
- .tpsOut(750)
- .build()
- ))
- .proxies(List.of(
- ProxyVO.builder()
- .addr("10.0.0.10:8081")
- .status(ClusterStatus.healthy)
- .connections(156)
- .grpcPort(8081)
- .remotingPort(10911)
- .build()
- ))
- .nameServers(List.of(
- NameServerVO.builder()
- .addr("10.0.0.20:9876")
- .status(ClusterStatus.healthy)
- .build(),
- NameServerVO.builder()
- .addr("10.0.0.21:9876")
- .status(ClusterStatus.healthy)
- .build()
- ))
- .build();
- cluster.setId(clusterId);
- return cluster;
- }
-}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/MqAdminExtFactory.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/MqAdminExtFactory.java
new file mode 100644
index 00000000..7023df35
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/MqAdminExtFactory.java
@@ -0,0 +1,141 @@
+/*
+ * 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 org.apache.rocketmq.remoting.RPCHook;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
+import org.apache.rocketmq.tools.admin.MQAdminExt;
+
+import jakarta.annotation.PreDestroy;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Central lifecycle owner for real {@link DefaultMQAdminExt} connections.
+ *
+ * <p>This is the single real-network entry point shared by every live
cluster/metadata provider.
+ * Admin clients are created lazily, started once and cached per NameServer
address so subsequent
+ * calls reuse the established connection. All cached clients are shut down on
context destruction.
+ *
+ * <p>The {@link RPCHook} parameter is reserved for the ACL / authentication
work (AUTH-01); it is
+ * currently always {@code null} but is threaded through so credentials can be
injected later
+ * without changing this contract.
+ */
+@Slf4j
+@Component
+public class MqAdminExtFactory {
+
+ /** Default admin RPC timeout in milliseconds. */
+ private static final long DEFAULT_TIMEOUT_MILLIS = 5000L;
+
+ private final Map<String, DefaultMQAdminExt> cache = new
ConcurrentHashMap<>();
+ private final AtomicInteger instanceCounter = new AtomicInteger();
+ private volatile boolean closed = false;
+
+ /**
+ * Runs an action against a started admin client bound to the given
NameServer address.
+ *
+ * @param namesrvAddr NameServer address list, e.g. {@code
host1:9876;host2:9876}
+ * @param rpcHook optional RPC hook for authentication, may be {@code
null}
+ * @param action the admin interaction to execute
+ * @param <T> result type
+ * @return the action result
+ * @throws BusinessException if the connection cannot be established or
the action fails
+ */
+ public <T> T execute(String namesrvAddr, RPCHook rpcHook, AdminAction<T>
action) {
+ if (namesrvAddr == null || namesrvAddr.isBlank()) {
+ throw new BusinessException(400, "NameServer address is required");
+ }
+ if (closed) {
+ throw new BusinessException(503, "Admin factory is shutting down");
+ }
+ DefaultMQAdminExt admin = cache.computeIfAbsent(namesrvAddr.trim(),
+ addr -> createAndStart(addr, rpcHook));
+ try {
+ return action.apply(admin);
+ } catch (BusinessException ex) {
+ throw ex;
+ } catch (Exception ex) {
+ log.warn("Admin action failed against namesrv {}: {}",
namesrvAddr, ex.getMessage());
+ throw new BusinessException(502, "RocketMQ admin call failed: " +
rootMessage(ex));
+ }
+ }
+
+ private DefaultMQAdminExt createAndStart(String namesrvAddr, RPCHook
rpcHook) {
+ DefaultMQAdminExt admin = newAdmin(rpcHook);
+ admin.setNamesrvAddr(namesrvAddr);
+ admin.setInstanceName(buildInstanceName(namesrvAddr));
+ try {
+ admin.start();
+ log.info("Started RocketMQ admin client for namesrv {}",
namesrvAddr);
+ return admin;
+ } catch (Exception ex) {
+ safeShutdown(admin);
+ throw new BusinessException(502,
+ "Failed to connect NameServer " + namesrvAddr + ": " +
rootMessage(ex));
+ }
+ }
+
+ /**
+ * Creates a new (not-yet-started) admin client. Extracted so tests can
inject a stub without a
+ * live cluster.
+ */
+ protected DefaultMQAdminExt newAdmin(RPCHook rpcHook) {
+ return new DefaultMQAdminExt(rpcHook, DEFAULT_TIMEOUT_MILLIS);
+ }
+
+ private String buildInstanceName(String namesrvAddr) {
+ return "rmq-studio-" + Integer.toHexString(namesrvAddr.hashCode())
+ + "-" + instanceCounter.incrementAndGet();
+ }
+
+ private void safeShutdown(MQAdminExt admin) {
+ try {
+ admin.shutdown();
+ } catch (Exception ex) {
+ log.debug("Ignoring admin shutdown error: {}", ex.getMessage());
+ }
+ }
+
+ private String rootMessage(Throwable ex) {
+ Throwable cause = ex;
+ while (cause.getCause() != null && cause.getCause() != cause) {
+ cause = cause.getCause();
+ }
+ String message = cause.getMessage();
+ return message == null ? cause.getClass().getSimpleName() : message;
+ }
+
+ @PreDestroy
+ public void shutdown() {
+ closed = true;
+ cache.values().forEach(this::safeShutdown);
+ cache.clear();
+ log.info("Shut down all RocketMQ admin clients");
+ }
+
+ /** Callback executed against a live admin client. */
+ @FunctionalInterface
+ public interface AdminAction<T> {
+ T apply(MQAdminExt admin) throws Exception;
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/MqAdminProperties.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/MqAdminProperties.java
new file mode 100644
index 00000000..fe5ece97
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/MqAdminProperties.java
@@ -0,0 +1,38 @@
+/*
+ * 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.Getter;
+import lombok.Setter;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.stereotype.Component;
+
+/**
+ * Configuration for the default RocketMQ cluster the studio connects to.
+ *
+ * <p>When {@code namesrvAddr} is left blank the {@link RealClusterProvider}
performs no discovery
+ * and callers rely on the interactive connection-test endpoint instead.
+ */
+@Getter
+@Setter
+@Component
+@ConfigurationProperties(prefix = "studio.cluster.admin")
+public class MqAdminProperties {
+
+ /** NameServer address list, e.g. {@code host1:9876;host2:9876}; blank
disables discovery. */
+ private String namesrvAddr;
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/RealClusterProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/RealClusterProvider.java
new file mode 100644
index 00000000..542b9b1a
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/RealClusterProvider.java
@@ -0,0 +1,136 @@
+/*
+ * 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 org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import org.apache.rocketmq.studio.cluster.nameserver.NameServerVO;
+import org.apache.rocketmq.studio.common.domain.enums.BrokerStatus;
+import org.apache.rocketmq.studio.common.domain.enums.ClusterStatus;
+import org.apache.rocketmq.studio.common.domain.enums.ClusterType;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * {@link ClusterProvider} backed by a live NameServer connection.
+ *
+ * <p>Replaces the former hard-coded stub: cluster topology is read on demand
from the RocketMQ
+ * NameServer through {@link MqAdminExtFactory#examineBrokerClusterInfo}.
Discovery honours the
+ * {@link MqAdminProperties#getNamesrvAddr() configured NameServer}; when none
is configured no
+ * clusters are returned and callers fall back to the interactive
connection-test flow.
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class RealClusterProvider implements ClusterProvider {
+
+ private final MqAdminExtFactory adminFactory;
+ private final MqAdminProperties properties;
+
+ @Override
+ public List<ClusterVO> discoverClusters() {
+ String namesrvAddr = properties.getNamesrvAddr();
+ if (namesrvAddr == null || namesrvAddr.isBlank()) {
+ log.info("No NameServer configured; skipping cluster discovery");
+ return List.of();
+ }
+ return List.of(describeCluster(namesrvAddr));
+ }
+
+ @Override
+ public ClusterVO refreshClusterDetail(String clusterId) {
+ String namesrvAddr = properties.getNamesrvAddr();
+ if (namesrvAddr == null || namesrvAddr.isBlank()) {
+ throw new BusinessException(400, "No NameServer configured for
cluster " + clusterId);
+ }
+ return describeCluster(namesrvAddr);
+ }
+
+ /**
+ * Connects to the given NameServer and maps its live topology into a
{@link ClusterVO}.
+ *
+ * @param namesrvAddr NameServer address list, e.g. {@code
host1:9876;host2:9876}
+ * @return the cluster snapshot reported by the NameServer
+ * @throws BusinessException if the NameServer is unreachable or returns
an error
+ */
+ public ClusterVO describeCluster(String namesrvAddr) {
+ return adminFactory.execute(namesrvAddr, null,
+ admin -> toClusterVO(namesrvAddr,
admin.examineBrokerClusterInfo()));
+ }
+
+ private ClusterVO toClusterVO(String namesrvAddr, ClusterInfo clusterInfo)
{
+ Map<String, BrokerData> brokerAddrTable =
+ clusterInfo.getBrokerAddrTable() == null ? Map.of() :
clusterInfo.getBrokerAddrTable();
+ Map<String, Set<String>> clusterAddrTable =
+ clusterInfo.getClusterAddrTable() == null ? Map.of() :
clusterInfo.getClusterAddrTable();
+
+ List<BrokerVO> brokers = brokerAddrTable.values().stream()
+ .map(this::toBrokerVO)
+ .sorted(Comparator.comparing(BrokerVO::getName,
+ Comparator.nullsLast(Comparator.naturalOrder())))
+ .toList();
+
+ String clusterName = clusterAddrTable.keySet().stream()
+ .sorted()
+ .findFirst()
+ .orElse("DefaultCluster");
+
+ List<NameServerVO> nameServers =
Arrays.stream(namesrvAddr.split("[;,]"))
+ .map(String::trim)
+ .filter(addr -> !addr.isEmpty())
+ .map(addr ->
NameServerVO.builder().addr(addr).status(ClusterStatus.healthy).build())
+ .toList();
+
+ ClusterVO cluster = ClusterVO.builder()
+ .name(clusterName)
+ .nsClusterName(clusterName)
+ .type(ClusterType.V4_DIRECT)
+ .endpoint(namesrvAddr)
+ .status(ClusterStatus.healthy)
+ .brokers(brokers)
+ .proxies(List.of())
+ .nameServers(nameServers)
+ .build();
+ cluster.setId(clusterName);
+ return cluster;
+ }
+
+ private BrokerVO toBrokerVO(BrokerData data) {
+ String addr = null;
+ if (data.getBrokerAddrs() != null) {
+ addr = data.getBrokerAddrs().get(MixAll.MASTER_ID);
+ if (addr == null) {
+ addr =
data.getBrokerAddrs().values().stream().findFirst().orElse(null);
+ }
+ }
+ return BrokerVO.builder()
+ .name(data.getBrokerName())
+ .addr(addr)
+ .status(BrokerStatus.running)
+ .build();
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/TestConnectionDTO.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/TestConnectionDTO.java
new file mode 100644
index 00000000..fd1c43e1
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/TestConnectionDTO.java
@@ -0,0 +1,37 @@
+/*
+ * 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 jakarta.validation.constraints.NotBlank;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * Request payload for probing a live RocketMQ cluster before registering it.
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class TestConnectionDTO {
+
+ /** NameServer address list, e.g. {@code host1:9876;host2:9876}. */
+ @NotBlank(message = "namesrvAddr is required")
+ private String namesrvAddr;
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterConnectionServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterConnectionServiceTest.java
new file mode 100644
index 00000000..e7c4986c
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterConnectionServiceTest.java
@@ -0,0 +1,79 @@
+/*
+ * 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 org.apache.rocketmq.studio.common.domain.enums.BrokerStatus;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class ClusterConnectionServiceTest {
+
+ @Mock
+ private RealClusterProvider clusterProvider;
+
+ @InjectMocks
+ private ClusterConnectionService service;
+
+ private ClusterVO clusterWith(String... brokerNames) {
+ List<BrokerVO> brokers = java.util.Arrays.stream(brokerNames)
+ .map(name ->
BrokerVO.builder().name(name).status(BrokerStatus.running).build())
+ .toList();
+ return ClusterVO.builder()
+ .name("DefaultCluster")
+ .brokers(brokers)
+ .build();
+ }
+
+ @Test
+ void testConnectionShouldSummariseTopology() {
+ when(clusterProvider.describeCluster(eq("10.0.0.1:9876")))
+ .thenReturn(clusterWith("broker-a", "broker-b"));
+
+ ClusterProbeResult result = service.testConnection(
+ TestConnectionDTO.builder().namesrvAddr(" 10.0.0.1:9876
").build());
+
+ assertThat(result.isConnected()).isTrue();
+ assertThat(result.getNamesrvAddr()).isEqualTo("10.0.0.1:9876");
+ assertThat(result.getClusterName()).isEqualTo("DefaultCluster");
+ assertThat(result.getBrokerCount()).isEqualTo(2);
+ assertThat(result.getBrokerNames()).containsExactly("broker-a",
"broker-b");
+ assertThat(result.getElapsedMillis()).isGreaterThanOrEqualTo(0L);
+ }
+
+ @Test
+ void testConnectionShouldPropagateProviderFailure() {
+ when(clusterProvider.describeCluster(eq("10.0.0.9:9876")))
+ .thenThrow(new BusinessException(502, "Failed to connect
NameServer"));
+
+ assertThatThrownBy(() -> service.testConnection(
+
TestConnectionDTO.builder().namesrvAddr("10.0.0.9:9876").build()))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("Failed to connect NameServer");
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterControllerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterControllerTest.java
index 1c23fcdb..cb33bbcd 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterControllerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterControllerTest.java
@@ -61,6 +61,51 @@ class ClusterControllerTest {
@MockBean
private ClusterService clusterService;
+ @MockBean
+ private ClusterConnectionService clusterConnectionService;
+
+ @Test
+ void testConnectionShouldReturnProbeResult() throws Exception {
+ ClusterProbeResult probe = ClusterProbeResult.builder()
+ .connected(true)
+ .namesrvAddr("10.0.0.1:9876")
+ .clusterName("DefaultCluster")
+ .brokerCount(2)
+ .brokerNames(Arrays.asList("broker-a", "broker-b"))
+ .elapsedMillis(12L)
+ .message("Connected to 2 broker(s) in 12ms")
+ .build();
+
when(clusterConnectionService.testConnection(any(TestConnectionDTO.class))).thenReturn(probe);
+
+ ObjectNode command =
objectMapper.createObjectNode().put("namesrvAddr", "10.0.0.1:9876");
+
+ mockMvc.perform(post("/api/clusters/test-connection")
+ .contentType(MediaType.APPLICATION_JSON)
+ .content(objectMapper.writeValueAsString(command)))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.code").value(200))
+ .andExpect(jsonPath("$.data.connected").value(true))
+
.andExpect(jsonPath("$.data.clusterName").value("DefaultCluster"))
+ .andExpect(jsonPath("$.data.brokerCount").value(2))
+ .andExpect(jsonPath("$.data.brokerNames.length()").value(2));
+
+
verify(clusterConnectionService).testConnection(any(TestConnectionDTO.class));
+ }
+
+ @Test
+ void testConnectionShouldRejectBlankNamesrvAddr() throws Exception {
+ ObjectNode command =
objectMapper.createObjectNode().put("namesrvAddr", " ");
+
+ mockMvc.perform(post("/api/clusters/test-connection")
+ .contentType(MediaType.APPLICATION_JSON)
+ .content(objectMapper.writeValueAsString(command)))
+ .andExpect(status().isBadRequest())
+ .andExpect(jsonPath("$.code").value(400))
+ .andExpect(jsonPath("$.message").value("namesrvAddr is
required"));
+
+ verifyNoInteractions(clusterConnectionService);
+ }
+
@Test
void listClustersShouldReturnAllClusters() throws Exception {
ClusterVO cluster1 = buildCluster("cluster-1", "production-cluster",
ClusterStatus.healthy);
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterProviderStubTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterProviderStubTest.java
deleted file mode 100644
index a9882330..00000000
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterProviderStubTest.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * 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 org.junit.jupiter.api.Test;
-
-import java.util.List;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-class ClusterProviderStubTest {
-
- private final ClusterProviderStub provider = new ClusterProviderStub();
-
- @Test
- void discoverClustersShouldReturnStableClusterId() {
- List<ClusterVO> clusters = provider.discoverClusters();
-
- assertThat(clusters).hasSize(1);
- assertThat(clusters.get(0).getId()).isEqualTo("cluster-001");
- assertThat(clusters.get(0).getBrokers()).isNotEmpty();
- assertThat(clusters.get(0).getProxies()).isNotEmpty();
- assertThat(clusters.get(0).getNameServers()).isNotEmpty();
- }
-
- @Test
- void refreshClusterDetailShouldPreserveRequestedClusterId() {
- ClusterVO detail = provider.refreshClusterDetail("cluster-prod");
-
- assertThat(detail.getId()).isEqualTo("cluster-prod");
- assertThat(detail.getName()).isEqualTo("rmq-cluster-01");
- assertThat(detail.getBrokers()).hasSize(2);
- }
-}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/MqAdminExtFactoryTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/MqAdminExtFactoryTest.java
new file mode 100644
index 00000000..89151866
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/MqAdminExtFactoryTest.java
@@ -0,0 +1,95 @@
+/*
+ * 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 org.apache.rocketmq.remoting.RPCHook;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+class MqAdminExtFactoryTest {
+
+ private static final class RecordingFactory extends MqAdminExtFactory {
+ private final DefaultMQAdminExt admin;
+ private final AtomicInteger created = new AtomicInteger();
+
+ private RecordingFactory(DefaultMQAdminExt admin) {
+ this.admin = admin;
+ }
+
+ @Override
+ protected DefaultMQAdminExt newAdmin(RPCHook rpcHook) {
+ created.incrementAndGet();
+ return admin;
+ }
+ }
+
+ @Test
+ void executeShouldRejectBlankNamesrvAddr() {
+ MqAdminExtFactory factory = new MqAdminExtFactory();
+
+ assertThatThrownBy(() -> factory.execute(" ", null, admin ->
"unused"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("NameServer address is required");
+ }
+
+ @Test
+ void executeShouldStartClientOnceAndCacheIt() throws Exception {
+ DefaultMQAdminExt admin = mock(DefaultMQAdminExt.class);
+ RecordingFactory factory = new RecordingFactory(admin);
+
+ String first = factory.execute("10.0.0.1:9876", null, a -> "first");
+ String second = factory.execute("10.0.0.1:9876", null, a -> "second");
+
+ assertThat(first).isEqualTo("first");
+ assertThat(second).isEqualTo("second");
+ assertThat(factory.created.get()).isEqualTo(1);
+ verify(admin, times(1)).start();
+ }
+
+ @Test
+ void executeShouldWrapConnectionFailure() throws Exception {
+ DefaultMQAdminExt admin = mock(DefaultMQAdminExt.class);
+ doThrow(new RuntimeException("connection
refused")).when(admin).start();
+ RecordingFactory factory = new RecordingFactory(admin);
+
+ assertThatThrownBy(() -> factory.execute("10.0.0.9:9876", null, a ->
"unused"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("connection refused");
+ verify(admin).shutdown();
+ }
+
+ @Test
+ void executeShouldRejectCallsAfterShutdown() {
+ DefaultMQAdminExt admin = mock(DefaultMQAdminExt.class);
+ RecordingFactory factory = new RecordingFactory(admin);
+ factory.shutdown();
+
+ assertThatThrownBy(() -> factory.execute("10.0.0.1:9876", null, a ->
"unused"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("shutting down");
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/RealClusterProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/RealClusterProviderTest.java
new file mode 100644
index 00000000..2ed9ddab
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/RealClusterProviderTest.java
@@ -0,0 +1,116 @@
+/*
+ * 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 org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import org.apache.rocketmq.studio.common.domain.enums.BrokerStatus;
+import org.apache.rocketmq.studio.common.domain.enums.ClusterType;
+import org.apache.rocketmq.tools.admin.MQAdminExt;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Set;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class RealClusterProviderTest {
+
+ @Mock
+ private MqAdminExtFactory adminFactory;
+
+ private final MqAdminProperties properties = new MqAdminProperties();
+
+ private RealClusterProvider provider;
+
+ @BeforeEach
+ void setUp() {
+ provider = new RealClusterProvider(adminFactory, properties);
+ }
+
+ @SuppressWarnings("unchecked")
+ private void stubClusterInfo(String namesrvAddr, ClusterInfo info) throws
Exception {
+ MQAdminExt admin = org.mockito.Mockito.mock(MQAdminExt.class);
+ when(admin.examineBrokerClusterInfo()).thenReturn(info);
+ when(adminFactory.execute(eq(namesrvAddr), isNull(),
any())).thenAnswer(invocation -> {
+ MqAdminExtFactory.AdminAction<Object> action =
invocation.getArgument(2);
+ return action.apply(admin);
+ });
+ }
+
+ private ClusterInfo sampleClusterInfo() {
+ ClusterInfo info = new ClusterInfo();
+ HashMap<Long, String> addrs = new HashMap<>();
+ addrs.put(0L, "10.0.0.11:10911");
+ addrs.put(1L, "10.0.0.12:10911");
+ HashMap<String, BrokerData> brokerAddrTable = new HashMap<>();
+ brokerAddrTable.put("broker-b",
+ new BrokerData("DefaultCluster", "broker-b", new
HashMap<>(addrs)));
+ brokerAddrTable.put("broker-a",
+ new BrokerData("DefaultCluster", "broker-a", new
HashMap<>(addrs)));
+ info.setBrokerAddrTable(brokerAddrTable);
+ HashMap<String, Set<String>> clusterAddrTable = new HashMap<>();
+ clusterAddrTable.put("DefaultCluster", Set.of("broker-a", "broker-b"));
+ info.setClusterAddrTable(clusterAddrTable);
+ return info;
+ }
+
+ @Test
+ void describeClusterShouldMapLiveTopology() throws Exception {
+ stubClusterInfo("10.0.0.1:9876;10.0.0.2:9876", sampleClusterInfo());
+
+ ClusterVO cluster =
provider.describeCluster("10.0.0.1:9876;10.0.0.2:9876");
+
+ assertThat(cluster.getName()).isEqualTo("DefaultCluster");
+ assertThat(cluster.getType()).isEqualTo(ClusterType.V4_DIRECT);
+
assertThat(cluster.getEndpoint()).isEqualTo("10.0.0.1:9876;10.0.0.2:9876");
+ assertThat(cluster.getNameServers()).extracting("addr")
+ .containsExactly("10.0.0.1:9876", "10.0.0.2:9876");
+ assertThat(cluster.getBrokers()).hasSize(2);
+
assertThat(cluster.getBrokers().get(0).getName()).isEqualTo("broker-a");
+
assertThat(cluster.getBrokers().get(0).getAddr()).isEqualTo("10.0.0.11:10911");
+
assertThat(cluster.getBrokers().get(0).getStatus()).isEqualTo(BrokerStatus.running);
+ }
+
+ @Test
+ void discoverClustersShouldReturnEmptyWhenNoNamesrvConfigured() {
+ properties.setNamesrvAddr(" ");
+
+ assertThat(provider.discoverClusters()).isEmpty();
+ }
+
+ @Test
+ void discoverClustersShouldUseConfiguredNamesrv() throws Exception {
+ properties.setNamesrvAddr("10.0.0.1:9876");
+ stubClusterInfo("10.0.0.1:9876", sampleClusterInfo());
+
+ List<ClusterVO> clusters = provider.discoverClusters();
+
+ assertThat(clusters).hasSize(1);
+ assertThat(clusters.get(0).getBrokers()).hasSize(2);
+ }
+}
diff --git a/web/src/api/cluster.ts b/web/src/api/cluster.ts
index 5ea7912d..cb8f4ea4 100644
--- a/web/src/api/cluster.ts
+++ b/web/src/api/cluster.ts
@@ -73,6 +73,16 @@ export interface ClusterConfig {
deleteWhen: string;
}
+export interface ClusterProbeResult {
+ connected: boolean;
+ namesrvAddr: string;
+ clusterName: string;
+ brokerCount: number;
+ brokerNames: string[];
+ elapsedMillis: number;
+ message: string;
+}
+
export interface K8sCertInfo {
id: string;
name: string;
@@ -93,6 +103,13 @@ export async function listClusters() {
return res.data.data;
}
+export async function testClusterConnection(namesrvAddr: string) {
+ const res = await client.post<{ data: ClusterProbeResult
}>('/clusters/test-connection', {
+ namesrvAddr,
+ });
+ return res.data.data;
+}
+
export async function getCluster(id: string) {
const res = await client.get<{ data: ClusterInfo
}>(`/clusters/${pathSegment(id)}`);
return res.data.data;
diff --git a/web/src/i18n/translations.ts b/web/src/i18n/translations.ts
index 321ba2e7..389016cb 100644
--- a/web/src/i18n/translations.ts
+++ b/web/src/i18n/translations.ts
@@ -501,10 +501,18 @@ const translations: Record<string, Record<Lang, string>>
= {
'cluster.totalTps': { zh: '总 TPS', en: 'Total TPS' },
'cluster.avgTps': { zh: '平均 TPS', en: 'Avg TPS' },
'cluster.consumerGroupCount': { zh: '消费者组数', en: 'Consumer Group Count' },
- 'cluster.createClusterWip': {
- zh: '新建集群功能开发中',
- en: 'Create cluster feature is in development',
- },
+ 'cluster.testConnection': { zh: '测试连接', en: 'Test Connection' },
+ 'cluster.testConnectionTitle': { zh: '测试集群连接', en: 'Test Cluster Connection'
},
+ 'cluster.testConnectionDesc': {
+ zh: '输入 NameServer 地址以验证连通性并预览集群拓扑',
+ en: 'Enter a NameServer address to verify connectivity and preview the
cluster topology',
+ },
+ 'cluster.testConnectionSuccess': { zh: '连接成功', en: 'Connection successful' },
+ 'cluster.testConnectionFailed': { zh: '连接失败', en: 'Connection failed' },
+ 'cluster.probeClusterName': { zh: '集群名称', en: 'Cluster Name' },
+ 'cluster.probeBrokerCount': { zh: 'Broker 数量', en: 'Broker Count' },
+ 'cluster.probeBrokers': { zh: 'Broker 列表', en: 'Brokers' },
+ 'cluster.probeElapsed': { zh: '耗时 (ms)', en: 'Elapsed (ms)' },
'cluster.configTitle': { zh: '配置 - {name}', en: 'Config - {name}' },
'cluster.configUpdated': { zh: '配置已更新', en: 'Configuration updated' },
'cluster.flushDiskType': { zh: '刷盘方式', en: 'Flush Disk Type' },
diff --git a/web/src/pages/cluster/__tests__/ClusterPage.test.tsx
b/web/src/pages/cluster/__tests__/ClusterPage.test.tsx
index 8ead1bdb..7d3aa093 100644
--- a/web/src/pages/cluster/__tests__/ClusterPage.test.tsx
+++ b/web/src/pages/cluster/__tests__/ClusterPage.test.tsx
@@ -27,6 +27,7 @@ const clusterServiceMocks = vi.hoisted(() => ({
createNameServer: vi.fn(),
listClusters: vi.fn(),
restartProxy: vi.fn(),
+ testClusterConnection: vi.fn(),
updateClusterConfig: vi.fn(),
updateNameServer: vi.fn(),
}));
@@ -134,6 +135,7 @@ describe('Cluster page', () => {
clusterServiceMocks.createNameServer.mockReset().mockResolvedValue(undefined);
clusterServiceMocks.listClusters.mockReset().mockResolvedValue([buildCluster()]);
clusterServiceMocks.restartProxy.mockReset().mockResolvedValue(undefined);
+ clusterServiceMocks.testClusterConnection.mockReset();
clusterServiceMocks.updateClusterConfig.mockReset().mockResolvedValue(undefined);
clusterServiceMocks.updateNameServer.mockReset().mockResolvedValue(undefined);
});
diff --git a/web/src/pages/cluster/index.tsx b/web/src/pages/cluster/index.tsx
index d2d53153..4d4e0a8e 100644
--- a/web/src/pages/cluster/index.tsx
+++ b/web/src/pages/cluster/index.tsx
@@ -53,11 +53,13 @@ import type {
NameServerInfo,
ClusterConfig,
ClusterInfo,
+ ClusterProbeResult,
} from '../../api/cluster';
import {
createNameServer,
listClusters,
restartProxy,
+ testClusterConnection,
updateClusterConfig,
updateNameServer,
} from '../../services/clusterService';
@@ -89,6 +91,44 @@ const ClusterPage = () => {
const [nsForm] = Form.useForm();
const [configForm] = Form.useForm();
+ // ─── Connection test ──────────────────────────────────────────────────────
+ const [connectModalOpen, setConnectModalOpen] = useState(false);
+ const [connectTesting, setConnectTesting] = useState(false);
+ const [probeResult, setProbeResult] = useState<ClusterProbeResult |
null>(null);
+ const [connectForm] = Form.useForm();
+
+ const openConnectModal = useCallback(() => {
+ setProbeResult(null);
+ setConnectModalOpen(true);
+ }, []);
+
+ const closeConnectModal = useCallback(() => {
+ setConnectModalOpen(false);
+ setConnectTesting(false);
+ setProbeResult(null);
+ connectForm.resetFields();
+ }, [connectForm]);
+
+ const handleTestConnection = useCallback(async () => {
+ let namesrvAddr: string;
+ try {
+ ({ namesrvAddr } = await connectForm.validateFields());
+ } catch {
+ return;
+ }
+ setConnectTesting(true);
+ setProbeResult(null);
+ try {
+ const result = await testClusterConnection(namesrvAddr);
+ setProbeResult(result);
+ message.success(t('cluster.testConnectionSuccess'));
+ } catch {
+ message.error(t('cluster.testConnectionFailed'));
+ } finally {
+ setConnectTesting(false);
+ }
+ }, [connectForm, t]);
+
// ─── Cluster refresh coordinator ──────────────────────────────────────────
const [autoRefresh, setAutoRefresh] = useState(true);
const [refreshFailed, setRefreshFailed] = useState(false);
@@ -451,11 +491,7 @@ const ClusterPage = () => {
options={nsClusterOptions}
/>
</Space>
- <Button
- type="primary"
- icon={<PlusOutlined />}
- onClick={() => message.info(t('cluster.createClusterWip'))}
- >
+ <Button type="primary" icon={<PlusOutlined />}
onClick={openConnectModal}>
{t('cluster.createCluster')}
</Button>
</Flex>
@@ -850,11 +886,7 @@ const ClusterPage = () => {
style={{ width: 240 }}
/>
</Space>
- <Button
- type="primary"
- icon={<PlusOutlined />}
- onClick={() => message.info(t('cluster.createClusterWip'))}
- >
+ <Button type="primary" icon={<PlusOutlined />}
onClick={openConnectModal}>
{t('cluster.createCluster')}
</Button>
</Flex>
@@ -1044,6 +1076,54 @@ const ClusterPage = () => {
</Descriptions>
)}
</Modal>
+ <Modal
+ title={t('cluster.testConnectionTitle')}
+ open={connectModalOpen}
+ onCancel={closeConnectModal}
+ okText={t('cluster.testConnection')}
+ cancelText={t('common.close')}
+ confirmLoading={connectTesting}
+ onOk={handleTestConnection}
+ width={560}
+ destroyOnClose
+ >
+ <Text type="secondary">{t('cluster.testConnectionDesc')}</Text>
+ <Form form={connectForm} layout="vertical" style={{ marginTop: 16 }}>
+ <Form.Item
+ name="namesrvAddr"
+ label={t('cluster.nsAddr')}
+ rules={[{ required: true, message: t('cluster.nsAddrPlaceholder')
}]}
+ >
+ <Input
+ placeholder={t('cluster.nsAddrPlaceholder')}
+ onPressEnter={handleTestConnection}
+ />
+ </Form.Item>
+ </Form>
+ {probeResult && (
+ <Descriptions column={1} bordered size="small">
+ <Descriptions.Item label={t('common.status')}>
+ <Tag color={probeResult.connected ? 'green' : 'red'}>
+ {probeResult.connected
+ ? t('cluster.testConnectionSuccess')
+ : t('cluster.testConnectionFailed')}
+ </Tag>
+ </Descriptions.Item>
+ <Descriptions.Item label={t('cluster.probeClusterName')}>
+ {probeResult.clusterName}
+ </Descriptions.Item>
+ <Descriptions.Item label={t('cluster.probeBrokerCount')}>
+ {probeResult.brokerCount}
+ </Descriptions.Item>
+ <Descriptions.Item label={t('cluster.probeBrokers')}>
+ {probeResult.brokerNames.length > 0 ?
probeResult.brokerNames.join(', ') : '-'}
+ </Descriptions.Item>
+ <Descriptions.Item label={t('cluster.probeElapsed')}>
+ {probeResult.elapsedMillis}
+ </Descriptions.Item>
+ </Descriptions>
+ )}
+ </Modal>
</div>
);
};
diff --git a/web/src/services/clusterService.ts
b/web/src/services/clusterService.ts
index 374ad614..a0958d7b 100644
--- a/web/src/services/clusterService.ts
+++ b/web/src/services/clusterService.ts
@@ -1,6 +1,6 @@
import { USE_MOCK } from '../config';
import * as clusterApi from '../api/cluster';
-import type { ClusterConfig, ClusterInfo, K8sCertInfo } from '../api/cluster';
+import type { ClusterConfig, ClusterInfo, ClusterProbeResult, K8sCertInfo }
from '../api/cluster';
import clusters, { mockK8sCerts } from '../mock/clusters';
const mockCertStore: K8sCertInfo[] = mockK8sCerts.map((cert) => ({
@@ -34,6 +34,24 @@ export async function listClusters(): Promise<ClusterInfo[]>
{
return clusterApi.listClusters();
}
+export async function testClusterConnection(namesrvAddr: string):
Promise<ClusterProbeResult> {
+ if (USE_MOCK) {
+ const trimmed = namesrvAddr.trim();
+ const cluster = clusters[0];
+ const brokerNames = cluster ? cluster.brokers.map((broker) => broker.name)
: [];
+ return {
+ connected: true,
+ namesrvAddr: trimmed,
+ clusterName: cluster?.nsClusterName ?? 'DefaultCluster',
+ brokerCount: brokerNames.length,
+ brokerNames,
+ elapsedMillis: 12,
+ message: `Connected to ${brokerNames.length} broker(s) (mock)`,
+ };
+ }
+ return clusterApi.testClusterConnection(namesrvAddr);
+}
+
export async function getCluster(id: string): Promise<ClusterInfo> {
if (USE_MOCK) {
const cluster = clusters.find((item) => item.id === id);
