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 7bbbea558 fix(consumer): filter the group page by subscription mode
server-side (#7310)
7bbbea558 is described below
commit 7bbbea5581890c6ebe2a9e61d1e8b4c39f7cb745
Author: xdz <[email protected]>
AuthorDate: Mon Oct 26 17:29:47 2026 +0800
fix(consumer): filter the group page by subscription mode server-side
(#7310)
* fix(consumer): filter the group page by subscription mode server-side
Filters the consumer group page by subscription mode on the server side,
instead of filtering only the currently loaded page in the browser.
* fix(consumer): match the subscription mode filter to what the list
displays
parseSubscriptionMode answers Push for every stored value that is not Pop —
a null,
a blank, or a legacy "CLUSTERING" — but the paged query filtered with
message_model = 'Push', so exactly those rows vanished from the filtered
page while
the unfiltered list still labelled them Push. Pop stays an equality; Push
becomes
"not Pop", and the requested mode is canonicalised so the in-memory
inventory path
and the SQL path cannot disagree on spelling.
Cover the two shapes in the provider test, and document subscriptionMode on
GET /api/groups/page.
---
docs/api-spec.md | 3 +-
.../instance/group/ConsumerGroupController.java | 4 +-
.../studio/instance/topic/MetadataService.java | 40 +++++++++++++++-
.../rocketmq/studio/provider/InstanceProvider.java | 22 +++++++++
.../provider/apache/ApacheInstanceProvider.java | 7 +++
.../studio/provider/apache/MetadataProvider.java | 11 +++++
.../provider/apache/RocketMQMetadataProvider.java | 23 +++++++++
.../group/ConsumerGroupControllerTest.java | 19 +++++++-
.../studio/instance/topic/MetadataServiceTest.java | 55 ++++++++++++++++++++--
.../apache/RocketMQMetadataProviderTest.java | 22 +++++++++
web/src/api/metadata.ts | 3 +-
.../pages/instance/__tests__/ConsumerPage.test.tsx | 21 +++++++++
web/src/pages/instance/consumer.tsx | 19 ++------
web/src/services/consumerService.ts | 6 +--
14 files changed, 228 insertions(+), 27 deletions(-)
diff --git a/docs/api-spec.md b/docs/api-spec.md
index 481f489b2..01f5fceba 100644
--- a/docs/api-spec.md
+++ b/docs/api-spec.md
@@ -1435,7 +1435,7 @@ GET /api/groups/export?names={name1,name2}
### 6.10 分页获取消费组列表
```
-GET
/api/groups/page?instanceId={instanceId}&clusterId={clusterId}&search={search}&page={page}&pageSize={pageSize}
+GET
/api/groups/page?instanceId={instanceId}&clusterId={clusterId}&search={search}&subscriptionMode={subscriptionMode}&page={page}&pageSize={pageSize}
```
**Query Parameters:**
@@ -1445,6 +1445,7 @@ GET
/api/groups/page?instanceId={instanceId}&clusterId={clusterId}&search={searc
| `instanceId` | `string` | 否 | 所属实例 ID，不传时按默认平台集群查询 |
| `clusterId` | `string` | 否 | 按集群过滤 |
| `search` | `string` | 否 | 按名称模糊搜索 |
+| `subscriptionMode` | `string` | 否 | 按订阅模式过滤：`Push` / `Pop`（大小写不敏感）。`Pop`
为等值匹配；`Push` 与列表展示同口径，同时匹配 `message_model` 为空或存历史遗留值（如 `CLUSTERING`）的记录。不传或传
`ALL` 表示不过滤 |
| `page` | `number` | 否 | 页码，默认 `1` |
| `pageSize` | `number` | 否 | 每页条数，默认 `20` |
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupController.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupController.java
index 1eb357f07..847eead74 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupController.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupController.java
@@ -56,9 +56,11 @@ public class ConsumerGroupController {
@RequestParam(required = false) String instanceId,
@RequestParam(required = false) String clusterId,
@RequestParam(required = false) String search,
+ @RequestParam(required = false) String subscriptionMode,
@RequestParam(defaultValue = "1") int page,
@RequestParam(defaultValue = "20") int pageSize) {
- return Result.ok(metadataService.listConsumerGroupsPage(instanceId,
clusterId, search, page, pageSize));
+ return Result.ok(metadataService.listConsumerGroupsPage(instanceId,
clusterId, search,
+ subscriptionMode, page, pageSize));
}
@GetMapping("/{name}")
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 039a45d5c..055439569 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
@@ -44,6 +44,7 @@ import
org.apache.rocketmq.studio.common.domain.enums.SubscriptionMode;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.common.util.CsvUtil;
import org.apache.rocketmq.studio.common.util.MessagePropertyDisplay;
+import org.apache.rocketmq.studio.common.util.Pagination;
import org.apache.rocketmq.studio.common.util.SystemTopicFilter;
import org.apache.rocketmq.studio.instance.group.CreateConsumerGroupDTO;
import org.apache.rocketmq.studio.instance.group.ImportConsumerGroupsResultVO;
@@ -504,14 +505,51 @@ public class MetadataService {
public PageResult<ConsumerGroupVO> listConsumerGroupsPage(String
instanceId, String clusterId, String search,
int page, int
pageSize) {
+ return listConsumerGroupsPage(instanceId, clusterId, search, null,
page, pageSize);
+ }
+
+ public PageResult<ConsumerGroupVO> listConsumerGroupsPage(String
instanceId, String clusterId, String search,
+ String
subscriptionMode, int page, int pageSize) {
validatePagination(page, pageSize);
instanceId = normalizeInstanceId(instanceId);
+ String normalizedMode = normalizeSubscriptionMode(subscriptionMode);
if (!StringUtils.hasText(instanceId) &&
StringUtils.hasText(clusterId)) {
+ // Legacy cluster-scoped read without an instance: filter the mode
over the full
+ // inventory so the total describes the filtered result set, not
one filtered page.
+ if (normalizedMode != null) {
+ List<ConsumerGroupVO> groups =
metadataProvider.listConsumerGroups(
+ normalizeFilter(clusterId),
normalizeFilter(search)).stream()
+ .filter(group -> group.getSubscriptionMode() != null
+ &&
normalizedMode.equals(group.getSubscriptionMode().name()))
+ .toList();
+ long offset = Pagination.pageOffset(page, pageSize);
+ int from = (int) Math.min(offset, groups.size());
+ int to = from + (int) Math.min(pageSize, groups.size() - from);
+ return PageResult.of(groups.subList(from, to), groups.size(),
page, pageSize);
+ }
return
metadataProvider.listConsumerGroupsPage(normalizeFilter(clusterId),
normalizeFilter(search), page, pageSize);
}
return resolve(instanceId).listConsumerGroupsPage(instanceId,
normalizeFilter(clusterId),
- normalizeFilter(search), page, pageSize);
+ normalizeFilter(search), normalizedMode, page, pageSize);
+ }
+
+ /**
+ * Canonicalises the requested mode so the in-memory path over a full
inventory and the SQL
+ * path over the page agree on the same spelling; an unknown value is
passed through and simply
+ * matches nothing on both.
+ */
+ private static String normalizeSubscriptionMode(String subscriptionMode) {
+ if (!StringUtils.hasText(subscriptionMode) ||
"ALL".equals(subscriptionMode.trim())) {
+ return null;
+ }
+ String trimmed = subscriptionMode.trim();
+ for (SubscriptionMode mode : SubscriptionMode.values()) {
+ if (mode.name().equalsIgnoreCase(trimmed)) {
+ return mode.name();
+ }
+ }
+ return trimmed;
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProvider.java
index 9f33c6f8a..926892fe2 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProvider.java
@@ -127,6 +127,28 @@ public interface InstanceProvider {
return listConsumerGroupsPage(instanceId, search, page, pageSize);
}
+ /**
+ * Instance-scoped consumer-group pagination with an optional
subscription-mode filter
+ * ("Push" / "Pop"; blank or "ALL" means no restriction). Providers
without a mode-aware
+ * store filter their full inventory before paginating, so the page and
its total describe
+ * the filtered result set instead of one filtered page of an unfiltered
count.
+ */
+ default PageResult<ConsumerGroupVO> listConsumerGroupsPage(String
instanceId, String clusterId,
+ String search, String subscriptionMode, int page, int pageSize) {
+ if (subscriptionMode == null || subscriptionMode.isBlank() ||
"ALL".equals(subscriptionMode)) {
+ return listConsumerGroupsPage(instanceId, clusterId, search, page,
pageSize);
+ }
+ List<ConsumerGroupVO> groups = listConsumerGroups(instanceId,
search).stream()
+ .filter(group -> group.getSubscriptionMode() != null
+ &&
subscriptionMode.equals(group.getSubscriptionMode().name()))
+ .toList();
+ int total = groups.size();
+ long offset = Pagination.pageOffset(page, pageSize);
+ int from = (int) Math.min(offset, total);
+ int to = from + (int) Math.min(pageSize, total - from);
+ return PageResult.of(groups.subList(from, to), total, page, pageSize);
+ }
+
ConsumerGroupVO createConsumerGroup(String instanceId, ConsumerGroupVO
group);
/** Imports an existing consumer group; defaults to creation, providers
may override to compare configurations. */
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 f67d95b2a..7dbbef239 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
@@ -154,6 +154,13 @@ public class ApacheInstanceProvider implements
InstanceProvider {
return metadataProvider.listConsumerGroupsPage(instanceId, clusterId,
search, page, pageSize);
}
+ @Override
+ public PageResult<ConsumerGroupVO> listConsumerGroupsPage(String
instanceId, String clusterId,
+ String search, String subscriptionMode, int page, int pageSize) {
+ return metadataProvider.listConsumerGroupsPage(instanceId, clusterId,
search, subscriptionMode,
+ page, pageSize);
+ }
+
@Override
public ConsumerGroupVO createConsumerGroup(String instanceId,
ConsumerGroupVO group) {
requireGroupInstance(instanceId, group);
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 66454d34e..dae09e15f 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
@@ -71,6 +71,17 @@ public interface MetadataProvider {
return listConsumerGroupsPage(clusterId, search, page, pageSize);
}
+ /**
+ * Instance-scoped consumer-group pagination with an optional
subscription-mode filter
+ * ("Push" / "Pop"; blank or "ALL" means no restriction). The default
drops the filter;
+ * implementors with a mode-aware store override it so the page and its
total describe the
+ * filtered result set.
+ */
+ default PageResult<ConsumerGroupVO> listConsumerGroupsPage(String
instanceId, String clusterId,
+ String search, String subscriptionMode, int page, int pageSize) {
+ return listConsumerGroupsPage(instanceId, clusterId, search, page,
pageSize);
+ }
+
List<BrokerRouteVO> getTopicRoutes(String instanceId, String name);
List<TopicConsumerVO> getTopicConsumers(String instanceId, String name);
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 2171f0550..acab23c8e 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
@@ -267,8 +267,19 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
@Override
public PageResult<ConsumerGroupVO> listConsumerGroupsPage(String
instanceId, String clusterId,
String search, int page, int pageSize) {
+ return listConsumerGroupsPage(instanceId, clusterId, search, null,
page, pageSize);
+ }
+
+ @Override
+ public PageResult<ConsumerGroupVO> listConsumerGroupsPage(String
instanceId, String clusterId,
+ String search, String subscriptionMode, int page, int pageSize) {
String configuredCluster = StringUtils.hasText(instanceId)
? runtimeAdminClientResolver.configuredClusterName(instanceId)
: null;
+ // message_model stores the subscription mode ("Push"/"Pop"); blank
and "ALL" mean no
+ // restriction, so the count and the page both describe the filtered
result set.
+ boolean modeRestricted = StringUtils.hasText(subscriptionMode)
+ && !"ALL".equals(subscriptionMode);
+ boolean popOnly = modeRestricted &&
SubscriptionMode.Pop.name().equalsIgnoreCase(subscriptionMode);
LambdaQueryWrapper<RmqGroup> query = new LambdaQueryWrapper<RmqGroup>()
.and(instanceId != null, scope -> {
scope.eq(RmqGroup::getInstanceId,
normalizeMetadataScope(instanceId));
@@ -280,6 +291,18 @@ public class RocketMQMetadataProvider implements
MetadataProvider {
})
.eq(StringUtils.hasText(clusterId), RmqGroup::getClusterId,
clusterId)
.like(StringUtils.hasText(search), RmqGroup::getName, search)
+ .and(modeRestricted, scope -> {
+ if (popOnly) {
+ scope.eq(RmqGroup::getMessageModel,
SubscriptionMode.Pop.name());
+ } else {
+ // parseSubscriptionMode answers Push for every value
that is not Pop — a
+ // null, a blank, or a legacy "CLUSTERING" — so the
Push filter has to match
+ // that same set. message_model = 'Push' would drop
exactly the rows the
+ // list shows as Push, and the page total would
disagree with the page.
+ scope.isNull(RmqGroup::getMessageModel)
+ .or().ne(RmqGroup::getMessageModel,
SubscriptionMode.Pop.name());
+ }
+ })
.orderByAsc(RmqGroup::getName, RmqGroup::getId);
Page<RmqGroup> result = groupMapper.selectPage(new Page<>(page,
pageSize), query);
List<ConsumerGroupVO> groups = result.getRecords().stream()
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 1e5226600..47c4bd6bd 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
@@ -87,7 +87,7 @@ class ConsumerGroupControllerTest extends
WebMvcAuthTestSupport {
@Test
void listConsumerGroupsPageShouldPassSelectedInstanceFiltersAndPaging()
throws Exception {
PageResult<ConsumerGroupVO> page = PageResult.of(List.of(), 3, 2, 20);
- when(metadataService.listConsumerGroupsPage("instance-a", "cluster-a",
"orders", 2, 20))
+ when(metadataService.listConsumerGroupsPage("instance-a", "cluster-a",
"orders", null, 2, 20))
.thenReturn(page);
mockMvc.perform(get("/api/groups/page")
@@ -101,7 +101,22 @@ class ConsumerGroupControllerTest extends
WebMvcAuthTestSupport {
.andExpect(jsonPath("$.data.page").value(2))
.andExpect(jsonPath("$.data.size").value(20));
- verify(metadataService).listConsumerGroupsPage("instance-a",
"cluster-a", "orders", 2, 20);
+ verify(metadataService).listConsumerGroupsPage("instance-a",
"cluster-a", "orders", null, 2, 20);
+ }
+
+ @Test
+ void listConsumerGroupsPageShouldPassTheSubscriptionModeFilter() throws
Exception {
+ PageResult<ConsumerGroupVO> page = PageResult.of(List.of(), 1, 1, 20);
+ when(metadataService.listConsumerGroupsPage("instance-a", null, null,
"Push", 1, 20))
+ .thenReturn(page);
+
+ mockMvc.perform(get("/api/groups/page")
+ .param("instanceId", "instance-a")
+ .param("subscriptionMode", "Push"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.data.total").value(1));
+
+ verify(metadataService).listConsumerGroupsPage("instance-a", null,
null, "Push", 1, 20);
}
@Test
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
index ddea630da..5276ecf2f 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
@@ -967,7 +967,7 @@ class MetadataServiceTest {
void listConsumerGroupsPageShouldPaginateFromOneBasedIndexes() {
ConsumerGroupVO third = new ConsumerGroupVO();
third.setName("cg-c");
- when(apacheProvider.listConsumerGroupsPage("instance-a", null,
"order", 2, 2))
+ when(apacheProvider.listConsumerGroupsPage("instance-a", null,
"order", null, 2, 2))
.thenReturn(PageResult.of(List.of(third), 3, 2, 2));
PageResult<ConsumerGroupVO> result =
@@ -977,7 +977,7 @@ class MetadataServiceTest {
assertThat(result.getTotal()).isEqualTo(3);
assertThat(result.getPage()).isEqualTo(2);
assertThat(result.getSize()).isEqualTo(2);
- verify(apacheProvider).listConsumerGroupsPage("instance-a", null,
"order", 2, 2);
+ verify(apacheProvider).listConsumerGroupsPage("instance-a", null,
"order", null, 2, 2);
verify(apacheProvider,
org.mockito.Mockito.never()).listConsumerGroups("instance-a", "order");
}
@@ -986,7 +986,7 @@ class MetadataServiceTest {
ConsumerGroupVO clusterAGroup = new ConsumerGroupVO();
clusterAGroup.setName("group-a");
clusterAGroup.setClusterId("cluster-a");
- when(apacheProvider.listConsumerGroupsPage("instance-a", "cluster-a",
null, 1, 20))
+ when(apacheProvider.listConsumerGroupsPage("instance-a", "cluster-a",
null, null, 1, 20))
.thenReturn(PageResult.of(List.of(clusterAGroup), 1, 1, 20));
PageResult<ConsumerGroupVO> result =
metadataService.listConsumerGroupsPage(
@@ -994,7 +994,54 @@ class MetadataServiceTest {
assertThat(result.getItems()).containsExactly(clusterAGroup);
assertThat(result.getTotal()).isEqualTo(1);
- verify(apacheProvider).listConsumerGroupsPage("instance-a",
"cluster-a", null, 1, 20);
+ verify(apacheProvider).listConsumerGroupsPage("instance-a",
"cluster-a", null, null, 1, 20);
+ }
+
+ @Test
+ void listConsumerGroupsPageShouldPassSubscriptionModeToTheProviderTest() {
+ ConsumerGroupVO pushGroup = new ConsumerGroupVO();
+ pushGroup.setName("group-push");
+ when(apacheProvider.listConsumerGroupsPage("instance-a", null, null,
"Push", 1, 20))
+ .thenReturn(PageResult.of(List.of(pushGroup), 1, 1, 20));
+
+ PageResult<ConsumerGroupVO> result =
metadataService.listConsumerGroupsPage(
+ "instance-a", null, null, " Push ", 1, 20);
+
+ assertThat(result.getItems()).containsExactly(pushGroup);
+ verify(apacheProvider).listConsumerGroupsPage("instance-a", null,
null, "Push", 1, 20);
+ }
+
+ @Test
+ void listConsumerGroupsPageShouldTreatAllAsNoSubscriptionModeTest() {
+ when(apacheProvider.listConsumerGroupsPage("instance-a", null, null,
null, 1, 20))
+ .thenReturn(PageResult.empty(1, 20));
+
+ metadataService.listConsumerGroupsPage("instance-a", null, null,
"ALL", 1, 20);
+
+ verify(apacheProvider).listConsumerGroupsPage("instance-a", null,
null, null, 1, 20);
+ }
+
+ @Test
+ void
listConsumerGroupsPageShouldFilterModeOverTheFullLegacyInventoryTest() {
+ ConsumerGroupVO pushGroup = new ConsumerGroupVO();
+ pushGroup.setName("group-push");
+ pushGroup.setSubscriptionMode(SubscriptionMode.Push);
+ ConsumerGroupVO popGroup = new ConsumerGroupVO();
+ popGroup.setName("group-pop");
+ popGroup.setSubscriptionMode(SubscriptionMode.Pop);
+ when(metadataProvider.listConsumerGroups("cluster-1", "order"))
+ .thenReturn(List.of(pushGroup, popGroup));
+
+ PageResult<ConsumerGroupVO> result =
metadataService.listConsumerGroupsPage(
+ null, "cluster-1", "order", "Push", 2, 1);
+
+ // Only one Push row exists, so page 2 with size 1 is past the
filtered total.
+ assertThat(result.getItems()).isEmpty();
+ assertThat(result.getTotal()).isEqualTo(1);
+ org.mockito.Mockito.verify(metadataProvider,
org.mockito.Mockito.never())
+
.listConsumerGroupsPage(org.mockito.ArgumentMatchers.anyString(),
+ org.mockito.ArgumentMatchers.anyString(),
org.mockito.ArgumentMatchers.anyInt(),
+ org.mockito.ArgumentMatchers.anyInt());
}
@Test
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 441fe4a4c..4ce06642d 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
@@ -268,6 +268,28 @@ class RocketMQMetadataProviderTest {
verify(runtimeAdminClientResolver, times(2)).execute(eq("instance-a"),
any());
}
+ @Test
+ void
listConsumerGroupsPageShouldFilterTheSubscriptionModeTheWayItIsDisplayedTest() {
+ TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new
MybatisConfiguration(), ""), RmqGroup.class);
+ when(groupMapper.selectPage(any(Page.class),
any(LambdaQueryWrapper.class)))
+ .thenReturn(new Page<>(1, 20, 0));
+ RocketMQMetadataProvider provider = newProvider();
+
+ provider.listConsumerGroupsPage("instance-a", null, null, "Pop", 1,
20);
+ provider.listConsumerGroupsPage("instance-a", null, null, "Push", 1,
20);
+
+ org.mockito.ArgumentCaptor<LambdaQueryWrapper<RmqGroup>> captor =
+ org.mockito.ArgumentCaptor.forClass(LambdaQueryWrapper.class);
+ verify(groupMapper, times(2)).selectPage(any(Page.class),
captor.capture());
+ // Pop is a stored value, so it stays an equality. Push is what the
read path shows for
+ // every row that is not Pop — a null, a blank, or a legacy
"CLUSTERING" — so its filter
+ // has to match that set; message_model = 'Push' would hide those rows
while the list
+ // still labels them Push.
+
assertThat(captor.getAllValues().get(0).getSqlSegment()).contains("message_model
=");
+ assertThat(captor.getAllValues().get(1).getSqlSegment())
+ .contains("message_model IS NULL", "message_model <>");
+ }
+
@Test
void getTopicRoutesShouldUseSelectedInstanceRuntimeClient() {
List<BrokerRouteVO> routes =
List.of(BrokerRouteVO.builder().brokerName("broker-a").build());
diff --git a/web/src/api/metadata.ts b/web/src/api/metadata.ts
index 92b1d9309..bfa08b729 100644
--- a/web/src/api/metadata.ts
+++ b/web/src/api/metadata.ts
@@ -174,6 +174,8 @@ export interface ConsumerGroupQuery {
instanceId?: string;
clusterId?: string;
search?: string;
+ /** Subscription mode filter ("Push" / "Pop"); absent or "ALL" means no
restriction. */
+ subscriptionMode?: string;
}
export interface ConsumerGroupPageQuery extends ConsumerGroupQuery {
@@ -183,7 +185,6 @@ export interface ConsumerGroupPageQuery extends
ConsumerGroupQuery {
export interface ConsumerGroupExportQuery extends ConsumerGroupQuery {
names?: string[];
- subscriptionMode?: string;
}
export interface ImportConsumerGroupsRequest {
diff --git a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
index 6e68ebfb8..6cfb89171 100644
--- a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
+++ b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
@@ -355,6 +355,27 @@ describe('Consumer page', () => {
});
});
+ it('applies the subscription-mode filter on the server', async () => {
+ const user = userEvent.setup({ pointerEventsCheck: 0 });
+ renderWithProviders(<ConsumerPage />);
+
+ expect(await screen.findByText('remote-cg')).toBeInTheDocument();
+
+ const modeSelect = screen.getAllByRole('combobox')[1];
+ fireEvent.mouseDown(modeSelect.parentElement!);
+ await user.click(await screen.findByText('Pop', { selector:
'.ant-select-item-option-content' }));
+
+ await waitFor(() =>
+ expect(consumerService.listConsumerGroupPage).toHaveBeenLastCalledWith({
+ instanceId: 'instance-1',
+ page: 1,
+ pageSize: 20,
+ search: undefined,
+ subscriptionMode: 'Pop',
+ }),
+ );
+ });
+
it('clears selected consumer groups when the search scope changes', async ()
=> {
const user = userEvent.setup({ pointerEventsCheck: 0 });
renderWithProviders(<ConsumerPage />);
diff --git a/web/src/pages/instance/consumer.tsx
b/web/src/pages/instance/consumer.tsx
index 724180f19..8578a65dc 100644
--- a/web/src/pages/instance/consumer.tsx
+++ b/web/src/pages/instance/consumer.tsx
@@ -130,16 +130,6 @@ const lagColor = (lag: number): string => {
return '#52c41a';
};
-const visibleConsumerGroups = (groups: ConsumerGroup[], modeFilter: string):
ConsumerGroup[] => {
- let data = groups;
-
- if (modeFilter !== 'ALL') {
- data = data.filter((group) => group.subscriptionMode === modeFilter);
- }
-
- return data;
-};
-
const normalizedConsistency = (value?: string | null): string =>
value?.trim().toLowerCase() ?? '';
const isConsistentValue = (value?: string | null): boolean =>
@@ -353,6 +343,7 @@ const ConsumerPageContent = ({
const result = await listConsumerGroupPage({
instanceId: selectedInstanceId,
search: search.trim() || undefined,
+ subscriptionMode: modeFilter !== 'ALL' ? modeFilter : undefined,
page: pageToLoad,
pageSize: pageSizeToLoad,
});
@@ -376,7 +367,7 @@ const ConsumerPageContent = ({
if (requestId === groupRequestIdRef.current) setLoading(false);
}
},
- [t, selectedInstanceId, search],
+ [t, selectedInstanceId, search, modeFilter],
);
const reloadConsumerGroupPage = useCallback(async () => {
@@ -518,9 +509,9 @@ const ConsumerPageContent = ({
}, [modalOpen, selectedGroupName, selectedInstanceId, loadProgress,
loadSubscriptions]);
/* ─── Filtered & sorted data ─── */
- const filtered = useMemo(() => {
- return visibleConsumerGroups(groups, modeFilter);
- }, [groups, modeFilter]);
+ // The subscription-mode filter is applied server-side (it is part of the
page query), so the
+ // loaded page is already the filtered inventory and the total matches what
is displayed.
+ const filtered = useMemo(() => groups, [groups]);
const handleExport = async () => {
setExporting(true);
diff --git a/web/src/services/consumerService.ts
b/web/src/services/consumerService.ts
index 53f57abdb..cd6a3edc0 100644
--- a/web/src/services/consumerService.ts
+++ b/web/src/services/consumerService.ts
@@ -102,6 +102,9 @@ function filterConsumerGroups(params?: ConsumerGroupQuery):
ConsumerGroup[] {
const kw = params.search.trim().toLowerCase();
if (kw) result = result.filter((group) =>
group.name.toLowerCase().includes(kw));
}
+ if (params?.subscriptionMode && params.subscriptionMode !== 'ALL') {
+ result = result.filter((group) => group.subscriptionMode ===
params.subscriptionMode);
+ }
return result;
}
@@ -111,9 +114,6 @@ function visibleConsumerGroups(groups: ConsumerGroup[],
params?: ConsumerGroupEx
const selectedNames = new Set(params.names);
result = result.filter((group) => selectedNames.has(group.name));
}
- if (params?.subscriptionMode && params.subscriptionMode !== 'ALL') {
- result = result.filter((group) => group.subscriptionMode ===
params.subscriptionMode);
- }
return [...result].sort((left, right) =>
left.name.localeCompare(right.name));
}
