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 5318e8d57 feat(consumer): refresh a single consumer group (#2521)
5318e8d57 is described below
commit 5318e8d57dee6bc0f52ea195a1421fde581170ff
Author: zhaohai <[email protected]>
AuthorDate: Sat Aug 22 16:00:00 2026 +0800
feat(consumer): refresh a single consumer group (#2521)
Adds a per-group refresh action so operators can re-read one consumer
group directly from the broker without reloading the whole list.
Backend:
- MetadataService.refreshConsumerGroup: resolves the group through the
provider's list path filtered to the requested name and returns the
latest ConsumerGroupVO (404 when the group no longer exists).
- ConsumerGroupController: new GET /api/groups/{name}/refresh endpoint.
Frontend:
- api/service: refreshConsumerGroup wrapper (mock returns the mock
group copy).
- Consumer page action column gains a 刷新 button that replaces the
matching row with the refreshed group and invalidates the cached
queue progress so the detail modal re-fetches it.
---
.../instance/group/ConsumerGroupController.java | 7 +++
.../studio/instance/topic/MetadataService.java | 20 +++++++++
.../studio/instance/topic/MetadataServiceTest.java | 22 ++++++++++
web/src/api/metadata.ts | 8 ++++
.../pages/instance/__tests__/ConsumerPage.test.tsx | 20 +++++++++
web/src/pages/instance/consumer.tsx | 50 +++++++++++++++++++++-
web/src/services/consumerService.ts | 12 ++++++
7 files changed, 137 insertions(+), 2 deletions(-)
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 d6d0733ca..751a734d3 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
@@ -79,6 +79,13 @@ public class ConsumerGroupController {
request.getRetryQueueNums(), request.getRetryMaxTimes()));
}
+ @GetMapping("/{name}/refresh")
+ public Result<ConsumerGroupVO> refreshConsumerGroup(
+ @PathVariable String name,
+ @RequestParam(required = false) String instanceId) {
+ return Result.ok(metadataService.refreshConsumerGroup(instanceId,
name));
+ }
+
@GetMapping("/{name}/progress")
public Result<List<QueueProgressVO>> getGroupProgress(
@PathVariable String 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 17d492a34..ae77ed0e9 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
@@ -208,6 +208,26 @@ public class MetadataService {
return adminClient.getConsumerGroup(instanceId, groupName);
}
+ /**
+ * Re-reads a single consumer group so an operator can refresh one row
without reloading the
+ * whole group list. Implementation queries the provider's list path with
the group name as
+ * the search filter and then exact-matches the name among the returned
entries (no direct
+ * per-group lookup exists that works for every vendor).
+ *
+ * <p>Returns {@code null} when the group no longer exists: a missing
group is an empty
+ * business state, not an RPC error, so the endpoint responds 200 with
empty data and the
+ * frontend keeps the existing row unchanged.
+ */
+ public ConsumerGroupVO refreshConsumerGroup(String instanceId, String
name) {
+ instanceId = normalizeInstanceId(instanceId);
+ String groupName = requireName(name, "consumer group name");
+ return listConsumerGroups(instanceId, null, normalizeFilter(groupName))
+ .stream()
+ .filter(group -> groupName.equals(group.getName()))
+ .findFirst()
+ .orElse(null);
+ }
+
public List<QueueProgressVO> getGroupProgress(String name) {
return getGroupProgress(null, name);
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 0812728e3..aa6e713ae 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
@@ -335,4 +335,26 @@ class MetadataServiceTest {
.hasMessage("pageSize must be between 1 and 100");
}
+ @Test
+ void refreshConsumerGroupShouldExactMatchWithinProviderSearchResults() {
+ ConsumerGroupVO similar = new ConsumerGroupVO();
+ similar.setName("cg-order-archive");
+ ConsumerGroupVO group = new ConsumerGroupVO();
+ group.setName("cg-order");
+ when(apacheProvider.listConsumerGroups("instance-a", "cg-order"))
+ .thenReturn(List.of(similar, group));
+
+ ConsumerGroupVO result =
metadataService.refreshConsumerGroup("instance-a", " cg-order ");
+
+ assertThat(result).isSameAs(group);
+ verify(apacheProvider).listConsumerGroups("instance-a", "cg-order");
+ }
+
+ @Test
+ void refreshConsumerGroupShouldReturnNullWhenGroupMissingInsteadOfError() {
+ when(apacheProvider.listConsumerGroups("instance-a",
"cg-gone")).thenReturn(List.of());
+
+ assertThat(metadataService.refreshConsumerGroup("instance-a",
"cg-gone")).isNull();
+ }
+
}
diff --git a/web/src/api/metadata.ts b/web/src/api/metadata.ts
index 662ea21ba..4096cd38d 100644
--- a/web/src/api/metadata.ts
+++ b/web/src/api/metadata.ts
@@ -251,6 +251,14 @@ export async function getConsumerGroup(name: string,
instanceId?: string) {
return res.data.data;
}
+export async function refreshConsumerGroup(name: string, instanceId?: string) {
+ const res = await client.get<{ data: ConsumerGroup | null }>(
+ `/groups/${encodeURIComponent(name)}/refresh`,
+ { params: instanceId ? { instanceId } : {} },
+ );
+ return res.data.data;
+}
+
export async function getConsumerProgress(name: string, instanceId?: string) {
const res = await client.get<{ data: QueueProgress[] }>(
`/groups/${encodeURIComponent(name)}/progress`,
diff --git a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
index 82da18f4f..ccc83e89a 100644
--- a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
+++ b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
@@ -36,6 +36,7 @@ vi.mock('../../../services/consumerService', () => ({
getConsumerStack: vi.fn(),
getConsumerSubscriptions: vi.fn(),
listConsumerGroupPage: vi.fn(),
+ refreshConsumerGroup: vi.fn(),
resetConsumerOffset: vi.fn(),
}));
const instanceServiceMocks = vi.hoisted(() => ({ listInstances: vi.fn() }));
@@ -121,6 +122,11 @@ describe('Consumer page', () => {
},
]);
vi.mocked(consumerService.listConsumerGroupPage).mockResolvedValue(groupPage([group]));
+ vi.mocked(consumerService.refreshConsumerGroup).mockResolvedValue({
+ ...group,
+ totalLag: 42,
+ delaySeconds: 7,
+ });
vi.mocked(consumerService.createConsumerGroup).mockImplementation(
async (data: Partial<ConsumerGroup>) =>
({
@@ -767,4 +773,18 @@ describe('Consumer page', () => {
expect(screen.getByRole('button', { name: /导入/ })).toBeDisabled();
expect(screen.getByRole('button', { name: '创建 Group' })).toBeDisabled();
});
+
+ it('refreshes a single consumer group row without reloading the list', async
() => {
+ const user = userEvent.setup();
+ renderWithProviders(<ConsumerPage />);
+
+ const row = await screen.findByRole('row', { name: /remote-cg/ });
+ expect(within(row).getByText('10')).toBeInTheDocument();
+ await user.click(within(row).getByRole('button', { name: /刷\s*新/ }));
+
+ await waitFor(() => {
+
expect(consumerService.refreshConsumerGroup).toHaveBeenCalledWith('remote-cg',
'instance-1');
+ });
+ expect(await within(row).findByText('42')).toBeInTheDocument();
+ });
});
diff --git a/web/src/pages/instance/consumer.tsx
b/web/src/pages/instance/consumer.tsx
index 6001176a1..cff253685 100644
--- a/web/src/pages/instance/consumer.tsx
+++ b/web/src/pages/instance/consumer.tsx
@@ -56,7 +56,13 @@ import {
ArrowsClockwise,
SlidersHorizontal,
} from '@phosphor-icons/react';
-import { ImportOutlined, ExportOutlined, DeleteOutlined, SyncOutlined } from
'@ant-design/icons';
+import {
+ ImportOutlined,
+ ExportOutlined,
+ DeleteOutlined,
+ SyncOutlined,
+ ReloadOutlined,
+} from '@ant-design/icons';
import type { ColumnsType } from 'antd/es/table';
import dayjs from 'dayjs';
import type { Dayjs } from 'dayjs';
@@ -81,6 +87,7 @@ import {
getConsumerStack,
getConsumerSubscriptions,
listConsumerGroupPage,
+ refreshConsumerGroup,
resetConsumerOffset,
getConsumerGroupSettings,
updateConsumerGroupSettings,
@@ -327,6 +334,33 @@ const ConsumerPageContent = ({
[progressByGroup, t, selectedInstanceId],
);
+ const [refreshingGroup, setRefreshingGroup] = useState<string | null>(null);
+
+ const handleRefreshGroup = async (record: ConsumerGroup) => {
+ if (refreshingGroup) return;
+ setRefreshingGroup(record.name);
+ try {
+ const refreshed = await refreshConsumerGroup(record.name,
selectedInstanceId || undefined);
+ if (!refreshed) {
+ // 组不存在（业务状态为空）：保留原行数据，不弹错。
+ return;
+ }
+ setGroups((prev) => prev.map((group) => (group.name === record.name ?
refreshed : group)));
+ // Invalidate the cached progress so the detail modal re-fetches it on
next open.
+ setProgressByGroup((prev) => {
+ const next = { ...prev };
+ delete next[diagnosticCacheKey(selectedInstanceId, record.name)];
+ return next;
+ });
+ message.success(`消费组 ${record.name} 已刷新`);
+ } catch (error) {
+ const detail = error instanceof Error ? error.message : '';
+ message.error(detail || `刷新消费组 ${record.name} 失败`);
+ } finally {
+ setRefreshingGroup(null);
+ }
+ };
+
/* ─── Filtered & sorted data ─── */
const filtered = useMemo(() => {
let data = groups;
@@ -683,7 +717,7 @@ const ConsumerPageContent = ({
{
title: '操作',
key: 'actions',
- width: 210,
+ width: 360,
render: (_: unknown, record: ConsumerGroup) => (
<Flex gap={6}>
<Button
@@ -697,6 +731,18 @@ const ConsumerPageContent = ({
>
配置
</Button>
+ <Button
+ size="small"
+ icon={<ReloadOutlined />}
+ loading={refreshingGroup === record.name}
+ disabled={refreshingGroup !== null && refreshingGroup !==
record.name}
+ onClick={(e) => {
+ e.stopPropagation();
+ void handleRefreshGroup(record);
+ }}
+ >
+ 刷新
+ </Button>
<Button
size="small"
icon={<Eye size={14} />}
diff --git a/web/src/services/consumerService.ts
b/web/src/services/consumerService.ts
index db00f75a5..eef0bc779 100644
--- a/web/src/services/consumerService.ts
+++ b/web/src/services/consumerService.ts
@@ -125,6 +125,18 @@ export async function updateConsumerGroupSettings(
return metadataApi.updateConsumerGroupSettings(data);
}
+export async function refreshConsumerGroup(
+ name: string,
+ instanceId?: string,
+): Promise<ConsumerGroup | null> {
+ if (isMockMode()) {
+ const group = mockConsumerGroups.find((item) => item.name === name);
+ return group ? copyConsumerGroup(group) : null;
+ }
+ const data = await metadataApi.refreshConsumerGroup(name, instanceId);
+ return data ? normalizeConsumerGroup(data) : null;
+}
+
export async function getConsumerSubscriptions(
name: string,
instanceId?: string,
