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 462cf5912 fix(proxy): scope discovery to the configured broker cluster
(#7282)
462cf5912 is described below
commit 462cf5912be3b5b8d6104a9f59cd0218397181fd
Author: Xiao Yang <[email protected]>
AuthorDate: Mon Oct 26 17:00:39 2026 +0800
fix(proxy): scope discovery to the configured broker cluster (#7282)
* fix(proxy): scope discovery to the configured broker cluster
* fix(proxy): answer an unreadable membership with an empty discovery
discoverClustersAt answers a topology it cannot read with an empty list,
while the
scoped proxy lookup threw 503 for the same missing data — one cluster the
NameServer
could not describe failed the whole page. Only an RPC failure throws now,
which is
the admin query caliber the rest of the provider keeps (a business-empty
result is
a 200 with an empty list).
---
.../provider/apache/RocketMQClusterProvider.java | 40 +++++++++++--
.../apache/RocketMQClusterProviderTest.java | 69 ++++++++++++++++++++++
2 files changed, 105 insertions(+), 4 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 ca6070929..dd463f9ed 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
@@ -46,6 +46,7 @@ import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.TreeMap;
import java.util.TreeSet;
/**
@@ -190,7 +191,37 @@ public class RocketMQClusterProvider implements
ClusterProvider {
@Override
public List<ProxyVO> discoverProxies(String instanceId) {
if (!StringUtils.hasText(instanceId)) throw new BusinessException(400,
"Instance is required");
- return runtimeAdminClientResolver.execute(instanceId,
this::discoverProxiesViaHeartbeatSyncer);
+ String configuredCluster =
runtimeAdminClientResolver.configuredClusterName(instanceId);
+ return runtimeAdminClientResolver.execute(instanceId, admin -> {
+ if (configuredCluster == null) {
+ return discoverProxiesViaHeartbeatSyncer(admin, null);
+ }
+ ClusterInfo info = admin.examineBrokerClusterInfo();
+ Map<String, Set<String>> clusters = info == null ? null :
info.getClusterAddrTable();
+ Set<String> members = clusters == null ? null :
clusters.get(configuredCluster);
+ if (info == null || info.getBrokerAddrTable() == null || members
== null) {
+ // The caliber discoverClustersAt keeps: a topology or a
cluster membership the
+ // NameServer does not report is an empty discovery, not a
failure — only the RPC
+ // failure below throws.
+ return Collections.emptyList();
+ }
+ // A NameServer may serve several physical clusters. Query only
the selected
+ // cluster's brokers instead of letting the global group lookup
choose one.
+ Map<String, ProxyVO> proxies = new TreeMap<>();
+ for (String brokerName : new TreeSet<>(members)) {
+ BrokerData broker = info.getBrokerAddrTable().get(brokerName);
+ if (broker == null || broker.getBrokerAddrs() == null ||
broker.getBrokerAddrs().isEmpty()) {
+ continue;
+ }
+ String address = broker.selectBrokerAddr();
+ if (StringUtils.hasText(address)) {
+ for (ProxyVO proxy :
discoverProxiesViaHeartbeatSyncer(admin, address)) {
+ proxies.putIfAbsent(proxy.getAddr(), proxy);
+ }
+ }
+ }
+ return new ArrayList<>(proxies.values());
+ });
}
@Override
@@ -392,10 +423,11 @@ public class RocketMQClusterProvider implements
ClusterProvider {
return adminFactory.execute(namesrvAddr, null, action);
}
- private List<ProxyVO> discoverProxiesViaHeartbeatSyncer(MQAdminExt admin) {
+ private List<ProxyVO> discoverProxiesViaHeartbeatSyncer(MQAdminExt admin,
String brokerAddress) {
try {
- ConsumerConnection connection =
-
admin.examineConsumerConnectionInfo(HEARTBEAT_SYNCER_CONSUMER_GROUP);
+ ConsumerConnection connection = brokerAddress == null
+ ?
admin.examineConsumerConnectionInfo(HEARTBEAT_SYNCER_CONSUMER_GROUP)
+ :
admin.examineConsumerConnectionInfo(HEARTBEAT_SYNCER_CONSUMER_GROUP,
brokerAddress);
if (connection == null || connection.getConnectionSet() == null) {
return Collections.emptyList();
}
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 aeb7270f9..16d2c22f7 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
@@ -582,6 +582,75 @@ class RocketMQClusterProviderTest {
verify(admin, never()).fetchBrokerRuntimeStats(anyString());
}
+ @Test
+ void scopedProxyDiscoveryAggregatesOnlySelectedClusterBrokersTest() throws
Exception {
+ DefaultMQAdminExt admin = mock(DefaultMQAdminExt.class);
+ RocketMQClusterProvider provider = scopedProvider(admin,
"DefaultCluster");
+ ClusterInfo topology = sharedTopology();
+ topology.getClusterAddrTable().put("DefaultCluster",
Set.of("broker-a", "broker-c"));
+ topology.getBrokerAddrTable().put("broker-c", new
BrokerData("DefaultCluster", "broker-c",
+ new HashMap<>(Map.of(0L, "10.0.0.13:10911"))));
+ when(admin.examineBrokerClusterInfo()).thenReturn(topology);
+
when(admin.examineConsumerConnectionInfo("CID_DefaultHeartBeatSyncerTopic",
"10.0.0.11:10911"))
+ .thenReturn(proxyHeartbeats("10.0.3.5:40001",
"10.0.3.6:40002"));
+
when(admin.examineConsumerConnectionInfo("CID_DefaultHeartBeatSyncerTopic",
"10.0.0.13:10911"))
+ .thenReturn(proxyHeartbeats("10.0.3.5:40003",
"10.0.3.7:40004"));
+
+ assertThat(provider.discoverProxies("selected"))
+ .extracting(proxy -> proxy.getAddr())
+ .containsExactly("10.0.3.5:8080", "10.0.3.6:8080",
"10.0.3.7:8080");
+
verify(admin).examineConsumerConnectionInfo("CID_DefaultHeartBeatSyncerTopic",
"10.0.0.11:10911");
+
verify(admin).examineConsumerConnectionInfo("CID_DefaultHeartBeatSyncerTopic",
"10.0.0.13:10911");
+ verify(admin,
never()).examineConsumerConnectionInfo("CID_DefaultHeartBeatSyncerTopic",
"10.0.0.12:10911");
+ verify(admin, never()).examineConsumerConnectionInfo(anyString());
+ }
+
+ @Test
+ void scopedProxyDiscoveryDoesNotFallBackWhenClusterIsAbsentTest() throws
Exception {
+ DefaultMQAdminExt admin = mock(DefaultMQAdminExt.class);
+ RocketMQClusterProvider provider = scopedProvider(admin,
"MissingCluster");
+ when(admin.examineBrokerClusterInfo()).thenReturn(sharedTopology());
+
+ assertThat(provider.discoverProxies("selected")).isEmpty();
+
+ verify(admin, never()).examineConsumerConnectionInfo(anyString());
+ verify(admin, never()).examineConsumerConnectionInfo(anyString(),
anyString());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"topology", "brokers", "clusters", "membership"})
+ void scopedProxyDiscoveryTreatsUnavailableMembershipAsEmptyTest(String
unavailable) throws Exception {
+ DefaultMQAdminExt admin = mock(DefaultMQAdminExt.class);
+ RocketMQClusterProvider provider = scopedProvider(admin,
"DefaultCluster");
+ ClusterInfo topology = sharedTopology();
+ switch (unavailable) {
+ case "topology" -> topology = null;
+ case "brokers" -> topology.setBrokerAddrTable(null);
+ case "clusters" -> topology.setClusterAddrTable(null);
+ case "membership" ->
topology.getClusterAddrTable().put("DefaultCluster", null);
+ default -> throw new IllegalArgumentException(unavailable);
+ }
+ when(admin.examineBrokerClusterInfo()).thenReturn(topology);
+
+ // discoverClustersAt answers a topology it cannot read with an empty
list, not an error;
+ // the scoped lookup keeps that caliber so one missing membership
cannot fail the page.
+ assertThat(provider.discoverProxies("selected")).isEmpty();
+ verify(admin, never()).examineConsumerConnectionInfo(anyString());
+ verify(admin, never()).examineConsumerConnectionInfo(anyString(),
anyString());
+ }
+
+ private ConsumerConnection proxyHeartbeats(String... clientAddresses) {
+ LinkedHashSet<Connection> connections = new LinkedHashSet<>();
+ for (String clientAddress : clientAddresses) {
+ Connection connection = new Connection();
+ connection.setClientAddr(clientAddress);
+ connections.add(connection);
+ }
+ ConsumerConnection result = new ConsumerConnection();
+ result.setConnectionSet(connections);
+ return result;
+ }
+
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");
