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
commit 06b1d933c6378b6b1e0226b29586218b90b82e1f Author: aias00 <[email protected]> AuthorDate: Mon Aug 10 14:17:13 2026 +0800 fix: scope consumer group diagnostics by selected instance (#1248) * [ISSUE #1247] Scope Consumer Group list enrichment by instance * [ISSUE #1249] Surface Topic consumer diagnostics failures * [ISSUE #1251] Surface Consumer Group subscription failures * fix: scope consumer diagnostics caches by instance Signed-off-by: liuhy <[email protected]> * fix: stop resource loading without an instance Signed-off-by: liuhy <[email protected]> * fix: preserve selected instance for group details --------- Signed-off-by: liuhy <[email protected]> --- .../provider/apache/ApacheInstanceProvider.java | 4 +- .../studio/provider/apache/MetadataProvider.java | 5 ++ .../provider/apache/RocketMQMetadataProvider.java | 31 +++++++++--- .../apache/ApacheInstanceProviderTest.java | 10 ++++ .../apache/RocketMQMetadataProviderTest.java | 39 +++++++++++++++ .../pages/instance/__tests__/ConsumerPage.test.tsx | 53 +++++++++++++++++++++ .../pages/instance/__tests__/TopicPage.test.tsx | 2 + web/src/pages/instance/consumer.tsx | 55 +++++++++++++++------- web/src/pages/instance/topic.tsx | 4 ++ web/src/services/consumerService.test.ts | 25 +++++++++- web/src/services/consumerService.ts | 7 ++- 11 files changed, 205 insertions(+), 30 deletions(-) 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 eb1faa50..045f1564 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 @@ -90,9 +90,7 @@ public class ApacheInstanceProvider implements InstanceProvider { @Override public List<ConsumerGroupVO> listConsumerGroups(String instanceId, String search) { - return metadataProvider.listConsumerGroups(null, search).stream() - .filter(group -> matchesInstance(group.getInstanceId(), instanceId)) - .toList(); + return metadataProvider.listConsumerGroups(instanceId, null, search); } @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 9acfe40e..7d243478 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 @@ -28,6 +28,11 @@ import java.util.List; public interface MetadataProvider { List<TopicVO> listTopics(String clusterId, String type, String search); List<ConsumerGroupVO> listConsumerGroups(String clusterId, String search); + + default List<ConsumerGroupVO> listConsumerGroups(String instanceId, String clusterId, String search) { + return listConsumerGroups(clusterId, search); + } + List<BrokerRouteVO> getTopicRoutes(String instanceId, String name); List<TopicConsumerVO> getTopicConsumers(String instanceId, String name); List<QueueProgressVO> getGroupProgress(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 f63a1128..992644d4 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 @@ -156,7 +156,13 @@ public class RocketMQMetadataProvider implements MetadataProvider { @Override public List<ConsumerGroupVO> listConsumerGroups(String clusterId, String search) { + return listConsumerGroups(null, clusterId, search); + } + + @Override + public List<ConsumerGroupVO> listConsumerGroups(String instanceId, String clusterId, String search) { LambdaQueryWrapper<RmqGroup> query = new LambdaQueryWrapper<RmqGroup>() + .eq(StringUtils.hasText(instanceId), RmqGroup::getInstanceId, instanceId) .eq(StringUtils.hasText(clusterId), RmqGroup::getClusterId, clusterId) .like(StringUtils.hasText(search), RmqGroup::getName, search) .orderByAsc(RmqGroup::getName); @@ -175,8 +181,10 @@ public class RocketMQMetadataProvider implements MetadataProvider { vo.setCreatedAt(entity.getCreatedAt()); vo.setUpdatedAt(entity.getUpdatedAt()); - if (hasAdmin()) { - enrichGroupWithConnectionInfo(vo, entity.getName()); + if (StringUtils.hasText(instanceId)) { + enrichGroupWithConnectionInfo(vo, entity.getName(), instanceId); + } else if (hasAdmin()) { + enrichGroupWithConnectionInfo(vo, entity.getName(), null); } result.add(vo); } @@ -321,7 +329,7 @@ public class RocketMQMetadataProvider implements MetadataProvider { return consumers; } catch (Exception e) { log.warn("Failed to get consumers for topic {}: {}", name, e.getMessage()); - return Collections.emptyList(); + throw new BusinessException(502, "Failed to get consumers for topic " + name + ": " + e.getMessage()); } } @@ -412,15 +420,17 @@ public class RocketMQMetadataProvider implements MetadataProvider { return subscriptions; } catch (Exception e) { log.warn("Failed to get subscriptions for group {}: {}", name, e.getMessage()); - return Collections.emptyList(); + throw new BusinessException(502, + "Failed to get subscriptions for group " + name + ": " + e.getMessage()); } } // ── Helper methods ────────────────────────────────────────────────── - private void enrichGroupWithConnectionInfo(ConsumerGroupVO vo, String groupName) { + private void enrichGroupWithConnectionInfo(ConsumerGroupVO vo, String groupName, String instanceId) { try { - ConsumerConnection conn = adminExecute(admin -> admin.examineConsumerConnectionInfo(groupName)); + ConsumerConnection conn = executeForInstance(instanceId, + admin -> admin.examineConsumerConnectionInfo(groupName)); if (conn != null) { if (conn.getConnectionSet() != null) { vo.setOnlineInstances(conn.getConnectionSet().size()); @@ -435,7 +445,7 @@ public class RocketMQMetadataProvider implements MetadataProvider { // Try to get lag info try { - ConsumeStats stats = adminExecute(admin -> admin.examineConsumeStats(groupName)); + ConsumeStats stats = executeForInstance(instanceId, admin -> admin.examineConsumeStats(groupName)); if (stats != null && stats.getOffsetTable() != null) { long totalLag = 0; for (OffsetWrapper ow : stats.getOffsetTable().values()) { @@ -448,6 +458,13 @@ public class RocketMQMetadataProvider implements MetadataProvider { } } + private <T> T executeForInstance(String instanceId, MqAdminExtFactory.AdminAction<T> action) { + if (StringUtils.hasText(instanceId)) { + return runtimeAdminClientResolver.execute(instanceId, action); + } + return adminExecute(action); + } + private String filterMode(String expressionType) { if ("SQL92".equals(expressionType)) { return "SQL"; diff --git a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProviderTest.java b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProviderTest.java index 62db28a8..c0fa9b78 100644 --- a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProviderTest.java +++ b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProviderTest.java @@ -26,6 +26,7 @@ import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @ExtendWith(MockitoExtension.class) @@ -64,4 +65,13 @@ class ApacheInstanceProviderTest { assertThat(provider.countGroups("inst-1")).isEqualTo(2); } + + @Test + void listConsumerGroupsShouldPassTheSelectedInstanceToMetadataProvider() { + when(metadataProvider.listConsumerGroups("inst-1", null, "orders")).thenReturn(java.util.List.of()); + + assertThat(provider.listConsumerGroups("inst-1", "orders")).isEmpty(); + + verify(metadataProvider).listConsumerGroups("inst-1", null, "orders"); + } } 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 19459427..41def9bc 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 @@ -44,6 +44,7 @@ import static org.mockito.Mockito.lenient; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -102,6 +103,21 @@ class RocketMQMetadataProviderTest { assertThat(groups.get(0).getConsumeType()).isEqualTo(ConsumeType.CLUSTERING); } + @Test + void listConsumerGroupsShouldUseSelectedInstanceForRuntimeEnrichment() { + RmqGroup entity = new RmqGroup(); + entity.setName("group-a"); + entity.setInstanceId("instance-a"); + entity.setClusterId("cluster-a"); + when(groupMapper.selectList(any())).thenReturn(List.of(entity)); + when(runtimeAdminClientResolver.execute(eq("instance-a"), any())).thenReturn(null); + + List<ConsumerGroupVO> groups = newProvider().listConsumerGroups("instance-a", null, null); + + assertThat(groups).singleElement().extracting(ConsumerGroupVO::getName).isEqualTo("group-a"); + verify(runtimeAdminClientResolver, times(2)).execute(eq("instance-a"), any()); + } + @Test void getTopicRoutesShouldUseSelectedInstanceRuntimeClient() { List<BrokerRouteVO> routes = List.of(BrokerRouteVO.builder().brokerName("broker-a").build()); @@ -147,6 +163,17 @@ class RocketMQMetadataProviderTest { .satisfies(error -> assertThat(((BusinessException) error).getCode()).isEqualTo(502)); } + @Test + void getTopicConsumersSurfacesAdminFailure() throws Exception { + DefaultMQAdminExt admin = org.mockito.Mockito.mock(DefaultMQAdminExt.class); + when(admin.queryTopicConsumeByWho("TopicA")).thenThrow(new IllegalStateException("broker unavailable")); + + assertThatThrownBy(() -> newLiveProvider(admin).getTopicConsumers(null, "TopicA")) + .isInstanceOf(BusinessException.class) + .hasMessage("Failed to get consumers for topic TopicA: broker unavailable") + .satisfies(error -> assertThat(((BusinessException) error).getCode()).isEqualTo(502)); + } + @Test void getGroupProgressSurfacesAdminFailure() throws Exception { DefaultMQAdminExt admin = org.mockito.Mockito.mock(DefaultMQAdminExt.class); @@ -158,6 +185,18 @@ class RocketMQMetadataProviderTest { .satisfies(error -> assertThat(((BusinessException) error).getCode()).isEqualTo(502)); } + @Test + void getGroupSubscriptionsSurfacesAdminFailure() throws Exception { + DefaultMQAdminExt admin = org.mockito.Mockito.mock(DefaultMQAdminExt.class); + when(admin.examineConsumerConnectionInfo("group-a")) + .thenThrow(new IllegalStateException("broker unavailable")); + + assertThatThrownBy(() -> newLiveProvider(admin).getGroupSubscriptions(null, "group-a")) + .isInstanceOf(BusinessException.class) + .hasMessage("Failed to get subscriptions for group group-a: broker unavailable") + .satisfies(error -> assertThat(((BusinessException) error).getCode()).isEqualTo(502)); + } + private RocketMQMetadataProvider newLiveProvider(MQAdminExt admin) throws Exception { MqAdminExtFactory factory = mock(MqAdminExtFactory.class); RocketMQProperties liveProperties = new RocketMQProperties(); diff --git a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx index 4bd93b8f..66605e8b 100644 --- a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx +++ b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx @@ -262,6 +262,57 @@ describe('Consumer page', () => { ); }); + it('reloads same-named group diagnostics after changing the selected instance', 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', + }, + { + id: 'instance-b', + name: 'Instance B', + remark: '', + type: 'DIRECT', + endpoint: '127.0.0.2:9876', + topicCount: 0, + consumerGroupCount: 0, + createdAt: '2026-07-23T00:00:00Z', + updatedAt: '2026-07-23T00:00:00Z', + }, + ]); + vi.mocked(consumerService.listConsumerGroups).mockImplementation(async (params) => [ + { ...group, instanceId: params?.instanceId ?? 'instance-a' }, + ]); + const user = userEvent.setup(); + renderWithProviders(<ConsumerPage />, '/instance/instance-a/consumer'); + + await user.click(await screen.findByRole('button', { name: /详情/ })); + await waitFor(() => + expect(consumerService.getConsumerSubscriptions).toHaveBeenCalledWith('remote-cg', 'instance-a'), + ); + + await user.click(screen.getByText('Instance A')); + await user.click(await screen.findByText('Instance B')); + await waitFor(() => + expect(consumerService.listConsumerGroups).toHaveBeenCalledWith({ instanceId: 'instance-b' }), + ); + await user.click(await screen.findByRole('button', { name: /详情/ })); + + await waitFor(() => + expect(consumerService.getConsumerSubscriptions).toHaveBeenCalledWith('remote-cg', 'instance-b'), + ); + await waitFor(() => + expect(consumerService.getConsumerProgress).toHaveBeenCalledWith('remote-cg', 'instance-b'), + ); + }); + it('highlights inconsistent subscriptions and refreshes the check result', async () => { vi.mocked(consumerService.getConsumerSubscriptions) .mockResolvedValueOnce([ @@ -421,6 +472,8 @@ describe('Consumer page', () => { renderWithProviders(<ConsumerPage />); expect(await screen.findByText('选择实例')).toBeInTheDocument(); + expect(consumerService.listConsumerGroups).not.toHaveBeenCalled(); + expect(document.querySelector('.ant-spin-spinning')).toBeNull(); expect(screen.getByRole('button', { name: /导入/ })).toBeDisabled(); expect(screen.getByRole('button', { name: '创建 Group' })).toBeDisabled(); }); diff --git a/web/src/pages/instance/__tests__/TopicPage.test.tsx b/web/src/pages/instance/__tests__/TopicPage.test.tsx index 8713a88c..fde8eaa9 100644 --- a/web/src/pages/instance/__tests__/TopicPage.test.tsx +++ b/web/src/pages/instance/__tests__/TopicPage.test.tsx @@ -391,6 +391,8 @@ describe('TopicPage', () => { renderWithProviders(); expect(await screen.findByText('共 0 个 Topic')).toBeInTheDocument(); + expect(topicServiceMocks.listTopics).not.toHaveBeenCalled(); + expect(document.querySelector('.ant-spin-spinning')).toBeNull(); expect(screen.getByRole('button', { name: /导入/ })).toBeDisabled(); expect(screen.getByRole('button', { name: /创建 Topic/ })).toBeDisabled(); }); diff --git a/web/src/pages/instance/consumer.tsx b/web/src/pages/instance/consumer.tsx index b6c2b117..a11b4cba 100644 --- a/web/src/pages/instance/consumer.tsx +++ b/web/src/pages/instance/consumer.tsx @@ -176,6 +176,9 @@ const isConsistentSubscription = (subscription: SubscriptionEntry): boolean => const isInconsistentSubscription = (subscription: SubscriptionEntry): boolean => isInconsistentValue(subscription.consistency); +export const diagnosticCacheKey = (instanceId: string, groupName: string) => + `${instanceId}\u0000${groupName}`; + /* ═══════════════════════════════════════════ ConsumerPage ═══════════════════════════════════════════ */ @@ -222,8 +225,19 @@ const ConsumerPage = () => { const groupRequestIdRef = useRef(0); + useEffect(() => { + setSelectedGroup(null); + setModalOpen(false); + setResetGroup(null); + setResetModalOpen(false); + }, [selectedInstanceId]); + useEffect(() => { if (!selectedInstanceId) { + groupRequestIdRef.current += 1; + setGroups([]); + setSelectedRowKeys([]); + setLoading(false); return; } const requestId = ++groupRequestIdRef.current; @@ -247,20 +261,21 @@ const ConsumerPage = () => { const loadSubscriptions = useCallback( async (groupName: string, force = false) => { - if (!force && subscriptionsByGroup[groupName]) return; - setSubscriptionLoadingByGroup((prev) => ({ ...prev, [groupName]: true })); - setSubscriptionErrorByGroup((prev) => ({ ...prev, [groupName]: false })); + const cacheKey = diagnosticCacheKey(selectedInstanceId, groupName); + if (!force && subscriptionsByGroup[cacheKey]) return; + setSubscriptionLoadingByGroup((prev) => ({ ...prev, [cacheKey]: true })); + setSubscriptionErrorByGroup((prev) => ({ ...prev, [cacheKey]: false })); try { const subscriptions = await getConsumerSubscriptions( groupName, selectedInstanceId || undefined, ); - setSubscriptionsByGroup((prev) => ({ ...prev, [groupName]: subscriptions })); + setSubscriptionsByGroup((prev) => ({ ...prev, [cacheKey]: subscriptions })); } catch { - setSubscriptionErrorByGroup((prev) => ({ ...prev, [groupName]: true })); + setSubscriptionErrorByGroup((prev) => ({ ...prev, [cacheKey]: true })); message.error(t('consumer.fetchSubscriptionsFailed', { name: groupName })); } finally { - setSubscriptionLoadingByGroup((prev) => ({ ...prev, [groupName]: false })); + setSubscriptionLoadingByGroup((prev) => ({ ...prev, [cacheKey]: false })); } }, [subscriptionsByGroup, t, selectedInstanceId], @@ -268,10 +283,11 @@ const ConsumerPage = () => { const loadProgress = useCallback( async (groupName: string) => { - if (progressByGroup[groupName]) return; + const cacheKey = diagnosticCacheKey(selectedInstanceId, groupName); + if (progressByGroup[cacheKey]) return; try { const progress = await getConsumerProgress(groupName, selectedInstanceId || undefined); - setProgressByGroup((prev) => ({ ...prev, [groupName]: progress })); + setProgressByGroup((prev) => ({ ...prev, [cacheKey]: progress })); } catch { message.error(t('consumer.fetchProgressFailed', { name: groupName })); } @@ -311,8 +327,11 @@ const ConsumerPage = () => { void loadProgress(group.name); }; + const selectedDiagnosticKey = selectedGroup + ? diagnosticCacheKey(selectedInstanceId, selectedGroup.name) + : ''; const selectedSubscriptions = selectedGroup - ? (subscriptionsByGroup[selectedGroup.name] ?? []) + ? (subscriptionsByGroup[selectedDiagnosticKey] ?? []) : []; const inconsistentSubscriptions = selectedSubscriptions.filter(isInconsistentSubscription); const unknownSubscriptions = selectedSubscriptions.filter( @@ -322,7 +341,7 @@ const ConsumerPage = () => { const visibleSubscriptions = showOnlyInconsistent ? inconsistentSubscriptions : selectedSubscriptions; - const selectedProgress = selectedGroup ? (progressByGroup[selectedGroup.name] ?? []) : []; + const selectedProgress = selectedGroup ? (progressByGroup[selectedDiagnosticKey] ?? []) : []; const handleImportFile = async (file: File) => { if (!selectedInstanceId) { @@ -895,9 +914,11 @@ const ConsumerPage = () => { <div style={{ padding: '8px 0' }}> <Table columns={subscriptionSubColumns} - dataSource={subscriptionsByGroup[record.name] ?? []} + dataSource={ + subscriptionsByGroup[diagnosticCacheKey(selectedInstanceId, record.name)] ?? [] + } rowKey="topic" - loading={subscriptionLoadingByGroup[record.name]} + loading={subscriptionLoadingByGroup[diagnosticCacheKey(selectedInstanceId, record.name)]} pagination={false} size="small" /> @@ -1073,7 +1094,7 @@ const ConsumerPage = () => { <Button size="small" icon={<ArrowsClockwise size={14} />} - loading={subscriptionLoadingByGroup[selectedGroup.name]} + loading={subscriptionLoadingByGroup[selectedDiagnosticKey]} onClick={() => { setShowOnlyInconsistent(false); void loadSubscriptions(selectedGroup.name, true); @@ -1085,7 +1106,7 @@ const ConsumerPage = () => { <Alert showIcon type={ - subscriptionErrorByGroup[selectedGroup.name] + subscriptionErrorByGroup[selectedDiagnosticKey] ? 'error' : inconsistentSubscriptions.length > 0 || unknownSubscriptions.length > 0 @@ -1095,9 +1116,9 @@ const ConsumerPage = () => { : 'info' } message={ - subscriptionErrorByGroup[selectedGroup.name] + subscriptionErrorByGroup[selectedDiagnosticKey] ? '订阅一致性检查失败，当前保留上次检查结果' - : subscriptionLoadingByGroup[selectedGroup.name] && + : subscriptionLoadingByGroup[selectedDiagnosticKey] && selectedSubscriptions.length === 0 ? '正在检查订阅一致性' : inconsistentSubscriptions.length > 0 @@ -1123,7 +1144,7 @@ const ConsumerPage = () => { columns={subscriptionSubColumns} dataSource={visibleSubscriptions} rowKey="topic" - loading={subscriptionLoadingByGroup[selectedGroup.name]} + loading={subscriptionLoadingByGroup[selectedDiagnosticKey]} pagination={false} size="small" /> diff --git a/web/src/pages/instance/topic.tsx b/web/src/pages/instance/topic.tsx index 19c05dd6..64f74386 100644 --- a/web/src/pages/instance/topic.tsx +++ b/web/src/pages/instance/topic.tsx @@ -307,6 +307,10 @@ const TopicPage = () => { useEffect(() => { if (!selectedInstanceId) { + topicRequestIdRef.current += 1; + setTopics([]); + setSelectedRowKeys([]); + setLoading(false); return; } const requestId = ++topicRequestIdRef.current; diff --git a/web/src/services/consumerService.test.ts b/web/src/services/consumerService.test.ts index 4d61b45d..27ef23b2 100644 --- a/web/src/services/consumerService.test.ts +++ b/web/src/services/consumerService.test.ts @@ -24,10 +24,16 @@ import { listConsumerGroups, } from './consumerService'; -vi.mock('./dataMode', () => ({ isMockMode: () => true })); +const { mode, metadataApi } = vi.hoisted(() => ({ + mode: { mock: true }, + metadataApi: { getConsumerGroup: vi.fn() }, +})); + +vi.mock('./dataMode', () => ({ isMockMode: () => mode.mock })); vi.mock('../config', () => ({ API_BASE_URL: '/api', })); +vi.mock('../api/metadata', () => metadataApi); describe('consumer service mock data', () => { it('returns copied consumer group rows', async () => { @@ -93,4 +99,21 @@ describe('consumer service mock data', () => { expect(detail.subscribedTopics).toEqual(['created-topic']); expect(detail).not.toBe(created); }); + + it('forwards the selected instance when loading consumer group details in API mode', async () => { + mode.mock = false; + const detail = { + name: 'cg-orders', + subscribedTopics: [], + instances: [], + }; + try { + metadataApi.getConsumerGroup.mockResolvedValue(detail); + + await expect(getConsumerGroup('cg-orders', 'instance-a')).resolves.toEqual(detail); + expect(metadataApi.getConsumerGroup).toHaveBeenCalledWith('cg-orders', 'instance-a'); + } finally { + mode.mock = true; + } + }); }); diff --git a/web/src/services/consumerService.ts b/web/src/services/consumerService.ts index b48f209a..dfabd684 100644 --- a/web/src/services/consumerService.ts +++ b/web/src/services/consumerService.ts @@ -61,13 +61,16 @@ export async function getConsumerProgress( return metadataApi.getConsumerProgress(name, instanceId); } -export async function getConsumerGroup(name: string): Promise<ConsumerGroupDetail> { +export async function getConsumerGroup( + name: string, + instanceId?: string, +): Promise<ConsumerGroupDetail> { if (isMockMode()) { const group = mockConsumerGroups.find((item) => item.name === name); if (!group) throw new Error(`Consumer group not found: ${name}`); return copyConsumerGroup(group as unknown as ConsumerGroupDetail) as ConsumerGroupDetail; } - return metadataApi.getConsumerGroup(name); + return metadataApi.getConsumerGroup(name, instanceId); } export async function getConsumerSubscriptions(
