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 786a308e7 fix(cluster): apply the configured cluster to detail and
brokers (#6196)
786a308e7 is described below
commit 786a308e7f2126a5a87ecdaf87ab4e9fadb42e9b
Author: zmuxuny <[email protected]>
AuthorDate: Mon Oct 12 00:18:30 2026 -0700
fix(cluster): apply the configured cluster to detail and brokers (#6196)
`discoverClusters(instanceId)` already narrowed the topology to the cluster
configured for the instance (`RocketMQClusterProvider.java:88-91`), but the
other two instance-scoped reads ignored that restriction.
`refreshClusterDetail(clusterId, instanceId)` (`:139-176`) looked the requested
id up in the shared `clusterAddrTable` and fetched its broker runtime stats,
and `discoverBrokers` (`:184-196`) listed every name in `brokerAddrTable`, so
an instance configured for cluster A on a sh [...]
Nine new instances — seven tests plus a two-case parameterized one — assert
the status codes, the broker-name sets, and `verify(admin,
never()).fetchBrokerRuntimeStats(...)` for the rejected detail read.
`globalDetailStillExposesOtherClusterTest` and
`unscopedInstanceStillExposesBothClustersTest` are the controls that fail if
the restriction is widened past instance scope;
`scopedBrokerLookupAcceptsMemberBrokerTest` and
`scopedBrokerListDoesNotFallBackWhenClusterIsAbsentTest` pin the [...]
---
.../provider/apache/RocketMQClusterProvider.java | 25 ++++-
.../apache/RocketMQClusterProviderTest.java | 115 +++++++++++++++++++++
2 files changed, 137 insertions(+), 3 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProvider.java
index adfef1aa2..77f57d395 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProvider.java
@@ -144,6 +144,12 @@ public class RocketMQClusterProvider implements
ClusterProvider {
return null;
}
+ String configuredCluster = StringUtils.hasText(instanceId)
+ ? runtimeAdminClientResolver.configuredClusterName(instanceId)
: null;
+ if (configuredCluster != null && !configuredCluster.equals(clusterId))
{
+ return null;
+ }
+
try {
return executeAdmin(instanceId, namesrvAddr, admin -> {
ClusterInfo clusterInfo = admin.examineBrokerClusterInfo();
@@ -183,14 +189,27 @@ public class RocketMQClusterProvider implements
ClusterProvider {
@Override
public List<BrokerVO> discoverBrokers(String instanceId, String
brokerName) {
+ String configuredCluster =
runtimeAdminClientResolver.configuredClusterName(instanceId);
return runtimeAdminClientResolver.execute(instanceId, admin -> {
ClusterInfo info = admin.examineBrokerClusterInfo();
if (info == null || info.getBrokerAddrTable() == null)
throw new BusinessException(503, "Broker topology is
unavailable");
Map<String, BrokerData> table = info.getBrokerAddrTable();
- if (brokerName != null && !table.containsKey(brokerName))
- throw new BusinessException(404, "Broker not found in
Instance: " + brokerName);
- Set<String> names = brokerName == null ? new
java.util.TreeSet<>(table.keySet()) : Set.of(brokerName);
+ Set<String> names = new TreeSet<>(table.keySet());
+ if (configuredCluster != null) {
+ Map<String, Set<String>> clusters = info.getClusterAddrTable();
+ if (clusters == null || clusters.containsKey(configuredCluster)
+ && clusters.get(configuredCluster) == null) {
+ throw new BusinessException(503, "Broker cluster
membership is unavailable");
+ }
+ names.retainAll(clusters.getOrDefault(configuredCluster,
Set.of()));
+ }
+ if (brokerName != null) {
+ if (!names.contains(brokerName)) {
+ throw new BusinessException(404, "Broker not found in
Instance: " + brokerName);
+ }
+ names = Set.of(brokerName);
+ }
return buildBrokerList(admin, names, table);
});
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProviderTest.java
index b10f378aa..2d8b62859 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProviderTest.java
@@ -36,10 +36,13 @@ import
org.apache.rocketmq.studio.instance.InstanceRepository;
import org.apache.rocketmq.studio.instance.InstanceVO;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.Set;
@@ -383,6 +386,118 @@ class RocketMQClusterProviderTest {
verify(adminFactory, never()).execute(eq("10.0.0.2:9876"), isNull(),
any());
}
+ @Test
+ void scopedDetailDoesNotReadAnotherClusterTest() throws Exception {
+ DefaultMQAdminExt admin = mock(DefaultMQAdminExt.class);
+ RocketMQClusterProvider provider = scopedProvider(admin,
"DefaultCluster");
+ when(admin.examineBrokerClusterInfo()).thenReturn(sharedTopology());
+
+ assertThat(provider.refreshClusterDetail("OtherCluster",
"selected")).isNull();
+ verify(admin, never()).fetchBrokerRuntimeStats("10.0.0.12:10911");
+ assertThat(provider.refreshClusterDetail("DefaultCluster",
"selected")).isNotNull();
+ }
+
+ @Test
+ void scopedBrokerListIncludesOnlySelectedClusterTest() throws Exception {
+ DefaultMQAdminExt admin = mock(DefaultMQAdminExt.class);
+ RocketMQClusterProvider provider = scopedProvider(admin,
"DefaultCluster");
+ when(admin.examineBrokerClusterInfo()).thenReturn(sharedTopology());
+
+ assertThat(provider.discoverBrokers("selected", null))
+
.extracting(org.apache.rocketmq.studio.cluster.broker.BrokerVO::getName)
+ .containsExactly("broker-a");
+ verify(admin, never()).fetchBrokerRuntimeStats("10.0.0.12:10911");
+ }
+
+ @Test
+ void scopedBrokerLookupAcceptsMemberBrokerTest() throws Exception {
+ DefaultMQAdminExt admin = mock(DefaultMQAdminExt.class);
+ RocketMQClusterProvider provider = scopedProvider(admin,
"DefaultCluster");
+ when(admin.examineBrokerClusterInfo()).thenReturn(sharedTopology());
+
+ assertThat(provider.discoverBrokers("selected", "broker-a"))
+
.extracting(org.apache.rocketmq.studio.cluster.broker.BrokerVO::getName)
+ .containsExactly("broker-a");
+ verify(admin, never()).fetchBrokerRuntimeStats("10.0.0.12:10911");
+ }
+
+ @Test
+ void globalDetailStillExposesOtherClusterTest() throws Exception {
+ DefaultMQAdminExt admin = mock(DefaultMQAdminExt.class);
+ RocketMQClusterProvider provider = newProvider(admin);
+ when(admin.examineBrokerClusterInfo()).thenReturn(sharedTopology());
+
+ assertThat(provider.refreshClusterDetail("OtherCluster")).isNotNull();
+ }
+
+ @Test
+ void scopedBrokerLookupRejectsOtherClusterTest() throws Exception {
+ DefaultMQAdminExt admin = mock(DefaultMQAdminExt.class);
+ RocketMQClusterProvider provider = scopedProvider(admin,
"DefaultCluster");
+ when(admin.examineBrokerClusterInfo()).thenReturn(sharedTopology());
+
+ assertThatThrownBy(() -> provider.discoverBrokers("selected",
"broker-b"))
+ .isInstanceOfSatisfying(BusinessException.class, error ->
assertThat(error.getCode()).isEqualTo(404));
+ verify(admin, never()).fetchBrokerRuntimeStats("10.0.0.12:10911");
+ }
+
+ @Test
+ void unscopedInstanceStillExposesBothClustersTest() throws Exception {
+ DefaultMQAdminExt admin = mock(DefaultMQAdminExt.class);
+ RocketMQClusterProvider provider = scopedProvider(admin, null);
+ when(admin.examineBrokerClusterInfo()).thenReturn(sharedTopology());
+
+ assertThat(provider.refreshClusterDetail("OtherCluster",
"selected")).isNotNull();
+ assertThat(provider.discoverBrokers("selected", null))
+
.extracting(org.apache.rocketmq.studio.cluster.broker.BrokerVO::getName)
+ .containsExactly("broker-a", "broker-b");
+ }
+
+ @Test
+ void scopedBrokerListDoesNotFallBackWhenClusterIsAbsentTest() throws
Exception {
+ DefaultMQAdminExt admin = mock(DefaultMQAdminExt.class);
+ RocketMQClusterProvider provider = scopedProvider(admin,
"MissingCluster");
+ when(admin.examineBrokerClusterInfo()).thenReturn(sharedTopology());
+
+ assertThat(provider.discoverBrokers("selected", null)).isEmpty();
+ verify(admin, never()).fetchBrokerRuntimeStats(anyString());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void scopedBrokerListRejectsMissingMembershipTest(boolean missingTable)
throws Exception {
+ DefaultMQAdminExt admin = mock(DefaultMQAdminExt.class);
+ RocketMQClusterProvider provider = scopedProvider(admin,
"DefaultCluster");
+ ClusterInfo topology = sharedTopology();
+ if (missingTable) {
+ topology.setClusterAddrTable(null);
+ } else {
+ topology.getClusterAddrTable().put("DefaultCluster", null);
+ }
+ when(admin.examineBrokerClusterInfo()).thenReturn(topology);
+
+ assertThatThrownBy(() -> provider.discoverBrokers("selected", null))
+ .isInstanceOfSatisfying(BusinessException.class, error ->
assertThat(error.getCode()).isEqualTo(503));
+ verify(admin, never()).fetchBrokerRuntimeStats(anyString());
+ }
+
+ private RocketMQClusterProvider scopedProvider(DefaultMQAdminExt admin,
String cluster) {
+ RuntimeAdminClientResolver runtime =
mock(RuntimeAdminClientResolver.class);
+
org.mockito.Mockito.lenient().when(runtime.resolveEndpoint("selected")).thenReturn("10.0.0.1:9876");
+
org.mockito.Mockito.lenient().when(runtime.configuredClusterName("selected")).thenReturn(cluster);
+ when(runtime.execute(eq("selected"), any())).thenAnswer(call ->
+
call.<MqAdminExtFactory.AdminAction<Object>>getArgument(1).apply(admin));
+ return new RocketMQClusterProvider(mock(MqAdminExtFactory.class), new
RocketMQProperties(), runtime);
+ }
+
+ private ClusterInfo sharedTopology() {
+ ClusterInfo info = clusterInfo();
+ info.getClusterAddrTable().put("OtherCluster", Set.of("broker-b"));
+ info.getBrokerAddrTable().put("broker-b", new
BrokerData("OtherCluster", "broker-b",
+ new HashMap<>(Map.of(0L, "10.0.0.12:10911"))));
+ return info;
+ }
+
private RocketMQClusterProvider
newAuthenticatedInstanceProvider(MqAdminExtFactory adminFactory,
DefaultMQAdminExt adminExt) {
InstanceRepository instanceRepository = mock(InstanceRepository.class);
