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 caf4bbc4 fix(consumer): require topic selection for offset reset
(#1633)
caf4bbc4 is described below
commit caf4bbc4af360d0f1c938177c1d30b882f6d4944
Author: youngkermit8-coder <[email protected]>
AuthorDate: Tue Aug 11 20:42:32 2026 +0800
fix(consumer): require topic selection for offset reset (#1633)
Signed-off-by: youngkermit8-coder <[email protected]>
---
web/src/api/metadata.ts | 4 +-
.../pages/instance/__tests__/ConsumerPage.test.tsx | 40 ++++++++++++
web/src/pages/instance/consumer.tsx | 75 +++++++++++++++++-----
3 files changed, 100 insertions(+), 19 deletions(-)
diff --git a/web/src/api/metadata.ts b/web/src/api/metadata.ts
index 966a71af..00f7fe0d 100644
--- a/web/src/api/metadata.ts
+++ b/web/src/api/metadata.ts
@@ -121,7 +121,7 @@ export interface ConsumerGroupQuery {
export interface ResetConsumerOffsetRequest {
name: string;
timestamp: number;
- topic?: string;
+ topic: string;
}
// ─── Topic API ──────────────────────────────────────────────────
@@ -231,7 +231,7 @@ export interface ResetConsumerOffsetRequest {
name: string;
instanceId?: string;
timestamp: number;
- topic?: string;
+ topic: 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 c3f2262b..019f1b37 100644
--- a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
+++ b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
@@ -280,6 +280,46 @@ describe('Consumer page', () => {
);
});
+ it('requires and submits a target topic when resetting consumer offsets',
async () => {
+ const user = userEvent.setup();
+ renderWithProviders(<ConsumerPage />);
+
+ await user.click(await screen.findByRole('button', { name: /重置位点/ }));
+
+ await waitFor(() =>
+ expect(consumerService.getConsumerSubscriptions).toHaveBeenCalledWith(
+ 'remote-cg',
+ 'instance-1',
+ ),
+ );
+ const confirm = screen.getByRole('button', { name: '确认重置' });
+ expect(confirm).toBeDisabled();
+
+ const topicSelect = screen.getByRole('combobox', { name: '目标 Topic' });
+ await user.click(topicSelect);
+ const option = await waitFor(() => {
+ const element = screen
+ .getAllByText('remote-topic')
+ .find((candidate) =>
candidate.classList.contains('ant-select-item-option-content'));
+ if (!element) throw new Error('Missing target Topic option');
+ return element;
+ });
+ await user.click(option);
+ await waitFor(() => expect(confirm).toBeEnabled());
+ await user.click(confirm);
+
+ await waitFor(() =>
+ expect(consumerService.resetConsumerOffset).toHaveBeenCalledWith(
+ expect.objectContaining({
+ name: 'remote-cg',
+ instanceId: 'instance-1',
+ topic: 'remote-topic',
+ timestamp: expect.any(Number),
+ }),
+ ),
+ );
+ });
+
it('reloads same-named group diagnostics after changing the selected
instance', async () => {
vi.mocked(instanceService.listInstances).mockResolvedValue([
{
diff --git a/web/src/pages/instance/consumer.tsx
b/web/src/pages/instance/consumer.tsx
index 2940cdd9..b1611145 100644
--- a/web/src/pages/instance/consumer.tsx
+++ b/web/src/pages/instance/consumer.tsx
@@ -206,6 +206,7 @@ const ConsumerPage = () => {
const [dataTypeValue, setDataTypeValue] = useState<string |
undefined>(undefined);
const [resetModalOpen, setResetModalOpen] = useState(false);
const [resetGroup, setResetGroup] = useState<ConsumerGroup | null>(null);
+ const [resetTopic, setResetTopic] = useState<string>();
const [resetTime, setResetTime] = useState<Dayjs>(dayjs().subtract(3,
'hour'));
const [subscriptionsByGroup, setSubscriptionsByGroup] = useState<
Record<string, SubscriptionEntry[]>
@@ -236,6 +237,7 @@ const ConsumerPage = () => {
setSelectedGroup(null);
setModalOpen(false);
setResetGroup(null);
+ setResetTopic(undefined);
setResetModalOpen(false);
}, [selectedInstanceId]);
@@ -338,6 +340,16 @@ const ConsumerPage = () => {
const selectedDiagnosticKey = selectedGroup
? diagnosticCacheKey(selectedInstanceId, selectedGroup.name)
: '';
+ const resetDiagnosticKey = resetGroup
+ ? diagnosticCacheKey(selectedInstanceId, resetGroup.name)
+ : '';
+ const resetTopicOptions = useMemo(() => {
+ const topics = new Set(resetGroup?.subscribedTopics ?? []);
+ for (const subscription of subscriptionsByGroup[resetDiagnosticKey] ?? [])
{
+ if (subscription.topic) topics.add(subscription.topic);
+ }
+ return Array.from(topics).map((topic) => ({ label: topic, value: topic }));
+ }, [resetDiagnosticKey, resetGroup, subscriptionsByGroup]);
const selectedSubscriptions = selectedGroup
? (subscriptionsByGroup[selectedDiagnosticKey] ?? [])
: [];
@@ -574,8 +586,10 @@ const ConsumerPage = () => {
onClick={(e) => {
e.stopPropagation();
setResetGroup(record);
+ setResetTopic(undefined);
setResetTime(dayjs().subtract(3, 'hour'));
setResetModalOpen(true);
+ void loadSubscriptions(record.name);
}}
>
重置位点
@@ -1574,30 +1588,36 @@ const ConsumerPage = () => {
onCancel={() => {
setResetModalOpen(false);
setResetGroup(null);
+ setResetTopic(undefined);
}}
onOk={async () => {
- if (resetGroup) {
- setResetSubmitting(true);
- try {
- await resetConsumerOffset({
- name: resetGroup.name,
- instanceId: selectedInstanceId || undefined,
- timestamp: resetTime.valueOf(),
- });
- message.success(
- `${resetGroup.name} 消费位点已重置到 ${resetTime.format('YYYY-MM-DD
HH:mm:ss')}`,
- );
- } catch {
- message.error(t('consumer.resetFailed'));
- return;
- } finally {
- setResetSubmitting(false);
- }
+ if (!resetGroup || !resetTopic) return;
+ setResetSubmitting(true);
+ try {
+ await resetConsumerOffset({
+ name: resetGroup.name,
+ instanceId: selectedInstanceId || undefined,
+ topic: resetTopic,
+ timestamp: resetTime.valueOf(),
+ });
+ message.success(
+ `${resetGroup.name} 在 ${resetTopic} 的消费位点已重置到
${resetTime.format('YYYY-MM-DD HH:mm:ss')}`,
+ );
+ } catch {
+ message.error(t('consumer.resetFailed'));
+ return;
+ } finally {
+ setResetSubmitting(false);
}
setResetModalOpen(false);
setResetGroup(null);
+ setResetTopic(undefined);
}}
confirmLoading={resetSubmitting}
+ okButtonProps={{
+ disabled:
+ !resetTopic ||
Boolean(subscriptionLoadingByGroup[resetDiagnosticKey]),
+ }}
okText="确认重置"
cancelText="取消"
width={480}
@@ -1626,6 +1646,27 @@ const ConsumerPage = () => {
{resetGroup.name}
</Text>
</div>
+ <div style={{ marginBottom: 16 }}>
+ <Text type="secondary" style={{ fontSize: 13, display: 'block',
marginBottom: 8 }}>
+ 目标 Topic
+ </Text>
+ <Select
+ aria-label="目标 Topic"
+ showSearch
+ optionFilterProp="label"
+ style={{ width: '100%' }}
+ value={resetTopic}
+ options={resetTopicOptions}
+ loading={subscriptionLoadingByGroup[resetDiagnosticKey]}
+ placeholder="选择要重置消费位点的 Topic"
+ onChange={setResetTopic}
+ notFoundContent={
+ subscriptionErrorByGroup[resetDiagnosticKey]
+ ? '订阅 Topic 加载失败'
+ : '该 Group 暂无订阅 Topic'
+ }
+ />
+ </div>
<div style={{ marginBottom: 16 }}>
<Text type="secondary" style={{ fontSize: 13, display: 'block',
marginBottom: 8 }}>
重置到以下时间点
