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 c2d2ed6e2 feat(consumer): make proxy-connected consumer groups
observable (#2504)
c2d2ed6e2 is described below
commit c2d2ed6e20d5950a8467da1081accfd7abd3f3ec
Author: lizhimins <[email protected]>
AuthorDate: Fri Aug 21 16:10:29 2026 +0800
feat(consumer): make proxy-connected consumer groups observable (#2504)
Clients that connect through a proxy keep their channel on the proxy and
never register on a broker, so the broker answers CODE 206 "not online" for
them and the group pages showed no clients, no lag and no delay at all.
Resolve the connection set, consume stats and running info through the proxy
whenever the broker reports the group as offline, and treat that state as an
empty business result rather than an RPC error. On top of that data the
group
list now fills online clients, total lag and consume delay, the detail
dialog
shows the online instance table in the overview tab, and queue progress
carries its topic so it can be filtered per subscription.
---
.../studio/instance/group/QueueProgressVO.java | 1 +
.../studio/provider/alibaba/AliyunConverters.java | 1 +
.../provider/apache/ConsumerConnections.java | 44 ++++
.../provider/apache/ProxyConsumerResolver.java | 222 +++++++++++++++++
.../provider/apache/RocketMQAdminClientImpl.java | 103 +++++++-
.../RocketMQConsumerDiagnosticsProvider.java | 46 +++-
.../provider/apache/RocketMQMetadataProvider.java | 174 ++++++++++++--
.../provider/tencent/TencentInstanceProvider.java | 1 +
.../provider/apache/ProxyConsumerResolverTest.java | 119 ++++++++++
.../apache/RocketMQAdminClientImplTest.java | 97 ++++++++
.../RocketMQConsumerDiagnosticsProviderTest.java | 47 +++-
.../apache/RocketMQMetadataProviderTest.java | 180 ++++++++++++++
web/src/api/metadata.ts | 1 +
.../pages/instance/__tests__/ConsumerPage.test.tsx | 49 +++-
web/src/pages/instance/consumer.tsx | 262 ++++++++++++++++-----
.../studio/__tests__/GroupManagement.test.tsx | 1 +
16 files changed, 1242 insertions(+), 106 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/QueueProgressVO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/QueueProgressVO.java
index 5d54b7d74..d7149b6df 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/QueueProgressVO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/QueueProgressVO.java
@@ -26,6 +26,7 @@ import lombok.NoArgsConstructor;
@NoArgsConstructor
@AllArgsConstructor
public class QueueProgressVO {
+ private String topic;
private String broker;
private int queueId;
private long brokerOffset;
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConverters.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConverters.java
index cfe1f2194..8fcfd06f2 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConverters.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConverters.java
@@ -182,6 +182,7 @@ final class AliyunConverters {
long ready = entry.getValue() == null ||
entry.getValue().getReadyCount() == null
? 0L : entry.getValue().getReadyCount();
rows.add(QueueProgressVO.builder()
+ .topic(entry.getKey())
.broker("topic:" + entry.getKey())
.queueId(0)
.brokerOffset(0L)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ConsumerConnections.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ConsumerConnections.java
new file mode 100644
index 000000000..f89296cde
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ConsumerConnections.java
@@ -0,0 +1,44 @@
+/*
+ * 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.
+ */
+package org.apache.rocketmq.studio.provider.apache;
+
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.studio.instance.group.ConsumerInstanceVO;
+
+import java.util.List;
+
+/**
+ * Maps a broker (or proxy) consumer connection set to the online instance
view, so the group
+ * listing and the group detail always derive the online instances from the
same source.
+ */
+final class ConsumerConnections {
+
+ private ConsumerConnections() {
+ }
+
+ static List<ConsumerInstanceVO> toInstances(ConsumerConnection connection)
{
+ if (connection == null || connection.getConnectionSet() == null) {
+ return List.of();
+ }
+ return connection.getConnectionSet().stream()
+ .map(conn -> ConsumerInstanceVO.builder()
+ .clientId(conn.getClientId())
+ .address(conn.getClientAddr())
+ .build())
+ .toList();
+ }
+}
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
new file mode 100644
index 000000000..192c55c75
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ProxyConsumerResolver.java
@@ -0,0 +1,222 @@
+/*
+ * 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.
+ */
+
+package org.apache.rocketmq.studio.provider.apache;
+
+import jakarta.annotation.PreDestroy;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.remoting.netty.NettyClientConfig;
+import org.apache.rocketmq.remoting.netty.NettyRemotingClient;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.RequestCode;
+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.remoting.protocol.body.ConsumerRunningInfo;
+import
org.apache.rocketmq.remoting.protocol.header.GetConsumerConnectionListRequestHeader;
+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 lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Service;
+import org.springframework.util.StringUtils;
+
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Resolves consumer connection info from the cluster proxies.
+ *
+ * <p>Clients connected through a proxy (remoting heartbeats stop at the
proxy, gRPC clients
+ * register via gRPC heartbeats) are invisible to broker-side consumer stats,
so the broker
+ * reports those groups as offline. Every proxy registers its remoting address
through the
+ * broadcast heartbeat-syncer group on the brokers, which lets us discover the
proxy nodes
+ * indirectly and query {@code GET_CONSUMER_CONNECTION_LIST} on the proxy's
remoting port,
+ * where the proxy answers from its own client manager.
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class ProxyConsumerResolver {
+
+ private static final String HEARTBEAT_SYNCER_CONSUMER_GROUP =
"CID_DefaultHeartBeatSyncerTopic";
+ private static final int PROXY_REMOTING_PORT = 8080;
+ private static final long PROXY_QUERY_TIMEOUT_MILLIS = 2_000L;
+ // Capturing a jstack on the client takes far longer than reading the
proxy's own client
+ // manager, so the running info query gets its own, more generous budget.
+ private static final long PROXY_RUNNING_INFO_TIMEOUT_MILLIS = 10_000L;
+ private static final long PROXY_ADDRESS_CACHE_TTL_MILLIS = 60_000L;
+ private static final String DEFAULT_INSTANCE_KEY = "__default__";
+
+ private final MqAdminExtFactory adminFactory;
+ private final RuntimeAdminClientResolver runtimeAdminClientResolver;
+ private final RocketMQProperties properties;
+
+ private final Map<String, CachedProxyAddresses> proxyAddressCache = new
ConcurrentHashMap<>();
+ private final AtomicBoolean clientStarted = new AtomicBoolean(false);
+ private volatile NettyRemotingClient remotingClient;
+
+ /**
+ * Queries the proxies of the given instance for the consumer connection
info of the group.
+ * Returns {@code null} when no proxy knows the group (offline everywhere)
or no proxy is
+ * reachable.
+ */
+ public ConsumerConnection resolveConsumerConnection(String instanceId,
String group) {
+ for (String addr : discoverProxyAddresses(instanceId)) {
+ try {
+ ConsumerConnection connection = queryProxy(addr, group);
+ if (connection != null) {
+ return connection;
+ }
+ } catch (Exception e) {
+ log.debug("Proxy consumer connection query failed for {} via
{}: {}",
+ group, addr, e.getMessage());
+ }
+ }
+ return null;
+ }
+
+ ConsumerConnection queryProxy(String proxyAddr, String group) throws
Exception {
+ GetConsumerConnectionListRequestHeader header = new
GetConsumerConnectionListRequestHeader();
+ header.setConsumerGroup(group);
+ 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;
+ }
+ return ConsumerConnection.decode(response.getBody(),
ConsumerConnection.class);
+ }
+
+ /**
+ * Asks the proxies for the running info (including the client jstack) of
one consumer client.
+ * Proxy-connected clients keep their channel on the proxy and never
register on a broker, so
+ * the proxy is the only component able to reach them. Returns {@code
null} when no proxy can
+ * answer, letting the caller fall back to the broker for directly
connected clients.
+ */
+ public ConsumerRunningInfo resolveConsumerRunningInfo(String instanceId,
String group, String clientId) {
+ for (String addr : discoverProxyAddresses(instanceId)) {
+ try {
+ ConsumerRunningInfo runningInfo = queryProxyRunningInfo(addr,
group, clientId);
+ if (runningInfo != null) {
+ return runningInfo;
+ }
+ } catch (Exception e) {
+ log.debug("Proxy consumer running info query failed for {}/{}
via {}: {}",
+ group, clientId, addr, e.getMessage());
+ }
+ }
+ return null;
+ }
+
+ ConsumerRunningInfo queryProxyRunningInfo(String proxyAddr, String group,
String clientId) throws Exception {
+ GetConsumerRunningInfoRequestHeader header = new
GetConsumerRunningInfoRequestHeader();
+ header.setConsumerGroup(group);
+ header.setClientId(clientId);
+ header.setJstackEnable(true);
+ RemotingCommand request =
+
RemotingCommand.createRequestCommand(RequestCode.GET_CONSUMER_RUNNING_INFO,
header);
+ RemotingCommand response =
+ remotingClient().invokeSync(proxyAddr, request,
PROXY_RUNNING_INFO_TIMEOUT_MILLIS);
+ if (response == null) {
+ return null;
+ }
+ if (response.getCode() != ResponseCode.SUCCESS || response.getBody()
== null) {
+ log.info("Proxy {} cannot report running info for {}/{}: code={}
remark={}",
+ proxyAddr, group, clientId, response.getCode(),
response.getRemark());
+ return null;
+ }
+ return ConsumerRunningInfo.decode(response.getBody(),
ConsumerRunningInfo.class);
+ }
+
+ List<String> discoverProxyAddresses(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();
+ }
+ Set<String> ips = new LinkedHashSet<>();
+ 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;
+ });
+ } catch (Exception e) {
+ log.debug("Proxy discovery via heartbeat syncer failed for
instance {}: {}",
+ instanceId, e.getMessage());
+ }
+ 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;
+ }
+
+ private <T> T executeAdmin(String instanceId,
MqAdminExtFactory.AdminAction<T> action) {
+ if (StringUtils.hasText(instanceId)) {
+ return runtimeAdminClientResolver.execute(instanceId, action);
+ }
+ return adminFactory.execute(properties.getNamesrvAddr(), null, action);
+ }
+
+ private NettyRemotingClient remotingClient() {
+ NettyRemotingClient client = remotingClient;
+ if (client == null) {
+ synchronized (this) {
+ if (remotingClient == null) {
+ remotingClient = new NettyRemotingClient(new
NettyClientConfig());
+ }
+ client = remotingClient;
+ }
+ }
+ if (clientStarted.compareAndSet(false, true)) {
+ client.start();
+ }
+ return client;
+ }
+
+ @PreDestroy
+ public void shutdownRemotingClient() {
+ if (clientStarted.get() && remotingClient != null) {
+ remotingClient.shutdown();
+ }
+ }
+
+ void setRemotingClientForTest(NettyRemotingClient client) {
+ this.remotingClient = client;
+ clientStarted.set(true);
+ }
+
+ 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 895bcd0f3..a333897f8 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
@@ -17,7 +17,9 @@
package org.apache.rocketmq.studio.provider.apache;
import org.apache.rocketmq.client.exception.MQBrokerException;
-import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
+import org.apache.rocketmq.remoting.protocol.admin.OffsetWrapper;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
@@ -79,6 +81,9 @@ public class RocketMQAdminClientImpl implements AdminClient {
private final AuditService auditService;
private final RuntimeAdminClientResolver runtimeAdminClientResolver;
+ @org.springframework.beans.factory.annotation.Autowired(required = false)
+ private ProxyConsumerResolver proxyConsumerResolver;
+
@Override
public TopicVO getTopic(String name) {
return adminFactory.execute(namesrvAddr(), null, admin -> {
@@ -104,36 +109,110 @@ public class RocketMQAdminClientImpl implements
AdminClient {
@Override
public ConsumerGroupVO getConsumerGroup(String instanceId, String name) {
if (StringUtils.hasText(instanceId)) {
- return runtimeAdminClientResolver.execute(instanceId, admin ->
getConsumerGroup(admin, name));
+ return runtimeAdminClientResolver.execute(instanceId, admin ->
getConsumerGroup(admin, instanceId, name));
}
- return adminFactory.execute(namesrvAddr(), null, admin ->
getConsumerGroup(admin, name));
+ return adminFactory.execute(namesrvAddr(), null, admin ->
getConsumerGroup(admin, null, name));
}
- private ConsumerGroupVO getConsumerGroup(MQAdminExt admin, String name) {
+ private ConsumerGroupVO getConsumerGroup(MQAdminExt admin, String
instanceId, String name) {
ConsumerGroupVO vo = new ConsumerGroupVO();
vo.setName(name);
try {
var conn = admin.examineConsumerConnectionInfo(name);
if (conn != null) {
if (conn.getConnectionSet() != null) {
- vo.setOnlineInstances(conn.getConnectionSet().size());
+ vo.setInstances(ConsumerConnections.toInstances(conn));
+ vo.setOnlineInstances(vo.getInstances().size());
}
if (conn.getSubscriptionTable() != null) {
vo.setSubscribedTopics(new
ArrayList<>(conn.getSubscriptionTable().keySet()));
}
}
- } catch (MQClientException exception) {
- if (exception.getResponseCode() ==
ResponseCode.CONSUMER_NOT_ONLINE) {
- log.debug("Consumer group {} is offline", name);
- return vo;
- }
- throw new BusinessException(502, "Failed to get consumer group: "
+ exception.getMessage());
} catch (Exception exception) {
- throw new BusinessException(502, "Failed to get consumer group: "
+ exception.getMessage());
+ if (isConsumerNotOnline(exception)) {
+ log.debug("Consumer group {} is offline on broker", name);
+ applyProxyFallback(instanceId, name, vo);
+ } else {
+ throw new BusinessException(502, "Failed to get consumer
group: " + exception.getMessage());
+ }
}
+ fillConsumeStats(admin, vo, name);
return vo;
}
+ /**
+ * Fills totalLag and delaySeconds from the broker consume stats.
Proxy-connected groups
+ * still maintain broker-side offset tables (the proxy forwards offset
updates), so this
+ * works even when the connection lookup reports the group offline; groups
without any
+ * offset table (e.g. pure POP) simply keep the zero defaults.
+ *
+ * <p>delaySeconds is derived from the newest consumed-message timestamp
(the consumption
+ * frontier). Using the oldest timestamp is misleading for POP groups,
where untouched
+ * queues keep frozen stale timestamps.
+ */
+ private void fillConsumeStats(MQAdminExt admin, ConsumerGroupVO vo, String
name) {
+ try {
+ ConsumeStats stats = admin.examineConsumeStats(name);
+ if (stats == null || stats.getOffsetTable() == null ||
stats.getOffsetTable().isEmpty()) {
+ return;
+ }
+ long totalLag = 0;
+ long newestConsumedTimestamp = 0;
+ for (OffsetWrapper wrapper : stats.getOffsetTable().values()) {
+ long diff = wrapper.getBrokerOffset() -
wrapper.getConsumerOffset();
+ if (diff > 0) {
+ totalLag += diff;
+ }
+ long lastTimestamp = wrapper.getLastTimestamp();
+ if (lastTimestamp > newestConsumedTimestamp) {
+ newestConsumedTimestamp = lastTimestamp;
+ }
+ }
+ vo.setTotalLag(totalLag);
+ if (newestConsumedTimestamp > 0) {
+ long delaySeconds = (System.currentTimeMillis() -
newestConsumedTimestamp) / 1000;
+ vo.setDelaySeconds((int) Math.max(delaySeconds, 0));
+ }
+ } catch (Exception e) {
+ log.debug("No consume stats for group {}: {}", name,
e.getMessage());
+ }
+ }
+
+ /**
+ * Groups whose clients connect through a proxy are invisible to
broker-side stats; the
+ * proxy's own client manager still knows them, so fill the offline detail
from there.
+ */
+ private void applyProxyFallback(String instanceId, String group,
ConsumerGroupVO vo) {
+ if (proxyConsumerResolver == null) {
+ return;
+ }
+ ConsumerConnection viaProxy =
proxyConsumerResolver.resolveConsumerConnection(instanceId, group);
+ if (viaProxy == null) {
+ return;
+ }
+ if (viaProxy.getConnectionSet() != null) {
+ vo.setInstances(ConsumerConnections.toInstances(viaProxy));
+ vo.setOnlineInstances(vo.getInstances().size());
+ }
+ if (viaProxy.getSubscriptionTable() != null) {
+ vo.setSubscribedTopics(new
ArrayList<>(viaProxy.getSubscriptionTable().keySet()));
+ }
+ }
+
+ /**
+ * examineConsumerConnectionInfo wraps the broker-side CODE 206 into an
MQClientException
+ * whose response code is lost (the code only survives in the message
text), so match on
+ * both the typed code and the message.
+ */
+ private static boolean isConsumerNotOnline(Exception exception) {
+ if (exception instanceof MQBrokerException brokerException
+ && brokerException.getResponseCode() ==
ResponseCode.CONSUMER_NOT_ONLINE) {
+ return true;
+ }
+ String message = exception.getMessage();
+ return message != null && (message.contains("not online") ||
message.contains("CODE: 206"));
+ }
+
@Override
public TopicVO createTopic(TopicVO topic) {
String topicName = topic.getName();
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQConsumerDiagnosticsProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQConsumerDiagnosticsProvider.java
index a41ca8904..19f6981d8 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQConsumerDiagnosticsProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQConsumerDiagnosticsProvider.java
@@ -55,8 +55,20 @@ public class RocketMQConsumerDiagnosticsProvider implements
ConsumerDiagnosticsP
private final MqAdminExtFactory adminFactory;
private final RocketMQProperties properties;
+ @org.springframework.beans.factory.annotation.Autowired(required = false)
+ private ProxyConsumerResolver proxyConsumerResolver;
+
@Override
public ConsumerStackTraceVO getConsumerStack(String instanceId, String
groupName, String clientId) {
+ // Clients that connect through a proxy keep their channel on the
proxy and never register
+ // on a broker, so ask the proxy first; the broker only knows directly
connected clients
+ // and answers "not online" for everyone else.
+ ConsumerRunningInfo viaProxy = proxyConsumerResolver == null
+ ? null
+ : proxyConsumerResolver.resolveConsumerRunningInfo(instanceId,
groupName, clientId);
+ if (viaProxy != null) {
+ return toStackTrace(groupName, clientId, viaProxy);
+ }
if (StringUtils.hasText(instanceId)) {
return runtimeAdminClientResolver.execute(instanceId,
admin -> getConsumerStack(admin, groupName, clientId));
@@ -74,19 +86,12 @@ public class RocketMQConsumerDiagnosticsProvider implements
ConsumerDiagnosticsP
if (runningInfo == null) {
throw new BusinessException(404, "Consumer client not found: "
+ clientId);
}
- List<ConsumerThreadStackVO> threads =
parseJstack(runningInfo.getJstack());
- return ConsumerStackTraceVO.builder()
- .groupName(groupName)
- .clientId(clientId)
- .capturedAt(LocalDateTime.now())
- .threadCount(threads.size())
- .threads(threads)
- .build();
+ return toStackTrace(groupName, clientId, runningInfo);
} catch (BusinessException e) {
throw e;
} catch (MQClientException e) {
if (e.getResponseCode() == ResponseCode.CONSUMER_NOT_ONLINE) {
- throw new BusinessException(404, "Consumer client is not
online: " + clientId);
+ throw new BusinessException(404, notReachable(clientId));
}
throw diagnosticsFailure(groupName, clientId, e);
} catch (Exception e) {
@@ -94,11 +99,32 @@ public class RocketMQConsumerDiagnosticsProvider implements
ConsumerDiagnosticsP
}
}
+ private ConsumerStackTraceVO toStackTrace(String groupName, String
clientId, ConsumerRunningInfo runningInfo) {
+ List<ConsumerThreadStackVO> threads =
parseJstack(runningInfo.getJstack());
+ return ConsumerStackTraceVO.builder()
+ .groupName(groupName)
+ .clientId(clientId)
+ .capturedAt(LocalDateTime.now())
+ .threadCount(threads.size())
+ .threads(threads)
+ .build();
+ }
+
+ private String notReachable(String clientId) {
+ return "Consumer client is not reachable from any proxy or broker: " +
clientId;
+ }
+
private BusinessException diagnosticsFailure(String groupName, String
clientId, Exception exception) {
log.warn("Failed to get consumer stack, groupName={}, clientId={}: {}",
groupName, clientId, exception.getMessage());
+ String rootMessage = rootMessage(exception);
+ // The broker answers "not online" for every proxy-connected client;
surface that as the
+ // business state it is instead of a raw broker error the operator
cannot act on.
+ if (rootMessage != null && rootMessage.contains("not online")) {
+ return new BusinessException(404, notReachable(clientId));
+ }
return new BusinessException(502,
- "Failed to get consumer stack for " + clientId + ": " +
rootMessage(exception));
+ "Failed to get consumer stack for " + clientId + ": " +
rootMessage);
}
private String rootMessage(Exception exception) {
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 d4e93d7e3..06f5cc28a 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
@@ -45,6 +45,7 @@ import
org.apache.rocketmq.studio.common.util.SystemTopicFilter;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.studio.common.domain.enums.TopicType;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
+import org.apache.rocketmq.studio.instance.group.ConsumerInstanceVO;
import org.apache.rocketmq.studio.instance.group.QueueProgressVO;
import org.apache.rocketmq.studio.instance.group.SubscriptionEntryVO;
import org.apache.rocketmq.studio.instance.topic.BrokerRouteVO;
@@ -68,6 +69,12 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
/**
* Real MetadataProvider implementation.
@@ -99,6 +106,9 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
private final RmqGroupMapper groupMapper;
private final RuntimeAdminClientResolver runtimeAdminClientResolver;
+ @org.springframework.beans.factory.annotation.Autowired(required = false)
+ private ProxyConsumerResolver proxyConsumerResolver;
+
/**
* Default proxy stats source until a real proxy transport is wired in. It
reports the unknown
* sentinel ({@link ConsumerLagResolver#UNKNOWN}) so a {@code -1} gRPC lag
is surfaced instead of
@@ -226,14 +236,107 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
vo.setGmtCreate(entity.getGmtCreate());
vo.setGmtModified(entity.getGmtModified());
- // Live connection info (online instances, lag) is intentionally
NOT fetched
- // during list operations to avoid N+1 admin API calls. It is
loaded on
- // demand when viewing a single group's detail page.
+ // Live stats (online clients, lag, delay) are enriched in
parallel below with a
+ // bounded executor and per-group timeout instead of blocking the
listing.
result.add(vo);
}
+ enrichLiveStats(instanceId, result);
return result;
}
+ private static final int ONLINE_ENRICHMENT_THREADS = 8;
+ private static final long ONLINE_ENRICHMENT_TIMEOUT_SECONDS = 3;
+
+ private final ExecutorService onlineEnrichmentExecutor =
Executors.newFixedThreadPool(
+ ONLINE_ENRICHMENT_THREADS, runnable -> {
+ Thread thread = new Thread(runnable,
"group-online-enrichment");
+ thread.setDaemon(true);
+ return thread;
+ });
+
+ @jakarta.annotation.PreDestroy
+ void shutdownOnlineEnrichmentExecutor() {
+ onlineEnrichmentExecutor.shutdownNow();
+ }
+
+ private void enrichLiveStats(String instanceId, List<ConsumerGroupVO>
groups) {
+ boolean noLiveSource = !StringUtils.hasText(instanceId) && !hasAdmin();
+ if (groups.isEmpty() || noLiveSource) {
+ return;
+ }
+ List<Future<?>> futures = new ArrayList<>(groups.size());
+ for (ConsumerGroupVO vo : groups) {
+ futures.add(onlineEnrichmentExecutor.submit(() ->
enrichGroupLiveStats(instanceId, vo)));
+ }
+ for (int i = 0; i < groups.size(); i++) {
+ try {
+ futures.get(i).get(ONLINE_ENRICHMENT_TIMEOUT_SECONDS,
TimeUnit.SECONDS);
+ } catch (TimeoutException e) {
+ futures.get(i).cancel(true);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
+ } catch (ExecutionException e) {
+ // leave the stats at zero
+ }
+ }
+ }
+
+ 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()));
+ vo.setInstances(instances);
+ vo.setOnlineInstances(instances.size());
+ try {
+ ConsumeStats stats;
+ if (StringUtils.hasText(instanceId)) {
+ stats = runtimeAdminClientResolver.execute(instanceId,
+ admin -> admin.examineConsumeStats(vo.getName()));
+ } else {
+ stats = adminExecute(admin ->
admin.examineConsumeStats(vo.getName()));
+ }
+ if (stats == null || stats.getOffsetTable() == null ||
stats.getOffsetTable().isEmpty()) {
+ return;
+ }
+ long totalLag = 0;
+ long newestConsumedTimestamp = 0;
+ for (OffsetWrapper wrapper : stats.getOffsetTable().values()) {
+ long diff = wrapper.getBrokerOffset() -
wrapper.getConsumerOffset();
+ if (diff > 0) {
+ totalLag += diff;
+ }
+ long lastTimestamp = wrapper.getLastTimestamp();
+ if (lastTimestamp > newestConsumedTimestamp) {
+ newestConsumedTimestamp = lastTimestamp;
+ }
+ }
+ vo.setTotalLag(totalLag);
+ if (newestConsumedTimestamp > 0) {
+ long delaySeconds = (System.currentTimeMillis() -
newestConsumedTimestamp) / 1000;
+ vo.setDelaySeconds((int) Math.max(delaySeconds, 0));
+ }
+ } catch (Exception e) {
+ // No consume stats (e.g. POP-only group without an offset table):
keep zeros.
+ }
+ }
+
+ private ConsumerConnection resolveConsumerConnection(String instanceId,
String group) {
+ try {
+ if (StringUtils.hasText(instanceId)) {
+ return runtimeAdminClientResolver.execute(instanceId,
+ admin -> admin.examineConsumerConnectionInfo(group));
+ }
+ return adminExecute(admin ->
admin.examineConsumerConnectionInfo(group));
+ } catch (Exception e) {
+ if (isGroupNotOnline(e) && proxyConsumerResolver != null) {
+ return
proxyConsumerResolver.resolveConsumerConnection(instanceId, group);
+ }
+ return null;
+ }
+ }
+
private String normalizeMetadataScope(String instanceId) {
return StringUtils.hasText(instanceId) ? instanceId.trim() : "";
}
@@ -439,6 +542,7 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
long diff = resolveDiff(ow.getBrokerOffset(),
ow.getConsumerOffset());
progress.add(QueueProgressVO.builder()
+ .topic(mq.getTopic())
.broker(mq.getBrokerName())
.queueId(mq.getQueueId())
.brokerOffset(ow.getBrokerOffset())
@@ -453,6 +557,12 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
});
return progress;
} catch (Exception e) {
+ if (isGroupNotOnline(e)) {
+ // The group has no consume stats on the broker (consumer not
online);
+ // that is a business state, not a connectivity failure.
+ log.info("Consumer group {} not online, no consume stats: {}",
name, e.getMessage());
+ return Collections.emptyList();
+ }
log.warn("Failed to get progress for group {}: {}", name,
e.getMessage());
throw new BusinessException(502, "Failed to get progress for group
" + name + ": " + e.getMessage());
}
@@ -461,15 +571,16 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
@Override
public List<SubscriptionEntryVO> getGroupSubscriptions(String instanceId,
String name) {
if (StringUtils.hasText(instanceId)) {
- return runtimeAdminClientResolver.execute(instanceId, admin ->
getGroupSubscriptions(admin, name));
+ return runtimeAdminClientResolver.execute(instanceId,
+ admin -> getGroupSubscriptions(admin, instanceId, name));
}
if (!hasAdmin()) {
return Collections.emptyList();
}
- return adminExecute(admin -> getGroupSubscriptions(admin, name));
+ return adminExecute(admin -> getGroupSubscriptions(admin, null, name));
}
- private List<SubscriptionEntryVO> getGroupSubscriptions(MQAdminExt admin,
String name) {
+ private List<SubscriptionEntryVO> getGroupSubscriptions(MQAdminExt admin,
String instanceId, String name) {
try {
ensureRetryTopicExists(admin, name);
ConsumerConnection conn =
admin.examineConsumerConnectionInfo(name);
@@ -477,24 +588,18 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
return Collections.emptyList();
}
- List<SubscriptionEntryVO> subscriptions = new ArrayList<>();
- for (Map.Entry<String, SubscriptionData> entry :
conn.getSubscriptionTable().entrySet()) {
- SubscriptionData sd = entry.getValue();
- subscriptions.add(SubscriptionEntryVO.builder()
- .topic(sd.getTopic())
- .expression(sd.getSubString())
- .type(sd.getExpressionType())
- .filterMode(filterMode(sd.getExpressionType()))
- .build());
- }
- return subscriptions;
+ return toSubscriptionEntries(conn, conn.getConnectionSet() != null
&& !conn.getConnectionSet().isEmpty());
} catch (Exception e) {
if (isGroupNotOnline(e)) {
- // Consumers connected through a proxy never register with the
broker, so
- // the broker reports the group as offline; surface an empty
subscription
- // list instead of an error.
+ // Consumers connected through a proxy never register with the
broker, so the
+ // broker reports the group as offline; the proxy's client
manager still knows
+ // them, so fall back to it before surfacing an empty
subscription list.
log.info("Consumer group {} not online on broker (likely
proxy-connected): {}",
name, e.getMessage());
+ List<SubscriptionEntryVO> viaProxy =
subscriptionsViaProxy(instanceId, name);
+ if (!viaProxy.isEmpty()) {
+ return viaProxy;
+ }
return Collections.emptyList();
}
log.warn("Failed to get subscriptions for group {}: {}", name,
e.getMessage());
@@ -503,6 +608,35 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
}
}
+ private List<SubscriptionEntryVO> subscriptionsViaProxy(String instanceId,
String group) {
+ if (proxyConsumerResolver == null) {
+ return Collections.emptyList();
+ }
+ ConsumerConnection conn =
proxyConsumerResolver.resolveConsumerConnection(instanceId, group);
+ if (conn == null || conn.getSubscriptionTable() == null) {
+ return Collections.emptyList();
+ }
+ return toSubscriptionEntries(conn, conn.getConnectionSet() != null &&
!conn.getConnectionSet().isEmpty());
+ }
+
+ private List<SubscriptionEntryVO> toSubscriptionEntries(ConsumerConnection
conn, boolean anyClientOnline) {
+ List<SubscriptionEntryVO> subscriptions = new ArrayList<>();
+ for (Map.Entry<String, SubscriptionData> entry :
conn.getSubscriptionTable().entrySet()) {
+ SubscriptionData sd = entry.getValue();
+ subscriptions.add(SubscriptionEntryVO.builder()
+ .topic(sd.getTopic())
+ .expression(sd.getSubString())
+ .type(sd.getExpressionType())
+ .filterMode(filterMode(sd.getExpressionType()))
+ // The broker/proxy expose the group's merged subscription
set only; with at
+ // least one client connected that merged view is the
consistent observable
+ // state. Without connections the consistency status stays
unknown (null).
+ .consistency(anyClientOnline ? "consistent" : null)
+ .build());
+ }
+ return subscriptions;
+ }
+
private boolean isGroupNotOnline(Exception e) {
if (e instanceof
org.apache.rocketmq.client.exception.MQBrokerException brokerException) {
return brokerException.getResponseCode()
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProvider.java
index 361b6ac7c..b75694a72 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProvider.java
@@ -497,6 +497,7 @@ public class TencentInstanceProvider implements
InstanceProvider {
continue;
}
rows.add(QueueProgressVO.builder()
+ .topic(subscription.getTopic())
.broker("topic:" + subscription.getTopic())
.queueId(0)
.brokerOffset(0L)
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
new file mode 100644
index 000000000..6957a1512
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ProxyConsumerResolverTest.java
@@ -0,0 +1,119 @@
+/*
+ * 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.
+ */
+
+package org.apache.rocketmq.studio.provider.apache;
+
+import org.apache.rocketmq.remoting.protocol.body.Connection;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
+import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
+import org.apache.rocketmq.tools.admin.MQAdminExt;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.util.HashSet;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class ProxyConsumerResolverTest {
+
+ @Mock
+ private MqAdminExtFactory adminFactory;
+
+ @Mock
+ private RuntimeAdminClientResolver runtimeAdminClientResolver;
+
+ @Mock
+ private MQAdminExt adminExt;
+
+ private ProxyConsumerResolver resolver;
+
+ @BeforeEach
+ void setUp() {
+ lenient().when(runtimeAdminClientResolver.execute(any(String.class),
any()))
+ .thenAnswer(invocation ->
+
invocation.<MqAdminExtFactory.AdminAction<Object>>getArgument(1).apply(adminExt));
+ lenient().when(adminFactory.execute(any(), any(), any()))
+ .thenAnswer(invocation ->
+
invocation.<MqAdminExtFactory.AdminAction<Object>>getArgument(2).apply(adminExt));
+ resolver = new ProxyConsumerResolver(adminFactory,
runtimeAdminClientResolver, new RocketMQProperties());
+ }
+
+ @Test
+ void
discoverProxyAddressesShouldDeriveRemotingAddressesFromHeartbeatSyncerTest()
throws Exception {
+ ConsumerConnection syncer = new ConsumerConnection();
+ HashSet<Connection> connections = new HashSet<>();
+ Connection proxyA = new Connection();
+ proxyA.setClientId("proxy-a");
+ proxyA.setClientAddr("10.0.4.66:10911");
+ Connection proxyB = new Connection();
+ proxyB.setClientId("proxy-b");
+ proxyB.setClientAddr("10.0.3.110:10911");
+ connections.add(proxyA);
+ connections.add(proxyB);
+ syncer.setConnectionSet(connections);
+
when(adminExt.examineConsumerConnectionInfo("CID_DefaultHeartBeatSyncerTopic")).thenReturn(syncer);
+
+ List<String> addresses = resolver.discoverProxyAddresses("instance-a");
+
+ assertThat(addresses).containsExactlyInAnyOrder("10.0.4.66:8080",
"10.0.3.110:8080");
+ }
+
+ @Test
+ void discoverProxyAddressesShouldCacheResultsTest() throws Exception {
+ ConsumerConnection syncer = new ConsumerConnection();
+ syncer.setConnectionSet(new HashSet<>());
+
when(adminExt.examineConsumerConnectionInfo("CID_DefaultHeartBeatSyncerTopic")).thenReturn(syncer);
+
+ resolver.discoverProxyAddresses("instance-a");
+ resolver.discoverProxyAddresses("instance-a");
+
+ org.mockito.Mockito.verify(adminExt, org.mockito.Mockito.times(1))
+
.examineConsumerConnectionInfo("CID_DefaultHeartBeatSyncerTopic");
+ }
+
+ @Test
+ void resolveConsumerConnectionShouldReturnNullWhenNoProxyDiscoveredTest()
throws Exception {
+
when(adminExt.examineConsumerConnectionInfo("CID_DefaultHeartBeatSyncerTopic"))
+ .thenThrow(new IllegalStateException("syncer group missing"));
+
+ assertThat(resolver.resolveConsumerConnection("instance-a",
"cg-orders")).isNull();
+ }
+
+ @Test
+ void resolveConsumerConnectionShouldReturnNullWhenProxyQueryFailsTest()
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);
+
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();
+ }
+}
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 26f31deb5..04cbdc210 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
@@ -16,6 +16,7 @@ import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
import org.apache.ibatis.builder.MapperBuilderAssistant;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
@@ -31,6 +32,7 @@ import
org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.common.domain.enums.TopicType;
import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
+import org.apache.rocketmq.studio.instance.group.ConsumerInstanceVO;
import org.apache.rocketmq.studio.instance.topic.TopicVO;
import org.apache.rocketmq.studio.instance.topic.SendMessageDTO;
import org.apache.rocketmq.studio.instance.topic.SendMessageVO;
@@ -113,6 +115,101 @@ class RocketMQAdminClientImplTest {
assertThat(group.getOnlineInstances()).isZero();
}
+ @Test
+ void
getConsumerGroupReturnsOfflineDetailWhenCode206OnlySurvivesInTheMessageTest()
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));
+
+ ConsumerGroupVO group = adminClient.getConsumerGroup(null, "orders");
+
+ assertThat(group.getName()).isEqualTo("orders");
+ assertThat(group.getOnlineInstances()).isZero();
+ }
+
+ @Test
+ void
getConsumerGroupFillsProxySideConnectionsWhenBrokerReportsOfflineTest() 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);
+ ConsumerConnection viaProxy = new ConsumerConnection();
+
java.util.HashSet<org.apache.rocketmq.remoting.protocol.body.Connection>
connections = new java.util.HashSet<>();
+ org.apache.rocketmq.remoting.protocol.body.Connection connection =
+ new org.apache.rocketmq.remoting.protocol.body.Connection();
+ connection.setClientId("client-1");
+ connection.setClientAddr("10.0.3.104:50124");
+ connections.add(connection);
+ viaProxy.setConnectionSet(connections);
+ java.util.concurrent.ConcurrentHashMap<String,
+
org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData> table =
+ new java.util.concurrent.ConcurrentHashMap<>();
+ org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData
subscription =
+ new
org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData();
+ subscription.setTopic("studio-normal");
+ table.put("studio-normal", subscription);
+ viaProxy.setSubscriptionTable(table);
+ when(resolver.resolveConsumerConnection("instance-a",
"orders")).thenReturn(viaProxy);
+
org.springframework.test.util.ReflectionTestUtils.setField(adminClient,
"proxyConsumerResolver", resolver);
+
+ ConsumerGroupVO group = adminClient.getConsumerGroup("instance-a",
"orders");
+
+ assertThat(group.getOnlineInstances()).isEqualTo(1);
+
assertThat(group.getSubscribedTopics()).containsExactly("studio-normal");
+ }
+
+ @Test
+ void getConsumerGroupComputesLagAndDelayFromConsumeStatsTest() throws
Exception {
+ org.apache.rocketmq.remoting.protocol.body.ConsumerConnection
connection =
+ new
org.apache.rocketmq.remoting.protocol.body.ConsumerConnection();
+ connection.setConnectionSet(new java.util.HashSet<>());
+
when(adminExt.examineConsumerConnectionInfo("orders")).thenReturn(connection);
+
+ org.apache.rocketmq.remoting.protocol.admin.ConsumeStats stats =
+ new org.apache.rocketmq.remoting.protocol.admin.ConsumeStats();
+ org.apache.rocketmq.common.message.MessageQueue queue =
+ new
org.apache.rocketmq.common.message.MessageQueue("orders-topic", "broker-a", 0);
+ org.apache.rocketmq.remoting.protocol.admin.OffsetWrapper wrapper =
+ new
org.apache.rocketmq.remoting.protocol.admin.OffsetWrapper();
+ wrapper.setBrokerOffset(100);
+ wrapper.setConsumerOffset(60);
+ wrapper.setLastTimestamp(System.currentTimeMillis() - 5_000);
+ stats.getOffsetTable().put(queue, wrapper);
+ when(adminExt.examineConsumeStats("orders")).thenReturn(stats);
+
+ ConsumerGroupVO group = adminClient.getConsumerGroup(null, "orders");
+
+ assertThat(group.getTotalLag()).isEqualTo(40);
+ assertThat(group.getDelaySeconds()).isBetween(4, 30);
+ }
+
+ @Test
+ void getConsumerGroupFillsOnlineInstanceListFromConnectionsTest() throws
Exception {
+ org.apache.rocketmq.remoting.protocol.body.ConsumerConnection
connection =
+ new
org.apache.rocketmq.remoting.protocol.body.ConsumerConnection();
+
java.util.HashSet<org.apache.rocketmq.remoting.protocol.body.Connection>
connections =
+ new java.util.HashSet<>();
+ org.apache.rocketmq.remoting.protocol.body.Connection conn =
+ new org.apache.rocketmq.remoting.protocol.body.Connection();
+ conn.setClientId("client-1");
+ conn.setClientAddr("10.0.3.104:50124");
+ connections.add(conn);
+ connection.setConnectionSet(connections);
+
when(adminExt.examineConsumerConnectionInfo("orders")).thenReturn(connection);
+
+ ConsumerGroupVO group = adminClient.getConsumerGroup(null, "orders");
+
+ assertThat(group.getOnlineInstances()).isEqualTo(1);
+ assertThat(group.getInstances())
+ .extracting(ConsumerInstanceVO::getClientId,
ConsumerInstanceVO::getAddress)
+
.containsExactly(org.assertj.core.groups.Tuple.tuple("client-1",
"10.0.3.104:50124"));
+ }
+
@Test
void getConsumerGroupUsesSelectedInstanceAdmin() throws Exception {
DefaultMQAdminExt selectedAdmin =
org.mockito.Mockito.mock(DefaultMQAdminExt.class);
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQConsumerDiagnosticsProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQConsumerDiagnosticsProviderTest.java
index 287559829..e4f78bbeb 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQConsumerDiagnosticsProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQConsumerDiagnosticsProviderTest.java
@@ -147,7 +147,52 @@ class RocketMQConsumerDiagnosticsProviderTest {
assertThatThrownBy(() -> provider.getConsumerStack("instance-a",
"cg-orders", "client-1"))
.isInstanceOf(BusinessException.class)
- .hasMessage("Consumer client is not online: client-1")
+ .hasMessage("Consumer client is not reachable from any proxy
or broker: client-1")
+ .satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(404));
+ }
+
+ @Test
+ void getConsumerStackShouldPreferProxyOverBrokerTest() {
+ ProxyConsumerResolver resolver =
org.mockito.Mockito.mock(ProxyConsumerResolver.class);
+ ConsumerRunningInfo runningInfo = new ConsumerRunningInfo();
+ runningInfo.setJstack("""
+ ConsumeMessageThread_1 TID: 7 STATE: RUNNABLE
+ ConsumeMessageThread_1
com.example.Listener.consume(Listener.java:20)
+ """);
+ when(resolver.resolveConsumerRunningInfo("instance-a", "cg-orders",
"client-1")).thenReturn(runningInfo);
+ org.springframework.test.util.ReflectionTestUtils.setField(provider,
"proxyConsumerResolver", resolver);
+
+ ConsumerStackTraceVO result = provider.getConsumerStack("instance-a",
"cg-orders", "client-1");
+
+ assertThat(result.getThreadCount()).isEqualTo(1);
+
assertThat(result.getThreads().get(0).getThreadName()).isEqualTo("ConsumeMessageThread_1");
+ verify(runtimeAdminClientResolver, never()).execute(anyString(),
any());
+ }
+
+ @Test
+ void getConsumerStackShouldFallBackToBrokerWhenNoProxyAnswersTest() throws
Exception {
+ ProxyConsumerResolver resolver =
org.mockito.Mockito.mock(ProxyConsumerResolver.class);
+ when(resolver.resolveConsumerRunningInfo("instance-a", "cg-orders",
"client-1")).thenReturn(null);
+ org.springframework.test.util.ReflectionTestUtils.setField(provider,
"proxyConsumerResolver", resolver);
+ ConsumerRunningInfo runningInfo = new ConsumerRunningInfo();
+ runningInfo.setJstack("PullMessageService TID: 9
STATE: WAITING\n");
+ when(adminExt.getConsumerRunningInfo("cg-orders", "client-1",
true)).thenReturn(runningInfo);
+
+ ConsumerStackTraceVO result = provider.getConsumerStack("instance-a",
"cg-orders", "client-1");
+
+ assertThat(result.getThreadCount()).isEqualTo(1);
+ verify(adminExt).getConsumerRunningInfo("cg-orders", "client-1", true);
+ }
+
+ @Test
+ void getConsumerStackShouldMapBrokerNotOnlineRemarkToNotFoundTest() throws
Exception {
+ when(adminExt.getConsumerRunningInfo("cg-orders", "client-1", true))
+ .thenThrow(new MQClientException(1,
+ "The Consumer <cg-orders> <client-1> not online"));
+
+ assertThatThrownBy(() -> provider.getConsumerStack("instance-a",
"cg-orders", "client-1"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Consumer client is not reachable from any proxy
or broker: client-1")
.satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(404));
}
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 f74e81cc1..543ddea0a 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
@@ -396,6 +396,18 @@ class RocketMQMetadataProviderTest {
.satisfies(error -> assertThat(((BusinessException)
error).getCode()).isEqualTo(502));
}
+ @Test
+ void getGroupProgressShouldReturnEmptyWhenConsumerNotOnlineTest() throws
Exception {
+ DefaultMQAdminExt admin =
org.mockito.Mockito.mock(DefaultMQAdminExt.class);
+ when(admin.examineConsumeStats("group-offline")).thenThrow(
+ new org.apache.rocketmq.client.exception.MQBrokerException(
+
org.apache.rocketmq.remoting.protocol.ResponseCode.CONSUMER_NOT_ONLINE,
+ "Not found the consumer group consume stats, because
return offset table is empty, "
+ + "maybe the consumer not online"));
+
+ assertThat(newLiveProvider(admin).getGroupProgress(null,
"group-offline")).isEmpty();
+ }
+
@Test
void getGroupSubscriptionsSurfacesAdminFailure() throws Exception {
DefaultMQAdminExt admin =
org.mockito.Mockito.mock(DefaultMQAdminExt.class);
@@ -468,6 +480,174 @@ class RocketMQMetadataProviderTest {
assertThat(newLiveProvider(admin).getGroupSubscriptions(null,
"group-proxy")).isEmpty();
}
+ @Test
+ void getGroupSubscriptionsShouldFallBackToProxyConnectionsTest() throws
Exception {
+ DefaultMQAdminExt admin =
org.mockito.Mockito.mock(DefaultMQAdminExt.class);
+ org.apache.rocketmq.remoting.protocol.route.TopicRouteData route =
+ new
org.apache.rocketmq.remoting.protocol.route.TopicRouteData();
+ java.util.HashMap<Long, String> brokerAddrs = new
java.util.HashMap<>();
+ brokerAddrs.put(0L, "10.0.0.11:10911");
+ route.setBrokerDatas(List.of(new
org.apache.rocketmq.remoting.protocol.route.BrokerData(
+ "cluster-a", "broker-a", brokerAddrs)));
+
when(admin.examineTopicRouteInfo("%RETRY%group-proxy")).thenReturn(route);
+ when(admin.examineConsumerConnectionInfo("group-proxy")).thenThrow(
+ new org.apache.rocketmq.client.exception.MQBrokerException(
+
org.apache.rocketmq.remoting.protocol.ResponseCode.CONSUMER_NOT_ONLINE,
+ "the consumer group[group-proxy] not online BROKER:
10.0.0.11:10911"));
+
+ ProxyConsumerResolver resolver =
org.mockito.Mockito.mock(ProxyConsumerResolver.class);
+ org.apache.rocketmq.remoting.protocol.body.ConsumerConnection viaProxy
=
+ new
org.apache.rocketmq.remoting.protocol.body.ConsumerConnection();
+ java.util.concurrent.ConcurrentHashMap<String,
+
org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData> table =
+ new java.util.concurrent.ConcurrentHashMap<>();
+ org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData
subscription =
+ new
org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData();
+ subscription.setTopic("studio-normal");
+ subscription.setSubString("*");
+ subscription.setExpressionType("TAG");
+ table.put("studio-normal", subscription);
+ viaProxy.setSubscriptionTable(table);
+
java.util.HashSet<org.apache.rocketmq.remoting.protocol.body.Connection>
proxyConnections =
+ new java.util.HashSet<>();
+ org.apache.rocketmq.remoting.protocol.body.Connection proxyConnection =
+ new org.apache.rocketmq.remoting.protocol.body.Connection();
+ proxyConnection.setClientId("client-1");
+ proxyConnection.setClientAddr("10.0.3.104:50124");
+ proxyConnections.add(proxyConnection);
+ viaProxy.setConnectionSet(proxyConnections);
+ when(resolver.resolveConsumerConnection(null,
"group-proxy")).thenReturn(viaProxy);
+
+ RocketMQMetadataProvider provider = newLiveProvider(admin);
+ org.springframework.test.util.ReflectionTestUtils.setField(
+ provider, "proxyConsumerResolver", resolver);
+
+ List<SubscriptionEntryVO> subscriptions =
provider.getGroupSubscriptions(null, "group-proxy");
+
+ assertThat(subscriptions).extracting(SubscriptionEntryVO::getTopic)
+ .containsExactly("studio-normal");
+
assertThat(subscriptions).extracting(SubscriptionEntryVO::getConsistency)
+ .containsExactly("consistent");
+ }
+
+ @Test
+ void
listConsumerGroupsShouldEnrichOnlineInstancesFromBrokerConnectionsTest() throws
Exception {
+ RmqGroup entity = new RmqGroup();
+ entity.setName("cg-online");
+ entity.setInstanceId("instance-a");
+ when(groupMapper.selectList(any())).thenReturn(List.of(entity));
+
+ DefaultMQAdminExt admin =
org.mockito.Mockito.mock(DefaultMQAdminExt.class);
+ org.apache.rocketmq.remoting.protocol.body.ConsumerConnection
connection =
+ new
org.apache.rocketmq.remoting.protocol.body.ConsumerConnection();
+
java.util.HashSet<org.apache.rocketmq.remoting.protocol.body.Connection>
connections =
+ new java.util.HashSet<>();
+ org.apache.rocketmq.remoting.protocol.body.Connection connA =
+ new org.apache.rocketmq.remoting.protocol.body.Connection();
+ connA.setClientId("client-a");
+ connA.setClientAddr("10.0.0.20:10000");
+ connections.add(connA);
+ org.apache.rocketmq.remoting.protocol.body.Connection connB =
+ new org.apache.rocketmq.remoting.protocol.body.Connection();
+ connB.setClientId("client-b");
+ connB.setClientAddr("10.0.0.21:10000");
+ connections.add(connB);
+ connection.setConnectionSet(connections);
+
when(admin.examineConsumerConnectionInfo("cg-online")).thenReturn(connection);
+
when(runtimeAdminClientResolver.execute(org.mockito.ArgumentMatchers.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).hasSize(1);
+ assertThat(groups.get(0).getOnlineInstances()).isEqualTo(2);
+ assertThat(groups.get(0).getInstances())
+
.extracting(org.apache.rocketmq.studio.instance.group.ConsumerInstanceVO::getClientId)
+ .containsExactlyInAnyOrder("client-a", "client-b");
+ }
+
+ @Test
+ void listConsumerGroupsShouldEnrichOnlineInstancesViaProxyFallbackTest()
throws Exception {
+ RmqGroup entity = new RmqGroup();
+ entity.setName("cg-proxy");
+ entity.setInstanceId("instance-a");
+ when(groupMapper.selectList(any())).thenReturn(List.of(entity));
+
+ DefaultMQAdminExt admin =
org.mockito.Mockito.mock(DefaultMQAdminExt.class);
+ when(admin.examineConsumerConnectionInfo("cg-proxy")).thenThrow(
+ new org.apache.rocketmq.client.exception.MQBrokerException(
+
org.apache.rocketmq.remoting.protocol.ResponseCode.CONSUMER_NOT_ONLINE,
+ "the consumer group[cg-proxy] not online BROKER:
10.0.0.11:10911"));
+
when(runtimeAdminClientResolver.execute(org.mockito.ArgumentMatchers.eq("instance-a"),
any()))
+ .thenAnswer(invocation ->
+
invocation.<MqAdminExtFactory.AdminAction<Object>>getArgument(1).apply(admin));
+
+ ProxyConsumerResolver resolver =
org.mockito.Mockito.mock(ProxyConsumerResolver.class);
+ org.apache.rocketmq.remoting.protocol.body.ConsumerConnection viaProxy
=
+ new
org.apache.rocketmq.remoting.protocol.body.ConsumerConnection();
+
java.util.HashSet<org.apache.rocketmq.remoting.protocol.body.Connection>
connections =
+ new java.util.HashSet<>();
+ org.apache.rocketmq.remoting.protocol.body.Connection conn =
+ new org.apache.rocketmq.remoting.protocol.body.Connection();
+ conn.setClientId("client-1");
+ conn.setClientAddr("10.0.3.104:50124");
+ connections.add(conn);
+ viaProxy.setConnectionSet(connections);
+ when(resolver.resolveConsumerConnection("instance-a",
"cg-proxy")).thenReturn(viaProxy);
+
+ RocketMQMetadataProvider provider = newLiveProvider(admin);
+ org.springframework.test.util.ReflectionTestUtils.setField(
+ provider, "proxyConsumerResolver", resolver);
+
+ List<ConsumerGroupVO> groups =
provider.listConsumerGroups("instance-a", null, null);
+
+ assertThat(groups).hasSize(1);
+ assertThat(groups.get(0).getOnlineInstances()).isEqualTo(1);
+ assertThat(groups.get(0).getInstances())
+
.extracting(org.apache.rocketmq.studio.instance.group.ConsumerInstanceVO::getAddress)
+ .containsExactly("10.0.3.104:50124");
+ }
+
+ @Test
+ void listConsumerGroupsShouldEnrichLagAndDelayFromConsumeStatsTest()
throws Exception {
+ RmqGroup entity = new RmqGroup();
+ entity.setName("cg-lag");
+ entity.setInstanceId("instance-a");
+ when(groupMapper.selectList(any())).thenReturn(List.of(entity));
+
+ DefaultMQAdminExt admin =
org.mockito.Mockito.mock(DefaultMQAdminExt.class);
+ org.apache.rocketmq.remoting.protocol.body.ConsumerConnection
connection =
+ new
org.apache.rocketmq.remoting.protocol.body.ConsumerConnection();
+ connection.setConnectionSet(new java.util.HashSet<>());
+
when(admin.examineConsumerConnectionInfo("cg-lag")).thenReturn(connection);
+
+ org.apache.rocketmq.remoting.protocol.admin.ConsumeStats stats =
+ new org.apache.rocketmq.remoting.protocol.admin.ConsumeStats();
+ org.apache.rocketmq.common.message.MessageQueue queue =
+ new
org.apache.rocketmq.common.message.MessageQueue("studio-normal", "broker-a", 0);
+ org.apache.rocketmq.remoting.protocol.admin.OffsetWrapper wrapper =
+ new
org.apache.rocketmq.remoting.protocol.admin.OffsetWrapper();
+ wrapper.setBrokerOffset(100);
+ wrapper.setConsumerOffset(60);
+ wrapper.setLastTimestamp(System.currentTimeMillis() - 5_000);
+ stats.getOffsetTable().put(queue, wrapper);
+ when(admin.examineConsumeStats("cg-lag")).thenReturn(stats);
+
when(runtimeAdminClientResolver.execute(org.mockito.ArgumentMatchers.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).hasSize(1);
+ assertThat(groups.get(0).getTotalLag()).isEqualTo(40);
+ assertThat(groups.get(0).getDelaySeconds()).isBetween(4, 30);
+ }
+
private RocketMQMetadataProvider newLiveProvider(MQAdminExt admin) throws
Exception {
MqAdminExtFactory factory = mock(MqAdminExtFactory.class);
RocketMQProperties liveProperties = new RocketMQProperties();
diff --git a/web/src/api/metadata.ts b/web/src/api/metadata.ts
index 953ae9ccc..003fc52cb 100644
--- a/web/src/api/metadata.ts
+++ b/web/src/api/metadata.ts
@@ -112,6 +112,7 @@ export interface ConsumerGroupDetail extends ConsumerGroup {
}
export interface QueueProgress {
+ topic: string;
broker: string;
queueId: number;
brokerOffset: number;
diff --git a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
index 1ee7fd6e2..82da18f4f 100644
--- a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
+++ b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
@@ -16,7 +16,7 @@
*/
import { App } from 'antd';
-import { act, render, screen, waitFor } from '@testing-library/react';
+import { act, render, screen, waitFor, within } from '@testing-library/react';
import userEvent from '@testing-library/user-event';
import type React from 'react';
import { MemoryRouter } from 'react-router-dom';
@@ -139,12 +139,21 @@ describe('Consumer page', () => {
);
vi.mocked(consumerService.getConsumerProgress).mockResolvedValue([
{
+ topic: 'remote-topic',
broker: 'broker-a',
queueId: 0,
brokerOffset: 100,
consumerOffset: 90,
diffTotal: 10,
},
+ {
+ topic: '%RETRY%remote-cg',
+ broker: 'broker-a',
+ queueId: 0,
+ brokerOffset: 5,
+ consumerOffset: 5,
+ diffTotal: 0,
+ },
]);
vi.mocked(consumerService.getConsumerSubscriptions).mockResolvedValue([
{
@@ -283,6 +292,40 @@ describe('Consumer page', () => {
await waitFor(() =>
expect(screen.getAllByText('remote-topic').length).toBeGreaterThan(0));
});
+ it('filters queue progress to the topic of the clicked distribution button',
async () => {
+ vi.mocked(consumerService.getConsumerSubscriptions).mockResolvedValue([
+ {
+ topic: 'remote-topic',
+ expression: '*',
+ type: 'NORMAL',
+ filterMode: '全量',
+ consistency: '一致',
+ },
+ {
+ topic: '%RETRY%remote-cg',
+ expression: '*',
+ type: 'RETRY',
+ filterMode: '全量',
+ consistency: '一致',
+ },
+ ]);
+ const user = userEvent.setup();
+ renderWithProviders(<ConsumerPage />);
+
+ await user.click(await screen.findByRole('button', { name: /详情/ }));
+ // Click 查看分布 on the retry-topic subscription row.
+ await waitFor(() =>
expect(screen.getAllByText('%RETRY%remote-cg').length).toBeGreaterThan(0));
+ const retryRow = screen.getByText('%RETRY%remote-cg').closest('tr') as
HTMLElement;
+ await user.click(within(retryRow).getByRole('button', { name: /查看分布/ }));
+
+ // The progress tab should now only show the retry topic's queue, not the
normal topic.
+ const progressPanel = await screen.findByRole('tabpanel', { name: /消费进度/
});
+ await waitFor(() =>
+
expect(within(progressPanel).getAllByText('%RETRY%remote-cg').length).toBeGreaterThan(0),
+ );
+
expect(within(progressPanel).queryByText('remote-topic')).not.toBeInTheDocument();
+ });
+
it('passes the selected instance to group diagnostics', async () => {
vi.mocked(instanceService.listInstances).mockResolvedValue([
{
@@ -494,7 +537,8 @@ describe('Consumer page', () => {
renderWithProviders(<ConsumerPage />);
await user.click(await screen.findByRole('button', { name: /详情/ }));
- await user.click(await screen.findByRole('tab', { name: /在线实例/ }));
+ // The online instance table now lives inside the overview tab, so its 线程栈
buttons are
+ // available as soon as the detail dialog opens — no separate tab click.
const stackButtons = await screen.findAllByRole('button', { name: /线程栈/ });
await user.click(stackButtons[0]);
await user.click(stackButtons[1]);
@@ -528,7 +572,6 @@ describe('Consumer page', () => {
renderWithProviders(<ConsumerPage />);
await user.click(await screen.findByRole('button', { name: /详情/ }));
- await user.click(await screen.findByRole('tab', { name: /在线实例/ }));
await user.click(await screen.findByRole('button', { name: /线程栈/ }));
await waitFor(() =>
diff --git a/web/src/pages/instance/consumer.tsx
b/web/src/pages/instance/consumer.tsx
index 2a5259f5c..ea364b57c 100644
--- a/web/src/pages/instance/consumer.tsx
+++ b/web/src/pages/instance/consumer.tsx
@@ -39,6 +39,7 @@ import {
Col,
Flex,
DatePicker,
+ Tooltip,
message,
} from 'antd';
import {
@@ -54,7 +55,7 @@ import {
Info,
ArrowsClockwise,
} from '@phosphor-icons/react';
-import { ImportOutlined, ExportOutlined, DeleteOutlined } from
'@ant-design/icons';
+import { ImportOutlined, ExportOutlined, DeleteOutlined, SyncOutlined } from
'@ant-design/icons';
import type { ColumnsType } from 'antd/es/table';
import dayjs from 'dayjs';
import type { Dayjs } from 'dayjs';
@@ -88,6 +89,7 @@ import {
type ResourceImportRow,
} from '../../utils/resourceCsvImport';
import { buildCsv, downloadCsv, type CsvColumn } from '../../utils/download';
+import { tableScrollX } from '../../utils/table';
const { Text } = Typography;
@@ -212,6 +214,7 @@ const ConsumerPageContent = ({
const [stackModalOpen, setStackModalOpen] = useState(false);
const [stackLoading, setStackLoading] = useState(false);
const [selectedStack, setSelectedStack] = useState<ConsumerStackTrace |
null>(null);
+ const [stackError, setStackError] = useState<string | null>(null);
const [selectedStackClient, setSelectedStackClient] =
useState<ConsumerInstance | null>(null);
const importInputRef = useRef<HTMLInputElement>(null);
const [importModalOpen, setImportModalOpen] = useState(false);
@@ -223,6 +226,14 @@ const ConsumerPageContent = ({
const groupRequestIdRef = useRef(0);
const stackRequestIdRef = useRef(0);
+ const [autoRefresh, setAutoRefresh] = useState(false);
+ const silentRefreshRef = useRef(false);
+ const [refreshKey, setRefreshKey] = useState(0);
+ const triggerRefresh = useCallback((silent: boolean) => {
+ silentRefreshRef.current = silent;
+ setRefreshKey((key) => key + 1);
+ }, []);
+
useEffect(() => {
if (!selectedInstanceId) {
groupRequestIdRef.current += 1;
@@ -236,9 +247,11 @@ const ConsumerPageContent = ({
window.clearTimeout(resetTimer);
};
}
+ const silent = silentRefreshRef.current;
+ silentRefreshRef.current = false;
const requestId = ++groupRequestIdRef.current;
const timer = window.setTimeout(() => {
- setLoading(true);
+ if (!silent) setLoading(true);
void listConsumerGroupPage({
instanceId: selectedInstanceId,
search: search.trim() || undefined,
@@ -261,7 +274,15 @@ const ConsumerPageContent = ({
return () => {
window.clearTimeout(timer);
};
- }, [t, selectedInstanceId, search, page, pageSize, instancesLoading]);
+ }, [t, selectedInstanceId, search, page, pageSize, instancesLoading,
refreshKey]);
+
+ useEffect(() => {
+ if (!autoRefresh || !selectedInstanceId) {
+ return undefined;
+ }
+ const interval = window.setInterval(() => triggerRefresh(true), 2000);
+ return () => window.clearInterval(interval);
+ }, [autoRefresh, selectedInstanceId, triggerRefresh]);
const loadSubscriptions = useCallback(
async (groupName: string, force = false) => {
@@ -317,9 +338,13 @@ const ConsumerPageContent = ({
}, [groups, modeFilter, sortKey]);
/* ─── Open detail modal ─── */
- const openModal = (group: ConsumerGroup) => {
+ const [detailTab, setDetailTab] = useState('overview');
+ const [progressTopic, setProgressTopic] = useState<string |
undefined>(undefined);
+ const openModal = (group: ConsumerGroup, tab = 'overview', topic?: string)
=> {
setSelectedGroup(group);
setShowOnlyInconsistent(false);
+ setDetailTab(tab);
+ setProgressTopic(topic);
setModalOpen(true);
void loadSubscriptions(group.name);
void loadProgress(group.name);
@@ -350,6 +375,24 @@ const ConsumerPageContent = ({
? inconsistentSubscriptions
: selectedSubscriptions;
const selectedProgress = selectedGroup ?
(progressByGroup[selectedDiagnosticKey] ?? []) : [];
+ const progressTopicOptions = useMemo(
+ () => Array.from(new Set(selectedProgress.map((q) =>
q.topic).filter(Boolean))).sort(),
+ [selectedProgress],
+ );
+ const visibleProgress = useMemo(() => {
+ const base =
+ progressTopic && progressTopicOptions.includes(progressTopic)
+ ? selectedProgress.filter((q) => q.topic === progressTopic)
+ : selectedProgress;
+ return [...base].sort((a, b) => {
+ const byTopic = (a.topic ?? '').localeCompare(b.topic ?? '');
+ if (byTopic !== 0) return byTopic;
+ const byBroker = (a.broker ?? '').localeCompare(b.broker ?? '');
+ if (byBroker !== 0) return byBroker;
+ return (a.queueId ?? 0) - (b.queueId ?? 0);
+ });
+ }, [selectedProgress, progressTopic, progressTopicOptions]);
+ const visibleProgressLag = visibleProgress.reduce((sum, q) => sum +
(q.diffTotal ?? 0), 0);
const openStackModal = async (consumerInstance: ConsumerInstance) => {
if (!selectedGroup) return;
@@ -357,6 +400,7 @@ const ConsumerPageContent = ({
const groupName = selectedGroup.name;
setSelectedStackClient(consumerInstance);
setSelectedStack(null);
+ setStackError(null);
setStackModalOpen(true);
setStackLoading(true);
try {
@@ -366,9 +410,11 @@ const ConsumerPageContent = ({
selectedInstanceId || undefined,
);
if (requestId === stackRequestIdRef.current) setSelectedStack(stack);
- } catch {
+ } catch (error) {
+ // The API client already surfaces the server message as a toast; keep
the reason in the
+ // modal so the operator can tell "capture unsupported" apart from
"client went offline".
if (requestId === stackRequestIdRef.current) {
- message.error(`客户端 ${consumerInstance.clientId} 线程栈获取失败`);
+ setStackError(error instanceof Error ? error.message : '');
}
} finally {
if (requestId === stackRequestIdRef.current) setStackLoading(false);
@@ -479,19 +525,49 @@ const ConsumerPageContent = ({
title: 'Group 名称',
dataIndex: 'name',
key: 'name',
- width: 220,
+ width: 190,
sorter: (a, b) => a.name.localeCompare(b.name),
render: (name: string) => (
- <Text strong style={{ fontSize: 14 }}>
- {name}
- </Text>
+ <Tooltip title="点击复制名称">
+ <Text
+ strong
+ style={{ fontSize: 14, cursor: 'pointer' }}
+ onClick={() => {
+ const done = () => message.success(`已复制：${name}`);
+ const failed = () => message.error('复制失败，请手动复制');
+ if (navigator.clipboard?.writeText) {
+ navigator.clipboard.writeText(name).then(done, failed);
+ } else {
+ const textarea = document.createElement('textarea');
+ textarea.value = name;
+ textarea.style.position = 'fixed';
+ textarea.style.opacity = '0';
+ document.body.appendChild(textarea);
+ textarea.select();
+ try {
+ if (document.execCommand('copy')) {
+ done();
+ } else {
+ failed();
+ }
+ } catch {
+ failed();
+ } finally {
+ document.body.removeChild(textarea);
+ }
+ }
+ }}
+ >
+ {name}
+ </Text>
+ </Tooltip>
),
},
{
title: '订阅组类型',
dataIndex: 'subscriptionDataType',
key: 'subscriptionDataType',
- width: 110,
+ width: 100,
sorter: (a, b) => (a.subscriptionDataType ??
'').localeCompare(b.subscriptionDataType ?? ''),
render: (type: string) => {
const config = TOPIC_TYPE_MAP[type] || { labelKey: type, color:
'default' };
@@ -502,7 +578,7 @@ const ConsumerPageContent = ({
title: '订阅模式',
dataIndex: 'subscriptionMode',
key: 'subscriptionMode',
- width: 90,
+ width: 84,
sorter: (a, b) => (a.subscriptionMode ??
'').localeCompare(b.subscriptionMode ?? ''),
render: (mode: string) => <Tag color={mode === 'Push' ? 'blue' :
'green'}>{mode}</Tag>,
},
@@ -510,7 +586,7 @@ const ConsumerPageContent = ({
title: '在线客户端',
dataIndex: 'onlineInstances',
key: 'onlineInstances',
- width: 130,
+ width: 100,
align: 'center',
sorter: (a, b) => (a.onlineInstances ?? 0) - (b.onlineInstances ?? 0),
},
@@ -518,7 +594,8 @@ const ConsumerPageContent = ({
title: '总堆积量',
dataIndex: 'totalLag',
key: 'totalLag',
- width: 120,
+ width: 96,
+ align: 'right',
sorter: (a, b) => (a.totalLag ?? 0) - (b.totalLag ?? 0),
render: (lag: number) => (lag ?? 0).toLocaleString(),
},
@@ -526,7 +603,8 @@ const ConsumerPageContent = ({
title: '消费延迟',
dataIndex: 'delaySeconds',
key: 'delaySeconds',
- width: 160,
+ width: 100,
+ align: 'right',
sorter: (a, b) => (a.delaySeconds ?? 0) - (b.delaySeconds ?? 0),
render: (seconds: number) => formatDelay(seconds ?? 0),
},
@@ -534,7 +612,7 @@ const ConsumerPageContent = ({
title: '创建时间',
dataIndex: 'gmtCreate',
key: 'gmtCreate',
- width: 170,
+ width: 156,
sorter: (a, b) => (a.gmtCreate ?? '').localeCompare(b.gmtCreate ?? ''),
render: (d: string) => (
<Text type="secondary" style={{ fontSize: 14 }}>
@@ -546,7 +624,7 @@ const ConsumerPageContent = ({
title: '修改时间',
dataIndex: 'gmtModified',
key: 'gmtModified',
- width: 170,
+ width: 156,
sorter: (a, b) => (a.gmtModified ?? '').localeCompare(b.gmtModified ??
''),
render: (d: string) => (
<Text type="secondary" style={{ fontSize: 14 }}>
@@ -557,7 +635,7 @@ const ConsumerPageContent = ({
{
title: '操作',
key: 'actions',
- width: 240,
+ width: 210,
render: (_: unknown, record: ConsumerGroup) => (
<Flex gap={6}>
<Button
@@ -617,7 +695,7 @@ const ConsumerPageContent = ({
/* ═══════════════════════════════════════════
Expandable Sub-table: Subscription Details
═══════════════════════════════════════════ */
- const subscriptionSubColumns: ColumnsType<SubscriptionEntry> = [
+ const subscriptionSubColumns = (groupName: string):
ColumnsType<SubscriptionEntry> => [
{
title: 'Topic 主题',
dataIndex: 'topic',
@@ -673,13 +751,16 @@ const ConsumerPageContent = ({
title: '',
key: 'action',
width: 100,
- render: () => (
+ render: (_: unknown, record: SubscriptionEntry) => (
<Button
size="small"
icon={<Eye size={14} />}
- disabled
- title="队列分布暂未支持"
+ title="查看该 Topic 的队列分布"
style={{ borderColor: '#1677ff', color: '#1677ff' }}
+ onClick={() => {
+ const group = groups.find((g) => g.name === groupName) ??
selectedGroup;
+ if (group) openModal(group, 'progress', record.topic);
+ }}
>
查看分布
</Button>
@@ -695,7 +776,7 @@ const ConsumerPageContent = ({
title: 'Client ID',
dataIndex: 'clientId',
key: 'clientId',
- width: 220,
+ width: 210,
render: (id: string) => (
<Text copyable style={{ fontSize: 14 }}>
{id}
@@ -706,7 +787,7 @@ const ConsumerPageContent = ({
title: '协议',
dataIndex: 'protocol',
key: 'protocol',
- width: 100,
+ width: 80,
render: (protocol: string) => {
const config = PROTOCOL_MAP[protocol] || { labelKey: protocol, color:
'default' };
return <Tag color={config.color}>{t(config.labelKey)}</Tag>;
@@ -716,7 +797,7 @@ const ConsumerPageContent = ({
title: '地址',
dataIndex: 'address',
key: 'address',
- width: 180,
+ width: 150,
render: (addr: string) => (
<Text code style={{ fontSize: 14 }}>
{addr}
@@ -727,7 +808,7 @@ const ConsumerPageContent = ({
title: '最后心跳',
dataIndex: 'lastHeartbeat',
key: 'lastHeartbeat',
- width: 170,
+ width: 150,
render: (time: string) => (
<Text type="secondary" style={{ fontSize: 14 }}>
{formatDateTime(time)}
@@ -737,7 +818,7 @@ const ConsumerPageContent = ({
{
title: '诊断',
key: 'diagnostics',
- width: 110,
+ width: 90,
render: (_: unknown, record: ConsumerInstance) => (
<Button
size="small"
@@ -754,6 +835,18 @@ const ConsumerPageContent = ({
Modal: Queue Progress Tab
═══════════════════════════════════════════ */
const queueColumns: ColumnsType<QueueProgress> = [
+ {
+ title: 'Topic 主题',
+ dataIndex: 'topic',
+ key: 'topic',
+ width: 280,
+ ellipsis: true,
+ render: (topic: string) => (
+ <Text strong style={{ fontSize: 14 }} title={topic || '-'}>
+ {topic || '-'}
+ </Text>
+ ),
+ },
{
title: 'Broker',
dataIndex: 'broker',
@@ -940,6 +1033,21 @@ const ConsumerPageContent = ({
>
创建 Group
</Button>
+ <Tooltip title="开启后每 2 秒自动刷新列表">
+ <Button
+ icon={<SyncOutlined spin={autoRefresh} />}
+ type={autoRefresh ? 'primary' : 'default'}
+ ghost={autoRefresh}
+ disabled={!hasSelectedInstance}
+ onClick={() => {
+ const next = !autoRefresh;
+ setAutoRefresh(next);
+ if (next) triggerRefresh(true);
+ }}
+ >
+ 自动刷新
+ </Button>
+ </Tooltip>
</Space>
</Flex>
@@ -967,6 +1075,7 @@ const ConsumerPageContent = ({
},
}}
size="small"
+ scroll={{ x: tableScrollX(columns, { selection: true, expandable:
true }) }}
expandable={{
onExpand: (expanded, record) => {
if (expanded) void loadSubscriptions(record.name);
@@ -974,7 +1083,7 @@ const ConsumerPageContent = ({
expandedRowRender: (record) => (
<div style={{ padding: '8px 0' }}>
<Table
- columns={subscriptionSubColumns}
+ columns={subscriptionSubColumns(record.name)}
dataSource={
subscriptionsByGroup[diagnosticCacheKey(selectedInstanceId, record.name)] ?? []
}
@@ -1011,13 +1120,14 @@ const ConsumerPageContent = ({
setSelectedGroup(null);
setShowOnlyInconsistent(false);
}}
- width={800}
+ width={detailTab === 'progress' ? 1080 : 800}
destroyOnHidden
footer={null}
>
{selectedGroup && (
<Tabs
- defaultActiveKey="overview"
+ activeKey={detailTab}
+ onChange={setDetailTab}
items={[
/* ─── 概览 Tab ─── */
{
@@ -1145,6 +1255,24 @@ const ConsumerPageContent = ({
</Descriptions.Item>
</Descriptions>
+ {/* 在线实例 */}
+ <div style={{ marginTop: 24 }}>
+ <Flex align="center" gap={6} style={{ marginBottom: 12
}}>
+ <Users size={15} color="#52c41a" />
+ <Text strong style={{ fontSize: 14 }}>
+ 在线实例 ({(selectedGroup.instances ?? []).length})
+ </Text>
+ </Flex>
+ <Table
+ columns={instanceColumns}
+ dataSource={selectedGroup.instances ?? []}
+ rowKey="clientId"
+ pagination={false}
+ size="small"
+ scroll={{ x: tableScrollX(instanceColumns) }}
+ />
+ </div>
+
{/* 订阅关系 */}
<div style={{ marginTop: 24 }}>
<Flex justify="space-between" align="center" style={{
marginBottom: 12 }}>
@@ -1204,7 +1332,7 @@ const ConsumerPageContent = ({
style={{ marginBottom: 12 }}
/>
<Table
- columns={subscriptionSubColumns}
+ columns={subscriptionSubColumns(selectedGroup?.name ??
'')}
dataSource={visibleSubscriptions}
rowKey={(record) =>
`${record.topic}-${record.filterMode}-${record.expression}`
@@ -1217,26 +1345,6 @@ const ConsumerPageContent = ({
</div>
),
},
- /* ─── 在线实例 Tab ─── */
- {
- key: 'instances',
- label: (
- <Space size={4}>
- <Users size={14} />
- <span>在线实例 ({(selectedGroup.instances ??
[]).length})</span>
- </Space>
- ),
- children: (
- <Table
- columns={instanceColumns}
- dataSource={selectedGroup.instances ?? []}
- rowKey="clientId"
- pagination={false}
- size="small"
- scroll={{ y: 400 }}
- />
- ),
- },
/* ─── 消费进度 Tab ─── */
{
key: 'progress',
@@ -1248,6 +1356,27 @@ const ConsumerPageContent = ({
),
children: (
<div>
+ {progressTopicOptions.length > 0 && (
+ <Flex align="center" gap={8} style={{ marginBottom: 12
}}>
+ <Text type="secondary">Topic 筛选:</Text>
+ <Select
+ size="small"
+ style={{ minWidth: 240 }}
+ allowClear
+ placeholder="全部 Topic"
+ value={
+ progressTopic &&
progressTopicOptions.includes(progressTopic)
+ ? progressTopic
+ : undefined
+ }
+ onChange={(value) => setProgressTopic(value)}
+ options={progressTopicOptions.map((topic) => ({
+ label: topic,
+ value: topic,
+ }))}
+ />
+ </Flex>
+ )}
<Card
size="small"
style={{
@@ -1260,21 +1389,21 @@ const ConsumerPageContent = ({
<Space size={24}>
<Space size={4}>
<Text type="secondary">总 Broker 数:</Text>
- <Text strong>{new Set(selectedProgress.map((q) =>
q.broker)).size}</Text>
+ <Text strong>{new Set(visibleProgress.map((q) =>
q.broker)).size}</Text>
</Space>
<Space size={4}>
<Text type="secondary">总 Queue 数:</Text>
- <Text strong>{selectedProgress.length}</Text>
+ <Text strong>{visibleProgress.length}</Text>
</Space>
<Space size={4}>
<Text type="secondary">总堆积:</Text>
<Text
strong
style={{
- color: lagColor(selectedGroup.totalLag),
+ color: lagColor(visibleProgressLag),
}}
>
- {selectedGroup.totalLag.toLocaleString()}
+ {visibleProgressLag.toLocaleString()}
</Text>
</Space>
</Space>
@@ -1282,11 +1411,12 @@ const ConsumerPageContent = ({
<Table
columns={queueColumns}
- dataSource={selectedProgress}
- rowKey={(r) => `${r.broker}-${r.queueId}`}
+ dataSource={visibleProgress}
+ rowKey={(r) => `${r.topic}-${r.broker}-${r.queueId}`}
pagination={false}
size="small"
- scroll={{ y: 380 }}
+ scroll={{ x: tableScrollX(queueColumns), y: 380 }}
+ locale={{ emptyText: '消费组不在线，暂无队列进度数据' }}
/>
</div>
),
@@ -1311,6 +1441,7 @@ const ConsumerPageContent = ({
stackRequestIdRef.current += 1;
setStackModalOpen(false);
setSelectedStack(null);
+ setStackError(null);
setSelectedStackClient(null);
}}
footer={null}
@@ -1375,8 +1506,19 @@ const ConsumerPageContent = ({
<Alert
type="info"
showIcon
- message="暂无线程栈数据"
- description="客户端在线但没有返回可展示的 jstack 内容，或该客户端暂时无法采集线程信息。"
+ message="暂不支持采集该客户端的线程栈"
+ description={
+ <>
+ <div>
+ 经 Proxy 接入的客户端（gRPC、经 Proxy 的 Remoting）只在 Proxy
侧保持连接，Broker
+ 看不到它们；而 Proxy 目前未开放线程栈采集接口，因此这类客户端暂时无法采集。 直连
+ Broker 的客户端可正常查看。
+ </div>
+ {stackError && (
+ <div style={{ marginTop: 8, color: 'rgba(0,0,0,0.45)'
}}>{stackError}</div>
+ )}
+ </>
+ }
/>
)}
</Space>
diff --git a/web/src/pages/studio/__tests__/GroupManagement.test.tsx
b/web/src/pages/studio/__tests__/GroupManagement.test.tsx
index 2e2d1d9c3..a42c6d804 100644
--- a/web/src/pages/studio/__tests__/GroupManagement.test.tsx
+++ b/web/src/pages/studio/__tests__/GroupManagement.test.tsx
@@ -225,6 +225,7 @@ describe('GroupManagement Page', () => {
vi.mocked(consumerService.getConsumerSubscriptions).mockRejectedValue(new
Error('unavailable'));
vi.mocked(consumerService.getConsumerProgress).mockResolvedValue([
{
+ topic: 'orders',
broker: 'broker-a',
queueId: 0,
brokerOffset: 20,
