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 3398fcce fix: route consumer group operations by selected instance
(#1130)
3398fcce is described below
commit 3398fccecc6c003b70a4e2fa4d64ee48060169a6
Author: aias00 <[email protected]>
AuthorDate: Fri Aug 7 00:13:05 2026 -0700
fix: route consumer group operations by selected instance (#1130)
* fix: scope group diagnostics to selected instance
* fix: scope offset resets to selected instance
* fix: propagate instance context for group diagnostics
* fix: route consumer group creation by instance
* fix: route consumer group deletion by instance
* fix: route consumer group detail by instance
---
.../instance/group/ResetConsumerOffsetDTO.java | 5 +-
.../studio/instance/topic/MetadataService.java | 2 +-
.../studio/instance/topic/NameSrvAdminClient.java | 6 +-
.../studio/provider/apache/AdminClient.java | 6 +-
.../provider/apache/ApacheInstanceProvider.java | 8 +-
.../studio/provider/apache/MetadataProvider.java | 4 +-
.../provider/apache/RocketMQAdminClientImpl.java | 241 ++++++++++++---------
.../provider/apache/RocketMQMetadataProvider.java | 114 +++++-----
.../group/ConsumerGroupControllerTest.java | 25 ++-
.../instance/topic/NameSrvAdminClientTest.java | 6 +-
.../apache/RocketMQAdminClientImplTest.java | 94 +++++++-
.../apache/RocketMQMetadataProviderTest.java | 16 ++
web/src/api/consumerGroups.test.ts | 17 +-
web/src/api/metadata.ts | 2 +-
.../pages/instance/__tests__/ConsumerPage.test.tsx | 47 ++++
web/src/pages/instance/consumer.tsx | 10 +-
16 files changed, 412 insertions(+), 191 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/ResetConsumerOffsetDTO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ResetConsumerOffsetDTO.java
index 64111393..80ea5f75 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/ResetConsumerOffsetDTO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ResetConsumerOffsetDTO.java
@@ -29,6 +29,9 @@ import lombok.NoArgsConstructor;
@NoArgsConstructor
@AllArgsConstructor
public class ResetConsumerOffsetDTO {
+ @NotBlank(message = "instanceId is required")
+ private String instanceId;
+
@NotBlank(message = "name is required")
private String name;
@@ -37,6 +40,4 @@ public class ResetConsumerOffsetDTO {
private Long timestamp;
private String topic;
-
- private String instanceId;
}
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 fa7b1eae..d67ace2a 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
@@ -132,7 +132,7 @@ public class MetadataService {
if (resolve(instanceId).vendor() != InstanceVendor.APACHE) {
throw new BusinessException(501, "Consumer group detail is not
supported for cloud instances");
}
- return adminClient.getConsumerGroup(name);
+ return adminClient.getConsumerGroup(instanceId, name);
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/NameSrvAdminClient.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/NameSrvAdminClient.java
index f91765b2..c4de8986 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/NameSrvAdminClient.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/NameSrvAdminClient.java
@@ -34,7 +34,7 @@ public class NameSrvAdminClient implements AdminClient {
}
@Override
- public ConsumerGroupVO getConsumerGroup(String name) {
+ public ConsumerGroupVO getConsumerGroup(String instanceId, String name) {
throw consumerGroupAdminUnavailable();
}
@@ -64,12 +64,12 @@ public class NameSrvAdminClient implements AdminClient {
}
@Override
- public void deleteConsumerGroup(String name) {
+ public void deleteConsumerGroup(String instanceId, String name) {
throw consumerGroupAdminUnavailable();
}
@Override
- public void resetOffset(String name, long timestamp, String topic) {
+ public void resetOffset(String instanceId, String name, long timestamp,
String topic) {
throw consumerGroupAdminUnavailable();
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/AdminClient.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/AdminClient.java
index 90e958b9..bc394080 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/AdminClient.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/AdminClient.java
@@ -24,12 +24,12 @@ import
org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
public interface AdminClient {
TopicVO getTopic(String name);
- ConsumerGroupVO getConsumerGroup(String name);
+ ConsumerGroupVO getConsumerGroup(String instanceId, String name);
TopicVO createTopic(TopicVO topic);
TopicVO updateTopic(TopicVO topic);
void deleteTopic(String name);
SendMessageVO sendMessage(SendMessageDTO request);
ConsumerGroupVO createConsumerGroup(ConsumerGroupVO group);
- void deleteConsumerGroup(String name);
- void resetOffset(String name, long timestamp, String topic);
+ void deleteConsumerGroup(String instanceId, String name);
+ void resetOffset(String instanceId, String name, long timestamp, String
topic);
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
index 4915303a..665ac771 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
@@ -102,22 +102,22 @@ public class ApacheInstanceProvider implements
InstanceProvider {
@Override
public void deleteConsumerGroup(String instanceId, String groupName) {
- adminClient.deleteConsumerGroup(groupName);
+ adminClient.deleteConsumerGroup(instanceId, groupName);
}
@Override
public List<QueueProgressVO> getGroupProgress(String instanceId, String
groupName) {
- return metadataProvider.getGroupProgress(groupName);
+ return metadataProvider.getGroupProgress(instanceId, groupName);
}
@Override
public List<SubscriptionEntryVO> getGroupSubscriptions(String instanceId,
String groupName) {
- return metadataProvider.getGroupSubscriptions(groupName);
+ return metadataProvider.getGroupSubscriptions(instanceId, groupName);
}
@Override
public void resetOffset(String instanceId, String groupName, long
timestamp, String topic) {
- adminClient.resetOffset(groupName, timestamp, topic);
+ adminClient.resetOffset(instanceId, groupName, timestamp, topic);
}
@Override
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/MetadataProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/MetadataProvider.java
index 173e7c8a..9acfe40e 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/MetadataProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/MetadataProvider.java
@@ -30,6 +30,6 @@ public interface MetadataProvider {
List<ConsumerGroupVO> listConsumerGroups(String clusterId, String search);
List<BrokerRouteVO> getTopicRoutes(String instanceId, String name);
List<TopicConsumerVO> getTopicConsumers(String instanceId, String name);
- List<QueueProgressVO> getGroupProgress(String name);
- List<SubscriptionEntryVO> getGroupSubscriptions(String name);
+ List<QueueProgressVO> getGroupProgress(String instanceId, String name);
+ List<SubscriptionEntryVO> getGroupSubscriptions(String instanceId, String
name);
}
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 10da98ce..f6fc162c 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
@@ -26,6 +26,7 @@ import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import
org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
+import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.common.domain.enums.TopicPerm;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
@@ -69,6 +70,7 @@ public class RocketMQAdminClientImpl implements AdminClient {
private final RmqTopicMapper topicMapper;
private final RmqGroupMapper groupMapper;
private final AuditService auditService;
+ private final RuntimeAdminClientResolver runtimeAdminClientResolver;
@Override
public TopicVO getTopic(String name) {
@@ -94,32 +96,37 @@ public class RocketMQAdminClientImpl implements AdminClient
{
}
@Override
- public ConsumerGroupVO getConsumerGroup(String name) {
- return adminFactory.execute(namesrvAddr(), null, admin -> {
- ConsumerGroupVO vo = new ConsumerGroupVO();
- vo.setId(name);
- vo.setName(name);
- try {
- var conn = admin.examineConsumerConnectionInfo(name);
- if (conn != null) {
- if (conn.getConnectionSet() != null) {
- vo.setOnlineInstances(conn.getConnectionSet().size());
- }
- if (conn.getSubscriptionTable() != null) {
- vo.setSubscribedTopics(new
ArrayList<>(conn.getSubscriptionTable().keySet()));
- }
+ public ConsumerGroupVO getConsumerGroup(String instanceId, String name) {
+ if (StringUtils.hasText(instanceId)) {
+ return runtimeAdminClientResolver.execute(instanceId, admin ->
getConsumerGroup(admin, name));
+ }
+ return adminFactory.execute(namesrvAddr(), null, admin ->
getConsumerGroup(admin, name));
+ }
+
+ private ConsumerGroupVO getConsumerGroup(MQAdminExt admin, String name) {
+ ConsumerGroupVO vo = new ConsumerGroupVO();
+ vo.setId(name);
+ vo.setName(name);
+ try {
+ var conn = admin.examineConsumerConnectionInfo(name);
+ if (conn != null) {
+ if (conn.getConnectionSet() != null) {
+ vo.setOnlineInstances(conn.getConnectionSet().size());
}
- } catch (MQClientException exception) {
- if (exception.getResponseCode() ==
ResponseCode.CONSUMER_NOT_ONLINE) {
- log.debug("Consumer group {} is offline", name);
- return vo;
+ if (conn.getSubscriptionTable() != null) {
+ vo.setSubscribedTopics(new
ArrayList<>(conn.getSubscriptionTable().keySet()));
}
- 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());
}
- return vo;
- });
+ } 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());
+ }
+ return vo;
}
@Override
@@ -355,106 +362,130 @@ public class RocketMQAdminClientImpl implements
AdminClient {
@Override
public ConsumerGroupVO createConsumerGroup(ConsumerGroupVO group) {
+ if (group != null && StringUtils.hasText(group.getInstanceId())) {
+ return runtimeAdminClientResolver.execute(group.getInstanceId(),
+ admin -> createConsumerGroup(admin, group));
+ }
+ return adminFactory.execute(namesrvAddr(), null, admin ->
createConsumerGroup(admin, group));
+ }
+
+ private ConsumerGroupVO createConsumerGroup(MQAdminExt admin,
ConsumerGroupVO group) {
String groupName = group.getName();
- return adminFactory.execute(namesrvAddr(), null, admin -> {
- try {
- Set<String> brokerAddrs = getAllMasterBrokerAddrs(admin);
- if (brokerAddrs.isEmpty()) {
- throw new BusinessException(500, "No broker available to
create consumer group");
- }
+ try {
+ Set<String> brokerAddrs = getAllMasterBrokerAddrs(admin);
+ if (brokerAddrs.isEmpty()) {
+ throw new BusinessException(500, "No broker available to
create consumer group");
+ }
- SubscriptionGroupConfig config = new SubscriptionGroupConfig();
- config.setGroupName(groupName);
- config.setConsumeEnable(true);
- config.setConsumeBroadcastEnable(true);
- config.setRetryQueueNums(1);
- config.setRetryMaxTimes(group.getRetryMaxTimes() > 0 ?
group.getRetryMaxTimes() : 16);
+ SubscriptionGroupConfig config = new SubscriptionGroupConfig();
+ config.setGroupName(groupName);
+ config.setConsumeEnable(true);
+ config.setConsumeBroadcastEnable(true);
+ config.setRetryQueueNums(1);
+ config.setRetryMaxTimes(group.getRetryMaxTimes() > 0 ?
group.getRetryMaxTimes() : 16);
- for (String addr : brokerAddrs) {
- admin.createAndUpdateSubscriptionGroupConfig(addr, config);
- }
+ for (String addr : brokerAddrs) {
+ admin.createAndUpdateSubscriptionGroupConfig(addr, config);
+ }
- // Persist to DB, upserting so re-creating an existing group
does not violate the
- // unique (cluster_id, name) key.
- String groupClusterName = getClusterName(admin);
- RmqGroup entity = groupMapper.selectOne(new
LambdaQueryWrapper<RmqGroup>()
- .eq(RmqGroup::getClusterId, groupClusterName)
- .eq(RmqGroup::getName, groupName));
- boolean isNewGroup = entity == null;
- if (isNewGroup) {
- entity = new RmqGroup();
- entity.setName(groupName);
- entity.setClusterId(groupClusterName);
- entity.setCreatedAt(LocalDateTime.now());
- }
- if (StringUtils.hasText(group.getInstanceId())) {
- entity.setInstanceId(group.getInstanceId());
- }
- entity.setConsumeType(group.getConsumeType() != null ?
group.getConsumeType().name() : "CLUSTERING");
- entity.setMessageModel(group.getSubscriptionMode() != null ?
group.getSubscriptionMode().name() : "Push");
- entity.setMaxRetry(config.getRetryMaxTimes());
- entity.setStatus("ACTIVE");
- entity.setUpdatedAt(LocalDateTime.now());
- if (isNewGroup) {
- groupMapper.insert(entity);
- } else {
- groupMapper.updateById(entity);
- }
+ // Persist to DB, upserting so re-creating an existing group does
not violate the
+ // unique (cluster_id, name) key.
+ String groupClusterName = getClusterName(admin);
+ RmqGroup entity = groupMapper.selectOne(new
LambdaQueryWrapper<RmqGroup>()
+ .eq(RmqGroup::getClusterId, groupClusterName)
+ .eq(RmqGroup::getName, groupName));
+ boolean isNewGroup = entity == null;
+ if (isNewGroup) {
+ entity = new RmqGroup();
+ entity.setName(groupName);
+ entity.setClusterId(groupClusterName);
+ entity.setCreatedAt(LocalDateTime.now());
+ }
+ if (StringUtils.hasText(group.getInstanceId())) {
+ entity.setInstanceId(group.getInstanceId());
+ }
+ entity.setConsumeType(group.getConsumeType() != null ?
group.getConsumeType().name() : "CLUSTERING");
+ entity.setMessageModel(group.getSubscriptionMode() != null ?
group.getSubscriptionMode().name() : "Push");
+ entity.setMaxRetry(config.getRetryMaxTimes());
+ entity.setStatus("ACTIVE");
+ entity.setUpdatedAt(LocalDateTime.now());
+ if (isNewGroup) {
+ groupMapper.insert(entity);
+ } else {
+ groupMapper.updateById(entity);
+ }
- auditService.record("CREATE_GROUP", groupName,
- "retryMaxTimes=" + config.getRetryMaxTimes(),
"SUCCESS");
+ auditService.record("CREATE_GROUP", groupName,
+ "retryMaxTimes=" + config.getRetryMaxTimes(), "SUCCESS");
- group.setId(groupName);
- return group;
- } catch (BusinessException e) {
- auditService.record("CREATE_GROUP", groupName, e.getMessage(),
"FAILED");
- throw e;
- } catch (Exception e) {
- auditService.record("CREATE_GROUP", groupName, e.getMessage(),
"FAILED");
- throw new BusinessException(500, "Failed to create consumer
group: " + e.getMessage());
- }
- });
+ group.setId(groupName);
+ return group;
+ } catch (BusinessException e) {
+ auditService.record("CREATE_GROUP", groupName, e.getMessage(),
"FAILED");
+ throw e;
+ } catch (Exception e) {
+ auditService.record("CREATE_GROUP", groupName, e.getMessage(),
"FAILED");
+ throw new BusinessException(500, "Failed to create consumer group:
" + e.getMessage());
+ }
}
@Override
- public void deleteConsumerGroup(String name) {
+ public void deleteConsumerGroup(String instanceId, String name) {
+ if (StringUtils.hasText(instanceId)) {
+ runtimeAdminClientResolver.execute(instanceId, admin -> {
+ doDeleteConsumerGroup(admin, name);
+ return null;
+ });
+ return;
+ }
adminFactory.execute(namesrvAddr(), null, admin -> {
- try {
- Set<String> brokerAddrs = getAllMasterBrokerAddrs(admin);
-
- for (String addr : brokerAddrs) {
- admin.deleteSubscriptionGroup(addr, name, true);
- }
+ doDeleteConsumerGroup(admin, name);
+ return null;
+ });
+ }
- // Delete from DB
- groupMapper.delete(new
LambdaQueryWrapper<RmqGroup>().eq(RmqGroup::getName, name));
+ private void doDeleteConsumerGroup(MQAdminExt admin, String name) {
+ try {
+ Set<String> brokerAddrs = getAllMasterBrokerAddrs(admin);
- auditService.record("DELETE_GROUP", name, "", "SUCCESS");
- return null;
- } catch (BusinessException e) {
- auditService.record("DELETE_GROUP", name, e.getMessage(),
"FAILED");
- throw e;
- } catch (Exception e) {
- auditService.record("DELETE_GROUP", name, e.getMessage(),
"FAILED");
- throw new BusinessException(500, "Failed to delete consumer
group: " + e.getMessage());
+ for (String addr : brokerAddrs) {
+ admin.deleteSubscriptionGroup(addr, name, true);
}
- });
+
+ // Delete from DB
+ groupMapper.delete(new
LambdaQueryWrapper<RmqGroup>().eq(RmqGroup::getName, name));
+
+ auditService.record("DELETE_GROUP", name, "", "SUCCESS");
+ } catch (BusinessException e) {
+ auditService.record("DELETE_GROUP", name, e.getMessage(),
"FAILED");
+ throw e;
+ } catch (Exception e) {
+ auditService.record("DELETE_GROUP", name, e.getMessage(),
"FAILED");
+ throw new BusinessException(500, "Failed to delete consumer group:
" + e.getMessage());
+ }
}
@Override
- public void resetOffset(String name, long timestamp, String topic) {
- adminFactory.execute(namesrvAddr(), null, admin -> {
- try {
- admin.resetOffsetByTimestamp(getClusterName(admin), topic,
name, timestamp, false);
- auditService.record("RESET_OFFSET", name,
- "topic=" + topic + ", timestamp=" + timestamp,
"SUCCESS");
- return null;
- } catch (Exception e) {
- auditService.record("RESET_OFFSET", name, e.getMessage(),
"FAILED");
- throw new BusinessException(500, "Failed to reset offset: " +
e.getMessage());
+ public void resetOffset(String instanceId, String name, long timestamp,
String topic) {
+ try {
+ if (StringUtils.hasText(instanceId)) {
+ runtimeAdminClientResolver.execute(instanceId, admin -> {
+ admin.resetOffsetByTimestamp(getClusterName(admin), topic,
name, timestamp, false);
+ return null;
+ });
+ } else {
+ adminFactory.execute(namesrvAddr(), null, admin -> {
+ admin.resetOffsetByTimestamp(getClusterName(admin), topic,
name, timestamp, false);
+ return null;
+ });
}
- });
+ auditService.record("RESET_OFFSET", name,
+ "instanceId=" + instanceId + ", topic=" + topic + ",
timestamp=" + timestamp, "SUCCESS");
+ } catch (Exception e) {
+ auditService.record("RESET_OFFSET", name, e.getMessage(),
"FAILED");
+ throw new BusinessException(500, "Failed to reset offset: " +
e.getMessage());
+ }
}
// ── Helper methods ──────────────────────────────────────────────────
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 b17703bb..06d43089 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
@@ -337,74 +337,82 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
}
@Override
- public List<QueueProgressVO> getGroupProgress(String name) {
+ public List<QueueProgressVO> getGroupProgress(String instanceId, String
name) {
+ if (StringUtils.hasText(instanceId)) {
+ return runtimeAdminClientResolver.execute(instanceId, admin ->
getGroupProgress(admin, name));
+ }
if (!hasAdmin()) {
return Collections.emptyList();
}
+ return adminExecute(admin -> getGroupProgress(admin, name));
+ }
- return adminExecute(admin -> {
- try {
- ConsumeStats stats = admin.examineConsumeStats(name);
- if (stats == null || stats.getOffsetTable() == null) {
- return Collections.emptyList();
- }
-
- List<QueueProgressVO> progress = new ArrayList<>();
- for (Map.Entry<MessageQueue, OffsetWrapper> entry :
stats.getOffsetTable().entrySet()) {
- MessageQueue mq = entry.getKey();
- OffsetWrapper ow = entry.getValue();
- long diff = Math.max(0, ow.getBrokerOffset() -
ow.getConsumerOffset());
-
- progress.add(QueueProgressVO.builder()
- .broker(mq.getBrokerName())
- .queueId(mq.getQueueId())
- .brokerOffset(ow.getBrokerOffset())
- .consumerOffset(ow.getConsumerOffset())
- .diffTotal(diff)
- .build());
- }
-
- progress.sort((a, b) -> {
- int cmp = a.getBroker().compareToIgnoreCase(b.getBroker());
- return cmp != 0 ? cmp : Integer.compare(a.getQueueId(),
b.getQueueId());
- });
- return progress;
- } catch (Exception e) {
- log.warn("Failed to get progress for group {}: {}", name,
e.getMessage());
+ private List<QueueProgressVO> getGroupProgress(MQAdminExt admin, String
name) {
+ try {
+ ConsumeStats stats = admin.examineConsumeStats(name);
+ if (stats == null || stats.getOffsetTable() == null) {
return Collections.emptyList();
}
- });
+
+ List<QueueProgressVO> progress = new ArrayList<>();
+ for (Map.Entry<MessageQueue, OffsetWrapper> entry :
stats.getOffsetTable().entrySet()) {
+ MessageQueue mq = entry.getKey();
+ OffsetWrapper ow = entry.getValue();
+ long diff = Math.max(0, ow.getBrokerOffset() -
ow.getConsumerOffset());
+
+ progress.add(QueueProgressVO.builder()
+ .broker(mq.getBrokerName())
+ .queueId(mq.getQueueId())
+ .brokerOffset(ow.getBrokerOffset())
+ .consumerOffset(ow.getConsumerOffset())
+ .diffTotal(diff)
+ .build());
+ }
+
+ progress.sort((a, b) -> {
+ int cmp = a.getBroker().compareToIgnoreCase(b.getBroker());
+ return cmp != 0 ? cmp : Integer.compare(a.getQueueId(),
b.getQueueId());
+ });
+ return progress;
+ } catch (Exception e) {
+ log.warn("Failed to get progress for group {}: {}", name,
e.getMessage());
+ return Collections.emptyList();
+ }
}
@Override
- public List<SubscriptionEntryVO> getGroupSubscriptions(String name) {
+ public List<SubscriptionEntryVO> getGroupSubscriptions(String instanceId,
String name) {
+ if (StringUtils.hasText(instanceId)) {
+ return runtimeAdminClientResolver.execute(instanceId, admin ->
getGroupSubscriptions(admin, name));
+ }
if (!hasAdmin()) {
return Collections.emptyList();
}
+ return adminExecute(admin -> getGroupSubscriptions(admin, name));
+ }
- return adminExecute(admin -> {
- try {
- ConsumerConnection conn =
admin.examineConsumerConnectionInfo(name);
- if (conn == null || conn.getSubscriptionTable() == null) {
- 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;
- } catch (Exception e) {
- log.warn("Failed to get subscriptions for group {}: {}", name,
e.getMessage());
+ private List<SubscriptionEntryVO> getGroupSubscriptions(MQAdminExt admin,
String name) {
+ try {
+ ConsumerConnection conn =
admin.examineConsumerConnectionInfo(name);
+ if (conn == null || conn.getSubscriptionTable() == null) {
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;
+ } catch (Exception e) {
+ log.warn("Failed to get subscriptions for group {}: {}", name,
e.getMessage());
+ return Collections.emptyList();
+ }
}
// ── Helper methods ──────────────────────────────────────────────────
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupControllerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupControllerTest.java
index 09b9f718..55b5fd67 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupControllerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupControllerTest.java
@@ -63,6 +63,7 @@ class ConsumerGroupControllerTest {
@Test
void createConsumerGroupShouldPassValidatedRequest() throws Exception {
Map<String, Object> body = Map.of(
+ "instanceId", "instance-a",
"name", "cg-orders",
"clusterId", "cluster-a",
"retryMaxTimes", 8,
@@ -110,6 +111,7 @@ class ConsumerGroupControllerTest {
@Test
void createConsumerGroupShouldRejectNegativeRetryMaxTimes() throws
Exception {
Map<String, Object> body = Map.of(
+ "instanceId", "instance-a",
"name", "cg-orders",
"retryMaxTimes", -1
);
@@ -155,9 +157,26 @@ class ConsumerGroupControllerTest {
.value("org.apache.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService.run"));
}
+ @Test
+ void groupRuntimeDiagnosticsShouldPassSelectedInstance() throws Exception {
+ when(metadataService.getGroupProgress("instance-a",
"cg-orders")).thenReturn(List.of());
+ when(metadataService.getGroupSubscriptions("instance-a",
"cg-orders")).thenReturn(List.of());
+
+
mockMvc.perform(get("/api/groups/cg-orders/progress").param("instanceId",
"instance-a"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.code").value(200));
+
mockMvc.perform(get("/api/groups/cg-orders/subscriptions").param("instanceId",
"instance-a"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.code").value(200));
+
+ verify(metadataService).getGroupProgress("instance-a", "cg-orders");
+ verify(metadataService).getGroupSubscriptions("instance-a",
"cg-orders");
+ }
+
@Test
void resetOffsetShouldPassValidatedRequest() throws Exception {
Map<String, Object> body = Map.of(
+ "instanceId", "instance-a",
"name", "cg-orders",
"topic", "orders",
"timestamp", 1784246400000L
@@ -170,7 +189,7 @@ class ConsumerGroupControllerTest {
.andExpect(jsonPath("$.code").value(200))
.andExpect(jsonPath("$.message").value("success"));
- verify(metadataService).resetOffset(isNull(), eq("cg-orders"),
eq(1784246400000L), eq("orders"));
+ verify(metadataService).resetOffset(eq("instance-a"), eq("cg-orders"),
eq(1784246400000L), eq("orders"));
}
@Test
@@ -212,6 +231,7 @@ class ConsumerGroupControllerTest {
@Test
void resetOffsetShouldRejectMissingName() throws Exception {
Map<String, Object> body = Map.of(
+ "instanceId", "instance-a",
"topic", "orders",
"timestamp", 1784246400000L
);
@@ -229,6 +249,7 @@ class ConsumerGroupControllerTest {
@Test
void resetOffsetShouldRejectMissingTimestamp() throws Exception {
Map<String, Object> body = Map.of(
+ "instanceId", "instance-a",
"name", "cg-orders",
"topic", "orders"
);
@@ -246,6 +267,7 @@ class ConsumerGroupControllerTest {
@Test
void resetOffsetShouldRejectNonPositiveTimestamp() throws Exception {
Map<String, Object> body = Map.of(
+ "instanceId", "instance-a",
"name", "cg-orders",
"topic", "orders",
"timestamp", 0L
@@ -264,6 +286,7 @@ class ConsumerGroupControllerTest {
@Test
void resetOffsetShouldRejectInvalidTimestampType() throws Exception {
Map<String, Object> body = Map.of(
+ "instanceId", "instance-a",
"name", "cg-orders",
"topic", "orders",
"timestamp", "invalid"
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/NameSrvAdminClientTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/NameSrvAdminClientTest.java
index 0b7f1928..5626d4e3 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/NameSrvAdminClientTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/NameSrvAdminClientTest.java
@@ -33,10 +33,10 @@ class NameSrvAdminClientTest {
ConsumerGroupVO group = new ConsumerGroupVO();
group.setName("cg-order");
- assertUnavailable(() -> adminClient.getConsumerGroup("cg-order"));
+ assertUnavailable(() -> adminClient.getConsumerGroup("instance-a",
"cg-order"));
assertUnavailable(() -> adminClient.createConsumerGroup(group));
- assertUnavailable(() -> adminClient.deleteConsumerGroup("cg-order"));
- assertUnavailable(() -> adminClient.resetOffset("cg-order",
1784246400000L, "order-topic"));
+ assertUnavailable(() -> adminClient.deleteConsumerGroup("instance-a",
"cg-order"));
+ assertUnavailable(() -> adminClient.resetOffset("instance-a",
"cg-order", 1784246400000L, "order-topic"));
}
private void assertUnavailable(ThrowableAssert.ThrowingCallable callable) {
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 4ac7f165..0ad5d522 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
@@ -26,11 +26,13 @@ import
org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
import org.apache.rocketmq.studio.instance.topic.TopicVO;
import org.apache.rocketmq.studio.instance.topic.SendMessageDTO;
import org.apache.rocketmq.studio.instance.topic.SendMessageVO;
import org.apache.rocketmq.studio.ops.audit.AuditService;
+import org.apache.rocketmq.studio.persistence.entity.RmqGroup;
import org.apache.rocketmq.studio.persistence.entity.RmqTopic;
import org.apache.rocketmq.studio.persistence.mapper.RmqGroupMapper;
import org.apache.rocketmq.studio.persistence.mapper.RmqTopicMapper;
@@ -59,6 +61,7 @@ import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
@@ -76,6 +79,8 @@ class RocketMQAdminClientImplTest {
private RmqGroupMapper groupMapper;
@Mock
private AuditService auditService;
+ @Mock
+ private RuntimeAdminClientResolver runtimeAdminClientResolver;
private RocketMQAdminClientImpl adminClient;
@@ -84,7 +89,8 @@ class RocketMQAdminClientImplTest {
lenient().when(properties.getNamesrvAddr()).thenReturn("10.0.0.1:9876");
lenient().when(adminFactory.execute(anyString(), any(),
any())).thenAnswer(invocation ->
invocation.<MqAdminExtFactory.AdminAction<Object>>getArgument(2).apply(adminExt));
- adminClient = new RocketMQAdminClientImpl(adminFactory, properties,
topicMapper, groupMapper, auditService);
+ adminClient = new RocketMQAdminClientImpl(adminFactory, properties,
topicMapper, groupMapper, auditService,
+ runtimeAdminClientResolver);
}
@Test
@@ -93,18 +99,44 @@ class RocketMQAdminClientImplTest {
.thenThrow(new
MQClientException(ResponseCode.CONSUMER_NOT_ONLINE,
"Not found the consumer group connection"));
- ConsumerGroupVO group = adminClient.getConsumerGroup("orders");
+ ConsumerGroupVO group = adminClient.getConsumerGroup(null, "orders");
assertThat(group.getId()).isEqualTo("orders");
assertThat(group.getOnlineInstances()).isZero();
}
+ @Test
+ void getConsumerGroupUsesSelectedInstanceAdmin() throws Exception {
+ DefaultMQAdminExt selectedAdmin =
org.mockito.Mockito.mock(DefaultMQAdminExt.class);
+
when(runtimeAdminClientResolver.execute(org.mockito.ArgumentMatchers.eq("instance-a"),
any()))
+ .thenAnswer(invocation -> {
+ MqAdminExtFactory.AdminAction<?> action =
invocation.getArgument(1);
+ return action.apply(selectedAdmin);
+ });
+
+ ConsumerGroupVO group = adminClient.getConsumerGroup("instance-a",
"cg-orders");
+
+ assertThat(group.getName()).isEqualTo("cg-orders");
+
verify(runtimeAdminClientResolver).execute(org.mockito.ArgumentMatchers.eq("instance-a"),
any());
+ verify(selectedAdmin).examineConsumerConnectionInfo("cg-orders");
+ verify(adminExt, never()).examineConsumerConnectionInfo(anyString());
+ }
+
+ @Test
+ void resetOffsetShouldUseSelectedInstanceRuntimeClient() {
+ adminClient.resetOffset("instance-a", "cg-orders", 1784246400000L,
"orders");
+
+
verify(runtimeAdminClientResolver).execute(org.mockito.ArgumentMatchers.eq("instance-a"),
any());
+ verify(auditService).record("RESET_OFFSET", "cg-orders",
+ "instanceId=instance-a, topic=orders,
timestamp=1784246400000", "SUCCESS");
+ }
+
@Test
void getConsumerGroupSurfacesAdminTimeout() throws Exception {
when(adminExt.examineConsumerConnectionInfo("orders"))
.thenThrow(new RemotingTimeoutException("broker-0", 3_000));
- assertThatThrownBy(() -> adminClient.getConsumerGroup("orders"))
+ assertThatThrownBy(() -> adminClient.getConsumerGroup(null, "orders"))
.isInstanceOf(BusinessException.class)
.hasMessageContaining("Failed to get consumer group");
}
@@ -114,7 +146,7 @@ class RocketMQAdminClientImplTest {
when(adminExt.examineConsumerConnectionInfo("orders"))
.thenThrow(new MQBrokerException(16, "ACL denied"));
- assertThatThrownBy(() -> adminClient.getConsumerGroup("orders"))
+ assertThatThrownBy(() -> adminClient.getConsumerGroup(null, "orders"))
.isInstanceOf(BusinessException.class)
.hasMessageContaining("ACL denied");
}
@@ -148,6 +180,60 @@ class RocketMQAdminClientImplTest {
}
}
+ @Test
+ void createConsumerGroupUsesSelectedInstanceAdmin() throws Exception {
+ TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new
MybatisConfiguration(), ""), RmqGroup.class);
+ DefaultMQAdminExt selectedAdmin =
org.mockito.Mockito.mock(DefaultMQAdminExt.class);
+ ClusterInfo clusterInfo = new ClusterInfo();
+ clusterInfo.setClusterAddrTable(new HashMap<>(Map.of("cluster-1", new
HashSet<>(List.of("broker-1")))));
+ BrokerData brokerData = new BrokerData();
+ brokerData.setBrokerName("broker-1");
+ brokerData.setBrokerAddrs(new HashMap<>(Map.of(0L, "10.0.0.1:10911")));
+ clusterInfo.setBrokerAddrTable(new HashMap<>(Map.of("broker-1",
brokerData)));
+ when(selectedAdmin.examineBrokerClusterInfo()).thenReturn(clusterInfo);
+ when(groupMapper.selectOne(any())).thenReturn(null);
+
doNothing().when(selectedAdmin).createAndUpdateSubscriptionGroupConfig(anyString(),
any());
+
when(runtimeAdminClientResolver.execute(org.mockito.ArgumentMatchers.eq("instance-a"),
any()))
+ .thenAnswer(invocation -> {
+ MqAdminExtFactory.AdminAction<?> action =
invocation.getArgument(1);
+ return action.apply(selectedAdmin);
+ });
+
+ ConsumerGroupVO group = new ConsumerGroupVO();
+ group.setName("cg-orders");
+ group.setInstanceId("instance-a");
+
+ adminClient.createConsumerGroup(group);
+
+
verify(runtimeAdminClientResolver).execute(org.mockito.ArgumentMatchers.eq("instance-a"),
any());
+ verify(selectedAdmin).createAndUpdateSubscriptionGroupConfig(
+ org.mockito.ArgumentMatchers.eq("10.0.0.1:10911"), any());
+ verify(adminExt,
never()).createAndUpdateSubscriptionGroupConfig(anyString(), any());
+ }
+
+ @Test
+ void deleteConsumerGroupUsesSelectedInstanceAdmin() throws Exception {
+ DefaultMQAdminExt selectedAdmin =
org.mockito.Mockito.mock(DefaultMQAdminExt.class);
+ ClusterInfo clusterInfo = new ClusterInfo();
+ BrokerData brokerData = new BrokerData();
+ brokerData.setBrokerName("broker-1");
+ brokerData.setBrokerAddrs(new HashMap<>(Map.of(0L, "10.0.0.1:10911")));
+ clusterInfo.setBrokerAddrTable(new HashMap<>(Map.of("broker-1",
brokerData)));
+ when(selectedAdmin.examineBrokerClusterInfo()).thenReturn(clusterInfo);
+ doNothing().when(selectedAdmin).deleteSubscriptionGroup(anyString(),
anyString(), org.mockito.ArgumentMatchers.anyBoolean());
+
when(runtimeAdminClientResolver.execute(org.mockito.ArgumentMatchers.eq("instance-a"),
any()))
+ .thenAnswer(invocation -> {
+ MqAdminExtFactory.AdminAction<?> action =
invocation.getArgument(1);
+ return action.apply(selectedAdmin);
+ });
+
+ adminClient.deleteConsumerGroup("instance-a", "cg-orders");
+
+
verify(runtimeAdminClientResolver).execute(org.mockito.ArgumentMatchers.eq("instance-a"),
any());
+ verify(selectedAdmin).deleteSubscriptionGroup("10.0.0.1:10911",
"cg-orders", true);
+ verify(adminExt, never()).deleteSubscriptionGroup(anyString(),
anyString(), org.mockito.ArgumentMatchers.anyBoolean());
+ }
+
@Test
void sendMessageShouldNotFailWhenAuditRecordingFails() throws Exception {
when(properties.getNamesrvAddr()).thenReturn("10.0.0.1:9876");
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 55991581..90d4607c 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
@@ -19,6 +19,8 @@ package org.apache.rocketmq.studio.provider.apache;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
import org.apache.rocketmq.studio.common.domain.enums.ConsumeType;
import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
+import org.apache.rocketmq.studio.instance.group.QueueProgressVO;
+import org.apache.rocketmq.studio.instance.group.SubscriptionEntryVO;
import org.apache.rocketmq.studio.instance.topic.BrokerRouteVO;
import org.apache.rocketmq.studio.instance.topic.TopicConsumerVO;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
@@ -113,4 +115,18 @@ class RocketMQMetadataProviderTest {
assertThat(provider.getTopicConsumers("instance-a",
"orders")).containsExactlyElementsOf(consumers);
verify(runtimeAdminClientResolver).execute(eq("instance-a"), any());
}
+
+ @Test
+ void groupRuntimeDiagnosticsShouldUseSelectedInstanceRuntimeClient() {
+ List<QueueProgressVO> progress =
List.of(QueueProgressVO.builder().broker("broker-a").build());
+ List<SubscriptionEntryVO> subscriptions =
List.of(SubscriptionEntryVO.builder().topic("orders").build());
+ when(runtimeAdminClientResolver.execute(eq("instance-a"), any()))
+ .thenReturn(progress, subscriptions);
+ RocketMQMetadataProvider provider = newProvider();
+
+ assertThat(provider.getGroupProgress("instance-a",
"cg-orders")).containsExactlyElementsOf(progress);
+ assertThat(provider.getGroupSubscriptions("instance-a", "cg-orders"))
+ .containsExactlyElementsOf(subscriptions);
+ verify(runtimeAdminClientResolver,
org.mockito.Mockito.times(2)).execute(eq("instance-a"), any());
+ }
}
diff --git a/web/src/api/consumerGroups.test.ts
b/web/src/api/consumerGroups.test.ts
index c997a4d0..cd6f578b 100644
--- a/web/src/api/consumerGroups.test.ts
+++ b/web/src/api/consumerGroups.test.ts
@@ -65,19 +65,28 @@ describe('consumer groups API contract', () => {
await expect(listConsumerGroups(params)).resolves.toEqual([group]);
});
- it('encodes consumer group names used in path segments', async () => {
+ it('encodes consumer group names and passes instance context for runtime
queries', async () => {
const groupName = '%RETRY%cg-order';
+ const instanceId = 'instance-a';
mock.onGet('/groups/%25RETRY%25cg-order').reply(200, { code: 200, data:
group });
mock.onGet('/groups/%25RETRY%25cg-order/progress').reply(200, { code: 200,
data: [] });
mock.onGet('/groups/%25RETRY%25cg-order/subscriptions').reply(200, { code:
200, data: [] });
await expect(getConsumerGroup(groupName)).resolves.toEqual(group);
- await expect(getConsumerProgress(groupName)).resolves.toEqual([]);
- await expect(getConsumerSubscriptions(groupName)).resolves.toEqual([]);
+ await expect(getConsumerProgress(groupName,
instanceId)).resolves.toEqual([]);
+ await expect(getConsumerSubscriptions(groupName,
instanceId)).resolves.toEqual([]);
+
+ expect(mock.history.get[1].params).toEqual({ instanceId });
+ expect(mock.history.get[2].params).toEqual({ instanceId });
});
it('unwraps detail records and sends numeric reset timestamps', async () => {
- const reset = { name: group.name, topic: 'orders', timestamp:
1784246400000 };
+ const reset = {
+ name: group.name,
+ instanceId: 'instance-a',
+ topic: 'orders',
+ timestamp: 1784246400000,
+ };
mock.onGet('/groups/orders').reply(200, { code: 200, data: group });
mock.onPost('/groups/reset-offset').reply((config) => {
expect(JSON.parse(config.data)).toEqual(reset);
diff --git a/web/src/api/metadata.ts b/web/src/api/metadata.ts
index 7d63cccc..1d027812 100644
--- a/web/src/api/metadata.ts
+++ b/web/src/api/metadata.ts
@@ -204,9 +204,9 @@ export async function deleteConsumerGroup(name: string,
instanceId?: string) {
export interface ResetConsumerOffsetRequest {
name: string;
+ instanceId?: string;
timestamp: number;
topic?: string;
- instanceId?: string;
}
export async function resetConsumerOffset(data: ResetConsumerOffsetRequest) {
diff --git a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
index c7758803..2d3b7e50 100644
--- a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
+++ b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
@@ -22,6 +22,7 @@ import type React from 'react';
import { MemoryRouter } from 'react-router-dom';
import { beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
import type { ConsumerGroup } from '../../../api/metadata';
+import * as instanceService from '../../../services/instanceService';
import { LangProvider } from '../../../i18n/LangContext';
import * as consumerService from '../../../services/consumerService';
import ConsumerPage from '../consumer';
@@ -106,6 +107,19 @@ const renderWithProviders = (ui: React.ReactElement) =>
describe('Consumer page', () => {
beforeEach(() => {
vi.clearAllMocks();
+ vi.mocked(instanceService.listInstances).mockResolvedValue([
+ {
+ id: 'instance-1',
+ name: 'instance-1',
+ remark: '',
+ type: 'PROXY',
+ endpoint: '10.0.0.1:8080',
+ topicCount: 0,
+ consumerGroupCount: 0,
+ createdAt: '2026-01-01T00:00:00Z',
+ updatedAt: '2026-01-01T00:00:00Z',
+ },
+ ]);
vi.mocked(consumerService.listConsumerGroups).mockResolvedValue([group]);
vi.mocked(consumerService.createConsumerGroup).mockImplementation(
async (data: Partial<ConsumerGroup>) =>
@@ -214,6 +228,39 @@ describe('Consumer page', () => {
await waitFor(() =>
expect(screen.getAllByText('remote-topic').length).toBeGreaterThan(0));
});
+ it('passes the selected instance to group diagnostics', async () => {
+ vi.mocked(instanceService.listInstances).mockResolvedValue([
+ {
+ id: 'instance-a',
+ name: 'Instance A',
+ remark: '',
+ type: 'DIRECT',
+ endpoint: '127.0.0.1:9876',
+ topicCount: 0,
+ consumerGroupCount: 0,
+ createdAt: '2026-07-23T00:00:00Z',
+ updatedAt: '2026-07-23T00:00:00Z',
+ },
+ ]);
+ vi.mocked(consumerService.listConsumerGroups).mockResolvedValue([
+ { ...group, instanceId: 'instance-a' },
+ ]);
+ const user = userEvent.setup();
+ renderWithProviders(<ConsumerPage />);
+
+ await user.click(await screen.findByRole('button', { name: /详情/ }));
+
+ await waitFor(() =>
+ expect(consumerService.getConsumerSubscriptions).toHaveBeenCalledWith(
+ 'remote-cg',
+ 'instance-a',
+ ),
+ );
+ await waitFor(() =>
+
expect(consumerService.getConsumerProgress).toHaveBeenCalledWith('remote-cg',
'instance-a'),
+ );
+ });
+
it('highlights inconsistent subscriptions and refreshes the check result',
async () => {
vi.mocked(consumerService.getConsumerSubscriptions)
.mockResolvedValueOnce([
diff --git a/web/src/pages/instance/consumer.tsx
b/web/src/pages/instance/consumer.tsx
index 3bc792eb..636363b3 100644
--- a/web/src/pages/instance/consumer.tsx
+++ b/web/src/pages/instance/consumer.tsx
@@ -1405,11 +1405,11 @@ const ConsumerPage = () => {
if (resetGroup) {
setResetSubmitting(true);
try {
- await resetConsumerOffset({
- name: resetGroup.name,
- timestamp: resetTime.valueOf(),
- instanceId: selectedInstanceId || undefined,
- });
+ await resetConsumerOffset({
+ name: resetGroup.name,
+ instanceId: selectedInstanceId || undefined,
+ timestamp: resetTime.valueOf(),
+ });
message.success(
`${resetGroup.name} 消费位点已重置到 ${resetTime.format('YYYY-MM-DD
HH:mm:ss')}`,
);
