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 24715633 fix(topic): persist type/remark, scope operations to target
cluster, reset-offset validation (#1399, #1429, #1442)
24715633 is described below
commit 24715633108a0c592fc80ca23311c371c263e3f8
Author: Yu Xinqiang <[email protected]>
AuthorDate: Tue Aug 11 00:28:16 2026 +0800
fix(topic): persist type/remark, scope operations to target cluster,
reset-offset validation (#1399, #1429, #1442)
* [ISSUE #1392] Persist topicType and remark in updateTopic, remove dead
variable
- updateTopic now persists topicType and remark to the database,
matching the behavior of createTopic
- Removed dead fullTopic variable in sendMessage that was computed
but never used
Fixes #1392
* [ISSUE #1428] Scope topic/group operations to target cluster brokers
getAllMasterBrokerAddrs returned all brokers across all clusters,
causing topics and groups to be created on brokers outside the
target cluster. This fix:
- Adds getMasterBrokerAddrsForCluster that filters by cluster name
using clusterAddrTable
- Updates createTopic, updateTopic, createConsumerGroup to use the
cluster-scoped method
- Falls back to all brokers when cluster name is unknown
Fixes #1428
* [ISSUE #1441] Validate topic field in reset offset
The reset offset endpoint accepted null/blank topic without validation,
causing a 500 error instead of a clear 400. This fix adds @NotBlank to
the DTO and runtime validation in the admin client.
Fixes #1441
---
.../instance/group/ResetConsumerOffsetDTO.java | 1 +
.../provider/apache/RocketMQAdminClientImpl.java | 52 ++++++++++++++++++++--
2 files changed, 49 insertions(+), 4 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/ResetConsumerOffsetDTO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ResetConsumerOffsetDTO.java
index 80ea5f75..3cc3c318 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/ResetConsumerOffsetDTO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ResetConsumerOffsetDTO.java
@@ -39,5 +39,6 @@ public class ResetConsumerOffsetDTO {
@Positive(message = "timestamp must be positive")
private Long timestamp;
+ @NotBlank(message = "topic is required")
private String topic;
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
index ec80dfda..c09b5b4c 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
@@ -149,7 +149,7 @@ public class RocketMQAdminClientImpl implements AdminClient
{
TopicPerm effectivePerm = topic.getPerm() != null
? topic.getPerm()
: existing == null ? TopicPerm.RW :
fromRocketMQPerm(existing.getPerm());
- Set<String> brokerAddrs = getAllMasterBrokerAddrs(admin);
+ Set<String> brokerAddrs =
getMasterBrokerAddrsForCluster(admin, clusterName);
if (brokerAddrs.isEmpty()) {
throw new BusinessException(500, "No broker available to
create topic");
}
@@ -230,7 +230,7 @@ public class RocketMQAdminClientImpl implements AdminClient
{
TopicPerm effectivePerm = topic.getPerm() != null
? topic.getPerm()
: existing == null ? TopicPerm.RW :
fromRocketMQPerm(existing.getPerm());
- Set<String> brokerAddrs = getAllMasterBrokerAddrs(admin);
+ Set<String> brokerAddrs =
getMasterBrokerAddrsForCluster(admin, clusterName);
if (brokerAddrs.isEmpty()) {
throw new BusinessException(500, "No broker available to
update topic");
}
@@ -250,6 +250,12 @@ public class RocketMQAdminClientImpl implements
AdminClient {
existing.setWriteQueueNums(writeQueues);
existing.setReadQueueNums(readQueues);
existing.setPerm(topicConfig.getPerm());
+ if (topic.getType() != null) {
+ existing.setTopicType(topic.getType().name());
+ }
+ if (StringUtils.hasText(topic.getRemark())) {
+ existing.setRemark(topic.getRemark());
+ }
existing.setUpdatedAt(LocalDateTime.now());
topicMapper.updateById(existing);
}
@@ -384,7 +390,8 @@ public class RocketMQAdminClientImpl implements AdminClient
{
String groupName = group.getName();
try {
- Set<String> brokerAddrs = getAllMasterBrokerAddrs(admin);
+ String groupClusterName = getClusterName(admin);
+ Set<String> brokerAddrs = getMasterBrokerAddrsForCluster(admin,
groupClusterName);
if (brokerAddrs.isEmpty()) {
throw new BusinessException(500, "No broker available to
create consumer group");
}
@@ -402,7 +409,6 @@ public class RocketMQAdminClientImpl implements AdminClient
{
// Persist to DB, upserting so re-creating an existing group does
not violate the
// unique (cluster_id, name) key.
- String groupClusterName = getClusterName(admin);
RmqGroup entity = groupMapper.selectOne(new
LambdaQueryWrapper<RmqGroup>()
.eq(RmqGroup::getClusterId, groupClusterName)
.eq(RmqGroup::getName, groupName));
@@ -481,6 +487,9 @@ public class RocketMQAdminClientImpl implements AdminClient
{
@Override
public void resetOffset(String instanceId, String name, long timestamp,
String topic) {
+ if (!StringUtils.hasText(topic)) {
+ throw new BusinessException(400, "topic is required for offset
reset");
+ }
try {
if (StringUtils.hasText(instanceId)) {
runtimeAdminClientResolver.execute(instanceId, admin -> {
@@ -560,6 +569,41 @@ public class RocketMQAdminClientImpl implements
AdminClient {
return addrs;
}
+ /**
+ * Returns master broker addresses for the specified cluster only, using
the
+ * clusterAddrTable to map cluster name to broker names. Falls back to all
+ * brokers when the cluster name is unknown or the cluster table is
missing.
+ */
+ private Set<String> getMasterBrokerAddrsForCluster(MQAdminExt admin,
String clusterName) throws Exception {
+ if (!StringUtils.hasText(clusterName)) {
+ return getAllMasterBrokerAddrs(admin);
+ }
+ ClusterInfo clusterInfo = admin.examineBrokerClusterInfo();
+ if (clusterInfo == null || clusterInfo.getClusterAddrTable() == null
+ || clusterInfo.getBrokerAddrTable() == null) {
+ return getAllMasterBrokerAddrs(admin);
+ }
+ Set<String> brokerNames =
clusterInfo.getClusterAddrTable().get(clusterName);
+ if (brokerNames == null || brokerNames.isEmpty()) {
+ return getAllMasterBrokerAddrs(admin);
+ }
+ Set<String> addrs = new HashSet<>();
+ for (String brokerName : brokerNames) {
+ BrokerData brokerData =
clusterInfo.getBrokerAddrTable().get(brokerName);
+ if (brokerData == null || brokerData.getBrokerAddrs() == null) {
+ continue;
+ }
+ 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.isEmpty() ? getAllMasterBrokerAddrs(admin) : addrs;
+ }
+
private String getClusterName(MQAdminExt admin) {
try {
ClusterInfo clusterInfo = admin.examineBrokerClusterInfo();
