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 64bae563 feat: read cluster, broker, client and dashboard data from a
live cluster (#797)
64bae563 is described below
commit 64bae563b532054764cec4d6646032fa8dc04d98
Author: lizhimins <[email protected]>
AuthorDate: Mon Aug 3 14:32:29 2026 +0800
feat: read cluster, broker, client and dashboard data from a live cluster
(#797)
---
.../studio/cluster/broker/ClusterService.java | 76 ++++
.../studio/cluster/client/ClientProviderStub.java | 2 -
.../ops/dashboard/DashboardProviderStub.java | 7 +-
.../studio/rocketmq/RocketMQAdminClientImpl.java | 486 +++++++++++++++++++++
.../studio/rocketmq/RocketMQAdminConfig.java | 44 ++
.../rocketmq/RocketMQBrokerConfigService.java | 113 +++++
.../studio/rocketmq/RocketMQClientProvider.java | 264 +++++++++++
.../studio/rocketmq/RocketMQClusterProvider.java | 250 +++++++++++
.../studio/rocketmq/RocketMQDashboardProvider.java | 280 ++++++++++++
.../RocketMQProperties.java} | 21 +-
10 files changed, 1525 insertions(+), 18 deletions(-)
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 45393dc1..a2f0eec3 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
@@ -29,11 +29,13 @@ import
org.apache.rocketmq.studio.cluster.proxy.RestartProxyDTO;
import org.apache.rocketmq.studio.common.domain.enums.ClusterStatus;
import org.apache.rocketmq.studio.common.domain.enums.FlushDiskType;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.rocketmq.RocketMQBrokerConfigService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import java.util.List;
+import java.util.Properties;
@Slf4j
@Service
@@ -42,18 +44,53 @@ public class ClusterService {
private final ClusterRepository clusterRepository;
private final ClusterProvider clusterProvider;
+ private final RocketMQBrokerConfigService brokerConfigService;
public List<ClusterVO> listClusters() {
log.info("Listing all clusters");
+ List<ClusterVO> discovered = clusterProvider.discoverClusters();
+ if (discovered != null && !discovered.isEmpty()) {
+ discovered.forEach(this::enrichWithLiveConfig);
+ return discovered;
+ }
return clusterRepository.findAll();
}
public ClusterVO getCluster(String id) {
log.info("Getting cluster detail: {}", id);
+ ClusterVO live = clusterProvider.refreshClusterDetail(id);
+ if (live != null) {
+ enrichWithLiveConfig(live);
+ return live;
+ }
return clusterRepository.findById(id)
.orElseThrow(() -> new BusinessException(404, "Cluster not
found: " + id));
}
+ /**
+ * Attach live broker configuration (read from the first reachable master
broker via the
+ * admin API) to a discovered cluster. Falls back to the persisted config,
if any, when the
+ * live read is unavailable.
+ */
+ private void enrichWithLiveConfig(ClusterVO cluster) {
+ if (cluster.getBrokers() != null) {
+ for (BrokerVO broker : cluster.getBrokers()) {
+ if (broker.getAddr() != null && !broker.getAddr().isEmpty()) {
+ try {
+
cluster.setConfig(brokerConfigService.getBrokerConfig(broker.getAddr()));
+ return;
+ } catch (Exception e) {
+ log.warn("Failed to read live config from broker {}:
{}",
+ broker.getAddr(), e.getMessage());
+ }
+ }
+ }
+ }
+ if (cluster.getConfig() == null && cluster.getId() != null) {
+ clusterRepository.findById(cluster.getId()).ifPresent(stored ->
cluster.setConfig(stored.getConfig()));
+ }
+ }
+
public ClusterVO updateClusterConfig(UpdateConfigDTO command) {
log.info("Updating cluster config for: {}", command.getId());
ClusterVO cluster = clusterRepository.findById(command.getId())
@@ -86,6 +123,16 @@ public class ClusterService {
config.setBrokerPermission(command.getBrokerPermission());
}
+ // Push config to live brokers via admin API
+ if (cluster.getBrokers() != null && !cluster.getBrokers().isEmpty()) {
+ Properties brokerProps = buildBrokerProperties(command);
+ for (BrokerVO broker : cluster.getBrokers()) {
+ if (broker.getAddr() != null && !broker.getAddr().isEmpty()) {
+ brokerConfigService.updateBrokerConfig(broker.getAddr(),
command.getId(), brokerProps);
+ }
+ }
+ }
+
clusterRepository.updateConfig(command.getId(), config);
cluster.setConfig(config);
log.info("Cluster config updated successfully for: {}",
command.getId());
@@ -110,6 +157,35 @@ public class ClusterService {
.build();
}
+ private Properties buildBrokerProperties(UpdateConfigDTO command) {
+ Properties props = new Properties();
+ if (command.getFlushDiskType() != null) {
+ props.setProperty("flushDiskType", command.getFlushDiskType());
+ }
+ if (command.getAutoCreateTopicEnable() != null) {
+ props.setProperty("autoCreateTopicEnable",
command.getAutoCreateTopicEnable().toString());
+ }
+ if (command.getAutoCreateSubscriptionGroup() != null) {
+ props.setProperty("autoCreateSubscriptionGroup",
command.getAutoCreateSubscriptionGroup().toString());
+ }
+ if (command.getMaxMessageSize() != null) {
+ props.setProperty("maxMessageSize",
command.getMaxMessageSize().toString());
+ }
+ if (command.getFileReservedTime() != null) {
+ props.setProperty("fileReservedTime",
command.getFileReservedTime().toString());
+ }
+ if (command.getWriteQueueNums() != null) {
+ props.setProperty("defaultTopicQueueNums",
command.getWriteQueueNums().toString());
+ }
+ if (command.getReadQueueNums() != null) {
+ props.setProperty("defaultTopicQueueNums",
command.getReadQueueNums().toString());
+ }
+ if (command.getBrokerPermission() != null) {
+ props.setProperty("brokerPermission",
command.getBrokerPermission().toString());
+ }
+ return props;
+ }
+
private FlushDiskType parseFlushDiskType(String value) {
try {
return FlushDiskType.valueOf(value);
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ClientProviderStub.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ClientProviderStub.java
index f399226f..8ca03c03 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ClientProviderStub.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/client/ClientProviderStub.java
@@ -18,12 +18,10 @@ package org.apache.rocketmq.studio.cluster.client;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import lombok.extern.slf4j.Slf4j;
-import org.springframework.stereotype.Component;
import java.util.List;
@Slf4j
-@Component
public class ClientProviderStub implements ClientProvider {
@Override
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/dashboard/DashboardProviderStub.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/dashboard/DashboardProviderStub.java
index 83c67f8a..1400f8ef 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/dashboard/DashboardProviderStub.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/dashboard/DashboardProviderStub.java
@@ -19,10 +19,13 @@ package org.apache.rocketmq.studio.ops.dashboard;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import lombok.extern.slf4j.Slf4j;
-import org.springframework.stereotype.Component;
+/**
+ * Fails explicitly when no real dashboard provider is configured (trunk #699).
+ *
+ * <p>Not registered as a Spring bean: {@code RocketMQDashboardProvider}
serves live data.
+ */
@Slf4j
-@Component
public class DashboardProviderStub implements DashboardProvider {
@Override
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQAdminClientImpl.java
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQAdminClientImpl.java
new file mode 100644
index 00000000..69a0d039
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQAdminClientImpl.java
@@ -0,0 +1,486 @@
+/*
+ * 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.rocketmq;
+
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.common.TopicConfig;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import
org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
+import org.apache.rocketmq.studio.instance.topic.AdminClient;
+import org.apache.rocketmq.studio.instance.topic.SendMessageDTO;
+import org.apache.rocketmq.studio.instance.topic.SendMessageVO;
+import org.apache.rocketmq.studio.instance.topic.TopicVO;
+import org.apache.rocketmq.studio.ops.audit.AuditService;
+import org.apache.rocketmq.studio.persistence.entity.RmqGroup;
+import org.apache.rocketmq.studio.persistence.entity.RmqTopic;
+import org.apache.rocketmq.studio.persistence.mapper.RmqGroupMapper;
+import org.apache.rocketmq.studio.persistence.mapper.RmqTopicMapper;
+import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Primary;
+import org.springframework.stereotype.Service;
+import org.springframework.util.StringUtils;
+
+import java.nio.charset.StandardCharsets;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Real AdminClient implementation backed by DefaultMQAdminExt.
+ * Provides topic CRUD, message sending, consumer group CRUD, and offset reset.
+ */
+@Service
+@Primary
+public class RocketMQAdminClientImpl implements AdminClient {
+
+ private static final Logger log =
LoggerFactory.getLogger(RocketMQAdminClientImpl.class);
+
+ private final DefaultMQAdminExt adminExt;
+ private final RocketMQProperties properties;
+ private final RmqTopicMapper topicMapper;
+ private final RmqGroupMapper groupMapper;
+ private final AuditService auditService;
+
+ @Autowired
+ public RocketMQAdminClientImpl(
+ @Autowired(required = false) DefaultMQAdminExt adminExt,
+ RocketMQProperties properties,
+ RmqTopicMapper topicMapper,
+ RmqGroupMapper groupMapper,
+ AuditService auditService) {
+ this.adminExt = adminExt;
+ this.properties = properties;
+ this.topicMapper = topicMapper;
+ this.groupMapper = groupMapper;
+ this.auditService = auditService;
+ }
+
+ @Override
+ public TopicVO getTopic(String name) {
+ if (adminExt == null) {
+ throw new BusinessException(503, "RocketMQ admin not connected");
+ }
+ try {
+ var routeData = adminExt.examineTopicRouteInfo(name);
+ if (routeData == null || routeData.getQueueDatas() == null ||
routeData.getQueueDatas().isEmpty()) {
+ throw new BusinessException(404, "Topic not found: " + name);
+ }
+ var qd = routeData.getQueueDatas().get(0);
+ TopicVO vo = new TopicVO();
+ vo.setId(name);
+ vo.setName(name);
+ vo.setWriteQueues(qd.getWriteQueueNums());
+ vo.setReadQueues(qd.getReadQueueNums());
+ return vo;
+ } catch (BusinessException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new BusinessException(500, "Failed to get topic: " +
e.getMessage());
+ }
+ }
+
+ @Override
+ public ConsumerGroupVO getConsumerGroup(String name) {
+ if (adminExt == null) {
+ throw new BusinessException(503, "RocketMQ admin not connected");
+ }
+ ConsumerGroupVO vo = new ConsumerGroupVO();
+ vo.setId(name);
+ vo.setName(name);
+ try {
+ var conn = adminExt.examineConsumerConnectionInfo(name);
+ if (conn != null) {
+ if (conn.getConnectionSet() != null) {
+ vo.setOnlineInstances(conn.getConnectionSet().size());
+ }
+ if (conn.getSubscriptionTable() != null) {
+ vo.setSubscribedTopics(new
ArrayList<>(conn.getSubscriptionTable().keySet()));
+ }
+ }
+ } catch (Exception ignored) {
+ // Group may be offline
+ }
+ return vo;
+ }
+
+ @Override
+ public TopicVO createTopic(TopicVO topic) {
+ if (adminExt == null) {
+ throw new BusinessException(503, "RocketMQ admin not connected");
+ }
+
+ String topicName = topic.getName();
+ int writeQueues = topic.getWriteQueues() > 0 ? topic.getWriteQueues()
: 8;
+ int readQueues = topic.getReadQueues() > 0 ? topic.getReadQueues() : 8;
+
+ try {
+ Set<String> brokerAddrs = getAllMasterBrokerAddrs();
+ if (brokerAddrs.isEmpty()) {
+ throw new BusinessException(500, "No broker available to
create topic");
+ }
+
+ TopicConfig topicConfig = new TopicConfig();
+ topicConfig.setTopicName(topicName);
+ topicConfig.setWriteQueueNums(writeQueues);
+ topicConfig.setReadQueueNums(readQueues);
+ topicConfig.setPerm(6); // RW
+
+ for (String addr : brokerAddrs) {
+ adminExt.createAndUpdateTopicConfig(addr, topicConfig);
+ }
+
+ // Persist to DB. Re-creating a topic that already has a record
(for example when
+ // rebuilding a broker route from the console) must update it
instead of failing on
+ // the unique (cluster_id, name) key.
+ String clusterName = getClusterName();
+ RmqTopic entity = topicMapper.selectOne(new
LambdaQueryWrapper<RmqTopic>()
+ .eq(RmqTopic::getClusterId, clusterName)
+ .eq(RmqTopic::getName, topicName));
+ boolean isNew = entity == null;
+ if (isNew) {
+ entity = new RmqTopic();
+ entity.setName(topicName);
+ entity.setClusterId(clusterName);
+ entity.setCreatedAt(LocalDateTime.now());
+ }
+ entity.setTopicType(topic.getType() != null ?
topic.getType().name() : "NORMAL");
+ entity.setReadQueueNums(readQueues);
+ entity.setWriteQueueNums(writeQueues);
+ entity.setPerm(6);
+ if (StringUtils.hasText(topic.getRemark())) {
+ entity.setRemark(topic.getRemark());
+ }
+ entity.setStatus("ACTIVE");
+ entity.setUpdatedAt(LocalDateTime.now());
+ if (isNew) {
+ topicMapper.insert(entity);
+ } else {
+ topicMapper.updateById(entity);
+ }
+
+ auditService.record("CREATE_TOPIC", topicName,
+ "queues=" + writeQueues + "/" + readQueues, "SUCCESS");
+
+ topic.setId(topicName);
+ topic.setWriteQueues(writeQueues);
+ topic.setReadQueues(readQueues);
+ return topic;
+ } catch (BusinessException e) {
+ auditService.record("CREATE_TOPIC", topicName, e.getMessage(),
"FAILED");
+ throw e;
+ } catch (Exception e) {
+ auditService.record("CREATE_TOPIC", topicName, e.getMessage(),
"FAILED");
+ throw new BusinessException(500, "Failed to create topic: " +
e.getMessage());
+ }
+ }
+
+ @Override
+ public TopicVO updateTopic(TopicVO topic) {
+ if (adminExt == null) {
+ throw new BusinessException(503, "RocketMQ admin not connected");
+ }
+
+ String topicName = topic.getName();
+ int writeQueues = topic.getWriteQueues() > 0 ? topic.getWriteQueues()
: 8;
+ int readQueues = topic.getReadQueues() > 0 ? topic.getReadQueues() : 8;
+
+ try {
+ Set<String> brokerAddrs = getAllMasterBrokerAddrs();
+ if (brokerAddrs.isEmpty()) {
+ throw new BusinessException(500, "No broker available to
update topic");
+ }
+
+ TopicConfig topicConfig = new TopicConfig();
+ topicConfig.setTopicName(topicName);
+ topicConfig.setWriteQueueNums(writeQueues);
+ topicConfig.setReadQueueNums(readQueues);
+ topicConfig.setPerm(6); // RW
+
+ for (String addr : brokerAddrs) {
+ adminExt.createAndUpdateTopicConfig(addr, topicConfig);
+ }
+
+ // Update DB record
+ RmqTopic existing = topicMapper.selectOne(
+ new LambdaQueryWrapper<RmqTopic>().eq(RmqTopic::getName,
topicName));
+ if (existing != null) {
+ existing.setWriteQueueNums(writeQueues);
+ existing.setReadQueueNums(readQueues);
+ existing.setUpdatedAt(LocalDateTime.now());
+ topicMapper.updateById(existing);
+ }
+
+ auditService.record("UPDATE_TOPIC", topicName,
+ "queues=" + writeQueues + "/" + readQueues, "SUCCESS");
+
+ topic.setId(topicName);
+ topic.setWriteQueues(writeQueues);
+ topic.setReadQueues(readQueues);
+ return topic;
+ } catch (BusinessException e) {
+ auditService.record("UPDATE_TOPIC", topicName, e.getMessage(),
"FAILED");
+ throw e;
+ } catch (Exception e) {
+ auditService.record("UPDATE_TOPIC", topicName, e.getMessage(),
"FAILED");
+ throw new BusinessException(500, "Failed to update topic: " +
e.getMessage());
+ }
+ }
+
+ @Override
+ public void deleteTopic(String name) {
+ if (adminExt == null) {
+ throw new BusinessException(503, "RocketMQ admin not connected");
+ }
+
+ try {
+ Set<String> brokerAddrs = getAllMasterBrokerAddrs();
+
+ // Delete from brokers
+ if (!brokerAddrs.isEmpty()) {
+ adminExt.deleteTopicInBroker(brokerAddrs, name);
+ }
+
+ // Delete from nameserver
+ String namesrvAddr = properties.getNamesrvAddr();
+ if (namesrvAddr != null && !namesrvAddr.isEmpty()) {
+ Set<String> nsAddrs = new HashSet<>();
+ for (String addr : namesrvAddr.split("[;,]")) {
+ String trimmed = addr.trim();
+ if (!trimmed.isEmpty()) {
+ nsAddrs.add(trimmed);
+ }
+ }
+ adminExt.deleteTopicInNameServer(nsAddrs, getClusterName(),
name);
+ }
+
+ // Delete from DB
+ topicMapper.delete(new
LambdaQueryWrapper<RmqTopic>().eq(RmqTopic::getName, name));
+
+ auditService.record("DELETE_TOPIC", name, "", "SUCCESS");
+ } catch (BusinessException e) {
+ auditService.record("DELETE_TOPIC", name, e.getMessage(),
"FAILED");
+ throw e;
+ } catch (Exception e) {
+ auditService.record("DELETE_TOPIC", name, e.getMessage(),
"FAILED");
+ throw new BusinessException(500, "Failed to delete topic: " +
e.getMessage());
+ }
+ }
+
+ @Override
+ public SendMessageVO sendMessage(SendMessageDTO request) {
+ if (adminExt == null) {
+ throw new BusinessException(503, "RocketMQ admin not connected");
+ }
+
+ String namesrvAddr = properties.getNamesrvAddr();
+ if (namesrvAddr == null || namesrvAddr.isEmpty()) {
+ throw new BusinessException(500, "Nameserver address not
configured");
+ }
+
+ DefaultMQProducer producer = new
DefaultMQProducer("studio_msg_sender_" + System.currentTimeMillis());
+ producer.setNamesrvAddr(namesrvAddr);
+ producer.setSendMsgTimeout(5000);
+
+ try {
+ producer.start();
+
+ String topic = request.getTopic();
+ String tag = request.getTag() != null ? request.getTag() : "";
+ String key = request.getKey() != null ? request.getKey() : "";
+ String body = request.getBody() != null ? request.getBody() : "";
+
+ String fullTopic = tag.isEmpty() ? topic : topic + ":" + tag;
+ Message msg = new Message(topic, tag, key,
body.getBytes(StandardCharsets.UTF_8));
+
+ // Add custom properties
+ if (request.getProperties() != null) {
+ for (Map.Entry<String, String> entry :
request.getProperties().entrySet()) {
+ msg.putUserProperty(entry.getKey(), entry.getValue());
+ }
+ }
+
+ SendResult sendResult = producer.send(msg);
+
+ auditService.record("SEND_MESSAGE", topic,
+ "tag=" + tag + ", key=" + key + ", msgId=" +
sendResult.getMsgId(), "SUCCESS");
+
+ return SendMessageVO.builder()
+ .msgId(sendResult.getMsgId())
+ .sendTime(System.currentTimeMillis())
+ .offsetMsgId(sendResult.getOffsetMsgId())
+ .build();
+ } catch (Exception e) {
+ auditService.record("SEND_MESSAGE", request.getTopic(),
e.getMessage(), "FAILED");
+ throw new BusinessException(500, "Failed to send message: " +
e.getMessage());
+ } finally {
+ producer.shutdown();
+ }
+ }
+
+ @Override
+ public ConsumerGroupVO createConsumerGroup(ConsumerGroupVO group) {
+ if (adminExt == null) {
+ throw new BusinessException(503, "RocketMQ admin not connected");
+ }
+
+ String groupName = group.getName();
+
+ try {
+ Set<String> brokerAddrs = getAllMasterBrokerAddrs();
+ if (brokerAddrs.isEmpty()) {
+ throw new BusinessException(500, "No broker available to
create consumer group");
+ }
+
+ SubscriptionGroupConfig config = new SubscriptionGroupConfig();
+ config.setGroupName(groupName);
+ config.setConsumeEnable(true);
+ config.setConsumeBroadcastEnable(true);
+ config.setRetryQueueNums(1);
+ config.setRetryMaxTimes(group.getRetryMaxTimes() > 0 ?
group.getRetryMaxTimes() : 16);
+
+ for (String addr : brokerAddrs) {
+ adminExt.createAndUpdateSubscriptionGroupConfig(addr, config);
+ }
+
+ // Persist to DB, upserting so re-creating an existing group does
not violate the
+ // unique (cluster_id, name) key.
+ String groupClusterName = getClusterName();
+ RmqGroup entity = groupMapper.selectOne(new
LambdaQueryWrapper<RmqGroup>()
+ .eq(RmqGroup::getClusterId, groupClusterName)
+ .eq(RmqGroup::getName, groupName));
+ boolean isNewGroup = entity == null;
+ if (isNewGroup) {
+ entity = new RmqGroup();
+ entity.setName(groupName);
+ entity.setClusterId(groupClusterName);
+ entity.setCreatedAt(LocalDateTime.now());
+ }
+ entity.setConsumeType(group.getConsumeType() != null ?
group.getConsumeType().name() : "CLUSTERING");
+ entity.setMessageModel(group.getSubscriptionMode() != null ?
group.getSubscriptionMode().name() : "Push");
+ entity.setMaxRetry(config.getRetryMaxTimes());
+ entity.setStatus("ACTIVE");
+ entity.setUpdatedAt(LocalDateTime.now());
+ if (isNewGroup) {
+ groupMapper.insert(entity);
+ } else {
+ groupMapper.updateById(entity);
+ }
+
+ auditService.record("CREATE_GROUP", groupName,
+ "retryMaxTimes=" + config.getRetryMaxTimes(), "SUCCESS");
+
+ group.setId(groupName);
+ return group;
+ } catch (BusinessException e) {
+ auditService.record("CREATE_GROUP", groupName, e.getMessage(),
"FAILED");
+ throw e;
+ } catch (Exception e) {
+ auditService.record("CREATE_GROUP", groupName, e.getMessage(),
"FAILED");
+ throw new BusinessException(500, "Failed to create consumer group:
" + e.getMessage());
+ }
+ }
+
+ @Override
+ public void deleteConsumerGroup(String name) {
+ if (adminExt == null) {
+ throw new BusinessException(503, "RocketMQ admin not connected");
+ }
+
+ try {
+ Set<String> brokerAddrs = getAllMasterBrokerAddrs();
+
+ for (String addr : brokerAddrs) {
+ adminExt.deleteSubscriptionGroup(addr, name, true);
+ }
+
+ // Delete from DB
+ groupMapper.delete(new
LambdaQueryWrapper<RmqGroup>().eq(RmqGroup::getName, name));
+
+ auditService.record("DELETE_GROUP", name, "", "SUCCESS");
+ } catch (BusinessException e) {
+ auditService.record("DELETE_GROUP", name, e.getMessage(),
"FAILED");
+ throw e;
+ } catch (Exception e) {
+ auditService.record("DELETE_GROUP", name, e.getMessage(),
"FAILED");
+ throw new BusinessException(500, "Failed to delete consumer group:
" + e.getMessage());
+ }
+ }
+
+ @Override
+ public void resetOffset(String name, long timestamp, String topic) {
+ if (adminExt == null) {
+ throw new BusinessException(503, "RocketMQ admin not connected");
+ }
+
+ try {
+ adminExt.resetOffsetByTimestamp(getClusterName(), topic, name,
timestamp, false);
+ auditService.record("RESET_OFFSET", name,
+ "topic=" + topic + ", timestamp=" + timestamp, "SUCCESS");
+ } catch (Exception e) {
+ auditService.record("RESET_OFFSET", name, e.getMessage(),
"FAILED");
+ throw new BusinessException(500, "Failed to reset offset: " +
e.getMessage());
+ }
+ }
+
+ // ── Helper methods ──────────────────────────────────────────────────
+
+ private Set<String> getAllMasterBrokerAddrs() throws Exception {
+ Set<String> addrs = new HashSet<>();
+ ClusterInfo clusterInfo = adminExt.examineBrokerClusterInfo();
+ if (clusterInfo == null || clusterInfo.getBrokerAddrTable() == null) {
+ return addrs;
+ }
+
+ for (BrokerData brokerData :
clusterInfo.getBrokerAddrTable().values()) {
+ if (brokerData.getBrokerAddrs() == null) {
+ continue;
+ }
+ // Use master address (brokerId = 0) preferentially
+ String masterAddr = brokerData.getBrokerAddrs().get(0L);
+ if (masterAddr == null && !brokerData.getBrokerAddrs().isEmpty()) {
+ masterAddr =
brokerData.getBrokerAddrs().values().iterator().next();
+ }
+ if (masterAddr != null) {
+ addrs.add(masterAddr);
+ }
+ }
+ return addrs;
+ }
+
+ private String getClusterName() {
+ try {
+ ClusterInfo clusterInfo = adminExt.examineBrokerClusterInfo();
+ if (clusterInfo != null && clusterInfo.getClusterAddrTable() !=
null
+ && !clusterInfo.getClusterAddrTable().isEmpty()) {
+ return
clusterInfo.getClusterAddrTable().keySet().iterator().next();
+ }
+ } catch (Exception ignored) {
+ }
+ return "DefaultCluster";
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQAdminConfig.java
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQAdminConfig.java
new file mode 100644
index 00000000..8cf4e7f5
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQAdminConfig.java
@@ -0,0 +1,44 @@
+/*
+ * 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.rocketmq;
+
+import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import
org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+@EnableConfigurationProperties(RocketMQProperties.class)
+public class RocketMQAdminConfig {
+
+ private static final Logger log =
LoggerFactory.getLogger(RocketMQAdminConfig.class);
+
+ @Bean(initMethod = "start", destroyMethod = "shutdown")
+ @ConditionalOnProperty(prefix = "studio.rocketmq", name = "namesrv-addr")
+ public DefaultMQAdminExt defaultMQAdminExt(
+ @Value("${studio.rocketmq.namesrv-addr:}") String namesrvAddr) {
+ log.info("Initializing DefaultMQAdminExt with namesrvAddr=[{}]",
namesrvAddr);
+ DefaultMQAdminExt adminExt = new DefaultMQAdminExt();
+ adminExt.setNamesrvAddr(namesrvAddr);
+ adminExt.setInstanceName("studio-admin-" + System.currentTimeMillis());
+ return adminExt;
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQBrokerConfigService.java
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQBrokerConfigService.java
new file mode 100644
index 00000000..67415674
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQBrokerConfigService.java
@@ -0,0 +1,113 @@
+/*
+ * 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.rocketmq;
+
+import org.apache.rocketmq.studio.cluster.config.ClusterConfigVO;
+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.tools.admin.DefaultMQAdminExt;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.lang.Nullable;
+import org.springframework.stereotype.Service;
+
+import java.util.Properties;
+
+@Slf4j
+@Service
+public class RocketMQBrokerConfigService {
+
+ private final DefaultMQAdminExt adminExt;
+ private final AuditService auditService;
+
+ public RocketMQBrokerConfigService(@Nullable DefaultMQAdminExt adminExt,
AuditService auditService) {
+ this.adminExt = adminExt;
+ this.auditService = auditService;
+ }
+
+ /**
+ * Read broker config from the live broker via admin API.
+ */
+ public ClusterConfigVO getBrokerConfig(String brokerAddr) {
+ requireAdmin();
+ try {
+ Properties props = adminExt.getBrokerConfig(brokerAddr);
+ return mapToClusterConfigVO(props);
+ } catch (Exception e) {
+ log.error("Failed to get broker config from {}", brokerAddr, e);
+ throw new BusinessException(500, "Failed to get broker config: " +
e.getMessage());
+ }
+ }
+
+ /**
+ * Update broker config on the live broker via admin API, then record
audit.
+ */
+ public void updateBrokerConfig(String brokerAddr, String clusterId,
Properties newConfig) {
+ requireAdmin();
+ try {
+ adminExt.updateBrokerConfig(brokerAddr, newConfig);
+ String detail = "brokerAddr=" + brokerAddr + ", config=" +
newConfig;
+ auditService.record("UPDATE_BROKER_CONFIG", "CLUSTER:" +
clusterId, detail, "SUCCESS");
+ log.info("Broker config updated successfully: {}", brokerAddr);
+ } catch (Exception e) {
+ log.error("Failed to update broker config at {}", brokerAddr, e);
+ String detail = "brokerAddr=" + brokerAddr + ", error=" +
e.getMessage();
+ auditService.record("UPDATE_BROKER_CONFIG", "CLUSTER:" +
clusterId, detail, "FAILURE");
+ throw new BusinessException(500, "Failed to update broker config:
" + e.getMessage());
+ }
+ }
+
+ private ClusterConfigVO mapToClusterConfigVO(Properties props) {
+ ClusterConfigVO vo = new ClusterConfigVO();
+
vo.setFlushDiskType(parseFlushDiskType(props.getProperty("flushDiskType",
"ASYNC_FLUSH")));
+
vo.setAutoCreateTopicEnable(Boolean.parseBoolean(props.getProperty("autoCreateTopicEnable",
"true")));
+
vo.setAutoCreateSubscriptionGroup(Boolean.parseBoolean(props.getProperty("autoCreateSubscriptionGroup",
"true")));
+ vo.setMaxMessageSize(parseIntSafe(props.getProperty("maxMessageSize"),
4194304));
+
vo.setWriteQueueNums(parseIntSafe(props.getProperty("defaultTopicQueueNums"),
8));
+
vo.setReadQueueNums(parseIntSafe(props.getProperty("defaultTopicQueueNums"),
8));
+
vo.setFileReservedTime(parseIntSafe(props.getProperty("fileReservedTime"), 72));
+
vo.setBrokerPermission(parseIntSafe(props.getProperty("brokerPermission"), 6));
+ vo.setDeleteWhen(props.getProperty("deleteWhen", "04"));
+ vo.setMsgTraceTopicName(props.getProperty("msgTraceTopicName",
"RMQ_SYS_TRACE_TOPIC"));
+ return vo;
+ }
+
+ private FlushDiskType parseFlushDiskType(String value) {
+ try {
+ return FlushDiskType.valueOf(value);
+ } catch (IllegalArgumentException e) {
+ return FlushDiskType.ASYNC_FLUSH;
+ }
+ }
+
+ private int parseIntSafe(String value, int defaultValue) {
+ if (value == null || value.isEmpty()) {
+ return defaultValue;
+ }
+ try {
+ return Integer.parseInt(value.trim());
+ } catch (NumberFormatException e) {
+ return defaultValue;
+ }
+ }
+
+ private void requireAdmin() {
+ if (adminExt == null) {
+ throw new BusinessException(503, "RocketMQ admin is not
configured. Set studio.rocketmq.namesrv-addr.");
+ }
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQClientProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQClientProvider.java
new file mode 100644
index 00000000..c2d3d389
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQClientProvider.java
@@ -0,0 +1,264 @@
+/*
+ * 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.rocketmq;
+
+import org.apache.rocketmq.remoting.protocol.LanguageCode;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.Connection;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ProducerConnection;
+import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupWrapper;
+import org.apache.rocketmq.remoting.protocol.body.TopicList;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import org.apache.rocketmq.studio.cluster.client.ClientConnectionVO;
+import org.apache.rocketmq.studio.cluster.client.ClientProvider;
+import org.apache.rocketmq.studio.common.domain.enums.ClientLanguage;
+import org.apache.rocketmq.studio.common.domain.enums.ClientType;
+import org.apache.rocketmq.studio.common.domain.enums.Protocol;
+import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.context.annotation.Primary;
+import org.springframework.stereotype.Service;
+
+import java.util.ArrayList;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Set;
+
+/**
+ * Live {@link ClientProvider} backed by the RocketMQ admin API. It discovers
producer
+ * connections by scanning non-system topics and consumer connections by
scanning
+ * subscription groups across all brokers in the cluster.
+ */
+@Slf4j
+@Service
+@Primary
+public class RocketMQClientProvider implements ClientProvider {
+
+ /**
+ * Upper bound on the number of non-system topics scanned for producer
connections,
+ * to avoid issuing an admin call per topic on clusters with a large topic
count.
+ */
+ private static final int MAX_PRODUCER_TOPIC_SCAN = 50;
+
+ private static final long SUBSCRIPTION_GROUP_TIMEOUT_MILLIS = 5000L;
+
+ private final ObjectProvider<DefaultMQAdminExt> adminExtProvider;
+
+ public RocketMQClientProvider(ObjectProvider<DefaultMQAdminExt>
adminExtProvider) {
+ this.adminExtProvider = adminExtProvider;
+ }
+
+ @Override
+ public List<ClientConnectionVO> findConnections(String clusterId, String
type) {
+ DefaultMQAdminExt adminExt = adminExtProvider.getIfAvailable();
+ if (adminExt == null) {
+ log.warn("DefaultMQAdminExt is not configured, returning empty
client connection list");
+ return List.of();
+ }
+
+ ClientType clientType = parseType(type);
+ List<ClientConnectionVO> connections = new ArrayList<>();
+ if (clientType == null || clientType == ClientType.Producer) {
+ connections.addAll(findProducerConnections(adminExt, clusterId));
+ }
+ if (clientType == null || clientType == ClientType.Consumer) {
+ connections.addAll(findConsumerConnections(adminExt, clusterId));
+ }
+ return connections;
+ }
+
+ private List<ClientConnectionVO> findProducerConnections(DefaultMQAdminExt
adminExt, String clusterId) {
+ List<ClientConnectionVO> result = new ArrayList<>();
+ Set<String> topics;
+ try {
+ TopicList topicList = adminExt.fetchAllTopicList();
+ topics = topicList == null ? Set.of() : topicList.getTopicList();
+ } catch (Exception e) {
+ log.warn("Failed to fetch topic list for producer connection
scan", e);
+ return result;
+ }
+
+ int scanned = 0;
+ for (String topic : topics) {
+ if (isSystemTopic(topic)) {
+ continue;
+ }
+ if (scanned >= MAX_PRODUCER_TOPIC_SCAN) {
+ log.info("Producer connection scan capped at {} non-system
topics", MAX_PRODUCER_TOPIC_SCAN);
+ break;
+ }
+ scanned++;
+ try {
+ ProducerConnection producerConnection =
adminExt.examineProducerConnectionInfo(null, topic);
+ if (producerConnection == null ||
producerConnection.getConnectionSet() == null) {
+ continue;
+ }
+ for (Connection connection :
producerConnection.getConnectionSet()) {
+ result.add(toConnectionVO(connection, ClientType.Producer,
topic, topic, clusterId));
+ }
+ } catch (Exception e) {
+ log.warn("Failed to examine producer connection for topic={},
skipping", topic, e);
+ }
+ }
+ return result;
+ }
+
+ private List<ClientConnectionVO> findConsumerConnections(DefaultMQAdminExt
adminExt, String clusterId) {
+ List<ClientConnectionVO> result = new ArrayList<>();
+ Set<String> groups = collectSubscriptionGroups(adminExt);
+ for (String group : groups) {
+ if (isSystemGroup(group)) {
+ continue;
+ }
+ try {
+ ConsumerConnection consumerConnection =
adminExt.examineConsumerConnectionInfo(group);
+ if (consumerConnection == null ||
consumerConnection.getConnectionSet() == null) {
+ continue;
+ }
+ for (Connection connection :
consumerConnection.getConnectionSet()) {
+ result.add(toConnectionVO(connection, ClientType.Consumer,
group, null, clusterId));
+ }
+ } catch (Exception e) {
+ log.warn("Failed to examine consumer connection for group={},
skipping", group, e);
+ }
+ }
+ return result;
+ }
+
+ private Set<String> collectSubscriptionGroups(DefaultMQAdminExt adminExt) {
+ Set<String> groups = new LinkedHashSet<>();
+ ClusterInfo clusterInfo;
+ try {
+ clusterInfo = adminExt.examineBrokerClusterInfo();
+ } catch (Exception e) {
+ log.warn("Failed to fetch cluster info for consumer connection
scan", e);
+ return groups;
+ }
+ if (clusterInfo == null || clusterInfo.getBrokerAddrTable() == null) {
+ return groups;
+ }
+ for (BrokerData brokerData :
clusterInfo.getBrokerAddrTable().values()) {
+ String brokerAddr = brokerData.selectBrokerAddr();
+ if (brokerAddr == null) {
+ continue;
+ }
+ try {
+ SubscriptionGroupWrapper wrapper =
+ adminExt.getAllSubscriptionGroup(brokerAddr,
SUBSCRIPTION_GROUP_TIMEOUT_MILLIS);
+ if (wrapper != null && wrapper.getSubscriptionGroupTable() !=
null) {
+
groups.addAll(wrapper.getSubscriptionGroupTable().keySet());
+ }
+ } catch (Exception e) {
+ log.warn("Failed to fetch subscription groups from broker={},
skipping", brokerAddr, e);
+ }
+ }
+ return groups;
+ }
+
+ private ClientConnectionVO toConnectionVO(Connection connection,
ClientType type, String groupOrTopic,
+ String producerGroup, String
clusterId) {
+ return ClientConnectionVO.builder()
+ .clientId(connection.getClientId())
+ .type(type)
+ .groupOrTopic(groupOrTopic)
+ .producerGroup(producerGroup)
+ .protocol(Protocol.Remoting)
+ .address(connection.getClientAddr())
+ .language(mapLanguage(connection.getLanguage()))
+ .version(String.valueOf(connection.getVersion()))
+ .clusterName(clusterId)
+ .build();
+ }
+
+ private ClientType parseType(String type) {
+ if (type == null || type.isBlank()) {
+ return null;
+ }
+ try {
+ return ClientType.valueOf(type.trim());
+ } catch (IllegalArgumentException e) {
+ String normalized = type.trim().toLowerCase(Locale.ROOT);
+ if (normalized.startsWith("prod")) {
+ return ClientType.Producer;
+ }
+ if (normalized.startsWith("cons")) {
+ return ClientType.Consumer;
+ }
+ log.warn("Unknown client type filter: {}", type);
+ return null;
+ }
+ }
+
+ private ClientLanguage mapLanguage(LanguageCode languageCode) {
+ if (languageCode == null) {
+ return null;
+ }
+ switch (languageCode) {
+ case JAVA:
+ return ClientLanguage.Java;
+ case GO:
+ return ClientLanguage.Go;
+ case PYTHON:
+ return ClientLanguage.Python;
+ case RUST:
+ return ClientLanguage.Rust;
+ case CPP:
+ return ClientLanguage.Cpp;
+ case DOTNET:
+ return ClientLanguage.CSharp;
+ case PHP:
+ return ClientLanguage.PHP;
+ default:
+ return null;
+ }
+ }
+
+ private boolean isSystemTopic(String topic) {
+ if (topic == null) {
+ return true;
+ }
+ return topic.startsWith("RMQ_SYS_")
+ || topic.startsWith("SCHEDULE_TOPIC_")
+ || topic.startsWith("%RETRY%")
+ || topic.startsWith("%DLQ%")
+ || topic.startsWith("TBW102")
+ || topic.startsWith("SELF_TEST_")
+ || topic.startsWith("DefaultCluster")
+ || topic.startsWith("broker_")
+ || topic.startsWith("OFFSET_MOVED_")
+ || topic.startsWith("CID_RMQ_SYS_")
+ || topic.startsWith("TRANS_CHECK_")
+ || topic.startsWith("BenchmarkTest");
+ }
+
+ private boolean isSystemGroup(String group) {
+ if (group == null) {
+ return true;
+ }
+ return group.startsWith("CID_RMQ_SYS_")
+ || group.startsWith("CID_ONSAPI_")
+ || group.startsWith("TOOLS_CONSUMER")
+ || group.startsWith("FILTERSRV_CONSUMER")
+ || group.startsWith("CID_SYS_")
+ || group.startsWith("%RETRY%")
+ || group.startsWith("SELF_TEST_")
+ || group.startsWith("CID_HOUSEKEEPING");
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQClusterProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQClusterProvider.java
new file mode 100644
index 00000000..f85d874b
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQClusterProvider.java
@@ -0,0 +1,250 @@
+/*
+ * 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.rocketmq;
+
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.KVTable;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import org.apache.rocketmq.studio.cluster.broker.BrokerVO;
+import org.apache.rocketmq.studio.cluster.broker.ClusterProvider;
+import org.apache.rocketmq.studio.cluster.broker.ClusterVO;
+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.tools.admin.DefaultMQAdminExt;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Primary;
+import org.springframework.stereotype.Service;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Real cluster discovery implementation using DefaultMQAdminExt.
+ * Falls back to returning empty results when adminExt is not configured or
connection fails.
+ */
+@Service
+@Primary
+public class RocketMQClusterProvider implements ClusterProvider {
+
+ private static final Logger log =
LoggerFactory.getLogger(RocketMQClusterProvider.class);
+
+ private final DefaultMQAdminExt adminExt;
+ private final RocketMQProperties properties;
+
+ @Autowired
+ public RocketMQClusterProvider(
+ @Autowired(required = false) DefaultMQAdminExt adminExt,
+ RocketMQProperties properties) {
+ this.adminExt = adminExt;
+ this.properties = properties;
+ }
+
+ @Override
+ public List<ClusterVO> discoverClusters() {
+ if (adminExt == null) {
+ log.debug("DefaultMQAdminExt not configured, returning empty
cluster list");
+ return Collections.emptyList();
+ }
+
+ try {
+ ClusterInfo clusterInfo = adminExt.examineBrokerClusterInfo();
+ if (clusterInfo == null || clusterInfo.getClusterAddrTable() ==
null) {
+ return Collections.emptyList();
+ }
+
+ Map<String, Set<String>> clusterAddrTable =
clusterInfo.getClusterAddrTable();
+ Map<String, BrokerData> brokerAddrTable =
clusterInfo.getBrokerAddrTable();
+
+ List<ClusterVO> clusters = new ArrayList<>();
+ for (Map.Entry<String, Set<String>> entry :
clusterAddrTable.entrySet()) {
+ String clusterName = entry.getKey();
+ Set<String> brokerNames = entry.getValue();
+
+ List<BrokerVO> brokers = buildBrokerList(brokerNames,
brokerAddrTable);
+ List<NameServerVO> nameServers = buildNameServerList();
+
+ ClusterVO cluster = ClusterVO.builder()
+ .name(clusterName)
+ .status(ClusterStatus.healthy)
+ .brokers(brokers)
+ .nameServers(nameServers)
+ .build();
+ cluster.setId(clusterName);
+ clusters.add(cluster);
+ }
+ return clusters;
+ } catch (Exception e) {
+ log.warn("Failed to discover clusters via NameServer: {}",
e.getMessage());
+ return Collections.emptyList();
+ }
+ }
+
+ @Override
+ public ClusterVO refreshClusterDetail(String clusterId) {
+ if (adminExt == null) {
+ log.debug("DefaultMQAdminExt not configured, cannot refresh
cluster detail");
+ return null;
+ }
+
+ try {
+ ClusterInfo clusterInfo = adminExt.examineBrokerClusterInfo();
+ if (clusterInfo == null || clusterInfo.getClusterAddrTable() ==
null) {
+ return null;
+ }
+
+ Map<String, Set<String>> clusterAddrTable =
clusterInfo.getClusterAddrTable();
+ Map<String, BrokerData> brokerAddrTable =
clusterInfo.getBrokerAddrTable();
+
+ Set<String> brokerNames = clusterAddrTable.get(clusterId);
+ if (brokerNames == null) {
+ return null;
+ }
+
+ List<BrokerVO> brokers = buildBrokerList(brokerNames,
brokerAddrTable);
+ List<NameServerVO> nameServers = buildNameServerList();
+
+ ClusterVO cluster = ClusterVO.builder()
+ .name(clusterId)
+ .status(ClusterStatus.healthy)
+ .brokers(brokers)
+ .nameServers(nameServers)
+ .build();
+ cluster.setId(clusterId);
+ return cluster;
+ } catch (Exception e) {
+ log.warn("Failed to refresh cluster detail for {}: {}", clusterId,
e.getMessage());
+ return null;
+ }
+ }
+
+ private List<BrokerVO> buildBrokerList(Set<String> brokerNames,
+ Map<String, BrokerData>
brokerAddrTable) {
+ List<BrokerVO> brokers = new ArrayList<>();
+ if (brokerNames == null || brokerAddrTable == null) {
+ return brokers;
+ }
+
+ for (String brokerName : brokerNames) {
+ BrokerData brokerData = brokerAddrTable.get(brokerName);
+ if (brokerData == null || brokerData.getBrokerAddrs() == null) {
+ continue;
+ }
+
+ // Use master address (brokerId = 0) preferentially
+ String masterAddr = brokerData.getBrokerAddrs().get(0L);
+ if (masterAddr == null && !brokerData.getBrokerAddrs().isEmpty()) {
+ masterAddr =
brokerData.getBrokerAddrs().values().iterator().next();
+ }
+ if (masterAddr == null) {
+ continue;
+ }
+
+ BrokerVO.BrokerVOBuilder builder = BrokerVO.builder()
+ .name(brokerName)
+ .addr(masterAddr)
+ .status(BrokerStatus.running);
+
+ // Try to get runtime info for version and TPS
+ enrichBrokerWithRuntimeInfo(builder, masterAddr);
+
+ brokers.add(builder.build());
+ }
+ return brokers;
+ }
+
+ private void enrichBrokerWithRuntimeInfo(BrokerVO.BrokerVOBuilder builder,
String brokerAddr) {
+ try {
+ KVTable runtimeInfo = adminExt.fetchBrokerRuntimeStats(brokerAddr);
+ if (runtimeInfo == null || runtimeInfo.getTable() == null) {
+ return;
+ }
+
+ Map<String, String> table = runtimeInfo.getTable();
+
+ String version = table.getOrDefault("brokerVersionDesc",
+ table.getOrDefault("rocketmqVersion", ""));
+ if (!version.isEmpty()) {
+ builder.version(version);
+ }
+
+ // Parse TPS from runtime stats
+ String putTps = table.get("putTps");
+ if (putTps != null && !putTps.isEmpty()) {
+ builder.tpsIn(parseFirstTpsValue(putTps));
+ }
+
+ String getTransferredTps = table.get("getTransferedTps");
+ if (getTransferredTps != null && !getTransferredTps.isEmpty()) {
+ builder.tpsOut(parseFirstTpsValue(getTransferredTps));
+ }
+
+ // Disk usage
+ String diskRatio = table.get("commitLogDiskRatio");
+ if (diskRatio != null && !diskRatio.isEmpty()) {
+ try {
+ builder.diskUsage(Double.parseDouble(diskRatio));
+ } catch (NumberFormatException ignored) {
+ // keep default
+ }
+ }
+ } catch (Exception e) {
+ log.debug("Failed to get runtime info for broker at {}: {}",
brokerAddr, e.getMessage());
+ }
+ }
+
+ /**
+ * TPS properties are space-separated values representing 10s/1min/10min
averages.
+ * Parse the first value (10s average).
+ */
+ private int parseFirstTpsValue(String tpsStr) {
+ try {
+ String[] parts = tpsStr.split(" ");
+ if (parts.length > 0) {
+ return (int) Double.parseDouble(parts[0]);
+ }
+ } catch (NumberFormatException ignored) {
+ // fall through
+ }
+ return 0;
+ }
+
+ private List<NameServerVO> buildNameServerList() {
+ List<NameServerVO> nameServers = new ArrayList<>();
+ String namesrvAddr = properties.getNamesrvAddr();
+ if (namesrvAddr == null || namesrvAddr.isEmpty()) {
+ return nameServers;
+ }
+
+ String[] addrs = namesrvAddr.split("[;,]");
+ for (String addr : addrs) {
+ String trimmed = addr.trim();
+ if (!trimmed.isEmpty()) {
+ nameServers.add(NameServerVO.builder()
+ .addr(trimmed)
+ .status(ClusterStatus.healthy)
+ .build());
+ }
+ }
+ return nameServers;
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQDashboardProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQDashboardProvider.java
new file mode 100644
index 00000000..c6365f5d
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQDashboardProvider.java
@@ -0,0 +1,280 @@
+/*
+ * 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.rocketmq;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.KVTable;
+import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupWrapper;
+import org.apache.rocketmq.remoting.protocol.body.TopicList;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import
org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
+import org.apache.rocketmq.studio.common.domain.enums.ClusterStatus;
+import org.apache.rocketmq.studio.common.domain.enums.ClusterType;
+import org.apache.rocketmq.studio.ops.dashboard.ClusterOverviewVO;
+import org.apache.rocketmq.studio.ops.dashboard.DashboardDataVO;
+import org.apache.rocketmq.studio.ops.dashboard.DashboardProvider;
+import org.apache.rocketmq.studio.ops.dashboard.DashboardStatsVO;
+import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Primary;
+import org.springframework.lang.Nullable;
+import org.springframework.stereotype.Service;
+
+@Service
+@Primary
+public class RocketMQDashboardProvider implements DashboardProvider {
+
+ private static final Logger log =
LoggerFactory.getLogger(RocketMQDashboardProvider.class);
+
+ private static final Set<String> SYSTEM_TOPIC_PREFIXES = Set.of(
+ "rmq_sys_", "SCHEDULE_TOPIC_", "RMQ_SYS_", "broker_", "%RETRY%",
"%DLQ%",
+ "TBW102", "SELF_TEST_TOPIC", "BenchmarkTest", "OFFSET_MOVED_EVENT",
+ "DefaultCluster", "broker-a", "broker-b"
+ );
+
+ private final DefaultMQAdminExt adminExt;
+
+ public RocketMQDashboardProvider(@Nullable DefaultMQAdminExt adminExt) {
+ this.adminExt = adminExt;
+ }
+
+ @Override
+ public DashboardDataVO getDashboardData() {
+ if (adminExt == null) {
+ log.warn("DefaultMQAdminExt not available, returning empty
dashboard");
+ return emptyDashboard();
+ }
+
+ int totalClusters = 0;
+ int totalBrokers = 0;
+ int totalTopics = 0;
+ int totalGroups = 0;
+ long tpsIn = 0;
+ long tpsOut = 0;
+ long messagesToday = 0;
+ List<ClusterOverviewVO> clusters = new ArrayList<>();
+
+ try {
+ ClusterInfo clusterInfo = adminExt.examineBrokerClusterInfo();
+ Map<String, Set<String>> clusterAddrTable =
clusterInfo.getClusterAddrTable();
+ Map<String, BrokerData> brokerAddrTable =
clusterInfo.getBrokerAddrTable();
+
+ totalClusters = clusterAddrTable.size();
+ totalBrokers = brokerAddrTable.size();
+
+ // Collect all unique broker addresses (master only, brokerId=0)
+ Set<String> masterAddrs = new HashSet<>();
+ for (BrokerData brokerData : brokerAddrTable.values()) {
+ String masterAddr = brokerData.getBrokerAddrs().get(0L);
+ if (masterAddr != null) {
+ masterAddrs.add(masterAddr);
+ }
+ }
+
+ // Count topics
+ try {
+ TopicList topicList = adminExt.fetchAllTopicList();
+ Set<String> topics = topicList.getTopicList();
+ totalTopics = (int) topics.stream()
+ .filter(t -> !isSystemTopic(t))
+ .count();
+ } catch (Exception e) {
+ log.warn("Failed to fetch topic list: {}", e.getMessage());
+ }
+
+ // Count subscription groups and collect TPS from each master
broker
+ Set<String> allGroups = new HashSet<>();
+ for (String brokerAddr : masterAddrs) {
+ try {
+ SubscriptionGroupWrapper subscriptionGroupWrapper =
adminExt.getAllSubscriptionGroup(brokerAddr, 5000);
+ if (subscriptionGroupWrapper != null &&
subscriptionGroupWrapper.getSubscriptionGroupTable() != null) {
+ for (Map.Entry<String, SubscriptionGroupConfig> entry :
+
subscriptionGroupWrapper.getSubscriptionGroupTable().entrySet()) {
+ String groupName = entry.getKey();
+ if (!isSystemGroup(groupName)) {
+ allGroups.add(groupName);
+ }
+ }
+ }
+ } catch (Exception e) {
+ log.warn("Failed to get subscription groups from broker
{}: {}", brokerAddr, e.getMessage());
+ }
+
+ // Get runtime stats for TPS
+ try {
+ KVTable runtimeInfo =
adminExt.fetchBrokerRuntimeStats(brokerAddr);
+ if (runtimeInfo != null && runtimeInfo.getTable() != null)
{
+ Map<String, String> table = runtimeInfo.getTable();
+ tpsIn += parseTps(table.get("putTps"));
+ tpsOut += parseTps(table.get("getTransferedTps"));
+
+ String msgPutToday =
table.get("msgPutTotalTodayMorning");
+ if (msgPutToday != null) {
+ try {
+ messagesToday +=
Long.parseLong(msgPutToday.trim());
+ } catch (NumberFormatException ignored) {
+ }
+ }
+ }
+ } catch (Exception e) {
+ log.warn("Failed to get runtime info from broker {}: {}",
brokerAddr, e.getMessage());
+ }
+ }
+ totalGroups = allGroups.size();
+
+ // Build per-cluster overview
+ for (Map.Entry<String, Set<String>> clusterEntry :
clusterAddrTable.entrySet()) {
+ String clusterName = clusterEntry.getKey();
+ Set<String> brokerNames = clusterEntry.getValue();
+ int clusterBrokers = 0;
+ int clusterTpsIn = 0;
+ int clusterTpsOut = 0;
+ String version = "unknown";
+
+ for (String brokerName : brokerNames) {
+ BrokerData brokerData = brokerAddrTable.get(brokerName);
+ if (brokerData != null) {
+ clusterBrokers++;
+ String masterAddr =
brokerData.getBrokerAddrs().get(0L);
+ if (masterAddr != null) {
+ try {
+ KVTable rt =
adminExt.fetchBrokerRuntimeStats(masterAddr);
+ if (rt != null && rt.getTable() != null) {
+ clusterTpsIn += (int)
parseTps(rt.getTable().get("putTps"));
+ clusterTpsOut += (int)
parseTps(rt.getTable().get("getTransferedTps"));
+ String v =
rt.getTable().get("brokerVersionDesc");
+ if (v != null &&
!"unknown".equals(version)) {
+ version = v;
+ }
+ }
+ } catch (Exception ignored) {
+ }
+ }
+ }
+ }
+
+ clusters.add(ClusterOverviewVO.builder()
+ .id(clusterName)
+ .name(clusterName)
+ .type(ClusterType.V5_PROXY_CLUSTER)
+ .status(ClusterStatus.healthy)
+ .brokers(clusterBrokers)
+ .proxies(0)
+ .topics(0)
+ .groups(0)
+ .tpsIn(clusterTpsIn)
+ .tpsOut(clusterTpsOut)
+ .version(version)
+ .throughput(List.of())
+ .build());
+ }
+
+ } catch (Exception e) {
+ log.error("Failed to collect dashboard data from RocketMQ
cluster", e);
+ return emptyDashboard();
+ }
+
+ long messagesPerSecond = tpsIn + tpsOut;
+
+ DashboardStatsVO stats = DashboardStatsVO.builder()
+ .totalClusters(totalClusters)
+ .healthyClusters(totalClusters)
+ .totalBrokers(totalBrokers)
+ .totalProxies(0)
+ .totalNameServers(0)
+ .totalTopics(totalTopics)
+ .totalConsumerGroups(totalGroups)
+ .totalMessagesToday(messagesToday)
+ .messagesPerSecond(messagesPerSecond)
+ .tpsIn(tpsIn)
+ .tpsOut(tpsOut)
+ .build();
+
+ return DashboardDataVO.builder()
+ .stats(stats)
+ .clusters(clusters)
+ .build();
+ }
+
+ private DashboardDataVO emptyDashboard() {
+ DashboardStatsVO stats = DashboardStatsVO.builder()
+ .totalClusters(0)
+ .healthyClusters(0)
+ .totalBrokers(0)
+ .totalProxies(0)
+ .totalNameServers(0)
+ .totalTopics(0)
+ .totalConsumerGroups(0)
+ .totalMessagesToday(0)
+ .messagesPerSecond(0)
+ .tpsIn(0)
+ .tpsOut(0)
+ .build();
+ return DashboardDataVO.builder()
+ .stats(stats)
+ .clusters(List.of())
+ .build();
+ }
+
+ /**
+ * Parse TPS value from RocketMQ runtime stats format: "10minAvg 1minAvg
10secAvg"
+ * Returns the 1-minute average (second value) as a long.
+ */
+ private long parseTps(String tpsStr) {
+ if (tpsStr == null || tpsStr.isBlank()) {
+ return 0;
+ }
+ try {
+ String[] parts = tpsStr.trim().split("\\s+");
+ if (parts.length >= 2) {
+ return (long) Double.parseDouble(parts[1]);
+ } else if (parts.length == 1) {
+ return (long) Double.parseDouble(parts[0]);
+ }
+ } catch (NumberFormatException e) {
+ log.debug("Failed to parse TPS value: {}", tpsStr);
+ }
+ return 0;
+ }
+
+ private boolean isSystemTopic(String topic) {
+ for (String prefix : SYSTEM_TOPIC_PREFIXES) {
+ if (topic.startsWith(prefix) || topic.equals(prefix)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private boolean isSystemGroup(String group) {
+ return group.startsWith("CID_RMQ_SYS_")
+ || group.startsWith("rmq_sys_")
+ || group.equals("TOOLS_CONSUMER")
+ || group.equals("FILTERSRV_CONSUMER")
+ || group.equals("CID_ONSAPI_OWNER")
+ || group.equals("CID_ONSAPI_PERMISSION")
+ || group.equals("CID_ONSAPI_PULL")
+ || group.startsWith("SELF_TEST_");
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/dashboard/DashboardProviderStub.java
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQProperties.java
similarity index 59%
copy from
server/src/main/java/org/apache/rocketmq/studio/ops/dashboard/DashboardProviderStub.java
copy to
server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQProperties.java
index 83c67f8a..5c4bbaac 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/dashboard/DashboardProviderStub.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/rocketmq/RocketMQProperties.java
@@ -14,20 +14,13 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package org.apache.rocketmq.studio.rocketmq;
-package org.apache.rocketmq.studio.ops.dashboard;
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
-import org.apache.rocketmq.studio.common.exception.BusinessException;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.stereotype.Component;
-
-@Slf4j
-@Component
-public class DashboardProviderStub implements DashboardProvider {
-
- @Override
- public DashboardDataVO getDashboardData() {
- log.warn("DashboardProviderStub.getDashboardData called without a real
dashboard provider");
- throw new BusinessException(501, "Dashboard provider is not
configured");
- }
+@Data
+@ConfigurationProperties(prefix = "studio.rocketmq")
+public class RocketMQProperties {
+ private String namesrvAddr;
}
