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 f730a827 feat: import topics and consumer groups from CSV (#1013)
f730a827 is described below
commit f730a8270d6e955aaecbbbaa16d991bed4c109d5
Author: yx9o <[email protected]>
AuthorDate: Wed Aug 5 17:46:01 2026 +0800
feat: import topics and consumer groups from CSV (#1013)
---
.../pages/instance/__tests__/ConsumerPage.test.tsx | 75 ++++-
.../pages/instance/__tests__/TopicPage.test.tsx | 102 ++++++
web/src/pages/instance/consumer.tsx | 175 +++++++++-
web/src/pages/instance/topic.tsx | 173 +++++++++-
web/src/utils/resourceCsvImport.test.ts | 111 +++++++
web/src/utils/resourceCsvImport.ts | 364 +++++++++++++++++++++
6 files changed, 995 insertions(+), 5 deletions(-)
diff --git a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
index 4324a96e..fd191da3 100644
--- a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
+++ b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
@@ -94,6 +94,22 @@ describe('Consumer page', () => {
beforeEach(() => {
vi.clearAllMocks();
vi.mocked(consumerService.listConsumerGroups).mockResolvedValue([group]);
+ vi.mocked(consumerService.createConsumerGroup).mockImplementation(
+ async (data: Partial<ConsumerGroup>) =>
+ ({
+ ...group,
+ ...data,
+ namespace: 'default',
+ clusterId: 'server-cluster',
+ onlineInstances: 0,
+ totalLag: 0,
+ delaySeconds: 0,
+ instances: [],
+ subscribedTopics: data.subscribedTopics ?? [],
+ createdAt: '2026-07-24T00:00:00Z',
+ updatedAt: '2026-07-24T00:00:00Z',
+ }) as ConsumerGroup,
+ );
vi.mocked(consumerService.getConsumerProgress).mockResolvedValue([
{
broker: 'broker-a',
@@ -154,7 +170,9 @@ describe('Consumer page', () => {
expect(URL.createObjectURL).toHaveBeenCalledTimes(1);
expect(clickSpy).toHaveBeenCalledTimes(1);
expect(URL.revokeObjectURL).toHaveBeenCalledWith('blob:consumer-group-export');
-
expect(document.querySelector('a[download^="rocketmq-consumer-groups-"]')).not.toBeInTheDocument();
+ expect(
+ document.querySelector('a[download^="rocketmq-consumer-groups-"]'),
+ ).not.toBeInTheDocument();
expect(exportedBlob).toBeDefined();
const csv = await exportedBlob!.text();
@@ -264,4 +282,59 @@ describe('Consumer page', () => {
expect(await
screen.findByText('订阅一致性检查失败，当前保留上次检查结果')).toBeInTheDocument();
});
+
+ it('keeps per-row state when consumer group CSV import partially fails',
async () => {
+ vi.mocked(consumerService.listConsumerGroups).mockResolvedValue([]);
+ vi.mocked(consumerService.createConsumerGroup).mockImplementation(
+ async (data: Partial<ConsumerGroup>) => {
+ if (data.name === 'cg-fail') throw new Error('broker rejected group');
+ return {
+ ...group,
+ ...data,
+ name: data.name ?? '',
+ namespace: 'default',
+ clusterId: 'server-cluster',
+ onlineInstances: 0,
+ totalLag: 0,
+ delaySeconds: 0,
+ instances: [],
+ subscribedTopics: data.subscribedTopics ?? [],
+ createdAt: '2026-07-24T00:00:00Z',
+ updatedAt: '2026-07-24T00:00:00Z',
+ } as ConsumerGroup;
+ },
+ );
+
+ const user = userEvent.setup();
+ renderWithProviders(<ConsumerPage />);
+
+ await screen.findByText(/共 0 个 Group/);
+ const csv = [
+ '"Name","Subscription Mode","Consume Type","Retry Max
Times","Subscription Data Type","Delivery Order Type"',
+ '"cg-ok","Push","CLUSTERING","16","NORMAL",""',
+ '"cg-fail","Pop","BROADCASTING","4","FIFO","PARTITON_ORDER"',
+ ].join('\n');
+ await user.upload(
+ screen.getByTestId('consumer-group-import-file'),
+ new File([csv], 'groups.csv'),
+ );
+ expect(await screen.findByText('检测到 2 个
Group，将按顺序调用创建接口')).toBeInTheDocument();
+ await user.click(screen.getByRole('button', { name: '开始导入' }));
+
+ await waitFor(() =>
expect(consumerService.createConsumerGroup).toHaveBeenCalledTimes(2));
+ expect(consumerService.createConsumerGroup).toHaveBeenNthCalledWith(1, {
+ name: 'cg-ok',
+ subscriptionMode: 'Push',
+ consumeType: 'CLUSTERING',
+ retryMaxTimes: 16,
+ subscriptionDataType: 'NORMAL',
+ subscribedTopics: [],
+ });
+ expect(await screen.findByText('已导入 1 个 Group，1 个失败')).toBeInTheDocument();
+ expect(screen.getByText('broker rejected group')).toBeInTheDocument();
+ expect(screen.getAllByText('cg-ok').length).toBeGreaterThan(0);
+ await waitFor(() =>
+ expect(screen.getByRole('button', { name: /重试失败项/
})).toBeInTheDocument(),
+ );
+ });
});
diff --git a/web/src/pages/instance/__tests__/TopicPage.test.tsx
b/web/src/pages/instance/__tests__/TopicPage.test.tsx
index 88b18a69..05ae8715 100644
--- a/web/src/pages/instance/__tests__/TopicPage.test.tsx
+++ b/web/src/pages/instance/__tests__/TopicPage.test.tsx
@@ -109,6 +109,17 @@ describe('TopicPage', () => {
beforeEach(() => {
topicServiceMocks.listTopics.mockResolvedValue(buildTopics(25));
topicServiceMocks.batchDeleteTopics.mockResolvedValue({ deleted: [],
failed: [] });
+ topicServiceMocks.createTopic.mockImplementation(async (data:
Partial<Topic>) => ({
+ ...buildTopics(1)[0],
+ ...data,
+ namespace: 'default',
+ clusterId: 'server-cluster',
+ messageCount: 0,
+ tps: 0,
+ consumerGroupCount: 0,
+ createdAt: '2026-01-02T00:00:00Z',
+ updatedAt: '2026-01-02T00:00:00Z',
+ }));
topicServiceMocks.getTopicRoutes.mockResolvedValue([]);
topicServiceMocks.getTopicConsumers.mockResolvedValue([]);
instanceServiceMocks.listInstances.mockResolvedValue([]);
@@ -245,4 +256,95 @@ describe('TopicPage', () => {
expect(screen.queryByText('topic-b')).not.toBeInTheDocument();
expect(screen.getByText('10.0.2.21:8080')).toBeInTheDocument();
});
+
+ it('imports valid topic CSV rows through the create service with the
selected instance', async () => {
+ const user = userEvent.setup();
+ topicServiceMocks.listTopics.mockResolvedValue([]);
+ instanceServiceMocks.listInstances.mockResolvedValue([
+ {
+ id: 'instance-proxy-1',
+ name: 'instance-proxy-1',
+ remark: '',
+ type: 'PROXY',
+ endpoint: '10.0.2.21:8080',
+ topicCount: 0,
+ consumerGroupCount: 0,
+ createdAt: '2026-01-01T00:00:00Z',
+ updatedAt: '2026-01-01T00:00:00Z',
+ },
+ ]);
+ renderWithProviders('/instance/instance-proxy-1/topic');
+
+ await screen.findByText(/共 0 个 Topic/);
+ const csv = [
+ '"Name","Namespace","Type","Cluster ID","Write Queues","Read
Queues","Permission","Remark"',
+
'"imported-topic","ignored","NORMAL","ignored-cluster","4","6","RW","orders"',
+ ].join('\n');
+ await user.upload(screen.getByTestId('topic-import-file'), new File([csv],
'topics.csv'));
+ expect(await screen.findByText('检测到 1 个
Topic，将按顺序调用创建接口')).toBeInTheDocument();
+ await user.click(screen.getByRole('button', { name: '开始导入' }));
+
+ await waitFor(() =>
+ expect(topicServiceMocks.createTopic).toHaveBeenCalledWith({
+ name: 'imported-topic',
+ type: 'NORMAL',
+ writeQueues: 4,
+ readQueues: 6,
+ perm: 'RW',
+ remark: 'orders',
+ instanceId: 'instance-proxy-1',
+ }),
+ );
+ expect(await screen.findByText('已导入 1 个 Topic')).toBeInTheDocument();
+ expect(screen.getAllByText('imported-topic').length).toBeGreaterThan(0);
+ });
+
+ it('does not call createTopic when imported topic CSV is invalid or
duplicated', async () => {
+ const user = userEvent.setup();
+ renderWithProviders();
+
+ expect(await screen.findByText('topic-01')).toBeInTheDocument();
+ const csv = [
+ '"Name","Type","Write Queues","Read Queues","Permission"',
+ '"bad topic","NORMAL","8","8","RW"',
+ '"bad topic","NORMAL","8","8","RW"',
+ ].join('\n');
+ await user.upload(screen.getByTestId('topic-import-file'), new File([csv],
'bad.csv'));
+
+ expect(await screen.findByText('检测到 2 行无效，将跳过这些行')).toBeInTheDocument();
+ expect(screen.getAllByText(/Name 仅支持/).length).toBeGreaterThan(0);
+ expect(screen.getByText(/重复/)).toBeInTheDocument();
+ expect(screen.getByRole('button', { name: '开始导入' })).toBeDisabled();
+ expect(topicServiceMocks.createTopic).not.toHaveBeenCalled();
+ });
+
+ it('imports valid topic rows while skipping duplicate rows', async () => {
+ const user = userEvent.setup();
+ topicServiceMocks.listTopics.mockResolvedValue([]);
+ renderWithProviders();
+
+ await screen.findByText(/共 0 个 Topic/);
+ const csv = [
+ '"Name","Type","Write Queues","Read Queues","Permission"',
+ '"topic-a","NORMAL","8","8","RW"',
+ '"topic-a","NORMAL","8","8","RW"',
+ '"topic-b","FIFO","4","4","RW"',
+ ].join('\n');
+ await user.upload(screen.getByTestId('topic-import-file'), new File([csv],
'dedup.csv'));
+
+ expect(await screen.findByText('检测到 1 行无效，将跳过这些行')).toBeInTheDocument();
+ expect(screen.getByText(/Name 与第 2 行重复/)).toBeInTheDocument();
+ await user.click(screen.getByRole('button', { name: '开始导入' }));
+
+ await waitFor(() =>
expect(topicServiceMocks.createTopic).toHaveBeenCalledTimes(2));
+ expect(topicServiceMocks.createTopic).toHaveBeenNthCalledWith(
+ 1,
+ expect.objectContaining({ name: 'topic-a' }),
+ );
+ expect(topicServiceMocks.createTopic).toHaveBeenNthCalledWith(
+ 2,
+ expect.objectContaining({ name: 'topic-b' }),
+ );
+ expect(await screen.findByText('已导入 2 个 Topic，1
行无效已跳过')).toBeInTheDocument();
+ });
});
diff --git a/web/src/pages/instance/consumer.tsx
b/web/src/pages/instance/consumer.tsx
index b3e57efe..742e0d45 100644
--- a/web/src/pages/instance/consumer.tsx
+++ b/web/src/pages/instance/consumer.tsx
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-import { useCallback, useEffect, useMemo, useState } from 'react';
+import { useCallback, useEffect, useMemo, useRef, useState } from 'react';
import {
Alert,
Table,
@@ -78,6 +78,11 @@ import {
resetConsumerOffset,
} from '../../services/consumerService';
import { useInstanceFilter } from '../../hooks/useInstanceFilter';
+import {
+ parseCsvTable,
+ validateConsumerGroupCsvImport,
+ type ResourceImportRow,
+} from '../../utils/resourceCsvImport';
const { Text } = Typography;
@@ -204,6 +209,12 @@ const ConsumerPage = () => {
);
const [showOnlyInconsistent, setShowOnlyInconsistent] = useState(false);
const [progressByGroup, setProgressByGroup] = useState<Record<string,
QueueProgress[]>>({});
+ const importInputRef = useRef<HTMLInputElement>(null);
+ const [importModalOpen, setImportModalOpen] = useState(false);
+ const [importFilename, setImportFilename] = useState('');
+ const [importRows, setImportRows] =
useState<ResourceImportRow<Partial<ConsumerGroup>>[]>([]);
+ const [importErrors, setImportErrors] = useState<string[]>([]);
+ const [importing, setImporting] = useState(false);
useEffect(() => {
let cancelled = false;
@@ -301,6 +312,94 @@ const ConsumerPage = () => {
: selectedSubscriptions;
const selectedProgress = selectedGroup ?
(progressByGroup[selectedGroup.name] ?? []) : [];
+ const handleImportFile = async (file: File) => {
+ setImportFilename(file.name);
+ setImporting(false);
+ setImportModalOpen(true);
+ try {
+ const records = parseCsvTable(await file.text());
+ const validation = validateConsumerGroupCsvImport(records,
selectedInstanceId || undefined);
+ setImportRows(validation.rows);
+ setImportErrors(validation.errors);
+ } catch (error) {
+ setImportRows([]);
+ setImportErrors([error instanceof Error ? error.message : 'CSV 解析失败']);
+ } finally {
+ if (importInputRef.current) importInputRef.current.value = '';
+ }
+ };
+
+ const handleImportConsumerGroups = async () => {
+ const targetIndexes = importRows
+ .map((row, index) => ({ row, index }))
+ .filter(({ row }) => row.status === 'pending' || row.status ===
'failed');
+ if (targetIndexes.length === 0 || importErrors.length > 0) return;
+
+ setImporting(true);
+ const nextRows = importRows.map((row) => ({ ...row }));
+ const createdGroups: ConsumerGroup[] = [];
+
+ for (const { row, index } of targetIndexes) {
+ try {
+ const created = await createConsumerGroup(row.payload);
+ createdGroups.push(created);
+ nextRows[index] = { ...nextRows[index], status: 'success', message:
'已创建' };
+ } catch (error) {
+ nextRows[index] = {
+ ...nextRows[index],
+ status: 'failed',
+ message: error instanceof Error ? error.message : '创建失败',
+ };
+ }
+ setImportRows([...nextRows]);
+ }
+
+ if (createdGroups.length > 0) {
+ setGroups((previous) => {
+ const createdNames = new Set(createdGroups.map((group) => group.name));
+ return [...createdGroups, ...previous.filter((group) =>
!createdNames.has(group.name))];
+ });
+ }
+
+ const failedCount = nextRows.filter((row) => row.status ===
'failed').length;
+ const invalidCount = nextRows.filter((row) => row.status ===
'invalid').length;
+ if (failedCount === 0) {
+ if (invalidCount > 0) {
+ message.warning(`已导入 ${createdGroups.length} 个 Group，${invalidCount}
行无效已跳过`);
+ } else {
+ message.success(`已导入 ${createdGroups.length} 个 Group`);
+ }
+ } else if (createdGroups.length > 0) {
+ message.warning(`已导入 ${createdGroups.length} 个 Group，${failedCount}
个失败`);
+ } else {
+ message.error(`${failedCount} 个 Group 导入失败`);
+ }
+ setImporting(false);
+ };
+
+ const consumerGroupImportColumns:
ColumnsType<ResourceImportRow<Partial<ConsumerGroup>>> = [
+ { title: '行号', dataIndex: 'lineNumber', key: 'lineNumber', width: 80 },
+ { title: 'Group 名称', dataIndex: 'name', key: 'name' },
+ {
+ title: '状态',
+ dataIndex: 'status',
+ key: 'status',
+ width: 100,
+ render: (status: ResourceImportRow<Partial<ConsumerGroup>>['status']) =>
{
+ if (status === 'success') return <Tag color="success">成功</Tag>;
+ if (status === 'failed') return <Tag color="error">失败</Tag>;
+ if (status === 'invalid') return <Tag color="warning">无效</Tag>;
+ return <Tag>待导入</Tag>;
+ },
+ },
+ {
+ title: '说明',
+ dataIndex: 'message',
+ key: 'message',
+ render: (text?: string) => text || '-',
+ },
+ ];
+
/* ═══════════════════════════════════════════
Main Table Columns
═══════════════════════════════════════════ */
@@ -698,7 +797,22 @@ const ConsumerPage = () => {
删除 ({selectedRowKeys.length})
</Button>
)}
- <Button icon={<ImportOutlined />} onClick={() =>
message.info('导入功能开发中')}>
+ <input
+ ref={importInputRef}
+ type="file"
+ accept=".csv,text/csv"
+ data-testid="consumer-group-import-file"
+ style={{ display: 'none' }}
+ onChange={(event) => {
+ const file = event.target.files?.[0];
+ if (file) void handleImportFile(file);
+ }}
+ />
+ <Button
+ icon={<ImportOutlined />}
+ disabled={importing}
+ onClick={() => importInputRef.current?.click()}
+ >
导入
</Button>
<Button
@@ -1200,6 +1314,63 @@ const ConsumerPage = () => {
</Form>
</Modal>
+ {/* ═══════════════════════════════════════════
+ Import Group Modal
+ ═══════════════════════════════════════════ */}
+ <Modal
+ title={`导入 Group${importFilename ? `：${importFilename}` : ''}`}
+ open={importModalOpen}
+ onCancel={() => {
+ if (!importing) setImportModalOpen(false);
+ }}
+ onOk={() => void handleImportConsumerGroups()}
+ okText={importRows.some((row) => row.status === 'failed') ? '重试失败项' :
'开始导入'}
+ cancelText="关闭"
+ confirmLoading={importing}
+ okButtonProps={{
+ disabled:
+ importErrors.length > 0 ||
+ importRows.length === 0 ||
+ importRows.every((row) => row.status === 'success' || row.status
=== 'invalid'),
+ }}
+ width={720}
+ destroyOnClose
+ >
+ <Space direction="vertical" style={{ width: '100%' }} size={12}>
+ {importErrors.length > 0 ? (
+ <Alert
+ type="error"
+ showIcon
+ message="CSV 无法导入"
+ description={importErrors.join('；')}
+ />
+ ) : importRows.some((row) => row.status === 'invalid') ? (
+ <Alert
+ type="warning"
+ showIcon
+ message={`检测到 ${
+ importRows.filter((row) => row.status === 'invalid').length
+ } 行无效，将跳过这些行`}
+ description="仅导入可创建字段；CSV 中的 Namespace、Cluster ID 和运行状态列会被忽略。"
+ />
+ ) : (
+ <Alert
+ type="info"
+ showIcon
+ message={`检测到 ${importRows.length} 个 Group，将按顺序调用创建接口`}
+ description="仅导入可创建字段；CSV 中的 Namespace、Cluster ID 和运行状态列会被忽略。"
+ />
+ )}
+ <Table<ResourceImportRow<Partial<ConsumerGroup>>>
+ columns={consumerGroupImportColumns}
+ dataSource={importRows}
+ rowKey="key"
+ size="small"
+ pagination={false}
+ />
+ </Space>
+ </Modal>
+
{/* ═══════════════════════════════════════════
Reset Offset Modal
═══════════════════════════════════════════ */}
diff --git a/web/src/pages/instance/topic.tsx b/web/src/pages/instance/topic.tsx
index e4581205..4dbe5120 100644
--- a/web/src/pages/instance/topic.tsx
+++ b/web/src/pages/instance/topic.tsx
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-import { useEffect, useState, useMemo } from 'react';
+import { useEffect, useState, useMemo, useRef } from 'react';
import {
Alert,
Table,
@@ -66,6 +66,11 @@ import {
sendTopicMessage,
} from '../../services/topicService';
import { useInstanceFilter } from '../../hooks/useInstanceFilter';
+import {
+ parseCsvTable,
+ validateTopicCsvImport,
+ type ResourceImportRow,
+} from '../../utils/resourceCsvImport';
const { Text } = Typography;
@@ -288,6 +293,12 @@ const TopicPage = () => {
const [sendForm] = Form.useForm();
const [propsMode, setPropsMode] = useState<'form' | 'text'>('form');
const { modal } = App.useApp();
+ const importInputRef = useRef<HTMLInputElement>(null);
+ const [importModalOpen, setImportModalOpen] = useState(false);
+ const [importFilename, setImportFilename] = useState('');
+ const [importRows, setImportRows] =
useState<ResourceImportRow<Partial<Topic>>[]>([]);
+ const [importErrors, setImportErrors] = useState<string[]>([]);
+ const [importing, setImporting] = useState(false);
useEffect(() => {
let cancelled = false;
@@ -660,6 +671,94 @@ const TopicPage = () => {
}
};
+ const handleImportFile = async (file: File) => {
+ setImportFilename(file.name);
+ setImporting(false);
+ setImportModalOpen(true);
+ try {
+ const records = parseCsvTable(await file.text());
+ const validation = validateTopicCsvImport(records, selectedInstanceId ||
undefined);
+ setImportRows(validation.rows);
+ setImportErrors(validation.errors);
+ } catch (error) {
+ setImportRows([]);
+ setImportErrors([error instanceof Error ? error.message : 'CSV 解析失败']);
+ } finally {
+ if (importInputRef.current) importInputRef.current.value = '';
+ }
+ };
+
+ const handleImportTopics = async () => {
+ const targetIndexes = importRows
+ .map((row, index) => ({ row, index }))
+ .filter(({ row }) => row.status === 'pending' || row.status ===
'failed');
+ if (targetIndexes.length === 0 || importErrors.length > 0) return;
+
+ setImporting(true);
+ const nextRows = importRows.map((row) => ({ ...row }));
+ const createdTopics: Topic[] = [];
+
+ for (const { row, index } of targetIndexes) {
+ try {
+ const created = await createTopic(row.payload);
+ createdTopics.push(created);
+ nextRows[index] = { ...nextRows[index], status: 'success', message:
'已创建' };
+ } catch (error) {
+ nextRows[index] = {
+ ...nextRows[index],
+ status: 'failed',
+ message: error instanceof Error ? error.message : '创建失败',
+ };
+ }
+ setImportRows([...nextRows]);
+ }
+
+ if (createdTopics.length > 0) {
+ setTopics((previous) => {
+ const createdNames = new Set(createdTopics.map((topic) => topic.name));
+ return [...createdTopics, ...previous.filter((topic) =>
!createdNames.has(topic.name))];
+ });
+ }
+
+ const failedCount = nextRows.filter((row) => row.status ===
'failed').length;
+ const invalidCount = nextRows.filter((row) => row.status ===
'invalid').length;
+ if (failedCount === 0) {
+ if (invalidCount > 0) {
+ message.warning(`已导入 ${createdTopics.length} 个 Topic，${invalidCount}
行无效已跳过`);
+ } else {
+ message.success(`已导入 ${createdTopics.length} 个 Topic`);
+ }
+ } else if (createdTopics.length > 0) {
+ message.warning(`已导入 ${createdTopics.length} 个 Topic，${failedCount}
个失败`);
+ } else {
+ message.error(`${failedCount} 个 Topic 导入失败`);
+ }
+ setImporting(false);
+ };
+
+ const topicImportColumns:
TableColumnsType<ResourceImportRow<Partial<Topic>>> = [
+ { title: '行号', dataIndex: 'lineNumber', key: 'lineNumber', width: 80 },
+ { title: 'Topic 名称', dataIndex: 'name', key: 'name' },
+ {
+ title: '状态',
+ dataIndex: 'status',
+ key: 'status',
+ width: 100,
+ render: (status: ResourceImportRow<Partial<Topic>>['status']) => {
+ if (status === 'success') return <Tag color="success">成功</Tag>;
+ if (status === 'failed') return <Tag color="error">失败</Tag>;
+ if (status === 'invalid') return <Tag color="warning">无效</Tag>;
+ return <Tag>待导入</Tag>;
+ },
+ },
+ {
+ title: '说明',
+ dataIndex: 'message',
+ key: 'message',
+ render: (text?: string) => text || '-',
+ },
+ ];
+
// ─── Send message modal submit ────────────────────────────────
const handleSend = async () => {
let values;
@@ -821,7 +920,22 @@ const TopicPage = () => {
删除 ({selectedRowKeys.length})
</Button>
)}
- <Button icon={<ImportOutlined />} onClick={() =>
message.info('导入功能开发中')}>
+ <input
+ ref={importInputRef}
+ type="file"
+ accept=".csv,text/csv"
+ data-testid="topic-import-file"
+ style={{ display: 'none' }}
+ onChange={(event) => {
+ const file = event.target.files?.[0];
+ if (file) void handleImportFile(file);
+ }}
+ />
+ <Button
+ icon={<ImportOutlined />}
+ disabled={importing}
+ onClick={() => importInputRef.current?.click()}
+ >
导入
</Button>
<Button
@@ -1023,6 +1137,61 @@ const TopicPage = () => {
</Form>
</Modal>
+ {/* ── Import Topic Modal ────────────────────────────────── */}
+ <Modal
+ title={`导入 Topic${importFilename ? `：${importFilename}` : ''}`}
+ open={importModalOpen}
+ onCancel={() => {
+ if (!importing) setImportModalOpen(false);
+ }}
+ onOk={() => void handleImportTopics()}
+ okText={importRows.some((row) => row.status === 'failed') ? '重试失败项' :
'开始导入'}
+ cancelText="关闭"
+ confirmLoading={importing}
+ okButtonProps={{
+ disabled:
+ importErrors.length > 0 ||
+ importRows.length === 0 ||
+ importRows.every((row) => row.status === 'success' || row.status
=== 'invalid'),
+ }}
+ width={720}
+ destroyOnClose
+ >
+ <Space direction="vertical" style={{ width: '100%' }} size={12}>
+ {importErrors.length > 0 ? (
+ <Alert
+ type="error"
+ showIcon
+ message="CSV 无法导入"
+ description={importErrors.join('；')}
+ />
+ ) : importRows.some((row) => row.status === 'invalid') ? (
+ <Alert
+ type="warning"
+ showIcon
+ message={`检测到 ${
+ importRows.filter((row) => row.status === 'invalid').length
+ } 行无效，将跳过这些行`}
+ description="仅导入可创建字段；CSV 中的 Namespace、Cluster ID 和运行状态列会被忽略。"
+ />
+ ) : (
+ <Alert
+ type="info"
+ showIcon
+ message={`检测到 ${importRows.length} 个 Topic，将按顺序调用创建接口`}
+ description="仅导入可创建字段；CSV 中的 Namespace、Cluster ID 和运行状态列会被忽略。"
+ />
+ )}
+ <Table<ResourceImportRow<Partial<Topic>>>
+ columns={topicImportColumns}
+ dataSource={importRows}
+ rowKey="key"
+ size="small"
+ pagination={false}
+ />
+ </Space>
+ </Modal>
+
{/* ── Send Message Modal ──────────────────────────────────── */}
<Modal
title={
diff --git a/web/src/utils/resourceCsvImport.test.ts
b/web/src/utils/resourceCsvImport.test.ts
new file mode 100644
index 00000000..44ebf8e0
--- /dev/null
+++ b/web/src/utils/resourceCsvImport.test.ts
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { describe, expect, it } from 'vitest';
+import {
+ parseCsvTable,
+ RESOURCE_IMPORT_ROW_LIMIT,
+ validateConsumerGroupCsvImport,
+ validateTopicCsvImport,
+} from './resourceCsvImport';
+
+describe('resourceCsvImport', () => {
+ it('parses RFC4180 CSV with BOM, CRLF, quoted commas, newlines, and escaped
quotes', () => {
+ const records = parseCsvTable(
+ '\uFEFF"Name","Remark"\r\n"topic-a","line1\r\nline2,
""quoted"""\r\n\r\n',
+ );
+
+ expect(records).toEqual([
+ {
+ lineNumber: 2,
+ values: {
+ Name: 'topic-a',
+ Remark: 'line1\r\nline2, "quoted"',
+ },
+ },
+ ]);
+ });
+
+ it('rejects malformed headers, unclosed quotes, and row caps', () => {
+ expect(() => parseCsvTable('"Name","Name"\n"a","b"')).toThrow('CSV 表头重复');
+ expect(() => parseCsvTable('"Name","Remark"\n"a,"broken"')).toThrow('CSV
引号格式错误');
+ expect(() => parseCsvTable('"Name","Remark"\n"a"broken,"x"')).toThrow('CSV
引号格式错误');
+ expect(() => parseCsvTable('"Name"\n"unterminated')).toThrow('CSV 引号未闭合');
+
+ const tooManyRows = [
+ '"Name"',
+ ...Array.from({ length: RESOURCE_IMPORT_ROW_LIMIT + 1 }, (_, index) =>
`"topic-${index}"`),
+ ].join('\n');
+ expect(() => parseCsvTable(tooManyRows)).toThrow(
+ `一次最多导入 ${RESOURCE_IMPORT_ROW_LIMIT} 行`,
+ );
+ });
+
+ it('round-trips formula-safe apostrophes from exported cells', () => {
+ const records =
parseCsvTable('"Name","Remark"\n"\'-topic","\'=keep-original"');
+ const validation = validateTopicCsvImport(records, 'instance-a');
+
+ expect(validation.errors).toEqual([]);
+ expect(validation.rows[0].payload).toMatchObject({
+ name: '-topic',
+ remark: '=keep-original',
+ instanceId: 'instance-a',
+ });
+ });
+
+ it('validates topic fields and duplicate names before import calls', () => {
+ const records = parseCsvTable(
+ [
+ '"Name","Type","Write Queues","Read Queues","Permission"',
+ '"topic-a","NORMAL","8","8","RW"',
+ '"topic-a","INVALID","0","257","BAD"',
+ ].join('\n'),
+ );
+ const validation = validateTopicCsvImport(records);
+
+ expect(validation.errors).toEqual([]);
+ expect(validation.rows[0]).toMatchObject({ name: 'topic-a', status:
'pending' });
+ expect(validation.rows[1]).toMatchObject({ name: 'topic-a', status:
'invalid' });
+ expect(validation.rows[1].message).toContain('重复');
+ expect(validation.rows[1].message).toContain('Type 不支持');
+ expect(validation.rows[1].message).toContain('Write Queues 必须在 1..256 之间');
+ expect(validation.rows[1].message).toContain('Read Queues 必须在 1..256 之间');
+ expect(validation.rows[1].message).toContain('Permission 不支持');
+ });
+
+ it('maps consumer group CSV fields to create payloads', () => {
+ const records = parseCsvTable(
+ [
+ '"Name","Subscription Mode","Consume Type","Retry Max
Times","Subscription Data Type","Delivery Order Type","Cluster ID"',
+
'"cg-orders","Push","CLUSTERING","16","FIFO","PARTITON_ORDER","ignored-cluster"',
+ ].join('\n'),
+ );
+ const validation = validateConsumerGroupCsvImport(records, 'instance-b');
+
+ expect(validation.errors).toEqual([]);
+ expect(validation.rows[0].payload).toEqual({
+ name: 'cg-orders',
+ subscriptionMode: 'Push',
+ consumeType: 'CLUSTERING',
+ retryMaxTimes: 16,
+ subscriptionDataType: 'FIFO',
+ deliveryOrderType: 'PARTITON_ORDER',
+ subscribedTopics: [],
+ instanceId: 'instance-b',
+ });
+ });
+});
diff --git a/web/src/utils/resourceCsvImport.ts
b/web/src/utils/resourceCsvImport.ts
new file mode 100644
index 00000000..a94385f4
--- /dev/null
+++ b/web/src/utils/resourceCsvImport.ts
@@ -0,0 +1,364 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import type { ConsumerGroup, Topic } from '../api/metadata';
+
+export const RESOURCE_IMPORT_ROW_LIMIT = 100;
+
+export interface CsvRecord {
+ lineNumber: number;
+ values: Record<string, string>;
+}
+
+export interface ResourceImportRow<T> {
+ key: string;
+ lineNumber: number;
+ name: string;
+ payload: T;
+ status: 'pending' | 'invalid' | 'success' | 'failed';
+ message?: string;
+}
+
+export interface ResourceImportValidation<T> {
+ rows: ResourceImportRow<T>[];
+ errors: string[];
+}
+
+interface ParsedCsvRow {
+ lineNumber: number;
+ cells: string[];
+}
+
+const FORMULA_SAFE_PREFIX_PATTERN = /^'(?=[=+\-@])/;
+const TOPIC_NAME_PATTERN = /^[a-zA-Z0-9_\-/*]+$/;
+const GROUP_NAME_PATTERN = /^[a-zA-Z][a-zA-Z0-9_-]*$/;
+
+const TOPIC_TYPES = new Set(['NORMAL', 'FIFO', 'DELAY', 'TRANSACTION',
'LITE']);
+const TOPIC_PERMISSIONS = new Set(['RW', 'RO', 'WO']);
+const GROUP_SUBSCRIPTION_MODES = new Set(['Push', 'Pop']);
+const GROUP_CONSUME_TYPES = new Set(['CLUSTERING', 'BROADCASTING']);
+const GROUP_SUBSCRIPTION_DATA_TYPES = new Set(['NORMAL', 'FIFO', 'DELAY',
'TRANSACTION']);
+const GROUP_DELIVERY_ORDER_TYPES = new Set(['PARTITON_ORDER',
'PARTITION_ORDER', 'MESSAGES ORDER']);
+
+const restoreFormulaSafeCell = (value: string): string =>
+ value.replace(FORMULA_SAFE_PREFIX_PATTERN, '');
+
+const normalizeHeader = (header: string): string =>
restoreFormulaSafeCell(header).trim();
+
+const normalizeValue = (value: string | undefined): string =>
+ restoreFormulaSafeCell(value ?? '').trim();
+
+const parseInteger = (
+ value: string,
+ fieldName: string,
+ min: number,
+ max: number,
+ fallback: number,
+ errors: string[],
+): number => {
+ if (!value) return fallback;
+ if (!/^-?\d+$/.test(value)) {
+ errors.push(`${fieldName} 必须是整数`);
+ return fallback;
+ }
+
+ const parsed = Number(value);
+ if (parsed < min || parsed > max) {
+ errors.push(`${fieldName} 必须在 ${min}..${max} 之间`);
+ return fallback;
+ }
+ return parsed;
+};
+
+const readCsvRows = (content: string): ParsedCsvRow[] => {
+ const rows: ParsedCsvRow[] = [];
+ const text = content.startsWith('\uFEFF') ? content.slice(1) : content;
+ let cells: string[] = [];
+ let cell = '';
+ let inQuotes = false;
+ let quoteJustClosed = false;
+ let rowStartLine = 1;
+ let lineNumber = 1;
+
+ const pushRow = () => {
+ const nextCells = [...cells, cell];
+ if (nextCells.some((value) => value.trim() !== '')) {
+ rows.push({
+ lineNumber: rowStartLine,
+ cells: nextCells.map(restoreFormulaSafeCell),
+ });
+ }
+ cells = [];
+ cell = '';
+ quoteJustClosed = false;
+ rowStartLine = lineNumber;
+ };
+
+ for (let index = 0; index < text.length; index += 1) {
+ const char = text[index];
+ const next = text[index + 1];
+
+ if (inQuotes) {
+ if (char === '"') {
+ if (next === '"') {
+ cell += '"';
+ index += 1;
+ } else {
+ inQuotes = false;
+ quoteJustClosed = true;
+ }
+ } else {
+ if (char === '\n') lineNumber += 1;
+ cell += char;
+ }
+ continue;
+ }
+
+ if (quoteJustClosed && char !== ',' && char !== '\r' && char !== '\n') {
+ throw new Error(`第 ${lineNumber} 行 CSV 引号格式错误`);
+ }
+
+ if (char === '"') {
+ if (cell.length > 0) {
+ throw new Error(`第 ${lineNumber} 行 CSV 引号格式错误`);
+ }
+ inQuotes = true;
+ quoteJustClosed = false;
+ continue;
+ }
+
+ if (char === ',') {
+ cells.push(cell);
+ cell = '';
+ quoteJustClosed = false;
+ continue;
+ }
+
+ if (char === '\r' || char === '\n') {
+ pushRow();
+ if (char === '\r' && next === '\n') index += 1;
+ lineNumber += 1;
+ rowStartLine = lineNumber;
+ continue;
+ }
+
+ cell += char;
+ quoteJustClosed = false;
+ }
+
+ if (inQuotes) {
+ throw new Error(`第 ${rowStartLine} 行 CSV 引号未闭合`);
+ }
+
+ pushRow();
+ return rows;
+};
+
+export const parseCsvTable = (content: string): CsvRecord[] => {
+ const rows = readCsvRows(content);
+ if (rows.length === 0) {
+ throw new Error('CSV 文件为空');
+ }
+
+ const headers = rows[0].cells.map(normalizeHeader);
+ const duplicateHeaders = headers.filter(
+ (header, index) => header && headers.indexOf(header) !== index,
+ );
+ if (headers.some((header) => !header)) {
+ throw new Error('CSV 表头不能为空');
+ }
+ if (duplicateHeaders.length > 0) {
+ throw new Error(`CSV 表头重复：${Array.from(new Set(duplicateHeaders)).join(',
')}`);
+ }
+
+ const records = rows.slice(1).map((row) => {
+ if (row.cells.length > headers.length) {
+ throw new Error(`第 ${row.lineNumber} 行字段数超过表头字段数`);
+ }
+
+ return {
+ lineNumber: row.lineNumber,
+ values: headers.reduce<Record<string, string>>((acc, header, index) => {
+ acc[header] = normalizeValue(row.cells[index]);
+ return acc;
+ }, {}),
+ };
+ });
+
+ if (records.length === 0) {
+ throw new Error('CSV 没有可导入的数据行');
+ }
+ if (records.length > RESOURCE_IMPORT_ROW_LIMIT) {
+ throw new Error(`一次最多导入 ${RESOURCE_IMPORT_ROW_LIMIT} 行`);
+ }
+
+ return records;
+};
+
+const buildDuplicateNameMessages = (records: CsvRecord[]): Map<number, string>
=> {
+ const firstLineByName = new Map<string, number>();
+ const messagesByLine = new Map<number, string>();
+
+ records.forEach((record) => {
+ const name = normalizeValue(record.values.Name);
+ if (!name) return;
+
+ const firstLine = firstLineByName.get(name);
+ if (firstLine != null) {
+ messagesByLine.set(record.lineNumber, `Name 与第 ${firstLine}
行重复：${name}`);
+ } else {
+ firstLineByName.set(name, record.lineNumber);
+ }
+ });
+
+ return messagesByLine;
+};
+
+export const validateTopicCsvImport = (
+ records: CsvRecord[],
+ selectedInstanceId?: string,
+): ResourceImportValidation<Partial<Topic>> => {
+ const duplicateMessages = buildDuplicateNameMessages(records);
+ const rows: ResourceImportRow<Partial<Topic>>[] = [];
+
+ records.forEach((record, index) => {
+ const rowErrors: string[] = [];
+ const name = normalizeValue(record.values.Name);
+ const type = normalizeValue(record.values.Type) || 'NORMAL';
+ const writeQueues = parseInteger(
+ normalizeValue(record.values['Write Queues']),
+ 'Write Queues',
+ 1,
+ 256,
+ 8,
+ rowErrors,
+ );
+ const readQueues = parseInteger(
+ normalizeValue(record.values['Read Queues']),
+ 'Read Queues',
+ 1,
+ 256,
+ 8,
+ rowErrors,
+ );
+ const perm = normalizeValue(record.values.Permission) || 'RW';
+ const remark = normalizeValue(record.values.Remark);
+ const duplicateMessage = duplicateMessages.get(record.lineNumber);
+ if (duplicateMessage) rowErrors.push(duplicateMessage);
+
+ if (!name) {
+ rowErrors.push('Name 不能为空');
+ } else if (!TOPIC_NAME_PATTERN.test(name)) {
+ rowErrors.push('Name 仅支持字母、数字、下划线、中划线、斜杠和星号');
+ }
+ if (!TOPIC_TYPES.has(type)) {
+ rowErrors.push(`Type 不支持：${type}`);
+ }
+ if (!TOPIC_PERMISSIONS.has(perm)) {
+ rowErrors.push(`Permission 不支持：${perm}`);
+ }
+
+ rows.push({
+ key: `${record.lineNumber}-${name || index}`,
+ lineNumber: record.lineNumber,
+ name,
+ payload: {
+ name,
+ type,
+ writeQueues,
+ readQueues,
+ perm,
+ remark,
+ ...(selectedInstanceId ? { instanceId: selectedInstanceId } : {}),
+ },
+ status: rowErrors.length > 0 ? 'invalid' : 'pending',
+ message: rowErrors.join('；') || undefined,
+ });
+ });
+
+ return { rows, errors: [] };
+};
+
+export const validateConsumerGroupCsvImport = (
+ records: CsvRecord[],
+ selectedInstanceId?: string,
+): ResourceImportValidation<Partial<ConsumerGroup>> => {
+ const duplicateMessages = buildDuplicateNameMessages(records);
+ const rows: ResourceImportRow<Partial<ConsumerGroup>>[] = [];
+
+ records.forEach((record, index) => {
+ const rowErrors: string[] = [];
+ const name = normalizeValue(record.values.Name);
+ const subscriptionMode = normalizeValue(record.values['Subscription
Mode']) || 'Push';
+ const consumeType = normalizeValue(record.values['Consume Type']) ||
'CLUSTERING';
+ const retryMaxTimes = parseInteger(
+ normalizeValue(record.values['Retry Max Times']),
+ 'Retry Max Times',
+ 0,
+ 128,
+ 16,
+ rowErrors,
+ );
+ const subscriptionDataType =
+ normalizeValue(record.values['Subscription Data Type']) || 'NORMAL';
+ const deliveryOrderType = normalizeValue(record.values['Delivery Order
Type']);
+ const duplicateMessage = duplicateMessages.get(record.lineNumber);
+ if (duplicateMessage) rowErrors.push(duplicateMessage);
+
+ if (!name) {
+ rowErrors.push('Name 不能为空');
+ } else if (!GROUP_NAME_PATTERN.test(name)) {
+ rowErrors.push('Name 需以字母开头，仅包含字母、数字、下划线和短横线');
+ }
+ if (!GROUP_SUBSCRIPTION_MODES.has(subscriptionMode)) {
+ rowErrors.push(`Subscription Mode 不支持：${subscriptionMode}`);
+ }
+ if (!GROUP_CONSUME_TYPES.has(consumeType)) {
+ rowErrors.push(`Consume Type 不支持：${consumeType}`);
+ }
+ if (!GROUP_SUBSCRIPTION_DATA_TYPES.has(subscriptionDataType)) {
+ rowErrors.push(`Subscription Data Type 不支持：${subscriptionDataType}`);
+ }
+ if (
+ deliveryOrderType &&
+ subscriptionDataType === 'FIFO' &&
+ !GROUP_DELIVERY_ORDER_TYPES.has(deliveryOrderType)
+ ) {
+ rowErrors.push(`Delivery Order Type 不支持：${deliveryOrderType}`);
+ }
+
+ rows.push({
+ key: `${record.lineNumber}-${name || index}`,
+ lineNumber: record.lineNumber,
+ name,
+ payload: {
+ name,
+ subscriptionMode,
+ consumeType,
+ retryMaxTimes,
+ subscriptionDataType,
+ ...(subscriptionDataType === 'FIFO' && deliveryOrderType ? {
deliveryOrderType } : {}),
+ subscribedTopics: [],
+ ...(selectedInstanceId ? { instanceId: selectedInstanceId } : {}),
+ },
+ status: rowErrors.length > 0 ? 'invalid' : 'pending',
+ message: rowErrors.join('；') || undefined,
+ });
+ });
+
+ return { rows, errors: [] };
+};
