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 38c569d38 fix(client): show proxy-connected consumers in the client
list (#5453)
38c569d38 is described below
commit 38c569d382702b319d7d6dcfb0e8d41a0723b9b0
Author: pilichoumao <[email protected]>
AuthorDate: Fri Oct 9 12:52:07 2026 +0800
fix(client): show proxy-connected consumers in the client list (#5453)
Consumers that connect through a Proxy never showed up in the client list:
the consumer scan only
asked brokers, so a group whose clients sit behind a Proxy looked offline.
The scan now walks the
brokers of the selected topology and, for each of them, the Proxy addresses
discovered through the
heartbeat-syncer group, queries both by explicit address, and de-duplicates
by
(cluster, group, clientId, clientAddr). A Proxy response carries no
protocol and only a
compatibility version, so both fields are null on Proxy rows and the page
renders `-`; `partial`
marks a scan where some query failed, and a scan that failed everywhere
with nothing to show is a
502. `docs/api-spec.md` documents the new `namesrvAddr` parameter and these
field semantics.
Maintainer edits on top of the contribution (review follow-ups):
- Offline detection now reuses the existing `isGroupConnectionAbsent`, so a
"group not online"
answer that does not carry code 206 no longer counts as a scan failure -
it used to mark every row
`partial`, and to 502 when it was the only group.
- Proxy discovery treats `TOPIC_NOT_EXIST` like `CONSUMER_NOT_ONLINE` ("no
Proxy observable right
now"), matching `ProxyConsumerResolver`.
- Failed discoveries are cached for the rest of the request, so one
unreachable broker is probed
once instead of once per remaining group. A batch deadline like #4877's
was deliberately not
added: this path is a serial blocking RPC rather than an executor batch,
and a time-based bound
cannot be pinned by a test.
- Extracted `toProxyRemotingAddresses` so both discovery paths share one
address mapping.
- `clients.partialScan` (zh + en) replaces the hardcoded Alert text, and
the "scan limit was
reached" wording is gone from both it and `docs/api-spec.md` - the
consumer path has no such limit.
The spec still documents `protocol` / `version` as `null` rather than
absent: `ClientConnectionVO`
carries no `@JsonInclude` and there is no global Jackson inclusion
setting, so nulls are emitted.
- The protocol column gains an "unknown" filter option, so the rows behind
the `-: N` statistic can
actually be filtered; the CSV export uses the same sentinel.
Verification: `RocketMQClientProviderTest` (43) +
`ProxyConsumerResolverTest` (6) green with the dev
profile, 0 checkstyle violations; `ClientsPage.test.tsx` 24/24, prettier
and eslint clean. Each new
behaviour was mutation-checked - reverting the offline detection, the
`TOPIC_NOT_EXIST` branch, the
failure cache or the unknown-protocol filter option turns its case red.
---
docs/api-spec.md | 15 +-
.../provider/apache/ProxyConsumerResolver.java | 51 ++--
.../provider/apache/RocketMQClientProvider.java | 125 ++++++---
.../apache/RocketMQClientProviderTest.java | 293 ++++++++++++++++++++-
web/src/api/connections.ts | 4 +-
web/src/i18n/translations.ts | 4 +
.../pages/cluster/__tests__/ClientsPage.test.tsx | 89 +++++++
web/src/pages/cluster/clients.tsx | 43 +--
8 files changed, 538 insertions(+), 86 deletions(-)
diff --git a/docs/api-spec.md b/docs/api-spec.md
index 2205e6233..a53e229ff 100644
--- a/docs/api-spec.md
+++ b/docs/api-spec.md
@@ -1839,13 +1839,14 @@ GET
/api/dlq/export-excel?instanceId={instanceId}&groupName={groupName}&startTim
### 10.1 获取客户端连接列表
```
-GET /api/clients?clusterId={clusterId}&type={type}
+GET /api/clients?namesrvAddr={namesrvAddr}&clusterId={clusterId}&type={type}
```
**Query Parameters:**
| 参数 | 类型 | 必填 | 说明 |
|------|------|------|------|
+| `namesrvAddr` | `string` | 是 | 页面选中的 NameServer 地址 |
| `clusterId` | `string` | 否 | 按集群过滤 |
| `type` | `string` | 否 | 按类型过滤: `Producer` / `Consumer` |
@@ -1856,13 +1857,19 @@ GET /api/clients?clusterId={clusterId}&type={type}
| `clientId` | `string` | 客户端 ID |
| `type` | `string` | 类型: `Producer` / `Consumer` |
| `groupOrTopic` | `string` | 消费组名或 Topic 名 |
-| `protocol` | `string` | 协议: `gRPC` / `Remoting` |
+| `protocol` | `string / null` | 可确认的协议: `gRPC` / `Remoting`；未知为 `null` |
| `address` | `string` | 客户端地址 |
-| `language` | `string` | 客户端语言: `Java` / `Go` / `Python` / `Rust` / `C++` /
`C#` / `Node.js` / `PHP` |
-| `version` | `string` | SDK 版本号 |
+| `language` | `string / null` | 客户端语言: `Java` / `Go` / `Python` / `Rust` /
`C++` / `C#` / `Node.js` / `PHP` |
+| `version` | `string / null` | 可确认的 SDK 版本号；未知为 `null` |
| `connectedAt` | `string` | 连接时间 |
+| `partial` | `boolean` | 连接扫描不完整（部分查询失败） |
| `clusterName` | `string` | 所属集群名称（显示在第一列） |
+消费者查询合并所选 NameServer、集群中的 Broker 和 Proxy 连接，并按客户端及地址去重。
+Proxy 地址沿用心跳同步消费组发现机制及 Remoting 管理端口 8080，仅在本次请求内缓存。
+现有 Proxy 响应不提供客户端协议，版本号也可能是兼容值；这两项返回 `null`，页面显示 `-`，语言保留实际返回的可识别值。
+确认离线返回空列表；查询失败时保留已有行并设置 `partial`，没有可返回的行时返回查询错误。
+
### 10.2 获取 Producer Group 候选列表
```
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ProxyConsumerResolver.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ProxyConsumerResolver.java
index ba0da98fc..d0f254188 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ProxyConsumerResolver.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ProxyConsumerResolver.java
@@ -18,6 +18,10 @@
package org.apache.rocketmq.studio.provider.apache;
import jakarta.annotation.PreDestroy;
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.remoting.exception.RemotingException;
+import org.apache.rocketmq.tools.admin.MQAdminExt;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.remoting.netty.NettyClientConfig;
import org.apache.rocketmq.remoting.netty.NettyRemotingClient;
@@ -183,22 +187,10 @@ public class ProxyConsumerResolver {
if (cached != null && cached.expiresAtMillis() >
System.currentTimeMillis()) {
return ProxyAddressResolution.available(cached.addresses());
}
- Set<String> ips = new LinkedHashSet<>();
+ List<String> addresses;
try {
- executeAdmin(instanceId, admin -> {
- ConsumerConnection connection =
admin.examineConsumerConnectionInfo(HEARTBEAT_SYNCER_CONSUMER_GROUP);
- if (connection != null && connection.getConnectionSet() !=
null) {
- for (Connection conn : connection.getConnectionSet()) {
- String clientAddr = conn.getClientAddr();
- if (clientAddr == null || clientAddr.isBlank()) {
- continue;
- }
- int separator = clientAddr.lastIndexOf(':');
- ips.add(separator > 0 ? clientAddr.substring(0,
separator) : clientAddr);
- }
- }
- return null;
- });
+ addresses = executeAdmin(instanceId, admin ->
toProxyRemotingAddresses(
+
admin.examineConsumerConnectionInfo(HEARTBEAT_SYNCER_CONSUMER_GROUP)));
} catch (Exception e) {
if (MqResponseCodes.hasResponseCode(e,
ResponseCode.CONSUMER_NOT_ONLINE, ResponseCode.TOPIC_NOT_EXIST)) {
// A known absent heartbeat-syncer group means no proxy
address is currently observable.
@@ -208,7 +200,6 @@ public class ProxyConsumerResolver {
log.debug("Proxy discovery via heartbeat syncer failed for
instance {}: {}", instanceId, e.getMessage());
return ProxyAddressResolution.unavailable();
}
- List<String> addresses = ips.stream().map(ip -> ip + ":" +
PROXY_REMOTING_PORT).toList();
proxyAddressCache.put(cacheKey, new CachedProxyAddresses(addresses,
System.currentTimeMillis() + PROXY_ADDRESS_CACHE_TTL_MILLIS));
return ProxyAddressResolution.available(addresses);
}
@@ -220,6 +211,34 @@ public class ProxyConsumerResolver {
return adminFactory.execute(properties.getNamesrvAddr(), null, action);
}
+ /**
+ * Discovers proxies through a broker in the selected topology, using the
caller's pooled
+ * admin (and credentials). Callers cache this only within their current
inventory request;
+ * the legacy instance cache must not mix NameServers or clusters with
identical group names.
+ */
+ List<String> discoverProxyAddresses(MQAdminExt admin, String brokerAddress)
+ throws MQClientException, MQBrokerException, RemotingException,
InterruptedException {
+ return toProxyRemotingAddresses(
+
admin.examineConsumerConnectionInfo(HEARTBEAT_SYNCER_CONSUMER_GROUP,
brokerAddress));
+ }
+
+ private static List<String> toProxyRemotingAddresses(ConsumerConnection
syncer) {
+ if (syncer == null || syncer.getConnectionSet() == null) {
+ return List.of();
+ }
+ Set<String> addresses = new LinkedHashSet<>();
+ for (Connection connection : syncer.getConnectionSet()) {
+ String clientAddr = connection == null ? null :
connection.getClientAddr();
+ if (clientAddr == null || clientAddr.isBlank()) {
+ continue;
+ }
+ int separator = clientAddr.lastIndexOf(':');
+ addresses.add((separator > 0 ? clientAddr.substring(0, separator)
: clientAddr)
+ + ":" + PROXY_REMOTING_PORT);
+ }
+ return List.copyOf(addresses);
+ }
+
private NettyRemotingClient remotingClient() {
NettyRemotingClient client = remotingClient;
if (client == null) {
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 f89a84401..cd9e9a6ed 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
@@ -17,6 +17,8 @@
package org.apache.rocketmq.studio.provider.apache;
import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.common.MQVersion;
import org.apache.rocketmq.remoting.protocol.LanguageCode;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
@@ -69,6 +71,7 @@ public class RocketMQClientProvider implements ClientProvider
{
private final RuntimeAdminClientResolver runtimeAdminClientResolver;
private final MqAdminExtFactory adminFactory;
+ private final ProxyConsumerResolver proxyConsumerResolver;
@Override
public List<ClientConnectionVO> findConnections(String instanceId, String
clusterId, String type) {
@@ -193,7 +196,7 @@ public class RocketMQClientProvider implements
ClientProvider {
return producerConnection.getConnectionSet().stream()
.filter(Objects::nonNull)
.map(connection -> toConnectionVO(
- connection, ClientType.Producer, topic,
producerGroup, null))
+ connection, ClientType.Producer, topic,
producerGroup, null, false))
.toList();
} catch (MQClientException e) {
if (isTopicNotExist(e) || isGroupConnectionAbsent(e)) {
@@ -210,13 +213,16 @@ public class RocketMQClientProvider implements
ClientProvider {
}
}
- private boolean isGroupConnectionAbsent(MQClientException e) {
+ private boolean isGroupConnectionAbsent(Throwable e) {
if (MqResponseCodes.hasResponseCode(e,
ResponseCode.CONSUMER_NOT_ONLINE)) {
// rocketmq-tools examineConsumerConnectionInfo throws
CONSUMER_NOT_ONLINE (206)
// for a group with no online connections.
return true;
}
- String message = e.getErrorMessage() == null ? e.getMessage() :
e.getErrorMessage();
+ String message = e instanceof MQClientException clientException
+ && clientException.getErrorMessage() != null
+ ? clientException.getErrorMessage()
+ : e.getMessage();
return message != null && (message.contains("Not found the consumer
group connection")
|| message.contains("Not found the producer group
connection"));
}
@@ -337,49 +343,94 @@ public class RocketMQClientProvider implements
ClientProvider {
}
private List<ClientConnectionVO> findConsumerConnections(MQAdminExt
adminExt, String clusterId) {
- List<ClientConnectionVO> result = new ArrayList<>();
- Map<String, String> groups = collectSubscriptionGroups(adminExt,
clusterId);
+ Map<String, ClientConnectionVO> result = new LinkedHashMap<>();
+ Map<ConsumerGroup, List<String>> groups =
collectSubscriptionGroups(adminExt, clusterId);
+ Map<String, List<String>> proxiesByBroker = new LinkedHashMap<>();
int attemptedGroupQueries = 0;
int successfulGroupQueries = 0;
- for (Map.Entry<String, String> groupEntry : groups.entrySet()) {
- String group = groupEntry.getKey();
- if (isSystemGroup(group)) {
+ for (Map.Entry<ConsumerGroup, List<String>> entry : groups.entrySet())
{
+ ConsumerGroup group = entry.getKey();
+ if (isSystemGroup(group.name())) {
continue;
}
attemptedGroupQueries++;
- try {
- ConsumerConnection consumerConnection =
adminExt.examineConsumerConnectionInfo(group);
- successfulGroupQueries++;
- if (consumerConnection == null ||
consumerConnection.getConnectionSet() == null) {
- continue;
+ boolean complete = true;
+ LinkedHashSet<String> proxies = new LinkedHashSet<>();
+ for (String broker : entry.getValue()) {
+ complete &= addConsumerConnections(adminExt, broker, group,
false, result);
+ try {
+ List<String> addresses = proxiesByBroker.get(broker);
+ if (addresses == null) {
+ addresses =
proxyConsumerResolver.discoverProxyAddresses(adminExt, broker);
+ proxiesByBroker.put(broker, addresses);
+ }
+ proxies.addAll(addresses);
+ } catch (MQClientException | MQBrokerException |
RemotingException e) {
+ // Cache failures for the rest of this request too, so an
unreachable
+ // broker is not re-probed once per remaining group.
+ proxiesByBroker.put(broker, List.of());
+ if (!MqResponseCodes.hasResponseCode(e,
ResponseCode.CONSUMER_NOT_ONLINE,
+ ResponseCode.TOPIC_NOT_EXIST)) {
+ log.warn("Failed to discover proxies for broker={}",
broker, e);
+ complete = false;
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new BusinessException(502, "Consumer connection
query interrupted");
}
- for (Connection connection :
consumerConnection.getConnectionSet()) {
+ }
+ for (String proxy : proxies) {
+ complete &= addConsumerConnections(adminExt, proxy, group,
true, result);
+ }
+ if (complete) {
+ successfulGroupQueries++;
+ }
+ }
+ // An empty response cannot carry the existing per-row partial flag.
+ if (successfulGroupQueries < attemptedGroupQueries &&
result.isEmpty()) {
+ throw new BusinessException(502, "Failed to query consumer
connections from all groups");
+ }
+ if (successfulGroupQueries < attemptedGroupQueries) {
+ result.values().forEach(connection -> connection.setPartial(true));
+ }
+ return new ArrayList<>(result.values());
+ }
+
+ private boolean addConsumerConnections(MQAdminExt admin, String address,
ConsumerGroup group,
+ boolean proxy, Map<String,
ClientConnectionVO> result) {
+ try {
+ // Explicit addresses keep both Broker and Proxy queries inside
the selected cluster.
+ ConsumerConnection connections =
admin.examineConsumerConnectionInfo(group.name(), address);
+ if (connections != null && connections.getConnectionSet() != null)
{
+ for (Connection connection : connections.getConnectionSet()) {
if (connection == null) {
continue;
}
- result.add(toConnectionVO(connection, ClientType.Consumer,
group, null,
- groupEntry.getValue()));
+ ClientConnectionVO row = toConnectionVO(connection,
ClientType.Consumer,
+ group.name(), null, group.cluster(), proxy);
+ String key = group.cluster() + '\0' + group.name() + '\0'
+ + connection.getClientId() + '\0' +
connection.getClientAddr();
+ result.putIfAbsent(key, row);
}
- } catch (Exception e) {
- if (e instanceof MQClientException clientException &&
isGroupConnectionAbsent(clientException)) {
- // An offline group is a normal "no connections right now"
answer, not a
- // broken scan — count it so an all-offline cluster stays
an empty result.
- successfulGroupQueries++;
- continue;
- }
- log.warn("Failed to examine consumer connection for group={},
skipping", group, e);
}
+ return true;
+ } catch (MQClientException | MQBrokerException | RemotingException e) {
+ if (isGroupConnectionAbsent(e)) {
+ return true;
+ }
+ log.warn("Failed to examine consumer connection for group={} via
{}", group.name(), address, e);
+ return false;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new BusinessException(502, "Consumer connection query
interrupted");
}
- // Only fail when at least one non-system group was actually attempted
and all of them
- // failed; a cluster whose subscription table holds only system groups
is not an error.
- if (attemptedGroupQueries > 0 && successfulGroupQueries == 0) {
- throw new BusinessException(502, "Failed to query consumer
connections from all groups");
- }
- return result;
}
- private Map<String, String> collectSubscriptionGroups(MQAdminExt adminExt,
String clusterId) {
- Map<String, String> groups = new LinkedHashMap<>();
+ private record ConsumerGroup(String name, String cluster) {
+ }
+
+ private Map<ConsumerGroup, List<String>>
collectSubscriptionGroups(MQAdminExt adminExt, String clusterId) {
+ Map<ConsumerGroup, List<String>> groups = new LinkedHashMap<>();
BrokerTopology topology = discoverBrokerTopology(adminExt, clusterId,
"consumer connections");
int attemptedBrokerQueries = 0;
int successfulBrokerQueries = 0;
@@ -391,7 +442,8 @@ public class RocketMQClientProvider implements
ClientProvider {
successfulBrokerQueries++;
if (wrapper != null && wrapper.getSubscriptionGroupTable() !=
null) {
wrapper.getSubscriptionGroupTable().keySet().forEach(group
->
- groups.putIfAbsent(group,
topology.clusterFor(brokerAddr)));
+ groups.computeIfAbsent(new ConsumerGroup(group,
topology.clusterFor(brokerAddr)),
+ key -> new ArrayList<>()).add(brokerAddr));
}
} catch (Exception e) {
log.warn("Failed to fetch subscription groups from broker={},
skipping", brokerAddr, e);
@@ -404,16 +456,17 @@ public class RocketMQClientProvider implements
ClientProvider {
}
private ClientConnectionVO toConnectionVO(Connection connection,
ClientType type, String groupOrTopic,
- String producerGroup, String
clusterId) {
+ String producerGroup, String
clusterId, boolean proxy) {
return ClientConnectionVO.builder()
.clientId(connection.getClientId())
.type(type)
.groupOrTopic(groupOrTopic)
.producerGroup(producerGroup)
- .protocol(Protocol.Remoting)
+ // The Proxy response has no protocol; gRPC may carry a fixed
compatibility version.
+ .protocol(proxy ? null : Protocol.Remoting)
.address(connection.getClientAddr())
.language(mapLanguage(connection.getLanguage()))
- .version(MQVersion.getVersionDesc(connection.getVersion()))
+ .version(proxy ? null :
MQVersion.getVersionDesc(connection.getVersion()))
.clusterName(clusterId)
.build();
}
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 a6ac4bff6..5464bbce6 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
@@ -17,6 +17,8 @@
package org.apache.rocketmq.studio.provider.apache;
import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.remoting.exception.RemotingConnectException;
import org.apache.rocketmq.remoting.protocol.LanguageCode;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
@@ -29,7 +31,14 @@ import
org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupWrapper;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import
org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.studio.cluster.client.ClientConnectionVO;
+import org.apache.rocketmq.studio.cluster.client.ClientController;
+import org.apache.rocketmq.studio.cluster.client.ClientService;
import org.apache.rocketmq.studio.cluster.client.ProducerConnectionScanResult;
+import org.springframework.test.web.servlet.setup.MockMvcBuilders;
+
+import static
org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
+import static
org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
+import static
org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
import org.apache.rocketmq.studio.common.domain.enums.ClientLanguage;
@@ -75,8 +84,11 @@ class RocketMQClientProviderTest {
private RocketMQClientProvider provider;
@BeforeEach
- void setUp() {
- provider = new RocketMQClientProvider(runtimeAdminClientResolver,
adminFactory);
+ void setUp() throws Exception {
+ lenient().when(adminExt.examineConsumerConnectionInfo(anyString(),
anyString()))
+ .thenReturn(new ConsumerConnection());
+ provider = new RocketMQClientProvider(runtimeAdminClientResolver,
adminFactory,
+ new ProxyConsumerResolver(adminFactory,
runtimeAdminClientResolver, new RocketMQProperties()));
lenient().when(runtimeAdminClientResolver.execute(anyString(),
any())).thenAnswer(invocation ->
invocation.<MqAdminExtFactory.AdminAction<Object>>
getArgument(1).apply(adminExt));
@@ -86,6 +98,261 @@ class RocketMQClientProviderTest {
getArgument(2).apply(adminExt));
}
+ @Test
+ void clientsEndpointShowsProxyConsumersFromSelectedNameserverTest() throws
Exception {
+
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfo("127.0.0.1:10911"));
+ when(adminExt.getAllSubscriptionGroup("127.0.0.1:10911", 5000L))
+ .thenReturn(subscriptionGroups("group-a"));
+ when(adminExt.examineConsumerConnectionInfo("group-a",
"127.0.0.1:10911"))
+ .thenThrow(new
MQClientException(ResponseCode.CONSUMER_NOT_ONLINE, "offline"));
+ when(adminExt.examineConsumerConnectionInfo(
+ "CID_DefaultHeartBeatSyncerTopic", "127.0.0.1:10911"))
+ .thenReturn(consumerConnections(connection("syncer",
"10.0.0.8:40000")));
+ Connection proxyClient = connection("proxy-client", "10.0.0.9:50000");
+
proxyClient.setVersion(org.apache.rocketmq.common.MQVersion.Version.V5_0_0.ordinal());
+ when(adminExt.examineConsumerConnectionInfo("group-a",
"10.0.0.8:8080"))
+ .thenReturn(consumerConnections(proxyClient));
+
+
org.apache.rocketmq.studio.cluster.nameserver.NameserverRegistryService
registry =
+
org.mockito.Mockito.mock(org.apache.rocketmq.studio.cluster.nameserver.NameserverRegistryService.class);
+
when(registry.requireRegisteredAddress("selected:9876")).thenReturn("selected:9876");
+ MockMvcBuilders.standaloneSetup(new ClientController(new
ClientService(provider, registry))).build()
+ .perform(get("/api/clients").param("namesrvAddr",
"selected:9876")
+ .param("clusterId", "cluster-a").param("type",
"Consumer"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.data.length()").value(1))
+
.andExpect(jsonPath("$.data[0].clientId").value("proxy-client"))
+
.andExpect(jsonPath("$.data[0].clusterName").value("cluster-a"))
+ .andExpect(jsonPath("$.data[0].language").value("Java"))
+ .andExpect(jsonPath("$.data[0].protocol").doesNotExist())
+ .andExpect(jsonPath("$.data[0].version").doesNotExist());
+ verify(adminFactory).execute(eq("selected:9876"), any(),
any(MqAdminExtFactory.AdminAction.class));
+ verify(runtimeAdminClientResolver, never()).execute(anyString(),
any());
+ }
+
+ @Test
+ void consumerScanMergesDirectAndAllProxyConnectionsTest() throws Exception
{
+ prepareProxyGroup(adminExt, "127.0.0.1:10911", "10.0.0.8", "10.0.0.9");
+ Connection direct = connection("direct", "10.0.0.1:40000");
+ when(adminExt.examineConsumerConnectionInfo("group-a",
"127.0.0.1:10911"))
+ .thenReturn(consumerConnections(direct));
+ Connection unknown = connection("proxy-two", "10.0.0.3:40000");
+ unknown.setLanguage(null);
+ when(adminExt.examineConsumerConnectionInfo("group-a",
"10.0.0.8:8080"))
+ .thenReturn(consumerConnections(direct,
connection("proxy-one", "10.0.0.2:40000")));
+ when(adminExt.examineConsumerConnectionInfo("group-a",
"10.0.0.9:8080"))
+ .thenReturn(consumerConnections(unknown));
+
+ List<ClientConnectionVO> rows = provider.findConnections("instance-a",
"cluster-a", "Consumer");
+
+ assertThat(rows).extracting(ClientConnectionVO::getClientId)
+ .containsExactlyInAnyOrder("direct", "proxy-one", "proxy-two");
+ assertThat(rows).filteredOn(row ->
row.getClientId().equals("direct")).singleElement()
+ .satisfies(row -> {
+
assertThat(row.getProtocol()).isEqualTo(org.apache.rocketmq.studio.common.domain.enums.Protocol.Remoting);
+
assertThat(row.getVersion()).isEqualTo(org.apache.rocketmq.common.MQVersion.getVersionDesc(direct.getVersion()));
+ });
+ assertThat(rows).filteredOn(row ->
row.getClientId().equals("proxy-two")).singleElement()
+ .satisfies(row -> {
+ assertThat(row.getLanguage()).isNull();
+ assertThat(row.getProtocol()).isNull();
+ assertThat(row.getVersion()).isNull();
+ assertThat(row.isPartial()).isFalse();
+ });
+ }
+
+ @Test
+ void proxyDiscoveryCannotLeakAcrossNameserversSharingBrokerAddressesTest()
throws Exception {
+ DefaultMQAdminExt other =
org.mockito.Mockito.mock(DefaultMQAdminExt.class);
+ prepareProxyGroup(adminExt, "127.0.0.1:10911", "10.0.0.8");
+ prepareProxyGroup(other, "127.0.0.1:10911", "10.1.0.8");
+ when(adminFactory.execute(eq("other:9876"), any(), any()))
+ .thenAnswer(invocation ->
invocation.<MqAdminExtFactory.AdminAction<Object>>getArgument(2).apply(other));
+ when(adminExt.examineConsumerConnectionInfo("group-a",
"10.0.0.8:8080"))
+ .thenReturn(consumerConnections(connection("selected-client",
"10.0.0.1:40000")));
+ when(other.examineConsumerConnectionInfo("group-a", "10.1.0.8:8080"))
+ .thenReturn(consumerConnections(connection("other-client",
"10.1.0.1:40000")));
+
+ assertThat(provider.findConnectionsAt("selected:9876", null,
"Consumer"))
+
.extracting(ClientConnectionVO::getClientId).containsExactly("selected-client");
+ assertThat(provider.findConnectionsAt("other:9876", null, "Consumer"))
+
.extracting(ClientConnectionVO::getClientId).containsExactly("other-client");
+ assertThat(provider.findConnectionsAt("selected:9876", null,
"Consumer"))
+
.extracting(ClientConnectionVO::getClientId).containsExactly("selected-client");
+ }
+
+ @Test
+ void proxyScanPreservesSameGroupInDifferentClustersAndFiltersTest() throws
Exception {
+ prepareProxyGroup(adminExt, "127.0.0.1:10911", "10.0.0.8");
+
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-a"));
+
when(adminExt.examineConsumerConnectionInfo("CID_DefaultHeartBeatSyncerTopic",
"127.0.0.2:10911"))
+ .thenReturn(consumerConnections(connection("syncer-b",
"10.1.0.8:40000")));
+ when(adminExt.examineConsumerConnectionInfo("group-a",
"10.0.0.8:8080"))
+ .thenReturn(consumerConnections(connection("client-a",
"10.0.0.1:40000")));
+ when(adminExt.examineConsumerConnectionInfo("group-a",
"10.1.0.8:8080"))
+ .thenReturn(consumerConnections(connection("client-b",
"10.1.0.1:40000")));
+
+ assertThat(provider.findConnectionsAt("selected:9876", null,
"Consumer"))
+
.extracting(ClientConnectionVO::getClusterName).containsExactlyInAnyOrder("cluster-a",
"cluster-b");
+ assertThat(provider.findConnectionsAt("selected:9876", "cluster-b",
"Consumer")).singleElement()
+ .satisfies(row -> {
+ assertThat(row.getClientId()).isEqualTo("client-b");
+ assertThat(row.getClusterName()).isEqualTo("cluster-b");
+ });
+ verify(adminExt, times(1)).examineConsumerConnectionInfo("group-a",
"10.0.0.8:8080");
+ }
+
+ @Test
+ void proxyFailureKeepsPartialRowsButCannotClaimOfflineTest() throws
Exception {
+ prepareProxyGroup(adminExt, "127.0.0.1:10911", "10.0.0.8");
+ when(adminExt.examineConsumerConnectionInfo("group-a",
"10.0.0.8:8080"))
+ .thenThrow(new RemotingConnectException("proxy down"));
+ when(adminExt.examineConsumerConnectionInfo("group-a",
"127.0.0.1:10911"))
+ .thenReturn(consumerConnections(connection("direct",
"10.0.0.1:40000")))
+ .thenThrow(new
MQBrokerException(ResponseCode.CONSUMER_NOT_ONLINE, "offline"));
+
+ assertThat(provider.findConnectionsAt("selected:9876", null,
"Consumer")).singleElement()
+ .satisfies(row -> {
+ assertThat(row.getClientId()).isEqualTo("direct");
+ assertThat(row.isPartial()).isTrue();
+ });
+ assertThatThrownBy(() -> provider.findConnectionsAt("selected:9876",
null, "Consumer"))
+ .isInstanceOf(BusinessException.class).hasMessage("Failed to
query consumer connections from all groups");
+ org.mockito.Mockito.doThrow(new
MQBrokerException(ResponseCode.CONSUMER_NOT_ONLINE, "offline"))
+ .when(adminExt).examineConsumerConnectionInfo("group-a",
"10.0.0.8:8080");
+ assertThat(provider.findConnectionsAt("selected:9876", null,
"Consumer")).isEmpty();
+ }
+
+ @Test
+ void proxyDiscoveryRetriesFailuresAndIsSharedOnlyWithinRequestTest()
throws Exception {
+ prepareProxyGroup(adminExt, "127.0.0.1:10911", "10.0.0.8");
+
when(adminExt.examineConsumerConnectionInfo("CID_DefaultHeartBeatSyncerTopic",
"127.0.0.1:10911"))
+ .thenThrow(new RemotingConnectException("discovery down"))
+ .thenReturn(consumerConnections(connection("syncer",
"10.0.0.8:40000")));
+ assertThatThrownBy(() -> provider.findConnectionsAt("selected:9876",
null, "Consumer"))
+ .isInstanceOf(BusinessException.class);
+ when(adminExt.getAllSubscriptionGroup("127.0.0.1:10911", 5000L))
+ .thenReturn(subscriptionGroups("group-a", "group-b"));
+ when(adminExt.examineConsumerConnectionInfo("group-a",
"10.0.0.8:8080"))
+ .thenReturn(consumerConnections(connection("proxy-client",
"10.0.0.1:40000")));
+ assertThat(provider.findConnectionsAt("selected:9876", null,
"Consumer"))
+
.extracting(ClientConnectionVO::getClientId).containsExactly("proxy-client");
+ verify(adminExt, times(2)).examineConsumerConnectionInfo(
+ "CID_DefaultHeartBeatSyncerTopic", "127.0.0.1:10911");
+ }
+
+ @Test
+ void directConsumerOnAnotherBrokerInTheClusterRemainsVisibleTest() throws
Exception {
+ ClusterInfo topology = clusterInfo("127.0.0.1:10911",
"127.0.0.2:10911");
+ String lastBroker = new
java.util.ArrayList<>(topology.getBrokerAddrTable().values())
+ .getLast().selectBrokerAddr();
+ when(adminExt.examineBrokerClusterInfo()).thenReturn(topology);
+ when(adminExt.getAllSubscriptionGroup(anyString(),
anyLong())).thenReturn(subscriptionGroups("group-a"));
+ when(adminExt.examineConsumerConnectionInfo("group-a", lastBroker))
+ .thenReturn(consumerConnections(connection("direct",
"10.0.0.1:40000")));
+ assertThat(provider.findConnectionsAt("selected:9876", "cluster-a",
"Consumer"))
+
.extracting(ClientConnectionVO::getClientId).containsExactly("direct");
+ }
+
+ @Test
+ void emptyPartialProxyInventoryIsAnErrorTest() throws Exception {
+ prepareProxyGroup(adminExt, "127.0.0.1:10911", "10.0.0.8");
+ when(adminExt.getAllSubscriptionGroup("127.0.0.1:10911", 5000L))
+ .thenReturn(subscriptionGroups("group-a", "group-b"));
+ when(adminExt.examineConsumerConnectionInfo("group-a",
"10.0.0.8:8080"))
+ .thenThrow(new RemotingConnectException("proxy down"));
+ assertThatThrownBy(() -> provider.findConnectionsAt("selected:9876",
null, "Consumer"))
+ .isInstanceOf(BusinessException.class);
+ }
+
+ @Test
+ void proxyQueriesPreserveInterruptsAndDoNotSwallowProgrammingErrorsTest()
throws Exception {
+ prepareProxyGroup(adminExt, "127.0.0.1:10911", "10.0.0.8");
+ when(adminExt.examineConsumerConnectionInfo("group-a",
"10.0.0.8:8080"))
+ .thenThrow(new InterruptedException("cancelled"))
+ .thenThrow(new IllegalArgumentException("invalid state"));
+ try {
+ assertThatThrownBy(() ->
provider.findConnectionsAt("selected:9876", null, "Consumer"))
+
.isInstanceOf(BusinessException.class).hasMessage("Consumer connection query
interrupted");
+ assertThat(Thread.currentThread().isInterrupted()).isTrue();
+ } finally {
+ Thread.interrupted();
+ }
+ assertThatThrownBy(() -> provider.findConnectionsAt("selected:9876",
null, "Consumer"))
+
.isInstanceOf(IllegalArgumentException.class).hasMessage("invalid state");
+ }
+
+ @Test
+ void consumerOfflineMessageWithoutNotOnlineCodeIsNotAScanFailureTest()
throws Exception {
+
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfo("127.0.0.1:10911"));
+ when(adminExt.getAllSubscriptionGroup("127.0.0.1:10911", 5000L))
+ .thenReturn(subscriptionGroups("group-a"));
+ when(adminExt.examineConsumerConnectionInfo("group-a",
"127.0.0.1:10911"))
+ .thenThrow(new MQBrokerException(ResponseCode.SYSTEM_ERROR,
+ "Not found the consumer group connection"));
+
+ assertThat(provider.findConnectionsAt("selected:9876", null,
"Consumer")).isEmpty();
+ }
+
+ @Test
+ void heartbeatSyncerTopicNotExistIsTreatedAsNoObservableProxyTest() throws
Exception {
+
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfo("127.0.0.1:10911"));
+ when(adminExt.getAllSubscriptionGroup("127.0.0.1:10911", 5000L))
+ .thenReturn(subscriptionGroups("group-a"));
+ when(adminExt.examineConsumerConnectionInfo("group-a",
"127.0.0.1:10911"))
+ .thenReturn(consumerConnections(connection("direct",
"10.0.0.1:40000")));
+
when(adminExt.examineConsumerConnectionInfo("CID_DefaultHeartBeatSyncerTopic",
"127.0.0.1:10911"))
+ .thenThrow(new MQBrokerException(ResponseCode.TOPIC_NOT_EXIST,
+ "No route info of this topic:
CID_DefaultHeartBeatSyncerTopic"));
+
+ assertThat(provider.findConnectionsAt("selected:9876", null,
"Consumer")).singleElement()
+ .satisfies(row -> {
+ assertThat(row.getClientId()).isEqualTo("direct");
+ assertThat(row.isPartial()).isFalse();
+ });
+ }
+
+ @Test
+ void proxyDiscoveryFailureIsCachedForTheRestOfTheRequestTest() throws
Exception {
+
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfo("127.0.0.1:10911"));
+ when(adminExt.getAllSubscriptionGroup("127.0.0.1:10911", 5000L))
+ .thenReturn(subscriptionGroups("group-a", "group-b"));
+
when(adminExt.examineConsumerConnectionInfo("CID_DefaultHeartBeatSyncerTopic",
"127.0.0.1:10911"))
+ .thenThrow(new RemotingConnectException("discovery down"));
+ when(adminExt.examineConsumerConnectionInfo("group-a",
"127.0.0.1:10911"))
+ .thenReturn(consumerConnections(connection("direct",
"10.0.0.1:40000")));
+ when(adminExt.examineConsumerConnectionInfo("group-b",
"127.0.0.1:10911"))
+ .thenReturn(new ConsumerConnection());
+
+ assertThat(provider.findConnectionsAt("selected:9876", null,
"Consumer")).singleElement()
+ .satisfies(row -> {
+ assertThat(row.getClientId()).isEqualTo("direct");
+ assertThat(row.isPartial()).isTrue();
+ });
+ verify(adminExt, times(1)).examineConsumerConnectionInfo(
+ "CID_DefaultHeartBeatSyncerTopic", "127.0.0.1:10911");
+ }
+
+ private void prepareProxyGroup(DefaultMQAdminExt admin, String broker,
String... proxies) throws Exception {
+ lenient().when(admin.examineConsumerConnectionInfo(anyString(),
anyString()))
+ .thenReturn(new ConsumerConnection());
+ when(admin.examineBrokerClusterInfo()).thenReturn(clusterInfo(broker));
+ when(admin.getAllSubscriptionGroup(broker,
5000L)).thenReturn(subscriptionGroups("group-a"));
+ Connection[] syncers = java.util.Arrays.stream(proxies)
+ .map(proxy -> connection("syncer-" + proxy, proxy +
":40000")).toArray(Connection[]::new);
+
when(admin.examineConsumerConnectionInfo("CID_DefaultHeartBeatSyncerTopic",
broker))
+ .thenReturn(consumerConnections(syncers));
+ }
+
+ private static ConsumerConnection consumerConnections(Connection...
connections) {
+ ConsumerConnection result = new ConsumerConnection();
+ result.setConnectionSet(new HashSet<>(List.of(connections)));
+ return result;
+ }
+
@Test
void findConnectionsAtShouldUseNameserverScopedAdminClientTest() throws
Exception {
ClusterInfo clusterInfo = new ClusterInfo();
@@ -496,7 +763,7 @@ class RocketMQClientProviderTest {
.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);
+ when(adminExt.examineConsumerConnectionInfo("group-b",
"127.0.0.2:10911")).thenReturn(consumerConnection);
List<ClientConnectionVO> connections =
provider.findConnections("instance-a", "cluster-b", "Consumer");
@@ -534,7 +801,7 @@ class RocketMQClientProviderTest {
consumerConnection.setConnectionSet(connectionSet);
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfo);
when(adminExt.getAllSubscriptionGroup("127.0.0.1:10911",
5000L)).thenReturn(wrapper);
-
when(adminExt.examineConsumerConnectionInfo("group-a")).thenReturn(consumerConnection);
+ when(adminExt.examineConsumerConnectionInfo("group-a",
"127.0.0.1:10911")).thenReturn(consumerConnection);
List<ClientConnectionVO> connections =
provider.findConnections("instance-a", "cluster-a", "Consumer");
@@ -560,8 +827,8 @@ class RocketMQClientProviderTest {
SubscriptionGroupWrapper wrapper = subscriptionGroups("group-a",
"group-b");
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfo("127.0.0.1:10911"));
when(adminExt.getAllSubscriptionGroup("127.0.0.1:10911",
5000L)).thenReturn(wrapper);
- when(adminExt.examineConsumerConnectionInfo(anyString()))
- .thenThrow(new IllegalStateException("broker unavailable"));
+ when(adminExt.examineConsumerConnectionInfo(anyString(), anyString()))
+ .thenThrow(new RemotingConnectException("broker unavailable"));
assertThatThrownBy(() -> provider.findConnections("instance-a",
"cluster-a", "Consumer"))
.isInstanceOf(BusinessException.class)
@@ -578,7 +845,7 @@ class RocketMQClientProviderTest {
List<ClientConnectionVO> connections =
provider.findConnections("instance-a", "cluster-a", "Consumer");
assertThat(connections).isEmpty();
- verify(adminExt, never()).examineConsumerConnectionInfo(anyString());
+ verify(adminExt, never()).examineConsumerConnectionInfo(anyString(),
anyString());
}
@Test
@@ -588,9 +855,9 @@ class RocketMQClientProviderTest {
consumerConnection.setConnectionSet(new
HashSet<>(List.of(connection("consumer-client", "10.0.0.2:1000"))));
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfo("127.0.0.1:10911"));
when(adminExt.getAllSubscriptionGroup("127.0.0.1:10911",
5000L)).thenReturn(wrapper);
- when(adminExt.examineConsumerConnectionInfo("group-a"))
- .thenThrow(new IllegalStateException("broker unavailable"));
-
when(adminExt.examineConsumerConnectionInfo("group-b")).thenReturn(consumerConnection);
+ when(adminExt.examineConsumerConnectionInfo("group-a",
"127.0.0.1:10911"))
+ .thenThrow(new RemotingConnectException("broker unavailable"));
+ when(adminExt.examineConsumerConnectionInfo("group-b",
"127.0.0.1:10911")).thenReturn(consumerConnection);
List<ClientConnectionVO> connections =
provider.findConnections("instance-a", "cluster-a", "Consumer");
@@ -605,7 +872,7 @@ class RocketMQClientProviderTest {
SubscriptionGroupWrapper wrapper = subscriptionGroups("group-a",
"group-b");
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfo("127.0.0.1:10911"));
when(adminExt.getAllSubscriptionGroup("127.0.0.1:10911",
5000L)).thenReturn(wrapper);
-
when(adminExt.examineConsumerConnectionInfo(anyString())).thenThrow(new
MQClientException(
+ when(adminExt.examineConsumerConnectionInfo(anyString(),
anyString())).thenThrow(new MQClientException(
206, "Not found the consumer group connection"));
List<ClientConnectionVO> connections =
provider.findConnections("instance-a", "cluster-a", null);
@@ -620,9 +887,9 @@ class RocketMQClientProviderTest {
consumerConnection.setConnectionSet(new
HashSet<>(List.of(connection("consumer-client", "10.0.0.2:1000"))));
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfo("127.0.0.1:10911"));
when(adminExt.getAllSubscriptionGroup("127.0.0.1:10911",
5000L)).thenReturn(wrapper);
- when(adminExt.examineConsumerConnectionInfo("group-a")).thenThrow(new
MQClientException(
+ when(adminExt.examineConsumerConnectionInfo("group-a",
"127.0.0.1:10911")).thenThrow(new MQClientException(
206, "Not found the consumer group connection"));
-
when(adminExt.examineConsumerConnectionInfo("group-b")).thenReturn(consumerConnection);
+ when(adminExt.examineConsumerConnectionInfo("group-b",
"127.0.0.1:10911")).thenReturn(consumerConnection);
List<ClientConnectionVO> connections =
provider.findConnections("instance-a", "cluster-a", "Consumer");
diff --git a/web/src/api/connections.ts b/web/src/api/connections.ts
index 38e163604..68e71d86b 100644
--- a/web/src/api/connections.ts
+++ b/web/src/api/connections.ts
@@ -5,10 +5,10 @@ export interface ClientConnection {
clientId?: string | null;
type: string;
groupOrTopic: string;
- protocol: string;
+ protocol: string | null;
address?: string | null;
language?: string | null;
- version: string;
+ version: string | null;
connectedAt?: string | null;
partial?: boolean;
clusterName: string;
diff --git a/web/src/i18n/translations.ts b/web/src/i18n/translations.ts
index ad9145450..558aaec0d 100644
--- a/web/src/i18n/translations.ts
+++ b/web/src/i18n/translations.ts
@@ -210,6 +210,10 @@ const translations: Record<string, Record<Lang, string>> =
{
},
'cluster.k8sNamespace': { zh: 'K8s Namespace', en: 'K8s Namespace' },
'cluster.k8sId': { zh: 'k8s ID', en: 'k8s ID' },
+ 'clients.partialScan': {
+ zh: '客户端连接列表不完整：部分查询失败。',
+ en: 'Client connections are incomplete: some queries failed.',
+ },
'clients.selectClusterPlaceholder': { zh: '选择 Broker 集群', en: 'Select Broker
cluster' },
'clients.selectNameserverPlaceholder': { zh: '选择 NameServer', en: 'Select
NameServer' },
'cluster.k8sIdExtra': {
diff --git a/web/src/pages/cluster/__tests__/ClientsPage.test.tsx
b/web/src/pages/cluster/__tests__/ClientsPage.test.tsx
index ecae50abd..86fd3adf7 100644
--- a/web/src/pages/cluster/__tests__/ClientsPage.test.tsx
+++ b/web/src/pages/cluster/__tests__/ClientsPage.test.tsx
@@ -350,6 +350,95 @@ describe('Clients page', () => {
expect(screen.queryByText('[email protected]:49154')).toBeNull();
});
+ it('renders missing Proxy metadata in the table, distributions and details',
async () => {
+ const user = userEvent.setup();
+ vi.mocked(connectionsService.listConnections).mockResolvedValue([
+ {
+ ...connection,
+ type: 'Consumer',
+ protocol: null,
+ language: null,
+ version: null,
+ partial: true,
+ },
+ { ...connections[2], clusterName: 'ns-prod' },
+ ]);
+ renderWithProviders(<ClientsPage />);
+
+ const rows = await screen.findAllByRole('row', { name:
/order-svc-0@10\.0\.1\.12:49152/ });
+ const row = rows.find((candidate) =>
+ within(candidate).queryByRole('button', { name: /详情/ }),
+ )!;
+ // Protocol and version keep the neutral '-' placeholder; a missing
language renders
+ // through the page's renderLanguageTag helper as the localized "unknown"
label (#4850).
+ expect(within(row).getAllByText('-')).toHaveLength(2);
+ expect(within(row).getByText('未知')).toBeInTheDocument();
+ expect(within(row).queryByText('gRPC')).not.toBeInTheDocument();
+ expect(within(row).queryByText('Remoting')).not.toBeInTheDocument();
+ expect(
+ within(screen.getByTestId('protocol-distribution')).getByText('-: 1'),
+ ).toBeInTheDocument();
+ expect(
+
within(screen.getByTestId('language-version-distribution')).getByText('未知
unknown: 1'),
+ ).toBeInTheDocument();
+ expect(screen.getByText('客户端连接列表不完整：部分查询失败。')).toBeInTheDocument();
+ await user.click(within(row).getByRole('button', { name: /详情/ }));
+ const dialog = await screen.findByRole('dialog');
+ expect(within(dialog).getAllByText('-')).toHaveLength(2);
+ expect(within(dialog).getByText('未知')).toBeInTheDocument();
+ expect(within(dialog).queryByText('null')).not.toBeInTheDocument();
+ });
+
+ it('filters protocol-less proxy rows through the unknown protocol option',
async () => {
+ const createObjectURL = vi.fn((blob: Blob | MediaSource) => {
+ expect(blob).toBeInstanceOf(Blob);
+ return 'blob:unknown-protocol-connections';
+ });
+ Object.defineProperty(URL, 'createObjectURL', {
+ writable: true,
+ value: createObjectURL,
+ });
+ Object.defineProperty(URL, 'revokeObjectURL', {
+ writable: true,
+ value: vi.fn(),
+ });
+ vi.spyOn(HTMLAnchorElement.prototype, 'click').mockImplementation(() =>
{});
+ const user = userEvent.setup({ pointerEventsCheck: 0 });
+ vi.mocked(connectionsService.listConnections).mockResolvedValue([
+ { ...connection, type: 'Consumer', protocol: null },
+ connections[1],
+ ]);
+ renderWithProviders(<ClientsPage />);
+
+ await screen.findAllByText('[email protected]:49153');
+ expect(
+ within(screen.getByTestId('protocol-distribution')).getByText('-: 1'),
+ ).toBeInTheDocument();
+ const filterTriggers =
document.querySelectorAll<HTMLElement>('.ant-table-filter-trigger');
+ await user.click(filterTriggers[2]);
+ const filterDropdown =
document.querySelector<HTMLElement>('.ant-table-filter-dropdown');
+ expect(filterDropdown).not.toBeNull();
+ await user.click(within(filterDropdown!).getByText('未知'));
+ await user.click(within(filterDropdown!).getByRole('button', { name: 'OK'
}));
+
+ const hasDetailButton = (candidate: HTMLElement) =>
+ within(candidate).queryByRole('button', { name: /详情/ }) !== null;
+ const orderRows = await screen.findAllByRole('row', {
+ name: /order-svc-0@10\.0\.1\.12:49152/,
+ });
+ expect(orderRows.some(hasDetailButton)).toBe(true);
+ const paymentRows = screen.queryAllByRole('row', {
+ name: /payment-svc-0@10\.0\.1\.13:49153/,
+ });
+ expect(paymentRows.some(hasDetailButton)).toBe(false);
+
+ await user.click(screen.getByRole('button', { name: /导出/ }));
+ const blob = createObjectURL.mock.calls[0][0] as Blob;
+ const csv = await blob.text();
+ expect(csv).toContain('[email protected]:49152');
+ expect(csv).not.toContain('[email protected]:49153');
+ });
+
it('keeps incomplete client metadata searchable by address', async () => {
const user = userEvent.setup({ pointerEventsCheck: 0 });
vi.mocked(connectionsService.listConnections).mockResolvedValue([
diff --git a/web/src/pages/cluster/clients.tsx
b/web/src/pages/cluster/clients.tsx
index c6e51538d..8c2e67b1f 100644
--- a/web/src/pages/cluster/clients.tsx
+++ b/web/src/pages/cluster/clients.tsx
@@ -133,12 +133,14 @@ const CLIENT_CONNECTION_EXPORT_COLUMNS:
CsvColumn<ClientConnection>[] = [
type ClientTableFilters =
Parameters<NonNullable<TableProps<ClientConnection>['onChange']>>[1];
-const countBy = (values: string[]) =>
+const countBy = (values: Array<string | null>) =>
[
- ...values.reduce(
- (counts, value) => counts.set(value, (counts.get(value) ?? 0) + 1),
- new Map<string, number>(),
- ),
+ ...values
+ .map(displayMetadata)
+ .reduce(
+ (counts, value) => counts.set(value, (counts.get(value) ?? 0) + 1),
+ new Map<string, number>(),
+ ),
]
.map(([label, count]) => ({ label, count }))
.sort((a, b) => b.count - a.count || a.label.localeCompare(b.label));
@@ -151,6 +153,9 @@ const displayMetadata = (value: string | null | undefined)
=> value || '-';
*/
const UNKNOWN_LANGUAGE = 'unknown';
+/** Filter sentinel for rows whose protocol the Proxy response could not
confirm. */
+const UNKNOWN_PROTOCOL = 'unknown';
+
/* ═══════════════════════════════════════════
ClientsPage
═══════════════════════════════════════════ */
@@ -301,7 +306,8 @@ const ClientsPage = () => {
protocols: countBy(instances.map((connection) => connection.protocol)),
languageVersions: countBy(
instances.map(
- (connection) => `${connection.language ?? UNKNOWN_LANGUAGE}
${connection.version}`,
+ (connection) =>
+ `${connection.language ?? UNKNOWN_LANGUAGE} ${connection.version
?? UNKNOWN_LANGUAGE}`,
),
),
};
@@ -356,7 +362,7 @@ const ClientsPage = () => {
);
const exportConnections = useMemo(() => {
- const matches = (key: string, value: string) => {
+ const matches = (key: string, value: string | null) => {
const selected = columnFilters[key];
return !selected?.length || selected.some((filterValue) =>
String(filterValue) === value);
};
@@ -364,7 +370,7 @@ const ClientsPage = () => {
(connection) =>
matches('clusterName', connection.clusterName) &&
matches('type', connection.type) &&
- matches('protocol', connection.protocol) &&
+ matches('protocol', connection.protocol ?? UNKNOWN_PROTOCOL) &&
matches('language', connection.language ?? ''),
);
}, [columnFilters, filtered]);
@@ -459,11 +465,16 @@ const ClientsPage = () => {
filters: [
{ text: 'gRPC', value: 'gRPC' },
{ text: 'Remoting', value: 'Remoting' },
+ { text: t('common.unknown'), value: UNKNOWN_PROTOCOL },
],
filteredValue: columnFilters.protocol ?? null,
- onFilter: (value, record) => record.protocol === value,
- render: (protocol: string) => {
- const cfg = protocolConfig[protocol] ?? { color: 'default', label:
protocol };
+ onFilter: (value, record) =>
+ value === UNKNOWN_PROTOCOL ? !record.protocol : record.protocol ===
value,
+ render: (protocol: string | null) => {
+ const cfg = protocolConfig[protocol ?? ''] ?? {
+ color: 'default',
+ label: displayMetadata(protocol),
+ };
return <Tag color={cfg.color}>{cfg.label}</Tag>;
},
},
@@ -494,6 +505,7 @@ const ClientsPage = () => {
dataIndex: 'version',
key: 'version',
width: 90,
+ render: displayMetadata,
},
{
title: t('cluster.heartbeat'),
@@ -721,7 +733,7 @@ const ClientsPage = () => {
<Alert
showIcon
type="warning"
- message="Producer connections are sampled because the topic scan
limit was reached."
+ message={t('clients.partialScan')}
style={{ marginBottom: 16 }}
/>
)}
@@ -995,8 +1007,9 @@ const ClientsPage = () => {
{selectedConnection.groupOrTopic}
</Descriptions.Item>
<Descriptions.Item label={t('clients.protocol')}>
- <Tag color={protocolConfig[selectedConnection.protocol]?.color
?? 'default'}>
- {protocolConfig[selectedConnection.protocol]?.label ??
selectedConnection.protocol}
+ <Tag color={protocolConfig[selectedConnection.protocol ??
'']?.color ?? 'default'}>
+ {protocolConfig[selectedConnection.protocol ?? '']?.label ??
+ displayMetadata(selectedConnection.protocol)}
</Tag>
</Descriptions.Item>
<Descriptions.Item label={t('common.address')}>
@@ -1008,7 +1021,7 @@ const ClientsPage = () => {
{renderLanguageTag(selectedConnection.language)}
</Descriptions.Item>
<Descriptions.Item label={t('common.version')}>
- {selectedConnection.version}
+ {displayMetadata(selectedConnection.version)}
</Descriptions.Item>
<Descriptions.Item label={t('cluster.heartbeat')}>
{selectedConnection.connectedAt ?? '-'}
