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 c818121d fix(cluster): report real cluster names in client connections
(#1699)
c818121d is described below
commit c818121da4cf485aac540f4f588f3649d621b7cd
Author: aias00 <[email protected]>
AuthorDate: Tue Aug 11 20:50:41 2026 +0800
fix(cluster): report real cluster names in client connections (#1699)
---
.../provider/apache/RocketMQClientProvider.java | 75 +++++++++++-----------
.../apache/RocketMQClientProviderTest.java | 62 +++++++++++++++++-
2 files changed, 96 insertions(+), 41 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQClientProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQClientProvider.java
index abe5f98a..2c259dfd 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQClientProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQClientProvider.java
@@ -43,12 +43,10 @@ import org.springframework.stereotype.Service;
import java.util.ArrayList;
import java.util.LinkedHashMap;
-import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
-import java.util.Set;
/**
* Live {@link ClientProvider} backed by the RocketMQ admin API. It discovers
producer
@@ -125,46 +123,49 @@ public class RocketMQClientProvider implements
ClientProvider {
private List<ClientConnectionVO> findAllProducerConnections(
MQAdminExt adminExt, String clusterId) {
- Set<String> brokerAddresses = collectProducerBrokerAddresses(adminExt);
+ BrokerTopology topology = discoverBrokerTopology(adminExt, clusterId,
"producer connections");
Map<String, ClientConnectionVO> connections = new LinkedHashMap<>();
int successfulBrokers = 0;
- for (String brokerAddress : brokerAddresses) {
+ for (String brokerAddress : topology.brokerAddresses()) {
try {
ProducerTableInfo producerTable =
adminExt.getAllProducerInfo(brokerAddress);
successfulBrokers++;
- addProducerConnections(connections, producerTable, clusterId);
+ addProducerConnections(connections, producerTable,
topology.clusterFor(brokerAddress));
} catch (Exception e) {
log.warn("Failed to fetch producer connections from broker={},
skipping", brokerAddress, e);
}
}
- if (!brokerAddresses.isEmpty() && successfulBrokers == 0) {
+ if (!topology.brokerAddresses().isEmpty() && successfulBrokers == 0) {
throw new BusinessException(502, "Failed to query producer
connections from all brokers");
}
return new ArrayList<>(connections.values());
}
- private Set<String> collectProducerBrokerAddresses(MQAdminExt adminExt) {
+ private BrokerTopology discoverBrokerTopology(MQAdminExt adminExt, String
clusterId, String operation) {
ClusterInfo clusterInfo;
try {
clusterInfo = adminExt.examineBrokerClusterInfo();
} catch (Exception e) {
throw new BusinessException(502,
- "Failed to discover brokers for producer connections: " +
rootMessage(e));
+ "Failed to discover brokers for " + operation + ": " +
rootMessage(e));
}
- Set<String> addresses = new LinkedHashSet<>();
+ Map<String, String> clusterByAddress = new LinkedHashMap<>();
if (clusterInfo == null || clusterInfo.getBrokerAddrTable() == null) {
- return addresses;
+ return new BrokerTopology(clusterByAddress);
}
for (BrokerData brokerData :
clusterInfo.getBrokerAddrTable().values()) {
- if (brokerData == null) {
+ if (brokerData == null || brokerData.getBrokerAddrs() == null
+ || brokerData.getBrokerAddrs().isEmpty()) {
continue;
}
String brokerAddress = brokerData.selectBrokerAddr();
- if (brokerAddress != null && !brokerAddress.isBlank()) {
- addresses.add(brokerAddress);
+ String brokerCluster = brokerData.getCluster();
+ if (brokerAddress != null && !brokerAddress.isBlank()
+ && (clusterId == null || clusterId.equals(brokerCluster)))
{
+ clusterByAddress.putIfAbsent(brokerAddress, brokerCluster);
}
}
- return addresses;
+ return new BrokerTopology(clusterByAddress);
}
private void addProducerConnections(
@@ -207,9 +208,10 @@ public class RocketMQClientProvider implements
ClientProvider {
private List<ClientConnectionVO> findConsumerConnections(MQAdminExt
adminExt, String clusterId) {
List<ClientConnectionVO> result = new ArrayList<>();
- Set<String> groups = collectSubscriptionGroups(adminExt);
+ Map<String, String> groups = collectSubscriptionGroups(adminExt,
clusterId);
int successfulGroupQueries = 0;
- for (String group : groups) {
+ for (Map.Entry<String, String> groupEntry : groups.entrySet()) {
+ String group = groupEntry.getKey();
if (isSystemGroup(group)) {
continue;
}
@@ -223,7 +225,8 @@ public class RocketMQClientProvider implements
ClientProvider {
if (connection == null) {
continue;
}
- result.add(toConnectionVO(connection, ClientType.Consumer,
group, null, clusterId));
+ result.add(toConnectionVO(connection, ClientType.Consumer,
group, null,
+ groupEntry.getValue()));
}
} catch (Exception e) {
log.warn("Failed to examine consumer connection for group={},
skipping", group, e);
@@ -235,36 +238,20 @@ public class RocketMQClientProvider implements
ClientProvider {
return result;
}
- private Set<String> collectSubscriptionGroups(MQAdminExt 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);
- throw new BusinessException(502, "Failed to discover brokers for
consumer connections: "
- + rootMessage(e));
- }
- if (clusterInfo == null || clusterInfo.getBrokerAddrTable() == null) {
- return groups;
- }
+ private Map<String, String> collectSubscriptionGroups(MQAdminExt adminExt,
String clusterId) {
+ Map<String, String> groups = new LinkedHashMap<>();
+ BrokerTopology topology = discoverBrokerTopology(adminExt, clusterId,
"consumer connections");
int attemptedBrokerQueries = 0;
int successfulBrokerQueries = 0;
- for (BrokerData brokerData :
clusterInfo.getBrokerAddrTable().values()) {
- if (brokerData == null) {
- continue;
- }
- String brokerAddr = brokerData.selectBrokerAddr();
- if (brokerAddr == null) {
- continue;
- }
+ for (String brokerAddr : topology.brokerAddresses()) {
try {
attemptedBrokerQueries++;
SubscriptionGroupWrapper wrapper =
adminExt.getAllSubscriptionGroup(brokerAddr,
SUBSCRIPTION_GROUP_TIMEOUT_MILLIS);
successfulBrokerQueries++;
if (wrapper != null && wrapper.getSubscriptionGroupTable() !=
null) {
-
groups.addAll(wrapper.getSubscriptionGroupTable().keySet());
+ wrapper.getSubscriptionGroupTable().keySet().forEach(group
->
+ groups.putIfAbsent(group,
topology.clusterFor(brokerAddr)));
}
} catch (Exception e) {
log.warn("Failed to fetch subscription groups from broker={},
skipping", brokerAddr, e);
@@ -347,4 +334,14 @@ public class RocketMQClientProvider implements
ClientProvider {
? current.getClass().getSimpleName()
: message;
}
+
+ private record BrokerTopology(Map<String, String> clusterByAddress) {
+ private List<String> brokerAddresses() {
+ return List.copyOf(clusterByAddress.keySet());
+ }
+
+ private String clusterFor(String brokerAddress) {
+ return clusterByAddress.get(brokerAddress);
+ }
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQClientProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQClientProviderTest.java
index ded1bb8a..e71932cd 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQClientProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQClientProviderTest.java
@@ -53,6 +53,7 @@ import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -114,6 +115,32 @@ class RocketMQClientProviderTest {
verify(adminExt, never()).examineProducerConnectionInfo(anyString(),
anyString());
}
+ @Test
+ void clientScanUsesActualBrokerClustersAndFiltersRequestedCluster() throws
Exception {
+
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfo(Map.of(
+ "127.0.0.1:10911", "cluster-a",
+ "127.0.0.2:10911", "cluster-b")));
+ when(adminExt.getAllProducerInfo("127.0.0.1:10911"))
+ .thenReturn(new ProducerTableInfo(Map.of(
+ "pg-a", List.of(producerInfo("producer-a",
"10.0.0.1:1000")))));
+ when(adminExt.getAllProducerInfo("127.0.0.2:10911"))
+ .thenReturn(new ProducerTableInfo(Map.of(
+ "pg-b", List.of(producerInfo("producer-b",
"10.0.0.2:1000")))));
+
+ List<ClientConnectionVO> allConnections =
provider.findConnections("instance-a", null, "Producer");
+ List<ClientConnectionVO> clusterBConnections =
provider.findConnections("instance-a", "cluster-b", "Producer");
+
+ assertThat(allConnections)
+ .extracting(ClientConnectionVO::getClusterName)
+ .containsExactlyInAnyOrder("cluster-a", "cluster-b");
+ assertThat(clusterBConnections).singleElement().satisfies(connection
-> {
+ assertThat(connection.getClientId()).isEqualTo("producer-b");
+ assertThat(connection.getClusterName()).isEqualTo("cluster-b");
+ });
+ verify(adminExt).getAllProducerInfo("127.0.0.1:10911");
+ verify(adminExt, times(2)).getAllProducerInfo("127.0.0.2:10911");
+ }
+
@Test
void producerScanReturnsPartialResultsWhenOneBrokerFails() throws
Exception {
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfo(
@@ -220,6 +247,27 @@ class RocketMQClientProviderTest {
verify(adminExt).getAllSubscriptionGroup("127.0.0.1:10911", 5000L);
}
+ @Test
+ void
consumerScanFiltersSubscriptionGroupsByClusterAndPreservesClusterName() throws
Exception {
+
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfo(Map.of(
+ "127.0.0.1:10911", "cluster-a",
+ "127.0.0.2:10911", "cluster-b")));
+ when(adminExt.getAllSubscriptionGroup("127.0.0.2:10911", 5000L))
+ .thenReturn(subscriptionGroups("group-b"));
+ ConsumerConnection consumerConnection = new ConsumerConnection();
+ consumerConnection.setConnectionSet(new
HashSet<>(List.of(connection("consumer-b", "10.0.0.2:1000"))));
+
when(adminExt.examineConsumerConnectionInfo("group-b")).thenReturn(consumerConnection);
+
+ List<ClientConnectionVO> connections =
provider.findConnections("instance-a", "cluster-b", "Consumer");
+
+ assertThat(connections).singleElement().satisfies(connection -> {
+ assertThat(connection.getClientId()).isEqualTo("consumer-b");
+ assertThat(connection.getClusterName()).isEqualTo("cluster-b");
+ });
+ verify(adminExt, never()).getAllSubscriptionGroup("127.0.0.1:10911",
5000L);
+ verify(adminExt).getAllSubscriptionGroup("127.0.0.2:10911", 5000L);
+ }
+
@Test
void consumerScanFailsWhenBrokerDiscoveryFails() throws Exception {
when(adminExt.examineBrokerClusterInfo()).thenThrow(new
IllegalStateException("broker unavailable"));
@@ -324,12 +372,22 @@ class RocketMQClientProviderTest {
}
private static ClusterInfo clusterInfo(String... brokerAddresses) {
+ Map<String, String> clusters = new HashMap<>();
+ for (int i = 0; i < brokerAddresses.length; i++) {
+ clusters.put(brokerAddresses[i], "cluster-a");
+ }
+ return clusterInfo(clusters);
+ }
+
+ private static ClusterInfo clusterInfo(Map<String, String>
clustersByAddress) {
ClusterInfo clusterInfo = new ClusterInfo();
Map<String, BrokerData> brokerAddrTable = new HashMap<>();
- for (int i = 0; i < brokerAddresses.length; i++) {
+ int i = 0;
+ for (Map.Entry<String, String> entry : clustersByAddress.entrySet()) {
String brokerName = "broker-" + i;
brokerAddrTable.put(brokerName, new BrokerData(
- "cluster-a", brokerName, new HashMap<>(Map.of(0L,
brokerAddresses[i]))));
+ entry.getValue(), brokerName, new HashMap<>(Map.of(0L,
entry.getKey()))));
+ i++;
}
clusterInfo.setBrokerAddrTable(brokerAddrTable);
return clusterInfo;
