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 50ba961ec feat(topic): support topic bulk import and export (#2640)
50ba961ec is described below
commit 50ba961ecd453cd371595ea64be225da81e49a26
Author: coder999o <[email protected]>
AuthorDate: Mon Aug 31 21:08:46 2026 +0800
feat(topic): support topic bulk import and export (#2640)
---
.../studio/instance/topic/ImportTopicsDTO.java | 37 +++++++++
.../instance/topic/ImportTopicsResultVO.java | 52 ++++++++++++
.../studio/instance/topic/MetadataService.java | 78 ++++++++++++++++++
.../studio/instance/topic/TopicController.java | 28 +++++++
.../studio/instance/topic/MetadataServiceTest.java | 92 +++++++++++++++++++++
.../studio/instance/topic/TopicControllerTest.java | 54 ++++++++++++
web/src/api/metadata.test.ts | 63 ++++++++++++++
web/src/api/metadata.ts | 34 ++++++++
.../pages/instance/__tests__/TopicPage.test.tsx | 96 +++++++++++++---------
web/src/pages/instance/topic.tsx | 69 +++++++---------
web/src/services/topicService.ts | 58 +++++++++++++
11 files changed, 585 insertions(+), 76 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/ImportTopicsDTO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/ImportTopicsDTO.java
new file mode 100644
index 000000000..0f2083f80
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/ImportTopicsDTO.java
@@ -0,0 +1,37 @@
+/*
+ * 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.
+ */
+package org.apache.rocketmq.studio.instance.topic;
+
+import jakarta.validation.Valid;
+import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.NotEmpty;
+import jakarta.validation.constraints.Size;
+import lombok.Data;
+
+import java.util.List;
+
+@Data
+public class ImportTopicsDTO {
+
+ @NotBlank(message = "instanceId is required")
+ private String instanceId;
+
+ @Valid
+ @NotEmpty(message = "topics is required")
+ @Size(max = 100, message = "At most 100 topics are allowed per import")
+ private List<CreateTopicDTO> topics;
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/ImportTopicsResultVO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/ImportTopicsResultVO.java
new file mode 100644
index 000000000..34994d9b3
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/ImportTopicsResultVO.java
@@ -0,0 +1,52 @@
+/*
+ * 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.
+ */
+package org.apache.rocketmq.studio.instance.topic;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.List;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class ImportTopicsResultVO {
+
+ private int imported;
+
+ private int failed;
+
+ private List<TopicVO> topics;
+
+ private List<Failure> failures;
+
+ @Data
+ @Builder
+ @NoArgsConstructor
+ @AllArgsConstructor
+ public static class Failure {
+
+ private int index;
+
+ private String name;
+
+ private String message;
+ }
+}
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 cd9a47b7b..f482401a4 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
@@ -27,6 +27,7 @@ import
org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
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.SystemTopicFilter;
import org.apache.rocketmq.studio.instance.group.CreateConsumerGroupDTO;
import org.apache.rocketmq.studio.instance.group.ImportConsumerGroupsResultVO;
import org.springframework.util.StringUtils;
@@ -112,6 +113,9 @@ public class MetadataService {
public TopicVO createTopic(TopicVO topic) {
requireTopic(topic);
+ if (SystemTopicFilter.isSystem(topic.getName())) {
+ throw new BusinessException(400, "System topics cannot be created:
" + topic.getName());
+ }
String instanceId = topic.getInstanceId();
InstanceProvider provider = resolve(instanceId);
return executeWithAudit(provider, Operation.CREATE_TOPIC,
ResourceType.TOPIC, topic.getName(),
@@ -457,6 +461,80 @@ public class MetadataService {
return value.toString();
}
+ private static final int MAX_IMPORT_TOPICS = 100;
+
+ public ImportTopicsResultVO importTopics(String instanceId,
List<CreateTopicDTO> topics) {
+ if (!StringUtils.hasText(instanceId)) {
+ throw new BusinessException(400, "instanceId is required");
+ }
+ if (topics == null || topics.isEmpty()) {
+ throw new BusinessException(400, "topics is required");
+ }
+ if (topics.size() > MAX_IMPORT_TOPICS) {
+ throw new BusinessException(400, "At most 100 topics are allowed
per import");
+ }
+
+ String normalizedInstanceId = normalizeInstanceId(instanceId);
+ List<TopicVO> imported = new ArrayList<>();
+ List<ImportTopicsResultVO.Failure> failures = new ArrayList<>();
+ for (int index = 0; index < topics.size(); index++) {
+ CreateTopicDTO request = topics.get(index);
+ String name = request == null ? null : request.getName();
+ try {
+ if (request == null) {
+ throw new BusinessException(400, "topic request is
required");
+ }
+ TopicVO topic = request.toTopicVO();
+ topic.setInstanceId(normalizedInstanceId);
+ imported.add(createTopic(topic));
+ } catch (Exception exception) {
+ failures.add(ImportTopicsResultVO.Failure.builder()
+ .index(index)
+ .name(name)
+ .message(StringUtils.hasText(exception.getMessage())
+ ? exception.getMessage() : "Failed to create
topic")
+ .build());
+ }
+ }
+
+ return ImportTopicsResultVO.builder()
+ .imported(imported.size())
+ .failed(failures.size())
+ .topics(imported)
+ .failures(failures)
+ .build();
+ }
+
+ public String exportTopics(String instanceId, String type, String search,
List<String> names) {
+ instanceId = normalizeInstanceId(instanceId);
+ List<TopicVO> topics = new ArrayList<>(listTopics(instanceId, null,
type, search));
+ Set<String> selectedNames = new HashSet<>(names == null ? List.of() :
names);
+ if (!selectedNames.isEmpty()) {
+ topics.removeIf(topic -> !selectedNames.contains(topic.getName()));
+ }
+ topics.sort((left, right) -> compareNames(left.getName(),
right.getName()));
+ return buildTopicCsv(topics);
+ }
+
+ private int compareNames(String left, String right) {
+ String leftName = left == null ? "" : left;
+ String rightName = right == null ? "" : right;
+ return leftName.compareTo(rightName);
+ }
+
+ private String buildTopicCsv(List<TopicVO> topics) {
+ StringBuilder csv = new StringBuilder();
+ CsvUtil.appendRow(csv, "Name", "Namespace", "Type", "Cluster ID",
"Write Queues", "Read Queues",
+ "Permission", "Message Count", "TPS", "Consumer Groups",
"Remark", "Created At", "Updated At");
+ for (TopicVO topic : topics) {
+ CsvUtil.appendRow(csv, topic.getName(), topic.getNamespace(),
toText(topic.getType()), topic.getClusterId(),
+ topic.getWriteQueues(), topic.getReadQueues(),
toText(topic.getPerm()), topic.getMessageCount(),
+ topic.getTps(), topic.getConsumerGroupCount(),
topic.getRemark(),
+ topic.getGmtCreate(), topic.getGmtModified());
+ }
+ return csv.toString();
+ }
+
private String requireName(String value, String fieldName) {
if (!StringUtils.hasText(value)) {
throw new BusinessException(400, fieldName + " is required");
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/TopicController.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/TopicController.java
index eab11d963..cb8f88291 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/TopicController.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/TopicController.java
@@ -29,6 +29,8 @@ import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
+import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
@RestController
@@ -59,6 +61,21 @@ public class TopicController {
return Result.ok(metadataService.listTopicsPage(instanceId, clusterId,
type, search, page, pageSize));
}
+ @GetMapping("/export")
+ public Result<String> exportTopics(
+ @RequestParam(required = false) String instanceId,
+ @RequestParam(required = false) String type,
+ @RequestParam(required = false) String search,
+ @RequestParam(required = false) String names) {
+ return Result.ok(metadataService.exportTopics(instanceId, type,
search, parseNames(names)));
+ }
+
+ @PostMapping("/import")
+ public Result<ImportTopicsResultVO> importTopics(@Valid @RequestBody
ImportTopicsDTO request) {
+ String instanceId =
instanceService.normalizeIdentifier(request.getInstanceId());
+ return Result.ok(metadataService.importTopics(instanceId,
request.getTopics()));
+ }
+
@PostMapping("/create")
public Result<TopicVO> createTopic(@Valid @RequestBody(required = false)
CreateTopicDTO topic) {
requireCreateTopicRequest(topic);
@@ -134,4 +151,15 @@ public class TopicController {
throw new BusinessException(400, "Topic send message request is
required");
}
}
+
+ private List<String> parseNames(String names) {
+ if (names == null || names.isBlank()) {
+ return Collections.emptyList();
+ }
+ return Arrays.stream(names.split(","))
+ .map(String::trim)
+ .filter(name -> !name.isEmpty())
+ .distinct()
+ .toList();
+ }
}
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 1c6250963..82207638a 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
@@ -22,6 +22,8 @@ import org.apache.rocketmq.studio.common.domain.PageResult;
import org.apache.rocketmq.studio.common.domain.enums.ConsumeType;
import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
import org.apache.rocketmq.studio.common.domain.enums.SubscriptionMode;
+import org.apache.rocketmq.studio.common.domain.enums.TopicPerm;
+import org.apache.rocketmq.studio.common.domain.enums.TopicType;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
import org.apache.rocketmq.studio.instance.group.CreateConsumerGroupDTO;
@@ -176,6 +178,20 @@ class MetadataServiceTest {
verifyNoInteractions(operationAuditService);
}
+ @Test
+ void createTopicShouldRejectSystemTopicNamesTest() {
+ TopicVO systemTopic = new TopicVO();
+ systemTopic.setName("TBW102");
+ systemTopic.setWriteQueues(8);
+ systemTopic.setReadQueues(8);
+
+ assertThatThrownBy(() -> metadataService.createTopic(systemTopic))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("System topics cannot be created");
+
+ verify(apacheProvider, org.mockito.Mockito.never()).createTopic(any(),
any(TopicVO.class));
+ }
+
@Test
void apacheTopicWriteOperationsShouldNotDuplicateProviderAudit() {
TopicVO topic = new TopicVO();
@@ -249,6 +265,52 @@ class MetadataServiceTest {
"type=-, writeQueues=4, readQueues=4, perm=-", "FAILED", "open
api unavailable");
}
+ @Test
+ void exportTopicsShouldApplyFiltersSelectedNamesSortingAndCsvEscaping() {
+ TopicVO low = topic("orders-low", "\t=orders", TopicType.NORMAL);
+ TopicVO high = topic("orders-high", "critical", TopicType.FIFO);
+ TopicVO hidden = topic("users-topic", "=formula", TopicType.NORMAL);
+ when(apacheProvider.listTopics("instance-a", "NORMAL",
"orders")).thenReturn(List.of(low, hidden, high));
+
+ String csv = metadataService.exportTopics("instance-a", " NORMAL ", "
orders ",
+ List.of("orders-high", "orders-low"));
+
+ assertThat(csv).contains("\"Name\",\"Namespace\",\"Type\"");
+ assertThat(csv).contains("\"orders-high\",\"critical\",\"FIFO\"");
+ assertThat(csv).contains("\"orders-low\",\"'\t=orders\",\"NORMAL\"");
+ assertThat(csv).doesNotContain("users-topic");
+
assertThat(csv.indexOf("\"orders-high\"")).isLessThan(csv.indexOf("\"orders-low\""));
+ verify(apacheProvider).listTopics("instance-a", "NORMAL", "orders");
+ }
+
+ @Test
+ void importTopicsShouldContinueAfterRowFailure() {
+ when(apacheProvider.createTopic(eq("instance-a"),
any(TopicVO.class))).thenAnswer(invocation -> {
+ TopicVO topic = invocation.getArgument(1);
+ if ("topic-fail".equals(topic.getName())) {
+ throw new BusinessException(500, "broker rejected topic");
+ }
+ topic.setClusterId("cluster-a");
+ return topic;
+ });
+
+ ImportTopicsResultVO result =
metadataService.importTopics("instance-a",
+ List.of(topicImportRequest("topic-ok", "other-instance"),
topicImportRequest("topic-fail", "other-instance")));
+
+ assertThat(result.getImported()).isEqualTo(1);
+ assertThat(result.getFailed()).isEqualTo(1);
+
assertThat(result.getTopics()).extracting(TopicVO::getName).containsExactly("topic-ok");
+ assertThat(result.getFailures()).hasSize(1);
+ assertThat(result.getFailures().get(0).getIndex()).isEqualTo(1);
+
assertThat(result.getFailures().get(0).getName()).isEqualTo("topic-fail");
+ assertThat(result.getFailures().get(0).getMessage()).isEqualTo("broker
rejected topic");
+
+ ArgumentCaptor<TopicVO> captor =
ArgumentCaptor.forClass(TopicVO.class);
+ verify(apacheProvider,
org.mockito.Mockito.times(2)).createTopic(eq("instance-a"), captor.capture());
+ assertThat(captor.getAllValues()).extracting(TopicVO::getInstanceId)
+ .containsExactly("instance-a", "instance-a");
+ }
+
@Test
void deleteTopicShouldDelegateToApacheProvider() {
metadataService.deleteTopic("topic-to-delete");
@@ -577,4 +639,34 @@ class MetadataServiceTest {
}
+ private TopicVO topic(String name, String namespace, TopicType type) {
+ TopicVO topic = new TopicVO();
+ topic.setName(name);
+ topic.setNamespace(namespace);
+ topic.setType(type);
+ topic.setClusterId("cluster-a");
+ topic.setWriteQueues(8);
+ topic.setReadQueues(8);
+ topic.setPerm(TopicPerm.RW);
+ topic.setMessageCount(100);
+ topic.setTps(2.5);
+ topic.setConsumerGroupCount(3);
+ topic.setRemark("remark");
+ topic.setGmtCreate(LocalDateTime.of(2026, 8, 27, 10, 0));
+ topic.setGmtModified(LocalDateTime.of(2026, 8, 27, 11, 0));
+ return topic;
+ }
+
+ private CreateTopicDTO topicImportRequest(String name, String instanceId) {
+ CreateTopicDTO request = new CreateTopicDTO();
+ request.setName(name);
+ request.setInstanceId(instanceId);
+ request.setType(TopicType.NORMAL);
+ request.setWriteQueues(8);
+ request.setReadQueues(8);
+ request.setPerm(TopicPerm.RW);
+ return request;
+
+ }
+
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/TopicControllerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/TopicControllerTest.java
index 31792bef2..b3c2c5dc5 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/TopicControllerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/TopicControllerTest.java
@@ -144,6 +144,22 @@ class TopicControllerTest {
verify(metadataService).getTopicConsumersPage("instance-a", "orders",
2, 20);
}
+ @Test
+ void exportTopicsShouldPassFiltersAndSelectedNames() throws Exception {
+ when(metadataService.exportTopics("instance-a", "FIFO", "orders",
List.of("topic-a", "topic-b")))
+ .thenReturn("\"Name\"\n\"topic-a\"");
+
+ mockMvc.perform(get("/api/topics/export")
+ .param("instanceId", "instance-a")
+ .param("type", "FIFO")
+ .param("search", "orders")
+ .param("names", "topic-a, topic-b,topic-a"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.data").value("\"Name\"\n\"topic-a\""));
+
+ verify(metadataService).exportTopics("instance-a", "FIFO", "orders",
List.of("topic-a", "topic-b"));
+ }
+
@Test
void createTopicShouldReturnCreatedTopic() throws Exception {
TopicVO input = new TopicVO();
@@ -176,6 +192,44 @@ class TopicControllerTest {
assertThat(captor.getValue().getReadQueues()).isEqualTo(16);
}
+ @Test
+ void importTopicsShouldNormalizeInstanceAndDelegateBatch() throws
Exception {
+ Map<String, Object> body = Map.of(
+ "instanceId", "1",
+ "topics", List.of(Map.of(
+ "name", "imported-topic",
+ "type", "NORMAL",
+ "writeQueues", 8,
+ "readQueues", 8,
+ "perm", "RW"
+ ))
+ );
+ TopicVO imported = new TopicVO();
+ imported.setName("imported-topic");
+
when(instanceService.normalizeIdentifier("1")).thenReturn("open-source-local");
+ when(metadataService.importTopics(eq("open-source-local"), any()))
+ .thenReturn(ImportTopicsResultVO.builder()
+ .imported(1)
+ .failed(0)
+ .topics(List.of(imported))
+ .failures(List.of())
+ .build());
+
+ mockMvc.perform(post("/api/topics/import")
+ .contentType(MediaType.APPLICATION_JSON)
+ .content(objectMapper.writeValueAsString(body)))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.data.imported").value(1))
+
.andExpect(jsonPath("$.data.topics[0].name").value("imported-topic"));
+
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor<List<CreateTopicDTO>> captor =
ArgumentCaptor.forClass(List.class);
+ verify(metadataService).importTopics(eq("open-source-local"),
captor.capture());
+ assertThat(captor.getValue()).hasSize(1);
+
assertThat(captor.getValue().get(0).getName()).isEqualTo("imported-topic");
+ assertThat(captor.getValue().get(0).getWriteQueues()).isEqualTo(8);
+ }
+
@Test
void createTopicShouldRejectMissingName() throws Exception {
mockMvc.perform(post("/api/topics/create")
diff --git a/web/src/api/metadata.test.ts b/web/src/api/metadata.test.ts
index 867995474..e83e19322 100644
--- a/web/src/api/metadata.test.ts
+++ b/web/src/api/metadata.test.ts
@@ -29,6 +29,8 @@ import {
importConsumerGroups,
listTopics,
sendTopicMessage,
+ exportTopics,
+ importTopics,
} from './metadata';
const mock = new MockAdapter(client);
@@ -202,4 +204,65 @@ describe('topic metadata API', () => {
}),
).resolves.toMatchObject({ imported: 1, failed: 0 });
});
+
+ it('passes topic import and export contracts through API endpoints', async
() => {
+ mock.onGet('/topics/export').reply((config) => {
+ expect(config.params).toEqual({
+ instanceId: 'instance-1',
+ type: 'FIFO',
+ search: 'orders',
+ names: 'topic-a,topic-b',
+ });
+ return [200, { code: 200, data: '"Name"\n"topic-a"' }];
+ });
+ mock.onPost('/topics/import').reply((config) => {
+ expect(JSON.parse(config.data)).toEqual({
+ instanceId: 'instance-1',
+ topics: [
+ {
+ name: 'topic-a',
+ type: 'NORMAL',
+ writeQueues: 8,
+ readQueues: 8,
+ perm: 'RW',
+ },
+ ],
+ });
+ return [
+ 200,
+ {
+ code: 200,
+ data: {
+ imported: 1,
+ failed: 0,
+ topics: [],
+ failures: [],
+ },
+ },
+ ];
+ });
+
+ await expect(
+ exportTopics({
+ instanceId: 'instance-1',
+ type: 'FIFO',
+ search: 'orders',
+ names: ['topic-a', 'topic-b'],
+ }),
+ ).resolves.toBe('"Name"\n"topic-a"');
+ await expect(
+ importTopics({
+ instanceId: 'instance-1',
+ topics: [
+ {
+ name: 'topic-a',
+ type: 'NORMAL',
+ writeQueues: 8,
+ readQueues: 8,
+ perm: 'RW',
+ },
+ ],
+ }),
+ ).resolves.toMatchObject({ imported: 1, failed: 0 });
+ });
});
diff --git a/web/src/api/metadata.ts b/web/src/api/metadata.ts
index b919f32b3..148a8c842 100644
--- a/web/src/api/metadata.ts
+++ b/web/src/api/metadata.ts
@@ -30,6 +30,10 @@ export interface TopicQuery {
search?: string;
}
+export interface TopicExportQuery extends TopicQuery {
+ names?: string[];
+}
+
export interface BrokerRoute {
brokerName: string;
brokerAddr: string;
@@ -69,6 +73,24 @@ export interface PageResult<T> {
size: number;
}
+export interface ImportTopicsRequest {
+ instanceId: string;
+ topics: Partial<Topic>[];
+}
+
+export interface ImportTopicsFailure {
+ index: number;
+ name?: string;
+ message: string;
+}
+
+export interface ImportTopicsResult {
+ imported: number;
+ failed: number;
+ topics: Topic[];
+ failures: ImportTopicsFailure[];
+}
+
// ─── Consumer Group (matches mock/consumers.ts) ─────────────────
export interface ConsumerGroup {
name: string;
@@ -194,6 +216,18 @@ export async function listTopicsPage(params?: TopicQuery &
{ page?: number; page
return res.data.data;
}
+export async function exportTopics(params?: TopicExportQuery) {
+ const res = await client.get<{ data: string }>('/topics/export', {
+ params: { ...params, names: params?.names?.join(',') },
+ });
+ return res.data.data;
+}
+
+export async function importTopics(data: ImportTopicsRequest) {
+ const res = await client.post<{ data: ImportTopicsResult
}>('/topics/import', data);
+ return res.data.data;
+}
+
export async function createTopic(data: Partial<Topic>) {
const res = await client.post<{ data: Topic }>('/topics/create', data);
return res.data.data;
diff --git a/web/src/pages/instance/__tests__/TopicPage.test.tsx
b/web/src/pages/instance/__tests__/TopicPage.test.tsx
index df2e54cd0..d2003577e 100644
--- a/web/src/pages/instance/__tests__/TopicPage.test.tsx
+++ b/web/src/pages/instance/__tests__/TopicPage.test.tsx
@@ -29,9 +29,11 @@ const topicServiceMocks = vi.hoisted(() => ({
batchDeleteTopics: vi.fn(),
createTopic: vi.fn(),
deleteTopic: vi.fn(),
+ exportTopics: vi.fn(),
getTopicConsumers: vi.fn(),
getTopicConsumerPage: vi.fn(),
getTopicRoutes: vi.fn(),
+ importTopics: vi.fn(),
listAllTopics: vi.fn(),
listTopics: vi.fn(),
listTopicsPage: vi.fn(),
@@ -134,6 +136,7 @@ describe('TopicPage', () => {
beforeEach(() => {
mockTopicsList(buildTopics(25));
topicServiceMocks.listAllTopics.mockResolvedValue(buildTopics(25));
+ topicServiceMocks.exportTopics.mockResolvedValue('"Name"\n"topic-01"');
topicServiceMocks.batchDeleteTopics.mockResolvedValue({ deleted: [],
failed: [] });
topicServiceMocks.createTopic.mockImplementation(async (data:
Partial<Topic>) => ({
...buildTopics(1)[0],
@@ -146,6 +149,24 @@ describe('TopicPage', () => {
gmtCreate: '2026-01-02T00:00:00Z',
gmtModified: '2026-01-02T00:00:00Z',
}));
+ topicServiceMocks.importTopics.mockResolvedValue({
+ imported: 1,
+ failed: 0,
+ topics: [
+ {
+ ...buildTopics(1)[0],
+ name: 'imported-topic',
+ namespace: 'default',
+ clusterId: 'server-cluster',
+ messageCount: 0,
+ tps: 0,
+ consumerGroupCount: 0,
+ gmtCreate: '2026-01-02T00:00:00Z',
+ gmtModified: '2026-01-02T00:00:00Z',
+ },
+ ],
+ failures: [],
+ });
topicServiceMocks.getTopicRoutes.mockResolvedValue([]);
topicServiceMocks.getTopicConsumers.mockResolvedValue([]);
topicServiceMocks.getTopicConsumerPage.mockResolvedValue({
@@ -218,24 +239,14 @@ describe('TopicPage', () => {
remark: '\t=orders, "critical"',
},
];
- const archivedTopic = {
- ...buildTopics(1)[0],
- name: 'orders-topic-archive',
- namespace: 'trade',
- remark: '=archive',
- };
- const allMatchingTopics = [
- ...currentPageTopics,
- archivedTopic,
- {
- ...buildTopics(1)[0],
- name: 'users-topic',
- namespace: 'user',
- remark: '=formula-risk',
- },
- ];
mockTopicsList(currentPageTopics);
- topicServiceMocks.listAllTopics.mockResolvedValue(allMatchingTopics);
+ topicServiceMocks.exportTopics.mockResolvedValue(
+ [
+ '"Name","Namespace","Remark"',
+ '"orders-topic","trade","\'\t=orders, ""critical"""',
+ '"orders-topic-archive","trade","\'=archive"',
+ ].join('\n'),
+ );
renderWithProviders();
expect(await screen.findByText('orders-topic')).toBeInTheDocument();
@@ -245,12 +256,13 @@ describe('TopicPage', () => {
await user.click(screen.getByRole('button', { name: /导出/ }));
await waitFor(() =>
- expect(topicServiceMocks.listAllTopics).toHaveBeenCalledWith({
+ expect(topicServiceMocks.exportTopics).toHaveBeenCalledWith({
instanceId: 'instance-proxy-1',
type: undefined,
search: 'orders',
}),
);
+ expect(topicServiceMocks.listAllTopics).not.toHaveBeenCalled();
expect(URL.createObjectURL).toHaveBeenCalledTimes(1);
expect(clickSpy).toHaveBeenCalledTimes(1);
expect(URL.revokeObjectURL).toHaveBeenCalledWith('blob:topic-export');
@@ -458,7 +470,7 @@ describe('TopicPage', () => {
expect(screen.getByText('10.0.2.21:8080')).toBeInTheDocument();
});
- it('imports valid topic CSV rows through the create service with the
selected instance', async () => {
+ it('imports valid topic CSV rows through the backend batch service with the
selected instance', async () => {
const user = userEvent.setup();
mockTopicsList([]);
instanceServiceMocks.listInstances.mockResolvedValue([selectedInstance]);
@@ -470,25 +482,28 @@ describe('TopicPage', () => {
'"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();
+ 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(topicServiceMocks.importTopics).toHaveBeenCalledWith('instance-proxy-1',
[
+ {
+ name: 'imported-topic',
+ type: 'NORMAL',
+ writeQueues: 4,
+ readQueues: 6,
+ perm: 'RW',
+ remark: 'orders',
+ instanceId: 'instance-proxy-1',
+ },
+ ]),
);
+ expect(topicServiceMocks.createTopic).not.toHaveBeenCalled();
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 () => {
+ it('does not call importTopics when imported topic CSV is invalid or
duplicated', async () => {
const user = userEvent.setup();
instanceServiceMocks.listInstances.mockResolvedValue([selectedInstance]);
mockTopicsList([{ ...buildTopics(1)[0], instanceId: 'instance-proxy-1' }]);
@@ -507,13 +522,22 @@ describe('TopicPage', () => {
expect(screen.getAllByText(/Name 仅支持/).length).toBeGreaterThan(0);
expect(screen.getByText(/重复/)).toBeInTheDocument();
expect(screen.getByRole('button', { name: '开始导入' })).toBeDisabled();
- expect(topicServiceMocks.createTopic).not.toHaveBeenCalled();
+ expect(topicServiceMocks.importTopics).not.toHaveBeenCalled();
});
it('imports valid topic rows while skipping duplicate rows', async () => {
const user = userEvent.setup();
mockTopicsList([]);
instanceServiceMocks.listInstances.mockResolvedValue([selectedInstance]);
+ topicServiceMocks.importTopics.mockResolvedValue({
+ imported: 2,
+ failed: 0,
+ topics: [
+ { ...buildTopics(1)[0], name: 'topic-a', instanceId:
'instance-proxy-1' },
+ { ...buildTopics(1)[0], name: 'topic-b', instanceId:
'instance-proxy-1', type: 'FIFO' },
+ ],
+ failures: [],
+ });
renderWithProviders('/instance/instance-proxy-1/topic');
await screen.findByText(/共 0 个 Topic/);
@@ -530,15 +554,11 @@ describe('TopicPage', () => {
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,
+ await waitFor(() =>
expect(topicServiceMocks.importTopics).toHaveBeenCalledTimes(1));
+
expect(topicServiceMocks.importTopics).toHaveBeenCalledWith('instance-proxy-1',
[
expect.objectContaining({ name: 'topic-a', instanceId:
'instance-proxy-1' }),
- );
- expect(topicServiceMocks.createTopic).toHaveBeenNthCalledWith(
- 2,
expect.objectContaining({ name: 'topic-b', instanceId:
'instance-proxy-1' }),
- );
+ ]);
expect(await screen.findByText('已导入 2 个 Topic，1
行无效已跳过')).toBeInTheDocument();
});
diff --git a/web/src/pages/instance/topic.tsx b/web/src/pages/instance/topic.tsx
index a93506ae1..437319059 100644
--- a/web/src/pages/instance/topic.tsx
+++ b/web/src/pages/instance/topic.tsx
@@ -67,9 +67,10 @@ import {
batchDeleteTopics,
createTopic,
deleteTopic,
+ exportTopics,
getTopicConsumerPage,
getTopicRoutes,
- listAllTopics,
+ importTopics,
listTopicsPage,
sendTopicMessage,
} from '../../services/topicService';
@@ -82,7 +83,7 @@ import {
validateTopicCsvImport,
type ResourceImportRow,
} from '../../utils/resourceCsvImport';
-import { buildCsv, downloadCsv, type CsvColumn } from '../../utils/download';
+import { downloadCsv } from '../../utils/download';
import { parseMessageProperties } from '../../utils/messageProperties';
import { tableScrollX } from '../../utils/table';
import {
@@ -147,24 +148,6 @@ const TOPIC_TYPE_CARDS = [
// ─── Perm label ───────────────────────────────────────────────────
const PERM_LABEL: Record<string, string> = { RW: '读写', RO: '只读', WO: '只写' };
-const TOPIC_EXPORT_COLUMNS: CsvColumn<Topic>[] = [
- { header: 'Name', value: (topic) => topic.name },
- { header: 'Namespace', value: (topic) => topic.namespace },
- { header: 'Type', value: (topic) => topic.type },
- { header: 'Cluster ID', value: (topic) => topic.clusterId },
- { header: 'Write Queues', value: (topic) => topic.writeQueues },
- { header: 'Read Queues', value: (topic) => topic.readQueues },
- { header: 'Permission', value: (topic) => topic.perm },
- { header: 'Message Count', value: (topic) => topic.messageCount },
- { header: 'TPS', value: (topic) => topic.tps },
- { header: 'Consumer Groups', value: (topic) => topic.consumerGroupCount },
- { header: 'Remark', value: (topic) => topic.remark },
- { header: 'Created At', value: (topic) => topic.gmtCreate },
- { header: 'Updated At', value: (topic) => topic.gmtModified },
-];
-
-const buildTopicCsv = (topics: Topic[]) => buildCsv(TOPIC_EXPORT_COLUMNS,
topics);
-
const visibleTopics = (
topics: Topic[],
selectedInstanceId: string | undefined,
@@ -593,18 +576,14 @@ const TopicPage = () => {
const handleExport = () => {
setExporting(true);
- void listAllTopics({
+ void exportTopics({
instanceId: selectedInstanceId || undefined,
type: typeFilter || undefined,
search: searchText.trim() || undefined,
})
- .then((allTopics) => {
- const exportTopics = visibleTopics(allTopics, selectedInstanceId,
searchText, typeFilter);
- downloadCsv(
- `rocketmq-topics-${new Date().toISOString().slice(0, 10)}.csv`,
- buildTopicCsv(exportTopics),
- );
- message.success(`已导出 ${exportTopics.length} 个 Topic`);
+ .then((csv) => {
+ downloadCsv(`rocketmq-topics-${new Date().toISOString().slice(0,
10)}.csv`, csv);
+ message.success('Topic 导出完成');
})
.catch(() => {
message.error('导出 Topic 失败，请稍后重试');
@@ -1121,22 +1100,37 @@ const TopicPage = () => {
setImporting(true);
const nextRows = importRows.map((row) => ({ ...row }));
- const createdTopics: Topic[] = [];
+ let 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) {
+ try {
+ const result = await importTopics(
+ selectedInstanceId,
+ targetIndexes.map(({ row }) => row.payload),
+ );
+ createdTopics = result.topics;
+ const failureByIndex = new Map(result.failures.map((failure) =>
[failure.index, failure]));
+ targetIndexes.forEach(({ index }, requestIndex) => {
+ const failure = failureByIndex.get(requestIndex);
+ nextRows[index] = failure
+ ? {
+ ...nextRows[index],
+ status: 'failed',
+ message: failure.message || '创建失败',
+ }
+ : { ...nextRows[index], status: 'success', message: '已创建' };
+ });
+ } catch (error) {
+ for (const { index } of targetIndexes) {
nextRows[index] = {
...nextRows[index],
status: 'failed',
message: error instanceof Error ? error.message : '创建失败',
};
}
- setImportRows([...nextRows]);
+ } finally {
+ setImporting(false);
}
+ setImportRows([...nextRows]);
if (createdTopics.length > 0) {
setTopics((previous) => {
@@ -1158,7 +1152,6 @@ const TopicPage = () => {
} else {
message.error(`${failedCount} 个 Topic 导入失败`);
}
- setImporting(false);
};
const topicImportColumns:
TableColumnsType<ResourceImportRow<Partial<Topic>>> = [
@@ -1624,7 +1617,7 @@ const TopicPage = () => {
<Alert
type="info"
showIcon
- message={`检测到 ${importRows.length} 个 Topic，将按顺序调用创建接口`}
+ message={`检测到 ${importRows.length} 个 Topic，将通过后端批量导入`}
description="仅导入可创建字段；CSV 中的 Namespace、Cluster ID 和运行状态列会被忽略。"
/>
)}
diff --git a/web/src/services/topicService.ts b/web/src/services/topicService.ts
index 2423dc36c..a82101874 100644
--- a/web/src/services/topicService.ts
+++ b/web/src/services/topicService.ts
@@ -2,6 +2,7 @@ import { isMockMode } from './dataMode';
import * as metadataApi from '../api/metadata';
import type {
Topic,
+ TopicExportQuery,
TopicQuery,
TopicPage,
BrokerRoute,
@@ -9,11 +10,28 @@ import type {
TopicConsumerPage,
SendTopicMessageRequest,
SendTopicMessageResult,
+ ImportTopicsResult,
} from '../api/metadata';
import { topics as mockTopics, topicRoutes, topicConsumers } from
'../mock/topics';
+import { buildCsv, type CsvColumn } from '../utils/download';
const EXPORT_PAGE_SIZE = 100;
const MAX_EXPORT_PAGES = 100;
+const TOPIC_EXPORT_COLUMNS: CsvColumn<Topic>[] = [
+ { header: 'Name', value: (topic) => topic.name },
+ { header: 'Namespace', value: (topic) => topic.namespace },
+ { header: 'Type', value: (topic) => topic.type },
+ { header: 'Cluster ID', value: (topic) => topic.clusterId },
+ { header: 'Write Queues', value: (topic) => topic.writeQueues },
+ { header: 'Read Queues', value: (topic) => topic.readQueues },
+ { header: 'Permission', value: (topic) => topic.perm },
+ { header: 'Message Count', value: (topic) => topic.messageCount },
+ { header: 'TPS', value: (topic) => topic.tps },
+ { header: 'Consumer Groups', value: (topic) => topic.consumerGroupCount },
+ { header: 'Remark', value: (topic) => topic.remark },
+ { header: 'Created At', value: (topic) => topic.gmtCreate },
+ { header: 'Updated At', value: (topic) => topic.gmtModified },
+];
const cloneTopic = (topic: Topic): Topic => ({ ...topic });
const cloneRoutes = (routes: BrokerRoute[]): BrokerRoute[] =>
@@ -37,6 +55,15 @@ function filterMockTopics(params?: TopicQuery): Topic[] {
return (result as unknown as Topic[]).map(cloneTopic);
}
+function visibleExportTopics(topics: Topic[], params?: TopicExportQuery):
Topic[] {
+ let result = topics;
+ if (params?.names?.length) {
+ const selectedNames = new Set(params.names);
+ result = result.filter((topic) => selectedNames.has(topic.name));
+ }
+ return [...result].sort((left, right) =>
left.name.localeCompare(right.name));
+}
+
export async function listTopics(params?: TopicQuery): Promise<Topic[]> {
if (isMockMode()) {
return filterMockTopics(params);
@@ -98,6 +125,37 @@ export async function createTopic(data: Partial<Topic>):
Promise<Topic> {
return metadataApi.createTopic(data);
}
+export async function importTopics(
+ instanceId: string,
+ topics: Partial<Topic>[],
+): Promise<ImportTopicsResult> {
+ if (isMockMode()) {
+ const imported: Topic[] = [];
+ const failures: ImportTopicsResult['failures'] = [];
+ for (const [index, topic] of topics.entries()) {
+ try {
+ imported.push(await createTopic({ ...topic, instanceId }));
+ } catch (error) {
+ failures.push({
+ index,
+ name: topic.name,
+ message: error instanceof Error ? error.message : '创建失败',
+ });
+ }
+ }
+ return { imported: imported.length, failed: failures.length, topics:
imported, failures };
+ }
+ return metadataApi.importTopics({ instanceId, topics });
+}
+
+export async function exportTopics(params: TopicExportQuery = {}):
Promise<string> {
+ if (isMockMode()) {
+ const topics = await listAllTopics(params);
+ return buildCsv(TOPIC_EXPORT_COLUMNS, visibleExportTopics(topics, params));
+ }
+ return metadataApi.exportTopics(params);
+}
+
export async function updateTopic(data: Partial<Topic>): Promise<Topic> {
if (isMockMode()) {
const idx = mockTopics.findIndex((t) => t.name === data.name);
