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 cd448e17f fix(group): preserve unavailable consumer connections (#4544)
cd448e17f is described below
commit cd448e17f14b4bd6e6d4ade87300caf19ca91bc2
Author: zmuxuny <[email protected]>
AuthorDate: Mon Sep 21 18:27:02 2026 +0800
fix(group): preserve unavailable consumer connections (#4544)
Consumer connection lookup failures were collapsed into
`onlineInstances=0`, so a group with positive lag and an unreachable proxy was
reported as "backlog with no online consumer". `ProxyConsumerResolver` now
resolves three states — available, unavailable (discovery or proxy lookup
failure) and a confirmed empty set (`CONSUMER_NOT_ONLINE`) — and the `-1`
sentinel is carried through the web table, both CSV exports, the diagnostics
and the AI group tools instead of being flattened to zero.
The tool catalog schema is part of the change: `rmq.group.detail` declared
`onlineInstances` with `minimum: 0`, and `ToolValidationFilter` validates every
tool result, so the sentinel would have failed the whole call at runtime. All
three schema sites now allow and document `-1`, and `catalog_gen.go` is
regenerated to keep the digest in sync.
Fixes #4543
---
docs/api-spec.md | 2 +-
rmqctl/internal/catalog/catalog_gen.go | 2 +-
.../studio/instance/group/ConsumerGroupVO.java | 1 +
.../studio/instance/topic/MetadataService.java | 9 ++-
.../ai/tool/contract/group/GroupDetailOutput.java | 1 +
.../ops/ai/tool/contract/group/GroupListItem.java | 1 +
.../tool/handler/group/GroupDetailToolHandler.java | 7 +-
.../provider/apache/ProxyConsumerResolver.java | 94 ++++++++++++++++------
.../provider/apache/RocketMQAdminClientImpl.java | 8 +-
.../provider/apache/RocketMQMetadataProvider.java | 28 ++++---
.../main/resources/tool-catalog/tools/group.yaml | 7 +-
.../studio/instance/topic/MetadataServiceTest.java | 16 ++++
.../group/ConsumerGroupReadToolHandlersTest.java | 17 ++++
.../tool/service/ToolOutputSchemaContractTest.java | 23 +++++-
.../provider/apache/ProxyConsumerResolverTest.java | 57 ++++++++++---
.../apache/RocketMQAdminClientImplTest.java | 22 ++++-
.../apache/RocketMQMetadataProviderTest.java | 35 +++++++-
web/src/api/metadata.ts | 1 +
.../pages/instance/__tests__/ConsumerPage.test.tsx | 33 ++++++++
web/src/pages/instance/consumer.tsx | 19 ++++-
web/src/services/consumerService.test.ts | 40 ++++++++-
web/src/services/consumerService.ts | 6 +-
web/src/utils/consumerConnections.test.ts | 51 ++++++++++++
web/src/utils/consumerConnections.ts | 35 ++++++++
web/src/utils/consumerGroupDiagnostics.test.ts | 12 +++
web/src/utils/consumerGroupDiagnostics.ts | 18 ++++-
26 files changed, 474 insertions(+), 71 deletions(-)
diff --git a/docs/api-spec.md b/docs/api-spec.md
index 199cb201d..a0754b29e 100644
--- a/docs/api-spec.md
+++ b/docs/api-spec.md
@@ -940,7 +940,7 @@ GET /api/groups?clusterId={clusterId}&search={keyword}
| `clusterId` | `string` | 所属集群 ID |
| `subscriptionMode` | `string` | 订阅模式: `Push` / `Pop` |
| `consumeType` | `string` | 消费类型: `CLUSTERING` / `BROADCASTING` |
-| `onlineInstances` | `number` | 在线实例数 |
+| `onlineInstances` | `number` | 在线实例数；`-1` 表示连接信息不可用 |
| `totalLag` | `number` | 总堆积消息数 |
| `subscribedTopics` | `string[]` | 订阅的 Topic 列表 |
| `subscriptionDataType` | `string` | 订阅数据类型: `NORMAL` / `FIFO` / `DELAY` /
`TRANSACTION` |
diff --git a/rmqctl/internal/catalog/catalog_gen.go
b/rmqctl/internal/catalog/catalog_gen.go
index 0a1531098..35db62557 100644
--- a/rmqctl/internal/catalog/catalog_gen.go
+++ b/rmqctl/internal/catalog/catalog_gen.go
@@ -21,7 +21,7 @@ package catalog
var defaultDocument = Document{
Version: "2.0.0",
MinimumClientVersion: "2.0.0",
- Digest:
"bc0da0891d07f943db08439900892d21984c27d873f00bee3cc2923a35a3dd8f",
+ Digest:
"71c2cc225321b7e83243ef77b7330f25800e9623c0a110f56b27a6834a21735e",
Tools: []Tool{
{
Name: "rmq.acl.list",
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupVO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupVO.java
index 11acd4c7f..fe624d63b 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupVO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupVO.java
@@ -34,6 +34,7 @@ public class ConsumerGroupVO extends BaseEntity {
private String instanceId;
private SubscriptionMode subscriptionMode;
private ConsumeType consumeType;
+ /** Number of confirmed online clients, or -1 when connection inventory is
unavailable. */
private int onlineInstances;
private long totalLag;
private List<String> subscribedTopics = new ArrayList<>();
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
index 2b551cd48..06a9a11bb 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
@@ -695,8 +695,9 @@ public class MetadataService {
for (ConsumerGroupVO group : groups) {
CsvUtil.appendRow(csv, group.getName(), group.getNamespace(),
group.getClusterId(),
toText(group.getSubscriptionMode()),
toText(group.getConsumeType()),
- group.getOnlineInstances(), lagText(group.getTotalLag()),
group.getDelaySeconds(),
- group.getSubscriptionDataType(),
group.getDeliveryOrderType(), group.getRetryMaxTimes(),
+ onlineInstancesText(group.getOnlineInstances()),
lagText(group.getTotalLag()),
+ group.getDelaySeconds(), group.getSubscriptionDataType(),
group.getDeliveryOrderType(),
+ group.getRetryMaxTimes(),
String.join(";", group.getSubscribedTopics() == null ?
List.of() : group.getSubscribedTopics()),
group.getGmtCreate(), group.getGmtModified());
}
@@ -707,6 +708,10 @@ public class MetadataService {
return totalLag == ConsumerLagResolver.UNKNOWN ? "unknown" :
String.valueOf(totalLag);
}
+ private static String onlineInstancesText(int onlineInstances) {
+ return onlineInstances < 0 ? "unknown" :
String.valueOf(onlineInstances);
+ }
+
private String toText(Object value) {
if (value == null) {
return "";
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/contract/group/GroupDetailOutput.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/contract/group/GroupDetailOutput.java
index 9a3a84e60..c54f291da 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/contract/group/GroupDetailOutput.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/contract/group/GroupDetailOutput.java
@@ -40,6 +40,7 @@ public record GroupDetailOutput(
String group,
SubscriptionMode subscriptionMode,
ConsumeType consumeType,
+ /** Confirmed online clients, or -1 when the connection inventory is
unavailable. */
int onlineInstances,
long totalLag,
List<String> subscribedTopics,
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/contract/group/GroupListItem.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/contract/group/GroupListItem.java
index 09d4211ac..c9ce4cce8 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/contract/group/GroupListItem.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/contract/group/GroupListItem.java
@@ -30,6 +30,7 @@ public record GroupListItem(
SubscriptionMode subscriptionMode,
ConsumeType consumeType,
int retryMaxTimes,
+ /** Confirmed online clients, or -1 when the connection inventory is
unavailable. */
int onlineInstances,
long totalLag,
List<String> subscribedTopics) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/handler/group/GroupDetailToolHandler.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/handler/group/GroupDetailToolHandler.java
index 0e8efbd8d..4a5ac0542 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/handler/group/GroupDetailToolHandler.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/handler/group/GroupDetailToolHandler.java
@@ -76,10 +76,13 @@ public class GroupDetailToolHandler implements
ToolHandler<GroupDetailInput, Gro
if (!group.isConsumeStatsAvailable()) {
status = "UNKNOWN";
reasons.add("Broker consume statistics are unavailable.");
- } else if (group.getOnlineInstances() <= 0 && group.getTotalLag() > 0)
{
+ } else if (group.getOnlineInstances() < 0) {
+ status = "UNKNOWN";
+ reasons.add("Consumer connection information is unavailable.");
+ } else if (group.getOnlineInstances() == 0 && group.getTotalLag() > 0)
{
status = "UNHEALTHY";
reasons.add("The group has accumulated messages but no online
consumer.");
- } else if (group.getOnlineInstances() <= 0) {
+ } else if (group.getOnlineInstances() == 0) {
status = "WARNING";
reasons.add("The group has no online consumer.");
} else if (group.getTotalLag() > 0) {
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 15d5e2ac0..ba0da98fc 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
@@ -31,6 +31,7 @@ import
org.apache.rocketmq.remoting.protocol.header.GetConsumerConnectionListReq
import
org.apache.rocketmq.remoting.protocol.header.GetConsumerRunningInfoRequestHeader;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
+import org.apache.rocketmq.studio.common.util.MqResponseCodes;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
@@ -80,30 +81,55 @@ public class ProxyConsumerResolver {
* reachable.
*/
public ConsumerConnection resolveConsumerConnection(String instanceId,
String group) {
- for (String addr : discoverProxyAddresses(instanceId)) {
+ return resolveConsumerConnectionStatus(instanceId, group).connection();
+ }
+
+ public ConsumerConnectionResolution resolveConsumerConnectionStatus(String
instanceId, String group) {
+ ProxyAddressResolution discovery =
discoverProxyAddressesStatus(instanceId);
+ if (!discovery.available()) {
+ return ConsumerConnectionResolution.unavailable();
+ }
+ if (discovery.addresses().isEmpty()) {
+ return ConsumerConnectionResolution.available(null);
+ }
+ boolean queryUnavailable = false;
+ for (String addr : discovery.addresses()) {
try {
- ConsumerConnection connection = queryProxy(addr, group);
- if (connection != null) {
- return connection;
+ ProxyQueryResolution result = queryProxyStatus(addr, group);
+ if (!result.available()) {
+ queryUnavailable = true;
+ continue;
+ }
+ if (result.connection() != null) {
+ return
ConsumerConnectionResolution.available(result.connection());
}
} catch (Exception e) {
- log.debug("Proxy consumer connection query failed for {} via
{}: {}",
- group, addr, e.getMessage());
+ queryUnavailable = true;
+ log.debug("Proxy consumer connection query failed for {} via
{}: {}", group, addr, e.getMessage());
}
}
- return null;
+ return queryUnavailable ? ConsumerConnectionResolution.unavailable() :
ConsumerConnectionResolution.available(null);
}
ConsumerConnection queryProxy(String proxyAddr, String group) throws
Exception {
+ return queryProxyStatus(proxyAddr, group).connection();
+ }
+
+ private ProxyQueryResolution queryProxyStatus(String proxyAddr, String
group) throws Exception {
GetConsumerConnectionListRequestHeader header = new
GetConsumerConnectionListRequestHeader();
header.setConsumerGroup(group);
- RemotingCommand request =
-
RemotingCommand.createRequestCommand(RequestCode.GET_CONSUMER_CONNECTION_LIST,
header);
+ RemotingCommand request =
RemotingCommand.createRequestCommand(RequestCode.GET_CONSUMER_CONNECTION_LIST,
header);
RemotingCommand response = remotingClient().invokeSync(proxyAddr,
request, PROXY_QUERY_TIMEOUT_MILLIS);
- if (response == null || response.getCode() != ResponseCode.SUCCESS ||
response.getBody() == null) {
- return null;
+ if (response == null) {
+ return ProxyQueryResolution.unavailable();
}
- return ConsumerConnection.decode(response.getBody(),
ConsumerConnection.class);
+ if (response.getCode() == ResponseCode.CONSUMER_NOT_ONLINE) {
+ return ProxyQueryResolution.available(null);
+ }
+ if (response.getCode() != ResponseCode.SUCCESS || response.getBody()
== null) {
+ return ProxyQueryResolution.unavailable();
+ }
+ return
ProxyQueryResolution.available(ConsumerConnection.decode(response.getBody(),
ConsumerConnection.class));
}
/**
@@ -148,16 +174,19 @@ public class ProxyConsumerResolver {
}
List<String> discoverProxyAddresses(String instanceId) {
+ return discoverProxyAddressesStatus(instanceId).addresses();
+ }
+
+ private ProxyAddressResolution discoverProxyAddressesStatus(String
instanceId) {
String cacheKey = StringUtils.hasText(instanceId) ? instanceId :
DEFAULT_INSTANCE_KEY;
CachedProxyAddresses cached = proxyAddressCache.get(cacheKey);
if (cached != null && cached.expiresAtMillis() >
System.currentTimeMillis()) {
- return cached.addresses();
+ return ProxyAddressResolution.available(cached.addresses());
}
Set<String> ips = new LinkedHashSet<>();
try {
executeAdmin(instanceId, admin -> {
- ConsumerConnection connection =
-
admin.examineConsumerConnectionInfo(HEARTBEAT_SYNCER_CONSUMER_GROUP);
+ ConsumerConnection connection =
admin.examineConsumerConnectionInfo(HEARTBEAT_SYNCER_CONSUMER_GROUP);
if (connection != null && connection.getConnectionSet() !=
null) {
for (Connection conn : connection.getConnectionSet()) {
String clientAddr = conn.getClientAddr();
@@ -171,17 +200,17 @@ public class ProxyConsumerResolver {
return null;
});
} catch (Exception e) {
- log.debug("Proxy discovery via heartbeat syncer failed for
instance {}: {}",
- instanceId, e.getMessage());
- // A failed lookup is transient and must not suppress discovery
for the full cache TTL.
- return List.of();
+ 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.
+ // Do not cache the empty result: a Proxy may register
immediately afterwards.
+ return ProxyAddressResolution.available(List.of());
+ }
+ 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 addresses;
+ 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);
}
private <T> T executeAdmin(String instanceId,
MqAdminExtFactory.AdminAction<T> action) {
@@ -219,6 +248,21 @@ public class ProxyConsumerResolver {
clientStarted.set(true);
}
+ public record ConsumerConnectionResolution(ConsumerConnection connection,
boolean available) {
+ static ConsumerConnectionResolution available(ConsumerConnection
connection) { return new ConsumerConnectionResolution(connection, true); }
+ static ConsumerConnectionResolution unavailable() { return new
ConsumerConnectionResolution(null, false); }
+ }
+
+ private record ProxyAddressResolution(List<String> addresses, boolean
available) {
+ static ProxyAddressResolution available(List<String> addresses) {
return new ProxyAddressResolution(addresses, true); }
+ static ProxyAddressResolution unavailable() { return new
ProxyAddressResolution(List.of(), false); }
+ }
+
+ private record ProxyQueryResolution(ConsumerConnection connection, boolean
available) {
+ static ProxyQueryResolution available(ConsumerConnection connection) {
return new ProxyQueryResolution(connection, true); }
+ static ProxyQueryResolution unavailable() { return new
ProxyQueryResolution(null, false); }
+ }
+
private record CachedProxyAddresses(List<String> addresses, long
expiresAtMillis) {
}
}
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 1a0cfd522..a95ede465 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
@@ -217,7 +217,13 @@ public class RocketMQAdminClientImpl implements
AdminClient {
if (proxyConsumerResolver == null) {
return;
}
- ConsumerConnection viaProxy =
proxyConsumerResolver.resolveConsumerConnection(instanceId, group);
+ ProxyConsumerResolver.ConsumerConnectionResolution resolution =
+
proxyConsumerResolver.resolveConsumerConnectionStatus(instanceId, group);
+ if (!resolution.available()) {
+ vo.setOnlineInstances(-1);
+ return;
+ }
+ ConsumerConnection viaProxy = resolution.connection();
if (viaProxy == null) {
return;
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
index 97fbbd5b1..1240d28a6 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
@@ -328,10 +328,11 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
private void enrichGroupLiveStats(String instanceId, ConsumerGroupVO vo) {
// The detail modal reuses the listed group as-is, so the online
instance list has to be
// filled here too; both fields come from the same connection set to
stay consistent.
- List<ConsumerInstanceVO> instances = ConsumerConnections.toInstances(
- resolveConsumerConnection(instanceId, vo.getName()));
+ ProxyConsumerResolver.ConsumerConnectionResolution connection =
+ resolveConsumerConnection(instanceId, vo.getName());
+ List<ConsumerInstanceVO> instances =
ConsumerConnections.toInstances(connection.connection());
vo.setInstances(instances);
- vo.setOnlineInstances(instances.size());
+ vo.setOnlineInstances(connection.available() ? instances.size() : -1);
try {
ConsumeStats stats;
if (StringUtils.hasText(instanceId)) {
@@ -375,18 +376,21 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
}
}
- private ConsumerConnection resolveConsumerConnection(String instanceId,
String group) {
+ private ProxyConsumerResolver.ConsumerConnectionResolution
resolveConsumerConnection(
+ String instanceId, String group) {
try {
- if (StringUtils.hasText(instanceId)) {
- return runtimeAdminClientResolver.execute(instanceId,
- admin -> admin.examineConsumerConnectionInfo(group));
- }
- return adminExecute(admin ->
admin.examineConsumerConnectionInfo(group));
+ ConsumerConnection connection = StringUtils.hasText(instanceId)
+ ? runtimeAdminClientResolver.execute(instanceId, admin ->
admin.examineConsumerConnectionInfo(group))
+ : adminExecute(admin ->
admin.examineConsumerConnectionInfo(group));
+ return
ProxyConsumerResolver.ConsumerConnectionResolution.available(connection);
} catch (Exception e) {
- if (isGroupNotOnline(e) && proxyConsumerResolver != null) {
- return
proxyConsumerResolver.resolveConsumerConnection(instanceId, group);
+ if (isGroupNotOnline(e)) {
+ return proxyConsumerResolver == null
+ ?
ProxyConsumerResolver.ConsumerConnectionResolution.available(null)
+ :
proxyConsumerResolver.resolveConsumerConnectionStatus(instanceId, group);
}
- return null;
+ log.debug("Consumer connection lookup unavailable for group {}:
{}", group, e.getMessage());
+ return
ProxyConsumerResolver.ConsumerConnectionResolution.unavailable();
}
}
diff --git a/server/src/main/resources/tool-catalog/tools/group.yaml
b/server/src/main/resources/tool-catalog/tools/group.yaml
index 3cfb96b50..adbc57bc8 100644
--- a/server/src/main/resources/tool-catalog/tools/group.yaml
+++ b/server/src/main/resources/tool-catalog/tools/group.yaml
@@ -56,6 +56,8 @@ tools:
- BROADCASTING
onlineInstances:
type: integer
+ minimum: -1
+ description: Confirmed online clients; -1 when the connection
inventory is unavailable.
totalLag:
type: integer
subscribedTopics:
@@ -122,7 +124,8 @@ tools:
- BROADCASTING
onlineInstances:
type: integer
- minimum: 0
+ minimum: -1
+ description: Confirmed online clients; -1 when the connection
inventory is unavailable.
totalLag:
type: integer
minimum: -1
@@ -232,6 +235,8 @@ tools:
- BROADCASTING
onlineInstances:
type: integer
+ minimum: -1
+ description: Confirmed online clients; -1 when the connection
inventory is unavailable.
totalLag:
type: integer
subscribedTopics:
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
index b13b809bc..e6148c9bc 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
@@ -913,6 +913,22 @@ class MetadataServiceTest {
verify(apacheProvider).listConsumerGroups("instance-a", "orders");
}
+ @Test
+ void
exportConsumerGroupsShouldRenderUnavailableOnlineInstancesAsUnknownTest() {
+ ConsumerGroupVO unavailable = consumerGroup("orders-unavailable",
"orders", 10, SubscriptionMode.Pop);
+ unavailable.setOnlineInstances(-1);
+ ConsumerGroupVO confirmedOffline = consumerGroup("orders-offline",
"orders", 10, SubscriptionMode.Pop);
+ confirmedOffline.setOnlineInstances(0);
+ when(apacheProvider.listConsumerGroups("instance-a", null))
+ .thenReturn(List.of(unavailable, confirmedOffline));
+
+ String csv = metadataService.exportConsumerGroups("instance-a", null,
null, List.of());
+
+ // The -1 sentinel must not reach the export, while a confirmed zero
still exports as 0.
+
assertThat(csv).contains("\"orders-unavailable\",\"orders\",\"cluster-a\",\"Pop\",\"CLUSTERING\",\"unknown\"");
+
assertThat(csv).contains("\"orders-offline\",\"orders\",\"cluster-a\",\"Pop\",\"CLUSTERING\",\"0\"");
+ }
+
@Test
void exportConsumerGroupsShouldEscapeFormulaCells() {
ConsumerGroupVO group = consumerGroup("orders-cg", "=formula", 10,
SubscriptionMode.Push);
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/handler/group/ConsumerGroupReadToolHandlersTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/handler/group/ConsumerGroupReadToolHandlersTest.java
index 32e3773b1..000b1be5a 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/handler/group/ConsumerGroupReadToolHandlersTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/handler/group/ConsumerGroupReadToolHandlersTest.java
@@ -155,6 +155,23 @@ class ConsumerGroupReadToolHandlersTest {
assertThat(output.clients().totalClients()).isEqualTo(1);
}
+ @Test
+ void detailMarksHealthUnknownWhenConsumerConnectionsAreUnavailableTest() {
+ group.setOnlineInstances(-1);
+ group.setTotalLag(12L);
+ when(metadataService.consumerGroupRuntimeView("instance-a",
"group-a")).thenReturn(group);
+ when(metadataService.consumerGroupConfigurations("instance-a",
"group-a")).thenReturn(List.of(group));
+ when(metadataService.getGroupSubscriptions("instance-a",
"group-a")).thenReturn(List.of());
+ when(metadataService.getGroupProgress("instance-a",
"group-a")).thenReturn(List.of());
+
+ GroupDetailOutput output = new GroupDetailToolHandler(metadataService)
+ .execute(new GroupDetailInput("instance-a", "group-a", null),
context());
+
+ assertThat(output.health().status()).isEqualTo("UNKNOWN");
+ assertThat(output.health().reasons()).contains("Consumer connection
information is unavailable.");
+ assertThat(output.onlineInstances()).isEqualTo(-1);
+ }
+
@Test
void detailWithoutOnlineConsumersKeepsProgressAndClientsEmpty() {
ConsumerGroupVO offline = new ConsumerGroupVO();
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/service/ToolOutputSchemaContractTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/service/ToolOutputSchemaContractTest.java
index 6cfee313d..17b7a9610 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/service/ToolOutputSchemaContractTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/service/ToolOutputSchemaContractTest.java
@@ -239,7 +239,15 @@ class ToolOutputSchemaContractTest {
GroupListItem groupItem = new GroupListItem(
"cg-orders", INSTANCE, SubscriptionMode.Push,
ConsumeType.CLUSTERING,
16, 2, 100L, List.of("orders"));
- samples.put("rmq.group.list", List.of(new
ListOutput<>(List.of(groupItem))));
+ // onlineInstances carries a -1 sentinel when the connection inventory
is unavailable.
+ // ToolValidationFilter validates every tool result, so a schema that
rejects -1 would
+ // fail the whole call at runtime instead of reporting the unknown
state.
+ GroupListItem unknownConnectionsItem = new GroupListItem(
+ "cg-orders", INSTANCE, SubscriptionMode.Push,
ConsumeType.CLUSTERING,
+ 16, -1, 100L, List.of("orders"));
+ samples.put("rmq.group.list", List.of(
+ new ListOutput<>(List.of(groupItem)),
+ new ListOutput<>(List.of(unknownConnectionsItem))));
samples.put("rmq.group.detail", List.of(new GroupDetailOutput(
INSTANCE, "cg-orders", SubscriptionMode.Push,
ConsumeType.CLUSTERING,
2, 100L, List.of("orders"), "TAG", "Concurrently", 16, 0,
@@ -256,7 +264,18 @@ class ToolOutputSchemaContractTest {
new GroupDetailOutput.Client(
"client-1", "gRPC", "127.0.0.1:50000", "JAVA",
"5.0.7",
true, List.of("orders"), "2026-08-22T09:30:00",
- Map.of("orders", 10L)))))));
+ Map.of("orders", 10L))))),
+ new GroupDetailOutput(
+ INSTANCE, "cg-orders", SubscriptionMode.Push,
ConsumeType.CLUSTERING,
+ -1, 100L, List.of("orders"), "TAG", "Concurrently",
16, 0,
+ List.of(new GroupDetailOutput.Subscription(
+ "orders", "*", "TAG", "STANDARD",
"CONSISTENT")),
+ List.of(),
+ new GroupDetailOutput.Health(
+ "UNKNOWN", List.of("Consumer connection
information is unavailable.")),
+ List.of(unknownConnectionsItem),
+ null,
+ null)));
samples.put("rmq.group.update", List.of(planned(),
executed(groupItem)));
samples.put("rmq.group.delete", List.of(planned(), executedVoid()));
samples.put("rmq.group.reset_offset", List.of(
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ProxyConsumerResolverTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ProxyConsumerResolverTest.java
index 137269cf9..c28a40041 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ProxyConsumerResolverTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ProxyConsumerResolverTest.java
@@ -17,6 +17,9 @@
package org.apache.rocketmq.studio.provider.apache;
+import org.apache.rocketmq.remoting.netty.NettyRemotingClient;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.body.Connection;
import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
@@ -33,6 +36,9 @@ import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.lenient;
import static org.mockito.Mockito.when;
@@ -113,25 +119,52 @@ class ProxyConsumerResolverTest {
}
@Test
- void resolveConsumerConnectionShouldReturnNullWhenNoProxyDiscoveredTest()
throws Exception {
+ void
resolveConsumerConnectionStatusShouldMarkDiscoveryFailureUnavailableTest()
throws Exception {
when(adminExt.examineConsumerConnectionInfo("CID_DefaultHeartBeatSyncerTopic"))
- .thenThrow(new IllegalStateException("syncer group missing"));
+ .thenThrow(new IllegalStateException("nameserver
unavailable"));
- assertThat(resolver.resolveConsumerConnection("instance-a",
"cg-orders")).isNull();
+ ProxyConsumerResolver.ConsumerConnectionResolution result =
+ resolver.resolveConsumerConnectionStatus("instance-a",
"cg-orders");
+
+ assertThat(result.available()).isFalse();
+ assertThat(result.connection()).isNull();
}
@Test
- void resolveConsumerConnectionShouldReturnNullWhenProxyQueryFailsTest()
throws Exception {
+ void
resolveConsumerConnectionStatusShouldTreatNoProxiesAsKnownOfflineTest() throws
Exception {
ConsumerConnection syncer = new ConsumerConnection();
- HashSet<Connection> connections = new HashSet<>();
- Connection proxyA = new Connection();
- proxyA.setClientId("proxy-a");
- proxyA.setClientAddr("192.0.2.1:10911");
- connections.add(proxyA);
- syncer.setConnectionSet(connections);
+ syncer.setConnectionSet(new HashSet<>());
when(adminExt.examineConsumerConnectionInfo("CID_DefaultHeartBeatSyncerTopic")).thenReturn(syncer);
- // 192.0.2.1 (TEST-NET) is unreachable, so the remoting query must
degrade to null
- assertThat(resolver.resolveConsumerConnection("instance-a",
"cg-orders")).isNull();
+ ProxyConsumerResolver.ConsumerConnectionResolution result =
+ resolver.resolveConsumerConnectionStatus("instance-a",
"cg-orders");
+
+ assertThat(result.available()).isTrue();
+ assertThat(result.connection()).isNull();
}
+
+ @Test
+ void
resolveConsumerConnectionStatusShouldDistinguishOfflineFromProxyFailureTest()
throws Exception {
+ ConsumerConnection syncer = new ConsumerConnection();
+ Connection proxy = new Connection();
+ proxy.setClientAddr("192.0.2.1:10911");
+ syncer.setConnectionSet(new HashSet<>(List.of(proxy)));
+
when(adminExt.examineConsumerConnectionInfo("CID_DefaultHeartBeatSyncerTopic")).thenReturn(syncer);
+ NettyRemotingClient client = mock(NettyRemotingClient.class);
+ resolver.setRemotingClientForTest(client);
+ when(client.invokeSync(anyString(), any(RemotingCommand.class),
anyLong()))
+
.thenReturn(RemotingCommand.createResponseCommand(ResponseCode.CONSUMER_NOT_ONLINE,
"not online"))
+ .thenReturn(null);
+
+ ProxyConsumerResolver.ConsumerConnectionResolution offline =
+ resolver.resolveConsumerConnectionStatus("instance-a",
"cg-orders");
+ ProxyConsumerResolver.ConsumerConnectionResolution unavailable =
+ resolver.resolveConsumerConnectionStatus("instance-a",
"cg-orders");
+
+ assertThat(offline.available()).isTrue();
+ assertThat(offline.connection()).isNull();
+ assertThat(unavailable.available()).isFalse();
+ assertThat(unavailable.connection()).isNull();
+ }
+
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
index d644b47c6..49114b2b0 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
@@ -225,7 +225,8 @@ class RocketMQAdminClientImplTest {
subscription.setTopic("studio-normal");
table.put("studio-normal", subscription);
viaProxy.setSubscriptionTable(table);
- when(resolver.resolveConsumerConnection("instance-a",
"orders")).thenReturn(viaProxy);
+ when(resolver.resolveConsumerConnectionStatus("instance-a", "orders"))
+
.thenReturn(ProxyConsumerResolver.ConsumerConnectionResolution.available(viaProxy));
org.springframework.test.util.ReflectionTestUtils.setField(adminClient,
"proxyConsumerResolver", resolver);
ConsumerGroupVO group = adminClient.getConsumerGroup("instance-a",
"orders");
@@ -234,6 +235,25 @@ class RocketMQAdminClientImplTest {
assertThat(group.getSubscribedTopics()).containsExactly("studio-normal");
}
+ @Test
+ void
getConsumerGroupShouldMarkConnectionsUnknownWhenProxyFallbackIsUnavailableTest()
throws Exception {
+ when(adminExt.examineConsumerConnectionInfo("orders"))
+ .thenThrow(new MQClientException(
+ "CODE: 206 DESC: the consumer group[orders] not
online BROKER: 10.0.4.69:10911",
+ (Throwable) null));
+
when(runtimeAdminClientResolver.execute(org.mockito.ArgumentMatchers.eq("instance-a"),
any()))
+ .thenAnswer(invocation ->
invocation.<MqAdminExtFactory.AdminAction<Object>>getArgument(1).apply(adminExt));
+ ProxyConsumerResolver resolver =
org.mockito.Mockito.mock(ProxyConsumerResolver.class);
+ when(resolver.resolveConsumerConnectionStatus("instance-a", "orders"))
+
.thenReturn(ProxyConsumerResolver.ConsumerConnectionResolution.unavailable());
+
org.springframework.test.util.ReflectionTestUtils.setField(adminClient,
"proxyConsumerResolver", resolver);
+
+ ConsumerGroupVO group = adminClient.getConsumerGroup("instance-a",
"orders");
+
+ assertThat(group.getOnlineInstances()).isEqualTo(-1);
+ assertThat(group.getInstances()).isEmpty();
+ }
+
@Test
void getConsumerGroupComputesLagAndDelayFromConsumeStatsTest() throws
Exception {
org.apache.rocketmq.remoting.protocol.body.ConsumerConnection
connection =
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
index d25b2c81d..8a4d47bf0 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
@@ -753,6 +753,38 @@ class RocketMQMetadataProviderTest {
.containsExactlyInAnyOrder("client-a", "client-b");
}
+ @Test
+ void
listConsumerGroupsShouldMarkOnlineInstancesUnknownWhenConnectionLookupFailsTest()
throws Exception {
+ RmqGroup entity = new RmqGroup();
+ entity.setName("cg-unknown-connections");
+ entity.setInstanceId("instance-a");
+ when(groupMapper.selectList(any())).thenReturn(List.of(entity));
+
+ DefaultMQAdminExt admin =
org.mockito.Mockito.mock(DefaultMQAdminExt.class);
+ when(admin.examineConsumerConnectionInfo("cg-unknown-connections"))
+ .thenThrow(new IllegalStateException("nameserver
unavailable"));
+ ConsumeStats stats = new ConsumeStats();
+ MessageQueue queue = new MessageQueue("orders", "broker-a", 0);
+ OffsetWrapper wrapper = new OffsetWrapper();
+ wrapper.setBrokerOffset(100L);
+ wrapper.setConsumerOffset(60L);
+ stats.getOffsetTable().put(queue, wrapper);
+
when(admin.examineConsumeStats("cg-unknown-connections")).thenReturn(stats);
+ when(runtimeAdminClientResolver.execute(eq("instance-a"), any()))
+ .thenAnswer(invocation ->
+
invocation.<MqAdminExtFactory.AdminAction<Object>>getArgument(1).apply(admin));
+
+ RocketMQMetadataProvider provider = newLiveProvider(admin);
+
+ List<ConsumerGroupVO> groups =
provider.listConsumerGroups("instance-a", null, null);
+
+ assertThat(groups).singleElement().satisfies(group -> {
+ assertThat(group.getOnlineInstances()).isEqualTo(-1);
+ assertThat(group.getTotalLag()).isEqualTo(40L);
+ assertThat(group.getInstances()).isEmpty();
+ });
+ }
+
@Test
void listConsumerGroupsShouldEnrichOnlineInstancesViaProxyFallbackTest()
throws Exception {
RmqGroup entity = new RmqGroup();
@@ -780,7 +812,8 @@ class RocketMQMetadataProviderTest {
conn.setClientAddr("10.0.3.104:50124");
connections.add(conn);
viaProxy.setConnectionSet(connections);
- when(resolver.resolveConsumerConnection("instance-a",
"cg-proxy")).thenReturn(viaProxy);
+ when(resolver.resolveConsumerConnectionStatus("instance-a",
"cg-proxy"))
+
.thenReturn(ProxyConsumerResolver.ConsumerConnectionResolution.available(viaProxy));
RocketMQMetadataProvider provider = newLiveProvider(admin);
org.springframework.test.util.ReflectionTestUtils.setField(
diff --git a/web/src/api/metadata.ts b/web/src/api/metadata.ts
index 60f93f20e..3dacf4c79 100644
--- a/web/src/api/metadata.ts
+++ b/web/src/api/metadata.ts
@@ -99,6 +99,7 @@ export interface ConsumerGroup {
instanceId?: string;
subscriptionMode: string;
consumeType: string;
+ /** Confirmed online clients; -1 means the connection inventory is
unavailable. */
onlineInstances: number;
totalLag: number;
subscribedTopics: string[];
diff --git a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
index 7f6722d11..eee564664 100644
--- a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
+++ b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
@@ -1517,6 +1517,39 @@ describe('Consumer page', () => {
});
});
+ it('sorts groups with unavailable connections after known client counts',
async () => {
+ const user = userEvent.setup();
+ vi.mocked(consumerService.listConsumerGroupPage).mockResolvedValue(
+ groupPage([
+ { ...group, name: 'unknown-conn-cg', onlineInstances: -1 },
+ { ...group, name: 'known-conn-cg', onlineInstances: 12 },
+ ]),
+ );
+ renderWithProviders(<ConsumerPage />);
+ await screen.findByRole('row', { name: /unknown-conn-cg/ });
+
+ // The sentinel must never reach the cell as a bare -1 ...
+ expect(screen.getByRole('row', { name: /unknown-conn-cg/
}).textContent).toContain('不可用');
+
+ // ... and must not sort as "fewer clients than zero".
+ const [clientsHeader] = screen.getAllByText('在线客户端');
+ await user.click(clientsHeader);
+ await waitFor(() => {
+ const rows = Array.from(document.querySelectorAll('tbody tr'));
+ const order = rows
+ .map((row) => row.textContent ?? '')
+ .map((text) =>
+ text.includes('unknown-conn-cg')
+ ? 'unknown'
+ : /\bknown-conn-cg\b/.test(text)
+ ? 'known'
+ : '',
+ )
+ .filter(Boolean);
+ expect(order).toEqual(['known', 'unknown']);
+ });
+ });
+
it('ignores settings responses from a previously closed group modal', async
() => {
const otherGroup = { ...group, name: 'other-cg' };
const firstSettings = deferred<{
diff --git a/web/src/pages/instance/consumer.tsx
b/web/src/pages/instance/consumer.tsx
index 07b59cca4..d09286875 100644
--- a/web/src/pages/instance/consumer.tsx
+++ b/web/src/pages/instance/consumer.tsx
@@ -104,6 +104,7 @@ import {
} from '../../utils/resourceCsvImport';
import { downloadCsv } from '../../utils/download';
import { formatLag, isLagAvailable, lagSortValue } from
'../../utils/consumerLag';
+import { formatOnlineInstances, onlineInstancesSortValue } from
'../../utils/consumerConnections';
import { tableScrollX } from '../../utils/table';
import {
analyzeConsumerGroupHealth,
@@ -946,7 +947,9 @@ const ConsumerPageContent = ({
key: 'onlineInstances',
width: 100,
align: 'center',
- sorter: (a, b) => (a.onlineInstances ?? 0) - (b.onlineInstances ?? 0),
+ sorter: (a, b) =>
+ onlineInstancesSortValue(a.onlineInstances) -
onlineInstancesSortValue(b.onlineInstances),
+ render: (value: number) => formatOnlineInstances(value,
UNAVAILABLE_LAG_LABEL),
},
{
title: '总堆积量',
@@ -1657,6 +1660,9 @@ const ConsumerPageContent = ({
<Statistic
title="在线实例"
value={selectedGroup.onlineInstances}
+ formatter={(value) =>
+ formatOnlineInstances(Number(value),
UNAVAILABLE_LAG_LABEL)
+ }
prefix={<Users size={18} color="#52c41a" />}
valueStyle={{ color: '#52c41a' }}
/>
@@ -1966,11 +1972,16 @@ const ConsumerPageContent = ({
<Statistic
title="客户端"
value={selectedGroupHealth.summary.onlineInstances}
+ formatter={(value) =>
+ formatOnlineInstances(Number(value),
UNAVAILABLE_LAG_LABEL)
+ }
/>
<Text type="secondary">
- {selectedGroupHealth.summary.staleClientCount > 0
- ?
`${selectedGroupHealth.summary.staleClientCount} 个心跳过期`
- : '心跳状态正常'}
+ {selectedGroupHealth.summary.onlineInstances < 0
+ ? '客户端连接信息不可用'
+ : selectedGroupHealth.summary.staleClientCount > 0
+ ?
`${selectedGroupHealth.summary.staleClientCount} 个心跳过期`
+ : '心跳状态正常'}
</Text>
</Card>
</Col>
diff --git a/web/src/services/consumerService.test.ts
b/web/src/services/consumerService.test.ts
index 7e5d59597..6052735da 100644
--- a/web/src/services/consumerService.test.ts
+++ b/web/src/services/consumerService.test.ts
@@ -16,9 +16,11 @@
*/
import { describe, expect, it, vi } from 'vitest';
+import { mockConsumerGroups } from '../mock/consumers';
import {
createConsumerGroup,
deleteConsumerGroup,
+ exportConsumerGroups,
getConsumerGroup,
getConsumerProgress,
getConsumerStack,
@@ -279,9 +281,9 @@ describe('consumer service mock data', () => {
await createConsumerGroup({ name, instanceId: 'instance-b', namespace:
'namespace-b' });
try {
- await expect(
- createConsumerGroup({ name, instanceId: 'instance-a' }),
- ).rejects.toThrow(`Consumer group already exists: ${name}`);
+ await expect(createConsumerGroup({ name, instanceId: 'instance-a'
})).rejects.toThrow(
+ `Consumer group already exists: ${name}`,
+ );
const instanceAGroups = await listConsumerGroups({ instanceId:
'instance-a', search: name });
const instanceBGroups = await listConsumerGroups({ instanceId:
'instance-b', search: name });
@@ -345,3 +347,35 @@ describe('consumer service mock data', () => {
}
});
});
+
+describe('consumer group CSV export', () => {
+ it('renders an unavailable connection count as unknown instead of the
sentinel', async () => {
+ const unavailable = {
+ ...mockConsumerGroups[0],
+ name: 'cg-conn-unavailable',
+ namespace: 'ns',
+ clusterId: 'cluster-a',
+ subscriptionMode: 'Push' as const,
+ consumeType: 'CLUSTERING' as const,
+ onlineInstances: -1,
+ totalLag: 10,
+ };
+ const confirmedOffline = { ...unavailable, name: 'cg-conn-offline',
onlineInstances: 0 };
+ mockConsumerGroups.push(unavailable, confirmedOffline);
+ try {
+ const csv = await exportConsumerGroups({
+ names: ['cg-conn-unavailable', 'cg-conn-offline'],
+ });
+
+ expect(csv).toContain(
+
'"cg-conn-unavailable","ns","cluster-a","Push","CLUSTERING","unknown","10"',
+ );
+ // A confirmed zero must stay numeric, otherwise the column loses its
meaning entirely.
+
expect(csv).toContain('"cg-conn-offline","ns","cluster-a","Push","CLUSTERING","0","10"');
+ expect(csv).not.toContain(`"'-1"`);
+ } finally {
+ mockConsumerGroups.pop();
+ mockConsumerGroups.pop();
+ }
+ });
+});
diff --git a/web/src/services/consumerService.ts
b/web/src/services/consumerService.ts
index 54ba4a3e0..1163e4211 100644
--- a/web/src/services/consumerService.ts
+++ b/web/src/services/consumerService.ts
@@ -17,6 +17,7 @@ import type {
SubscriptionEntry,
} from '../api/metadata';
import { mockConsumerGroups, mockQueueProgress, mockSubscriptions } from
'../mock/consumers';
+import { formatOnlineInstances } from '../utils/consumerConnections';
import { buildCsv, type CsvColumn } from '../utils/download';
const consumerGroupsState = mockConsumerGroups as unknown as ConsumerGroup[];
@@ -28,7 +29,10 @@ const GROUP_EXPORT_COLUMNS: CsvColumn<ConsumerGroup>[] = [
{ header: 'Cluster ID', value: (group) => group.clusterId },
{ header: 'Subscription Mode', value: (group) => group.subscriptionMode },
{ header: 'Consume Type', value: (group) => group.consumeType },
- { header: 'Online Instances', value: (group) => group.onlineInstances },
+ {
+ header: 'Online Instances',
+ value: (group) => formatOnlineInstances(group.onlineInstances, 'unknown'),
+ },
{ header: 'Total Lag', value: (group) => group.totalLag },
{ header: 'Delay Seconds', value: (group) => group.delaySeconds },
{ header: 'Subscription Data Type', value: (group) =>
group.subscriptionDataType },
diff --git a/web/src/utils/consumerConnections.test.ts
b/web/src/utils/consumerConnections.test.ts
new file mode 100644
index 000000000..f0c273acf
--- /dev/null
+++ b/web/src/utils/consumerConnections.test.ts
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { describe, expect, it } from 'vitest';
+import {
+ UNKNOWN_ONLINE_INSTANCES,
+ formatOnlineInstances,
+ isOnlineInstancesAvailable,
+ onlineInstancesSortValue,
+} from './consumerConnections';
+
+describe('consumer connection helpers', () => {
+ it('treats the -1 sentinel and missing values as unavailable', () => {
+ expect(UNKNOWN_ONLINE_INSTANCES).toBe(-1);
+ expect(isOnlineInstancesAvailable(UNKNOWN_ONLINE_INSTANCES)).toBe(false);
+ expect(isOnlineInstancesAvailable(-5)).toBe(false);
+ expect(isOnlineInstancesAvailable(0)).toBe(true);
+ expect(isOnlineInstancesAvailable(8)).toBe(true);
+ expect(isOnlineInstancesAvailable(undefined)).toBe(false);
+ expect(isOnlineInstancesAvailable(null)).toBe(false);
+ expect(isOnlineInstancesAvailable(Number.NaN)).toBe(false);
+ });
+
+ it('formats known counts numerically and unavailable counts with the label',
() => {
+ expect(formatOnlineInstances(0, '不可用')).toBe('0');
+ expect(formatOnlineInstances(1000,
'不可用')).toBe(formatOnlineInstances(1000));
+ expect(formatOnlineInstances(UNKNOWN_ONLINE_INSTANCES, '不可用')).toBe('不可用');
+ expect(formatOnlineInstances(undefined, 'unknown')).toBe('unknown');
+ });
+
+ it('sorts unavailable connection counts after every known count', () => {
+ expect(onlineInstancesSortValue(0)).toBe(0);
+ expect(onlineInstancesSortValue(64)).toBe(64);
+
expect(onlineInstancesSortValue(UNKNOWN_ONLINE_INSTANCES)).toBe(Number.MAX_SAFE_INTEGER);
+ expect(onlineInstancesSortValue(null)).toBe(Number.MAX_SAFE_INTEGER);
+ });
+});
diff --git a/web/src/utils/consumerConnections.ts
b/web/src/utils/consumerConnections.ts
new file mode 100644
index 000000000..220dfadc5
--- /dev/null
+++ b/web/src/utils/consumerConnections.ts
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Sentinel the backend reports when the consumer connection inventory could
not be
+ * established (instance discovery or proxy lookup failure), as opposed to a
confirmed
+ * zero online clients.
+ */
+export const UNKNOWN_ONLINE_INSTANCES = -1;
+
+export const isOnlineInstancesAvailable = (value: number | null | undefined):
value is number =>
+ typeof value === 'number' && Number.isFinite(value) && value >= 0;
+
+export const formatOnlineInstances = (
+ value: number | null | undefined,
+ unavailableLabel: string = String(UNKNOWN_ONLINE_INSTANCES),
+): string => (isOnlineInstancesAvailable(value) ? value.toLocaleString() :
unavailableLabel);
+
+/** Sort key that pushes unavailable connection counts to the end of an
ascending list. */
+export const onlineInstancesSortValue = (value: number | null | undefined):
number =>
+ isOnlineInstancesAvailable(value) ? value : Number.MAX_SAFE_INTEGER;
diff --git a/web/src/utils/consumerGroupDiagnostics.test.ts
b/web/src/utils/consumerGroupDiagnostics.test.ts
index b6a09bfcc..ad491615d 100644
--- a/web/src/utils/consumerGroupDiagnostics.test.ts
+++ b/web/src/utils/consumerGroupDiagnostics.test.ts
@@ -89,6 +89,18 @@ describe('consumer group diagnostics', () => {
expect(diagnostics.issues).toEqual([]);
});
+ it('does not treat unavailable connection data as zero online clients', ()
=> {
+ const diagnostics = analyzeConsumerGroupHealth(
+ group({ onlineInstances: -1, totalLag: 2400, instances: [] }),
+ [subscription()],
+ [queue({ diffTotal: 2400 })],
+ { now: '2026-08-31T12:01:00Z' },
+ );
+ expect(diagnostics.issues.map((item) =>
item.code)).toContain('CONNECTION_STATUS_UNKNOWN');
+ expect(diagnostics.issues.map((item) =>
item.code)).not.toContain('NO_ACTIVE_CLIENTS_WITH_LAG');
+ expect(diagnostics.summary.onlineInstances).toBe(-1);
+ });
+
it('flags critical subscription, queue and runtime risks', () => {
const diagnostics = analyzeConsumerGroupHealth(
group({
diff --git a/web/src/utils/consumerGroupDiagnostics.ts
b/web/src/utils/consumerGroupDiagnostics.ts
index 4e1d8c52a..bb0848ecf 100644
--- a/web/src/utils/consumerGroupDiagnostics.ts
+++ b/web/src/utils/consumerGroupDiagnostics.ts
@@ -21,6 +21,7 @@ import { isLagAvailable } from './consumerLag';
export type ConsumerGroupHealthStatus = 'healthy' | 'warning' | 'critical';
export type ConsumerGroupHealthIssueCode =
+ | 'CONNECTION_STATUS_UNKNOWN'
| 'NO_ACTIVE_CLIENTS_WITH_LAG'
| 'NO_SUBSCRIPTION_DATA'
| 'SUBSCRIPTION_INCONSISTENT'
@@ -298,7 +299,16 @@ const runtimeIssues = (
>,
): ConsumerGroupHealthIssue[] => {
const issues: ConsumerGroupHealthIssue[] = [];
- if ((group.onlineInstances ?? 0) === 0 && (lag ?? 0) > 0) {
+ if (group.onlineInstances < 0) {
+ issues.push(
+ issue(
+ 'CONNECTION_STATUS_UNKNOWN',
+ 'warning',
+ '客户端连接状态不可用',
+ '无法确认当前在线客户端数量，请先检查 Broker/Proxy 连接信息查询。',
+ ),
+ );
+ } else if ((group.onlineInstances ?? 0) === 0 && (lag ?? 0) > 0) {
issues.push(
issue(
'NO_ACTIVE_CLIENTS_WITH_LAG',
@@ -350,7 +360,11 @@ const runtimeIssues = (
const recommendations = (issues: ConsumerGroupHealthIssue[]): string[] => {
const codes = new Set(issues.map((item) => item.code));
const result: string[] = [];
- if (codes.has('NO_ACTIVE_CLIENTS_WITH_LAG') || codes.has('STALE_HEARTBEAT'))
{
+ if (
+ codes.has('CONNECTION_STATUS_UNKNOWN') ||
+ codes.has('NO_ACTIVE_CLIENTS_WITH_LAG') ||
+ codes.has('STALE_HEARTBEAT')
+ ) {
result.push('先确认消费者进程、Proxy/Broker 网络连通性和客户端心跳是否恢复。');
}
if (codes.has('SUBSCRIPTION_INCONSISTENT') ||
codes.has('SUBSCRIPTION_UNKNOWN')) {
