You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by li...@apache.org on 2026/07/08 08:06:39 UTC
(rocketmq-dashboard) 07/09: feat: implement backend instance management module
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 18f34fb89e81798a16db0f79e886aa018bd6eee3
Author: lizhimins <70...@qq.com>
AuthorDate: Tue Jul 7 13:41:00 2026 +0800
feat: implement backend instance management module
- Instance CRUD (InstanceController, InstanceService)
- Topic management with metadata (TopicController, MetadataService)
- Consumer group monitoring (ConsumerGroupController)
- Message query and trace (MessageController)
- Dead letter queue management (DLQController)
- ACL rule management (AclController)
- In-memory repository implementations
- Unit tests for all services and controllers
Co-Authored-By: Claude <no...@anthropic.com>
---
.../instance/InMemoryInstanceRepository.java | 115 +++++++++
.../studio/instance/InstanceController.java | 62 +++++
.../studio/instance/InstanceRepository.java | 39 +++
.../rocketmq/studio/instance/InstanceService.java | 104 ++++++++
.../com/rocketmq/studio/instance/InstanceVO.java | 40 +++
.../studio/instance/acl/AclController.java | 71 ++++++
.../studio/instance/acl/AclRepository.java | 34 +++
.../rocketmq/studio/instance/acl/AclRuleVO.java | 42 ++++
.../rocketmq/studio/instance/acl/AclService.java | 75 ++++++
.../rocketmq/studio/instance/acl/AclUserVO.java | 39 +++
.../studio/instance/acl/InMemoryAclRepository.java | 73 ++++++
.../studio/instance/dlq/DLQController.java | 52 ++++
.../rocketmq/studio/instance/dlq/DLQGroupVO.java | 37 +++
.../rocketmq/studio/instance/dlq/DLQProvider.java | 25 ++
.../studio/instance/dlq/DLQProviderStub.java | 39 +++
.../rocketmq/studio/instance/dlq/DLQService.java | 41 ++++
.../instance/group/ConsumerGroupController.java | 82 +++++++
.../studio/instance/group/ConsumerGroupVO.java | 43 ++++
.../studio/instance/group/ConsumerInstanceVO.java | 40 +++
.../studio/instance/group/QueueProgressVO.java | 34 +++
.../studio/instance/group/SubscriptionEntryVO.java | 34 +++
.../studio/instance/message/ConsumerStatusVO.java | 34 +++
.../studio/instance/message/MessageController.java | 50 ++++
.../studio/instance/message/MessageProvider.java | 26 ++
.../instance/message/MessageProviderStub.java | 43 ++++
.../studio/instance/message/MessageRecordVO.java | 41 ++++
.../studio/instance/message/MessageService.java | 41 ++++
.../studio/instance/message/TraceNodeVO.java | 34 +++
.../studio/instance/message/TraceRecordVO.java | 33 +++
.../studio/instance/topic/AdminClient.java | 32 +++
.../studio/instance/topic/BrokerRouteVO.java | 35 +++
.../instance/topic/CloudMetadataProvider.java | 61 +++++
.../studio/instance/topic/LiteTopicVO.java | 26 ++
.../studio/instance/topic/MetadataProvider.java | 32 +++
.../studio/instance/topic/MetadataService.java | 117 +++++++++
.../studio/instance/topic/NameSrvAdminClient.java | 72 ++++++
.../studio/instance/topic/NamespaceController.java | 39 +++
.../studio/instance/topic/NamespaceVO.java | 29 +++
.../studio/instance/topic/ProxyAdminClient.java | 25 ++
.../studio/instance/topic/SendMessageDTO.java | 36 +++
.../studio/instance/topic/SendMessageVO.java | 32 +++
.../studio/instance/topic/TopicConsumerVO.java | 35 +++
.../studio/instance/topic/TopicController.java | 77 ++++++
.../rocketmq/studio/instance/topic/TopicVO.java | 39 +++
.../studio/instance/InstanceControllerTest.java | 191 +++++++++++++++
.../studio/instance/InstanceServiceTest.java | 271 +++++++++++++++++++++
.../studio/instance/acl/AclControllerTest.java | 140 +++++++++++
.../studio/instance/acl/AclServiceTest.java | 136 +++++++++++
.../studio/instance/dlq/DLQControllerTest.java | 125 ++++++++++
.../studio/instance/dlq/DLQServiceTest.java | 90 +++++++
.../studio/instance/topic/MetadataServiceTest.java | 144 +++++++++++
.../studio/instance/topic/TopicControllerTest.java | 133 ++++++++++
52 files changed, 3340 insertions(+)
diff --git a/server/src/main/java/com/rocketmq/studio/instance/InMemoryInstanceRepository.java b/server/src/main/java/com/rocketmq/studio/instance/InMemoryInstanceRepository.java
new file mode 100644
index 0000000..36d1dd6
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/InMemoryInstanceRepository.java
@@ -0,0 +1,115 @@
+/*
+ * 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 com.rocketmq.studio.instance;
+
+import com.rocketmq.studio.common.domain.enums.InstanceType;
+import org.springframework.stereotype.Component;
+
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+@Component
+public class InMemoryInstanceRepository implements InstanceRepository {
+
+ private final Map<String, InstanceVO> store = new ConcurrentHashMap<>();
+
+ public InMemoryInstanceRepository() {
+ initData();
+ }
+
+ private void initData() {
+ addInstance("inst-1", "production-proxy", "Production Proxy InstanceVO",
+ InstanceType.PROXY, "10.0.1.100:8080", 42, 18,
+ LocalDateTime.now().minusDays(90));
+ addInstance("inst-2", "staging-proxy", "Staging Proxy InstanceVO",
+ InstanceType.PROXY, "10.0.2.100:8080", 15, 8,
+ LocalDateTime.now().minusDays(60));
+ addInstance("inst-3", "dev-direct", "Development Direct InstanceVO",
+ InstanceType.DIRECT, "10.0.3.100:10911", 8, 3,
+ LocalDateTime.now().minusDays(30));
+ }
+
+ private void addInstance(String id, String name, String remark, InstanceType type,
+ String endpoint, int topicCount, int consumerGroupCount,
+ LocalDateTime createdAt) {
+ InstanceVO instance = InstanceVO.builder()
+ .name(name)
+ .remark(remark)
+ .type(type)
+ .endpoint(endpoint)
+ .topicCount(topicCount)
+ .consumerGroupCount(consumerGroupCount)
+ .build();
+ instance.setId(id);
+ instance.setCreatedAt(createdAt);
+ instance.setUpdatedAt(createdAt);
+ store.put(id, instance);
+ }
+
+ @Override
+ public List<InstanceVO> findAll() {
+ return new ArrayList<>(store.values());
+ }
+
+ @Override
+ public List<InstanceVO> findByType(InstanceType type) {
+ return store.values().stream()
+ .filter(i -> i.getType() == type)
+ .collect(Collectors.toList());
+ }
+
+ @Override
+ public List<InstanceVO> search(String keyword) {
+ String lower = keyword.toLowerCase();
+ return store.values().stream()
+ .filter(i -> i.getName().toLowerCase().contains(lower)
+ || i.getRemark() != null && i.getRemark().toLowerCase().contains(lower))
+ .collect(Collectors.toList());
+ }
+
+ @Override
+ public List<InstanceVO> findByTypeAndSearch(InstanceType type, String keyword) {
+ String lower = keyword.toLowerCase();
+ return store.values().stream()
+ .filter(i -> i.getType() == type)
+ .filter(i -> i.getName().toLowerCase().contains(lower)
+ || i.getRemark() != null && i.getRemark().toLowerCase().contains(lower))
+ .collect(Collectors.toList());
+ }
+
+ @Override
+ public Optional<InstanceVO> findById(String id) {
+ return Optional.ofNullable(store.get(id));
+ }
+
+ @Override
+ public InstanceVO save(InstanceVO instance) {
+ store.put(instance.getId(), instance);
+ return instance;
+ }
+
+ @Override
+ public void deleteById(String id) {
+ store.remove(id);
+ }
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/InstanceController.java b/server/src/main/java/com/rocketmq/studio/instance/InstanceController.java
new file mode 100644
index 0000000..a7d1186
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/InstanceController.java
@@ -0,0 +1,62 @@
+/*
+ * 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 com.rocketmq.studio.instance;
+
+import com.rocketmq.studio.common.domain.Result;
+import com.rocketmq.studio.common.domain.enums.InstanceType;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.List;
+import java.util.Map;
+
+@RestController
+@RequestMapping("/api/instances")
+@RequiredArgsConstructor
+public class InstanceController {
+
+ private final InstanceService instanceService;
+
+ @GetMapping
+ public Result<List<InstanceVO>> listInstances(
+ @RequestParam(required = false) InstanceType type,
+ @RequestParam(required = false) String search) {
+ return Result.ok(instanceService.listInstances(type, search));
+ }
+
+ @PostMapping("/create")
+ public Result<InstanceVO> createInstance(@RequestBody InstanceVO instance) {
+ return Result.ok(instanceService.createInstance(instance));
+ }
+
+ @PostMapping("/update")
+ public Result<InstanceVO> updateInstance(@RequestBody InstanceVO instance) {
+ return Result.ok(instanceService.updateInstance(instance));
+ }
+
+ @PostMapping("/delete")
+ public Result<Void> deleteInstance(@RequestBody Map<String, String> body) {
+ instanceService.deleteInstance(body.get("id"));
+ return Result.ok();
+ }
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/InstanceRepository.java b/server/src/main/java/com/rocketmq/studio/instance/InstanceRepository.java
new file mode 100644
index 0000000..2361465
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/InstanceRepository.java
@@ -0,0 +1,39 @@
+/*
+ * 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 com.rocketmq.studio.instance;
+
+import com.rocketmq.studio.common.domain.enums.InstanceType;
+
+import java.util.List;
+import java.util.Optional;
+
+public interface InstanceRepository {
+ List<InstanceVO> findAll();
+
+ List<InstanceVO> findByType(InstanceType type);
+
+ List<InstanceVO> search(String keyword);
+
+ List<InstanceVO> findByTypeAndSearch(InstanceType type, String keyword);
+
+ Optional<InstanceVO> findById(String id);
+
+ InstanceVO save(InstanceVO instance);
+
+ void deleteById(String id);
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/InstanceService.java b/server/src/main/java/com/rocketmq/studio/instance/InstanceService.java
new file mode 100644
index 0000000..628d190
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/InstanceService.java
@@ -0,0 +1,104 @@
+/*
+ * 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 com.rocketmq.studio.instance;
+
+import com.rocketmq.studio.common.domain.enums.InstanceType;
+import com.rocketmq.studio.common.exception.BusinessException;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.UUID;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class InstanceService {
+
+ private final InstanceRepository instanceRepository;
+
+ public List<InstanceVO> listInstances(InstanceType type, String search) {
+ log.debug("Listing instances, type={}, search={}", type, search);
+
+ if (type != null && search != null && !search.isBlank()) {
+ return instanceRepository.findByTypeAndSearch(type, search);
+ } else if (type != null) {
+ return instanceRepository.findByType(type);
+ } else if (search != null && !search.isBlank()) {
+ return instanceRepository.search(search);
+ }
+ return instanceRepository.findAll();
+ }
+
+ public InstanceVO createInstance(InstanceVO instance) {
+ log.info("Creating instance: {}", instance.getName());
+
+ if (instance.getName() == null || instance.getName().isBlank()) {
+ throw new BusinessException(400, "InstanceVO name is required");
+ }
+ if (instance.getEndpoint() == null || instance.getEndpoint().isBlank()) {
+ throw new BusinessException(400, "InstanceVO endpoint is required");
+ }
+
+ instance.setId(UUID.randomUUID().toString());
+ instance.setCreatedAt(LocalDateTime.now());
+ instance.setUpdatedAt(LocalDateTime.now());
+ return instanceRepository.save(instance);
+ }
+
+ public InstanceVO updateInstance(InstanceVO instance) {
+ log.info("Updating instance: {}", instance.getId());
+
+ if (instance.getId() == null || instance.getId().isBlank()) {
+ throw new BusinessException(400, "InstanceVO ID is required");
+ }
+
+ InstanceVO existing = instanceRepository.findById(instance.getId())
+ .orElseThrow(() -> new BusinessException(404, "InstanceVO not found: " + instance.getId()));
+
+ if (instance.getName() != null) {
+ existing.setName(instance.getName());
+ }
+ if (instance.getType() != null) {
+ existing.setType(instance.getType());
+ }
+ if (instance.getEndpoint() != null) {
+ existing.setEndpoint(instance.getEndpoint());
+ }
+ if (instance.getRemark() != null) {
+ existing.setRemark(instance.getRemark());
+ }
+ existing.setUpdatedAt(LocalDateTime.now());
+
+ return instanceRepository.save(existing);
+ }
+
+ public void deleteInstance(String id) {
+ log.info("Deleting instance: {}", id);
+
+ if (id == null || id.isBlank()) {
+ throw new BusinessException(400, "InstanceVO ID is required");
+ }
+
+ instanceRepository.findById(id)
+ .orElseThrow(() -> new BusinessException(404, "InstanceVO not found: " + id));
+ instanceRepository.deleteById(id);
+ }
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/InstanceVO.java b/server/src/main/java/com/rocketmq/studio/instance/InstanceVO.java
new file mode 100644
index 0000000..dd4f1c9
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/InstanceVO.java
@@ -0,0 +1,40 @@
+/*
+ * 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 com.rocketmq.studio.instance;
+
+import com.rocketmq.studio.common.domain.BaseEntity;
+import com.rocketmq.studio.common.domain.enums.InstanceType;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+@EqualsAndHashCode(callSuper = true)
+public class InstanceVO extends BaseEntity {
+ private String name;
+ private String remark;
+ private InstanceType type;
+ private String endpoint;
+ private int topicCount;
+ private int consumerGroupCount;
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/acl/AclController.java b/server/src/main/java/com/rocketmq/studio/instance/acl/AclController.java
new file mode 100644
index 0000000..3d1d83c
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/acl/AclController.java
@@ -0,0 +1,71 @@
+/*
+ * 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 com.rocketmq.studio.instance.acl;
+
+import com.rocketmq.studio.common.domain.Result;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.List;
+import java.util.Map;
+
+@RestController
+@RequestMapping("/api/acl")
+@RequiredArgsConstructor
+public class AclController {
+
+ private final AclService aclService;
+
+ @GetMapping("/rules")
+ public Result<List<AclRuleVO>> listRules(
+ @RequestParam(required = false) String clusterId,
+ @RequestParam(required = false) String principal) {
+ return Result.ok(aclService.listRules(clusterId, principal));
+ }
+
+ @PostMapping("/rules/create")
+ public Result<AclRuleVO> createRule(@RequestBody AclRuleVO rule) {
+ return Result.ok(aclService.createRule(rule));
+ }
+
+ @PostMapping("/rules/delete")
+ public Result<Void> deleteRule(@RequestBody Map<String, String> request) {
+ aclService.deleteRule(request.get("id"));
+ return Result.ok();
+ }
+
+ @GetMapping("/users")
+ public Result<List<AclUserVO>> listUsers() {
+ return Result.ok(aclService.listUsers());
+ }
+
+ @PostMapping("/users/create")
+ public Result<AclUserVO> createUser(@RequestBody AclUserVO user) {
+ return Result.ok(aclService.createUser(user));
+ }
+
+ @PostMapping("/users/delete")
+ public Result<Void> deleteUser(@RequestBody Map<String, String> request) {
+ aclService.deleteUser(request.get("id"));
+ return Result.ok();
+ }
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/acl/AclRepository.java b/server/src/main/java/com/rocketmq/studio/instance/acl/AclRepository.java
new file mode 100644
index 0000000..fab77cc
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/acl/AclRepository.java
@@ -0,0 +1,34 @@
+/*
+ * 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 com.rocketmq.studio.instance.acl;
+
+
+import java.util.List;
+
+public interface AclRepository {
+ List<AclRuleVO> findRules(String clusterId, String principal);
+
+ AclRuleVO saveRule(AclRuleVO rule);
+
+ void deleteRule(String id);
+
+ List<AclUserVO> findUsers();
+
+ AclUserVO saveUser(AclUserVO user);
+
+ void deleteUser(String id);
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/acl/AclRuleVO.java b/server/src/main/java/com/rocketmq/studio/instance/acl/AclRuleVO.java
new file mode 100644
index 0000000..2e36e17
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/acl/AclRuleVO.java
@@ -0,0 +1,42 @@
+/*
+ * 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 com.rocketmq.studio.instance.acl;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.time.LocalDateTime;
+import java.util.List;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class AclRuleVO {
+ private String id;
+ private String principal;
+ private String resource;
+ private String resourceType;
+ private String resourcePattern;
+ private List<String> actions;
+ private String decision;
+ private String scope;
+ private Integer aclVersion;
+ private LocalDateTime createdAt;
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/acl/AclService.java b/server/src/main/java/com/rocketmq/studio/instance/acl/AclService.java
new file mode 100644
index 0000000..537b81e
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/acl/AclService.java
@@ -0,0 +1,75 @@
+/*
+ * 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 com.rocketmq.studio.instance.acl;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.UUID;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class AclService {
+
+ private final AclRepository aclRepository;
+
+
+ public List<AclRuleVO> listRules(String clusterId, String principal) {
+ log.info("Listing ACL rules for clusterId={}, principal={}", clusterId, principal);
+ return aclRepository.findRules(clusterId, principal);
+ }
+
+
+ public AclRuleVO createRule(AclRuleVO rule) {
+ log.info("Creating ACL rule for principal={}", rule.getPrincipal());
+ rule.setId(UUID.randomUUID().toString());
+ rule.setCreatedAt(LocalDateTime.now());
+ return aclRepository.saveRule(rule);
+ }
+
+
+ public void deleteRule(String id) {
+ log.info("Deleting ACL rule id={}", id);
+ aclRepository.deleteRule(id);
+ }
+
+
+ public List<AclUserVO> listUsers() {
+ log.info("Listing ACL users");
+ return aclRepository.findUsers();
+ }
+
+
+ public AclUserVO createUser(AclUserVO user) {
+ log.info("Creating ACL user username={}", user.getUsername());
+ user.setId(UUID.randomUUID().toString());
+ user.setAccessKey(UUID.randomUUID().toString().replace("-", ""));
+ user.setSecretKey(UUID.randomUUID().toString().replace("-", ""));
+ user.setCreatedAt(LocalDateTime.now());
+ return aclRepository.saveUser(user);
+ }
+
+
+ public void deleteUser(String id) {
+ log.info("Deleting ACL user id={}", id);
+ aclRepository.deleteUser(id);
+ }
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/acl/AclUserVO.java b/server/src/main/java/com/rocketmq/studio/instance/acl/AclUserVO.java
new file mode 100644
index 0000000..5fbea0b
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/acl/AclUserVO.java
@@ -0,0 +1,39 @@
+/*
+ * 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 com.rocketmq.studio.instance.acl;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.time.LocalDateTime;
+import java.util.List;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class AclUserVO {
+ private String id;
+ private String username;
+ private String accessKey;
+ private String secretKey;
+ private boolean admin;
+ private List<String> clusters;
+ private LocalDateTime createdAt;
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/acl/InMemoryAclRepository.java b/server/src/main/java/com/rocketmq/studio/instance/acl/InMemoryAclRepository.java
new file mode 100644
index 0000000..909b775
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/acl/InMemoryAclRepository.java
@@ -0,0 +1,73 @@
+/*
+ * 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 com.rocketmq.studio.instance.acl;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+@Slf4j
+@Component
+public class InMemoryAclRepository implements AclRepository {
+
+ private final Map<String, AclRuleVO> rules = new ConcurrentHashMap<>();
+ private final Map<String, AclUserVO> users = new ConcurrentHashMap<>();
+
+ @Override
+ public List<AclRuleVO> findRules(String clusterId, String principal) {
+ return rules.values().stream()
+ .filter(r -> clusterId == null || clusterId.equals(r.getScope()))
+ .filter(r -> principal == null || principal.equals(r.getPrincipal()))
+ .collect(Collectors.toList());
+ }
+
+ @Override
+ public AclRuleVO saveRule(AclRuleVO rule) {
+ rules.put(rule.getId(), rule);
+ log.debug("Saved ACL rule id={}", rule.getId());
+ return rule;
+ }
+
+ @Override
+ public void deleteRule(String id) {
+ rules.remove(id);
+ log.debug("Deleted ACL rule id={}", id);
+ }
+
+ @Override
+ public List<AclUserVO> findUsers() {
+ return new ArrayList<>(users.values());
+ }
+
+ @Override
+ public AclUserVO saveUser(AclUserVO user) {
+ users.put(user.getId(), user);
+ log.debug("Saved ACL user id={}", user.getId());
+ return user;
+ }
+
+ @Override
+ public void deleteUser(String id) {
+ users.remove(id);
+ log.debug("Deleted ACL user id={}", id);
+ }
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/dlq/DLQController.java b/server/src/main/java/com/rocketmq/studio/instance/dlq/DLQController.java
new file mode 100644
index 0000000..4cef343
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/dlq/DLQController.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 com.rocketmq.studio.instance.dlq;
+
+import com.rocketmq.studio.common.domain.Result;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.List;
+import java.util.Map;
+
+@RestController
+@RequestMapping("/api/dlq")
+@RequiredArgsConstructor
+public class DLQController {
+
+ private final DLQService dlqService;
+
+ @GetMapping
+ public Result<List<DLQGroupVO>> listDLQGroups(@RequestParam(required = false) String clusterId) {
+ return Result.ok(dlqService.listDLQGroups(clusterId));
+ }
+
+ @PostMapping("/resend")
+ public Result<Void> resendMessages(@RequestBody Map<String, Object> request) {
+ String groupName = (String) request.get("groupName");
+ Long startTime = request.get("startTime") != null ? ((Number) request.get("startTime")).longValue() : null;
+ Long endTime = request.get("endTime") != null ? ((Number) request.get("endTime")).longValue() : null;
+ String targetTopic = (String) request.get("targetTopic");
+ dlqService.resendMessages(groupName, startTime, endTime, targetTopic);
+ return Result.ok();
+ }
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/dlq/DLQGroupVO.java b/server/src/main/java/com/rocketmq/studio/instance/dlq/DLQGroupVO.java
new file mode 100644
index 0000000..9782b75
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/dlq/DLQGroupVO.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 com.rocketmq.studio.instance.dlq;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.time.LocalDateTime;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class DLQGroupVO {
+ private String groupName;
+ private String dlqTopic; // format: %DLQ%{groupName}
+ private long messageCount;
+ private LocalDateTime lastEnqueueTime;
+ private int retryCount;
+ private String status;
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/dlq/DLQProvider.java b/server/src/main/java/com/rocketmq/studio/instance/dlq/DLQProvider.java
new file mode 100644
index 0000000..45f940c
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/dlq/DLQProvider.java
@@ -0,0 +1,25 @@
+/*
+ * 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 com.rocketmq.studio.instance.dlq;
+
+
+import java.util.List;
+
+public interface DLQProvider {
+ List<DLQGroupVO> listDLQGroups(String clusterId);
+ void resendMessages(String groupName, Long startTime, Long endTime, String targetTopic);
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/dlq/DLQProviderStub.java b/server/src/main/java/com/rocketmq/studio/instance/dlq/DLQProviderStub.java
new file mode 100644
index 0000000..76b48ba
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/dlq/DLQProviderStub.java
@@ -0,0 +1,39 @@
+/*
+ * 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 com.rocketmq.studio.instance.dlq;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.util.Collections;
+import java.util.List;
+
+@Component
+@Slf4j
+public class DLQProviderStub implements DLQProvider {
+
+ @Override
+ public List<DLQGroupVO> listDLQGroups(String clusterId) {
+ log.warn("DLQProviderStub.listDLQGroups called - returning empty list");
+ return Collections.emptyList();
+ }
+
+ @Override
+ public void resendMessages(String groupName, Long startTime, Long endTime, String targetTopic) {
+ log.warn("DLQProviderStub.resendMessages called - no-op. group={}, targetTopic={}", groupName, targetTopic);
+ }
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/dlq/DLQService.java b/server/src/main/java/com/rocketmq/studio/instance/dlq/DLQService.java
new file mode 100644
index 0000000..f2c0037
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/dlq/DLQService.java
@@ -0,0 +1,41 @@
+/*
+ * 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 com.rocketmq.studio.instance.dlq;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+
+@Service
+@RequiredArgsConstructor
+@Slf4j
+public class DLQService {
+
+ private final DLQProvider dlqProvider;
+
+ public List<DLQGroupVO> listDLQGroups(String clusterId) {
+ log.info("Listing DLQ groups for cluster: {}", clusterId);
+ return dlqProvider.listDLQGroups(clusterId);
+ }
+
+ public void resendMessages(String groupName, Long startTime, Long endTime, String targetTopic) {
+ log.info("Resending DLQ messages: group={}, targetTopic={}", groupName, targetTopic);
+ dlqProvider.resendMessages(groupName, startTime, endTime, targetTopic);
+ }
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerGroupController.java b/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerGroupController.java
new file mode 100644
index 0000000..395744f
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerGroupController.java
@@ -0,0 +1,82 @@
+/*
+ * 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 com.rocketmq.studio.instance.group;
+
+import com.rocketmq.studio.instance.topic.MetadataService;
+
+import com.rocketmq.studio.common.domain.Result;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.List;
+import java.util.Map;
+
+@RestController
+@RequestMapping("/api/groups")
+@RequiredArgsConstructor
+public class ConsumerGroupController {
+
+ private final MetadataService metadataService;
+
+ @GetMapping
+ public Result<List<ConsumerGroupVO>> listConsumerGroups(
+ @RequestParam(required = false) String clusterId,
+ @RequestParam(required = false) String search) {
+ return Result.ok(metadataService.listConsumerGroups(clusterId, search));
+ }
+
+ @GetMapping("/{name}")
+ public Result<ConsumerGroupVO> getConsumerGroup(@PathVariable String name) {
+ return Result.ok(metadataService.getConsumerGroup(name));
+ }
+
+ @GetMapping("/{name}/progress")
+ public Result<List<QueueProgressVO>> getGroupProgress(@PathVariable String name) {
+ return Result.ok(metadataService.getGroupProgress(name));
+ }
+
+ @GetMapping("/{name}/subscriptions")
+ public Result<List<SubscriptionEntryVO>> getGroupSubscriptions(@PathVariable String name) {
+ return Result.ok(metadataService.getGroupSubscriptions(name));
+ }
+
+ @PostMapping("/create")
+ public Result<ConsumerGroupVO> createConsumerGroup(@RequestBody ConsumerGroupVO group) {
+ return Result.ok(metadataService.createConsumerGroup(group));
+ }
+
+ @PostMapping("/delete")
+ public Result<Void> deleteConsumerGroup(@RequestBody Map<String, String> request) {
+ metadataService.deleteConsumerGroup(request.get("name"));
+ return Result.ok();
+ }
+
+ @PostMapping("/reset-offset")
+ public Result<Void> resetOffset(@RequestBody Map<String, Object> request) {
+ String name = (String) request.get("name");
+ long timestamp = ((Number) request.get("timestamp")).longValue();
+ String topic = (String) request.get("topic");
+ metadataService.resetOffset(name, timestamp, topic);
+ return Result.ok();
+ }
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerGroupVO.java b/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerGroupVO.java
new file mode 100644
index 0000000..88d283e
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerGroupVO.java
@@ -0,0 +1,43 @@
+/*
+ * 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 com.rocketmq.studio.instance.group;
+
+import com.rocketmq.studio.common.domain.BaseEntity;
+import com.rocketmq.studio.common.domain.enums.ConsumeType;
+import com.rocketmq.studio.common.domain.enums.SubscriptionMode;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+import java.util.List;
+
+@Data
+@EqualsAndHashCode(callSuper = true)
+public class ConsumerGroupVO extends BaseEntity {
+ private String name;
+ private String namespace;
+ private String clusterId;
+ private SubscriptionMode subscriptionMode;
+ private ConsumeType consumeType;
+ private int onlineInstances;
+ private long totalLag;
+ private List<String> subscribedTopics;
+ private String subscriptionDataType;
+ private String deliveryOrderType;
+ private int retryMaxTimes;
+ private int delaySeconds;
+ private List<ConsumerInstanceVO> instances;
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerInstanceVO.java b/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerInstanceVO.java
new file mode 100644
index 0000000..9f2acae
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/group/ConsumerInstanceVO.java
@@ -0,0 +1,40 @@
+/*
+ * 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 com.rocketmq.studio.instance.group;
+
+import com.rocketmq.studio.common.domain.enums.Protocol;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Map;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class ConsumerInstanceVO {
+ private String clientId;
+ private Protocol protocol;
+ private String address;
+ private List<String> subscribedTopics;
+ private LocalDateTime lastHeartbeat;
+ private Map<String, Long> topicLag;
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/group/QueueProgressVO.java b/server/src/main/java/com/rocketmq/studio/instance/group/QueueProgressVO.java
new file mode 100644
index 0000000..dd29e26
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/group/QueueProgressVO.java
@@ -0,0 +1,34 @@
+/*
+ * 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 com.rocketmq.studio.instance.group;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class QueueProgressVO {
+ private String broker;
+ private int queueId;
+ private long brokerOffset;
+ private long consumerOffset;
+ private long diffTotal;
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/group/SubscriptionEntryVO.java b/server/src/main/java/com/rocketmq/studio/instance/group/SubscriptionEntryVO.java
new file mode 100644
index 0000000..b77241c
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/group/SubscriptionEntryVO.java
@@ -0,0 +1,34 @@
+/*
+ * 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 com.rocketmq.studio.instance.group;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class SubscriptionEntryVO {
+ private String topic;
+ private String expression;
+ private String type;
+ private String filterMode;
+ private String consistency;
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/message/ConsumerStatusVO.java b/server/src/main/java/com/rocketmq/studio/instance/message/ConsumerStatusVO.java
new file mode 100644
index 0000000..a84d2c1
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/message/ConsumerStatusVO.java
@@ -0,0 +1,34 @@
+/*
+ * 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 com.rocketmq.studio.instance.message;
+
+import com.rocketmq.studio.common.domain.enums.DeliveryStatus;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class ConsumerStatusVO {
+ private String group;
+ private DeliveryStatus deliveryStatus;
+ private long consumeTime;
+ private int retryCount;
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/message/MessageController.java b/server/src/main/java/com/rocketmq/studio/instance/message/MessageController.java
new file mode 100644
index 0000000..657e85c
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/message/MessageController.java
@@ -0,0 +1,50 @@
+/*
+ * 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 com.rocketmq.studio.instance.message;
+
+import com.rocketmq.studio.common.domain.Result;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.List;
+
+@RestController
+@RequestMapping("/api/messages")
+@RequiredArgsConstructor
+public class MessageController {
+
+ private final MessageService messageService;
+
+ @GetMapping
+ public Result<List<MessageRecordVO>> queryMessages(
+ @RequestParam(required = false) String topic,
+ @RequestParam(required = false) String msgId,
+ @RequestParam(required = false) String key,
+ @RequestParam(required = false) Long startTime,
+ @RequestParam(required = false) Long endTime) {
+ return Result.ok(messageService.queryMessages(topic, msgId, key, startTime, endTime));
+ }
+
+ @GetMapping("/{msgId}/trace")
+ public Result<TraceRecordVO> getMessageTrace(@PathVariable String msgId) {
+ return Result.ok(messageService.getMessageTrace(msgId));
+ }
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/message/MessageProvider.java b/server/src/main/java/com/rocketmq/studio/instance/message/MessageProvider.java
new file mode 100644
index 0000000..14afe46
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/message/MessageProvider.java
@@ -0,0 +1,26 @@
+/*
+ * 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 com.rocketmq.studio.instance.message;
+
+
+import java.util.List;
+
+public interface MessageProvider {
+ List<MessageRecordVO> queryMessages(String topic, String msgId, String key, Long startTime, Long endTime);
+
+ TraceRecordVO getMessageTrace(String msgId);
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/message/MessageProviderStub.java b/server/src/main/java/com/rocketmq/studio/instance/message/MessageProviderStub.java
new file mode 100644
index 0000000..cdeb7a7
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/message/MessageProviderStub.java
@@ -0,0 +1,43 @@
+/*
+ * 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 com.rocketmq.studio.instance.message;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.util.Collections;
+import java.util.List;
+
+@Component
+@Slf4j
+public class MessageProviderStub implements MessageProvider {
+
+ @Override
+ public List<MessageRecordVO> queryMessages(String topic, String msgId, String key, Long startTime, Long endTime) {
+ log.warn("MessageProviderStub.queryMessages called - returning empty list");
+ return Collections.emptyList();
+ }
+
+ @Override
+ public TraceRecordVO getMessageTrace(String msgId) {
+ log.warn("MessageProviderStub.getMessageTrace called - returning empty trace");
+ return TraceRecordVO.builder()
+ .nodes(Collections.emptyList())
+ .consumerStatus(Collections.emptyList())
+ .build();
+ }
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/message/MessageRecordVO.java b/server/src/main/java/com/rocketmq/studio/instance/message/MessageRecordVO.java
new file mode 100644
index 0000000..17a44b4
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/message/MessageRecordVO.java
@@ -0,0 +1,41 @@
+/*
+ * 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 com.rocketmq.studio.instance.message;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.Map;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class MessageRecordVO {
+ private String msgId;
+ private String topic;
+ private String tag;
+ private String key;
+ private String body;
+ private long storeTime;
+ private String bornHost;
+ private String storeHost;
+ private Map<String, String> properties;
+ private int size;
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/message/MessageService.java b/server/src/main/java/com/rocketmq/studio/instance/message/MessageService.java
new file mode 100644
index 0000000..1f3c500
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/message/MessageService.java
@@ -0,0 +1,41 @@
+/*
+ * 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 com.rocketmq.studio.instance.message;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+
+@Service
+@RequiredArgsConstructor
+@Slf4j
+public class MessageService {
+
+ private final MessageProvider messageProvider;
+
+ public List<MessageRecordVO> queryMessages(String topic, String msgId, String key, Long startTime, Long endTime) {
+ log.info("Querying messages: topic={}, msgId={}, key={}", topic, msgId, key);
+ return messageProvider.queryMessages(topic, msgId, key, startTime, endTime);
+ }
+
+ public TraceRecordVO getMessageTrace(String msgId) {
+ log.info("Getting message trace: msgId={}", msgId);
+ return messageProvider.getMessageTrace(msgId);
+ }
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/message/TraceNodeVO.java b/server/src/main/java/com/rocketmq/studio/instance/message/TraceNodeVO.java
new file mode 100644
index 0000000..b1926bc
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/message/TraceNodeVO.java
@@ -0,0 +1,34 @@
+/*
+ * 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 com.rocketmq.studio.instance.message;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class TraceNodeVO {
+ private String title;
+ private long timestamp;
+ private String status;
+ private long costTime;
+ private String description;
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/message/TraceRecordVO.java b/server/src/main/java/com/rocketmq/studio/instance/message/TraceRecordVO.java
new file mode 100644
index 0000000..e9e7f29
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/message/TraceRecordVO.java
@@ -0,0 +1,33 @@
+/*
+ * 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 com.rocketmq.studio.instance.message;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.List;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class TraceRecordVO {
+ private List<TraceNodeVO> nodes;
+ private List<ConsumerStatusVO> consumerStatus;
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/topic/AdminClient.java b/server/src/main/java/com/rocketmq/studio/instance/topic/AdminClient.java
new file mode 100644
index 0000000..bc85263
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/topic/AdminClient.java
@@ -0,0 +1,32 @@
+/*
+ * 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 com.rocketmq.studio.instance.topic;
+
+import com.rocketmq.studio.instance.group.ConsumerGroupVO;
+
+
+public interface AdminClient {
+ TopicVO getTopic(String name);
+ ConsumerGroupVO getConsumerGroup(String name);
+ TopicVO createTopic(TopicVO topic);
+ TopicVO updateTopic(TopicVO topic);
+ void deleteTopic(String name);
+ SendMessageVO sendMessage(SendMessageDTO request);
+ ConsumerGroupVO createConsumerGroup(ConsumerGroupVO group);
+ void deleteConsumerGroup(String name);
+ void resetOffset(String name, long timestamp, String topic);
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/topic/BrokerRouteVO.java b/server/src/main/java/com/rocketmq/studio/instance/topic/BrokerRouteVO.java
new file mode 100644
index 0000000..e17bbe4
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/topic/BrokerRouteVO.java
@@ -0,0 +1,35 @@
+/*
+ * 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 com.rocketmq.studio.instance.topic;
+
+import com.rocketmq.studio.common.domain.enums.TopicPerm;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class BrokerRouteVO {
+ private String brokerName;
+ private String brokerAddr;
+ private int writeQueues;
+ private int readQueues;
+ private TopicPerm perm;
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/topic/CloudMetadataProvider.java b/server/src/main/java/com/rocketmq/studio/instance/topic/CloudMetadataProvider.java
new file mode 100644
index 0000000..c8e4b78
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/topic/CloudMetadataProvider.java
@@ -0,0 +1,61 @@
+/*
+ * 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 com.rocketmq.studio.instance.topic;
+
+import com.rocketmq.studio.instance.group.ConsumerGroupVO;
+import com.rocketmq.studio.instance.group.QueueProgressVO;
+import com.rocketmq.studio.instance.group.SubscriptionEntryVO;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+
+@Slf4j
+@Component
+public class CloudMetadataProvider implements MetadataProvider {
+
+ @Override
+ public List<TopicVO> listTopics(String clusterId, String type, String search) {
+ throw new UnsupportedOperationException("Not implemented");
+ }
+
+ @Override
+ public List<ConsumerGroupVO> listConsumerGroups(String clusterId, String search) {
+ throw new UnsupportedOperationException("Not implemented");
+ }
+
+ @Override
+ public List<BrokerRouteVO> getTopicRoutes(String name) {
+ throw new UnsupportedOperationException("Not implemented");
+ }
+
+ @Override
+ public List<TopicConsumerVO> getTopicConsumers(String name) {
+ throw new UnsupportedOperationException("Not implemented");
+ }
+
+ @Override
+ public List<QueueProgressVO> getGroupProgress(String name) {
+ throw new UnsupportedOperationException("Not implemented");
+ }
+
+ @Override
+ public List<SubscriptionEntryVO> getGroupSubscriptions(String name) {
+ throw new UnsupportedOperationException("Not implemented");
+ }
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/topic/LiteTopicVO.java b/server/src/main/java/com/rocketmq/studio/instance/topic/LiteTopicVO.java
new file mode 100644
index 0000000..8748673
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/topic/LiteTopicVO.java
@@ -0,0 +1,26 @@
+/*
+ * 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 com.rocketmq.studio.instance.topic;
+
+import lombok.Data;
+
+@Data
+public class LiteTopicVO {
+ private String name;
+ private String namespace;
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/topic/MetadataProvider.java b/server/src/main/java/com/rocketmq/studio/instance/topic/MetadataProvider.java
new file mode 100644
index 0000000..d0c04d5
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/topic/MetadataProvider.java
@@ -0,0 +1,32 @@
+/*
+ * 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 com.rocketmq.studio.instance.topic;
+
+import com.rocketmq.studio.instance.group.ConsumerGroupVO;
+import com.rocketmq.studio.instance.group.QueueProgressVO;
+import com.rocketmq.studio.instance.group.SubscriptionEntryVO;
+
+import java.util.List;
+
+public interface MetadataProvider {
+ List<TopicVO> listTopics(String clusterId, String type, String search);
+ List<ConsumerGroupVO> listConsumerGroups(String clusterId, String search);
+ List<BrokerRouteVO> getTopicRoutes(String name);
+ List<TopicConsumerVO> getTopicConsumers(String name);
+ List<QueueProgressVO> getGroupProgress(String name);
+ List<SubscriptionEntryVO> getGroupSubscriptions(String name);
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/topic/MetadataService.java b/server/src/main/java/com/rocketmq/studio/instance/topic/MetadataService.java
new file mode 100644
index 0000000..5518068
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/topic/MetadataService.java
@@ -0,0 +1,117 @@
+/*
+ * 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 com.rocketmq.studio.instance.topic;
+
+import com.rocketmq.studio.instance.group.ConsumerGroupVO;
+import com.rocketmq.studio.instance.group.QueueProgressVO;
+import com.rocketmq.studio.instance.group.SubscriptionEntryVO;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class MetadataService {
+
+ private final MetadataProvider metadataProvider;
+ private final AdminClient adminClient;
+
+ // ── TopicVO ───────────────────────────────────────────────────────
+
+
+ public List<TopicVO> listTopics(String clusterId, String type, String search) {
+ return metadataProvider.listTopics(clusterId, type, search);
+ }
+
+
+ public TopicVO createTopic(TopicVO topic) {
+ return adminClient.createTopic(topic);
+ }
+
+
+ public TopicVO updateTopic(TopicVO topic) {
+ return adminClient.updateTopic(topic);
+ }
+
+
+ public void deleteTopic(String name) {
+ adminClient.deleteTopic(name);
+ }
+
+
+ public List<BrokerRouteVO> getTopicRoutes(String name) {
+ return metadataProvider.getTopicRoutes(name);
+ }
+
+
+ public List<TopicConsumerVO> getTopicConsumers(String name) {
+ return metadataProvider.getTopicConsumers(name);
+ }
+
+
+ public SendMessageVO sendMessage(SendMessageDTO request) {
+ return adminClient.sendMessage(request);
+ }
+
+ // ── ConsumerGroupVO ───────────────────────────────────────────────
+
+
+ public List<ConsumerGroupVO> listConsumerGroups(String clusterId, String search) {
+ return metadataProvider.listConsumerGroups(clusterId, search);
+ }
+
+
+ public ConsumerGroupVO getConsumerGroup(String name) {
+ return adminClient.getConsumerGroup(name);
+ }
+
+
+ public List<QueueProgressVO> getGroupProgress(String name) {
+ return metadataProvider.getGroupProgress(name);
+ }
+
+
+ public List<SubscriptionEntryVO> getGroupSubscriptions(String name) {
+ return metadataProvider.getGroupSubscriptions(name);
+ }
+
+
+ public ConsumerGroupVO createConsumerGroup(ConsumerGroupVO group) {
+ return adminClient.createConsumerGroup(group);
+ }
+
+
+ public void deleteConsumerGroup(String name) {
+ adminClient.deleteConsumerGroup(name);
+ }
+
+
+ public void resetOffset(String name, long timestamp, String topic) {
+ adminClient.resetOffset(name, timestamp, topic);
+ }
+
+ // ── NamespaceVO ───────────────────────────────────────────────────
+
+
+ public List<NamespaceVO> listNamespaces() {
+ throw new UnsupportedOperationException("Not implemented");
+ }
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/topic/NameSrvAdminClient.java b/server/src/main/java/com/rocketmq/studio/instance/topic/NameSrvAdminClient.java
new file mode 100644
index 0000000..f306afc
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/topic/NameSrvAdminClient.java
@@ -0,0 +1,72 @@
+/*
+ * 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 com.rocketmq.studio.instance.topic;
+
+import com.rocketmq.studio.instance.group.ConsumerGroupVO;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+public class NameSrvAdminClient implements AdminClient {
+
+ @Override
+ public TopicVO getTopic(String name) {
+ throw new UnsupportedOperationException("Not implemented");
+ }
+
+ @Override
+ public ConsumerGroupVO getConsumerGroup(String name) {
+ throw new UnsupportedOperationException("Not implemented");
+ }
+
+ @Override
+ public TopicVO createTopic(TopicVO topic) {
+ throw new UnsupportedOperationException("Not implemented");
+ }
+
+ @Override
+ public TopicVO updateTopic(TopicVO topic) {
+ throw new UnsupportedOperationException("Not implemented");
+ }
+
+ @Override
+ public void deleteTopic(String name) {
+ throw new UnsupportedOperationException("Not implemented");
+ }
+
+ @Override
+ public SendMessageVO sendMessage(SendMessageDTO request) {
+ throw new UnsupportedOperationException("Not implemented");
+ }
+
+ @Override
+ public ConsumerGroupVO createConsumerGroup(ConsumerGroupVO group) {
+ throw new UnsupportedOperationException("Not implemented");
+ }
+
+ @Override
+ public void deleteConsumerGroup(String name) {
+ throw new UnsupportedOperationException("Not implemented");
+ }
+
+ @Override
+ public void resetOffset(String name, long timestamp, String topic) {
+ throw new UnsupportedOperationException("Not implemented");
+ }
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/topic/NamespaceController.java b/server/src/main/java/com/rocketmq/studio/instance/topic/NamespaceController.java
new file mode 100644
index 0000000..06c1753
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/topic/NamespaceController.java
@@ -0,0 +1,39 @@
+/*
+ * 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 com.rocketmq.studio.instance.topic;
+
+import com.rocketmq.studio.common.domain.Result;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.List;
+
+@RestController
+@RequestMapping("/api/namespaces")
+@RequiredArgsConstructor
+public class NamespaceController {
+
+ private final MetadataService metadataService;
+
+ @GetMapping
+ public Result<List<NamespaceVO>> listNamespaces() {
+ return Result.ok(metadataService.listNamespaces());
+ }
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/topic/NamespaceVO.java b/server/src/main/java/com/rocketmq/studio/instance/topic/NamespaceVO.java
new file mode 100644
index 0000000..061cb24
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/topic/NamespaceVO.java
@@ -0,0 +1,29 @@
+/*
+ * 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 com.rocketmq.studio.instance.topic;
+
+import com.rocketmq.studio.common.domain.BaseEntity;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+@Data
+@EqualsAndHashCode(callSuper = true)
+public class NamespaceVO extends BaseEntity {
+ private String name;
+ private String clusterId;
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/topic/ProxyAdminClient.java b/server/src/main/java/com/rocketmq/studio/instance/topic/ProxyAdminClient.java
new file mode 100644
index 0000000..acc1641
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/topic/ProxyAdminClient.java
@@ -0,0 +1,25 @@
+/*
+ * 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 com.rocketmq.studio.instance.topic;
+
+import org.springframework.stereotype.Component;
+
+@Component
+public class ProxyAdminClient {
+ // TODO: Implement proxy admin client for RocketMQ 5.x proxy mode
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/topic/SendMessageDTO.java b/server/src/main/java/com/rocketmq/studio/instance/topic/SendMessageDTO.java
new file mode 100644
index 0000000..ed1d799
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/topic/SendMessageDTO.java
@@ -0,0 +1,36 @@
+/*
+ * 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 com.rocketmq.studio.instance.topic;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.Map;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class SendMessageDTO {
+ private String topic;
+ private String tag;
+ private String key;
+ private String body;
+ private Map<String, String> properties;
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/topic/SendMessageVO.java b/server/src/main/java/com/rocketmq/studio/instance/topic/SendMessageVO.java
new file mode 100644
index 0000000..51ebad3
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/topic/SendMessageVO.java
@@ -0,0 +1,32 @@
+/*
+ * 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 com.rocketmq.studio.instance.topic;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class SendMessageVO {
+ private String msgId;
+ private long sendTime;
+ private String offsetMsgId;
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/topic/TopicConsumerVO.java b/server/src/main/java/com/rocketmq/studio/instance/topic/TopicConsumerVO.java
new file mode 100644
index 0000000..4de25b4
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/topic/TopicConsumerVO.java
@@ -0,0 +1,35 @@
+/*
+ * 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 com.rocketmq.studio.instance.topic;
+
+import com.rocketmq.studio.common.domain.enums.ConsumeType;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class TopicConsumerVO {
+ private String group;
+ private ConsumeType consumeType;
+ private String messageModel;
+ private double consumeTps;
+ private long diffTotal;
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/topic/TopicController.java b/server/src/main/java/com/rocketmq/studio/instance/topic/TopicController.java
new file mode 100644
index 0000000..486bbed
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/topic/TopicController.java
@@ -0,0 +1,77 @@
+/*
+ * 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 com.rocketmq.studio.instance.topic;
+
+import com.rocketmq.studio.common.domain.Result;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.List;
+import java.util.Map;
+
+@RestController
+@RequestMapping("/api/topics")
+@RequiredArgsConstructor
+public class TopicController {
+
+ private final MetadataService metadataService;
+
+ @GetMapping
+ public Result<List<TopicVO>> listTopics(
+ @RequestParam(required = false) String clusterId,
+ @RequestParam(required = false) String type,
+ @RequestParam(required = false) String search) {
+ return Result.ok(metadataService.listTopics(clusterId, type, search));
+ }
+
+ @PostMapping("/create")
+ public Result<TopicVO> createTopic(@RequestBody TopicVO topic) {
+ return Result.ok(metadataService.createTopic(topic));
+ }
+
+ @PostMapping("/update")
+ public Result<TopicVO> updateTopic(@RequestBody TopicVO topic) {
+ return Result.ok(metadataService.updateTopic(topic));
+ }
+
+ @PostMapping("/delete")
+ public Result<Void> deleteTopic(@RequestBody Map<String, String> request) {
+ metadataService.deleteTopic(request.get("name"));
+ return Result.ok();
+ }
+
+ @GetMapping("/{name}/routes")
+ public Result<List<BrokerRouteVO>> getTopicRoutes(@PathVariable String name) {
+ return Result.ok(metadataService.getTopicRoutes(name));
+ }
+
+ @GetMapping("/{name}/consumers")
+ public Result<List<TopicConsumerVO>> getTopicConsumers(@PathVariable String name) {
+ return Result.ok(metadataService.getTopicConsumers(name));
+ }
+
+ @PostMapping("/send")
+ public Result<SendMessageVO> sendMessage(@RequestBody SendMessageDTO request) {
+ return Result.ok(metadataService.sendMessage(request));
+ }
+}
diff --git a/server/src/main/java/com/rocketmq/studio/instance/topic/TopicVO.java b/server/src/main/java/com/rocketmq/studio/instance/topic/TopicVO.java
new file mode 100644
index 0000000..fde407e
--- /dev/null
+++ b/server/src/main/java/com/rocketmq/studio/instance/topic/TopicVO.java
@@ -0,0 +1,39 @@
+/*
+ * 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 com.rocketmq.studio.instance.topic;
+
+import com.rocketmq.studio.common.domain.BaseEntity;
+import com.rocketmq.studio.common.domain.enums.TopicPerm;
+import com.rocketmq.studio.common.domain.enums.TopicType;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+@Data
+@EqualsAndHashCode(callSuper = true)
+public class TopicVO extends BaseEntity {
+ private String name;
+ private String namespace;
+ private String clusterId;
+ private TopicType type;
+ private int writeQueues;
+ private int readQueues;
+ private TopicPerm perm;
+ private long messageCount;
+ private double tps;
+ private int consumerGroupCount;
+ private String remark;
+}
diff --git a/server/src/test/java/com/rocketmq/studio/instance/InstanceControllerTest.java b/server/src/test/java/com/rocketmq/studio/instance/InstanceControllerTest.java
new file mode 100644
index 0000000..e4400b1
--- /dev/null
+++ b/server/src/test/java/com/rocketmq/studio/instance/InstanceControllerTest.java
@@ -0,0 +1,191 @@
+/*
+ * 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 com.rocketmq.studio.instance;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.rocketmq.studio.common.domain.enums.InstanceType;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
+import org.springframework.boot.test.autoconfigure.web.servlet.WebMvcTest;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.http.MediaType;
+import org.springframework.test.web.servlet.MockMvc;
+
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Map;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+
+@WebMvcTest(InstanceController.class)
+@AutoConfigureMockMvc(addFilters = false)
+class InstanceControllerTest {
+
+ @Autowired
+ private MockMvc mockMvc;
+
+ @MockBean
+ private InstanceService instanceService;
+
+ @Autowired
+ private ObjectMapper objectMapper;
+
+ @Test
+ void listInstancesShouldReturnAllInstances() throws Exception {
+ InstanceVO inst = buildInstance("inst-1", "production-proxy", InstanceType.PROXY, "10.0.1.1:8080");
+
+ when(instanceService.listInstances(isNull(), isNull())).thenReturn(List.of(inst));
+
+ mockMvc.perform(get("/api/instances"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.code").value(200))
+ .andExpect(jsonPath("$.data").isArray())
+ .andExpect(jsonPath("$.data[0].id").value("inst-1"))
+ .andExpect(jsonPath("$.data[0].name").value("production-proxy"))
+ .andExpect(jsonPath("$.data[0].type").value("PROXY"))
+ .andExpect(jsonPath("$.data[0].endpoint").value("10.0.1.1:8080"));
+ }
+
+ @Test
+ void listInstancesShouldFilterByType() throws Exception {
+ InstanceVO inst = buildInstance("inst-1", "proxy-1", InstanceType.PROXY, "10.0.1.1:8080");
+
+ when(instanceService.listInstances(eq(InstanceType.PROXY), isNull())).thenReturn(List.of(inst));
+
+ mockMvc.perform(get("/api/instances")
+ .param("type", "PROXY"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.data").isArray())
+ .andExpect(jsonPath("$.data[0].type").value("PROXY"));
+
+ verify(instanceService).listInstances(eq(InstanceType.PROXY), isNull());
+ }
+
+ @Test
+ void listInstancesShouldFilterBySearch() throws Exception {
+ InstanceVO inst = buildInstance("inst-1", "production", InstanceType.PROXY, "10.0.1.1:8080");
+
+ when(instanceService.listInstances(isNull(), eq("prod"))).thenReturn(List.of(inst));
+
+ mockMvc.perform(get("/api/instances")
+ .param("search", "prod"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.data[0].name").value("production"));
+
+ verify(instanceService).listInstances(isNull(), eq("prod"));
+ }
+
+ @Test
+ void createInstanceShouldReturnCreatedInstance() throws Exception {
+ InstanceVO input = InstanceVO.builder()
+ .name("new-instance")
+ .endpoint("10.0.2.1:8080")
+ .type(InstanceType.DIRECT)
+ .build();
+
+ InstanceVO created = InstanceVO.builder()
+ .name("new-instance")
+ .endpoint("10.0.2.1:8080")
+ .type(InstanceType.DIRECT)
+ .topicCount(0)
+ .consumerGroupCount(0)
+ .build();
+ created.setId("new-id");
+ created.setCreatedAt(LocalDateTime.of(2026, 1, 1, 0, 0));
+ created.setUpdatedAt(LocalDateTime.of(2026, 1, 1, 0, 0));
+
+ when(instanceService.createInstance(any(InstanceVO.class))).thenReturn(created);
+
+ mockMvc.perform(post("/api/instances/create")
+ .contentType(MediaType.APPLICATION_JSON)
+ .content(objectMapper.writeValueAsString(input)))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.code").value(200))
+ .andExpect(jsonPath("$.data.id").value("new-id"))
+ .andExpect(jsonPath("$.data.name").value("new-instance"))
+ .andExpect(jsonPath("$.data.endpoint").value("10.0.2.1:8080"))
+ .andExpect(jsonPath("$.data.type").value("DIRECT"));
+ }
+
+ @Test
+ void updateInstanceShouldReturnUpdatedInstance() throws Exception {
+ InstanceVO update = InstanceVO.builder()
+ .name("updated-name")
+ .build();
+ update.setId("inst-1");
+
+ InstanceVO updated = InstanceVO.builder()
+ .name("updated-name")
+ .endpoint("10.0.1.1:8080")
+ .type(InstanceType.PROXY)
+ .build();
+ updated.setId("inst-1");
+ updated.setCreatedAt(LocalDateTime.of(2026, 1, 1, 0, 0));
+ updated.setUpdatedAt(LocalDateTime.of(2026, 7, 8, 12, 0));
+
+ when(instanceService.updateInstance(any(InstanceVO.class))).thenReturn(updated);
+
+ mockMvc.perform(post("/api/instances/update")
+ .contentType(MediaType.APPLICATION_JSON)
+ .content(objectMapper.writeValueAsString(update)))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.code").value(200))
+ .andExpect(jsonPath("$.data.id").value("inst-1"))
+ .andExpect(jsonPath("$.data.name").value("updated-name"));
+ }
+
+ @Test
+ void deleteInstanceShouldReturnSuccess() throws Exception {
+ doNothing().when(instanceService).deleteInstance("inst-1");
+
+ Map<String, String> body = Map.of("id", "inst-1");
+
+ mockMvc.perform(post("/api/instances/delete")
+ .contentType(MediaType.APPLICATION_JSON)
+ .content(objectMapper.writeValueAsString(body)))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.code").value(200))
+ .andExpect(jsonPath("$.message").value("success"));
+
+ verify(instanceService).deleteInstance("inst-1");
+ }
+
+ private InstanceVO buildInstance(String id, String name, InstanceType type, String endpoint) {
+ InstanceVO instance = InstanceVO.builder()
+ .name(name)
+ .type(type)
+ .endpoint(endpoint)
+ .topicCount(10)
+ .consumerGroupCount(5)
+ .build();
+ instance.setId(id);
+ instance.setCreatedAt(LocalDateTime.of(2026, 1, 1, 0, 0));
+ instance.setUpdatedAt(LocalDateTime.of(2026, 1, 1, 0, 0));
+ return instance;
+ }
+}
diff --git a/server/src/test/java/com/rocketmq/studio/instance/InstanceServiceTest.java b/server/src/test/java/com/rocketmq/studio/instance/InstanceServiceTest.java
new file mode 100644
index 0000000..3e47087
--- /dev/null
+++ b/server/src/test/java/com/rocketmq/studio/instance/InstanceServiceTest.java
@@ -0,0 +1,271 @@
+/*
+ * 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 com.rocketmq.studio.instance;
+
+import com.rocketmq.studio.common.domain.enums.InstanceType;
+import com.rocketmq.studio.common.exception.BusinessException;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.util.List;
+import java.util.Optional;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class InstanceServiceTest {
+
+ @Mock
+ private InstanceRepository instanceRepository;
+
+ @InjectMocks
+ private InstanceService instanceService;
+
+ @Test
+ void listInstancesShouldReturnAllWhenNoFilters() {
+ List<InstanceVO> instances = List.of(
+ InstanceVO.builder().name("inst-1").build(),
+ InstanceVO.builder().name("inst-2").build()
+ );
+ when(instanceRepository.findAll()).thenReturn(instances);
+
+ List<InstanceVO> result = instanceService.listInstances(null, null);
+
+ assertThat(result).hasSize(2);
+ verify(instanceRepository).findAll();
+ }
+
+ @Test
+ void listInstancesShouldFilterByType() {
+ List<InstanceVO> instances = List.of(
+ InstanceVO.builder().name("proxy-1").type(InstanceType.PROXY).build()
+ );
+ when(instanceRepository.findByType(InstanceType.PROXY)).thenReturn(instances);
+
+ List<InstanceVO> result = instanceService.listInstances(InstanceType.PROXY, null);
+
+ assertThat(result).hasSize(1);
+ verify(instanceRepository).findByType(InstanceType.PROXY);
+ }
+
+ @Test
+ void listInstancesShouldSearchByKeyword() {
+ List<InstanceVO> instances = List.of(
+ InstanceVO.builder().name("production").build()
+ );
+ when(instanceRepository.search("prod")).thenReturn(instances);
+
+ List<InstanceVO> result = instanceService.listInstances(null, "prod");
+
+ assertThat(result).hasSize(1);
+ verify(instanceRepository).search("prod");
+ }
+
+ @Test
+ void listInstancesShouldFilterByTypeAndSearch() {
+ List<InstanceVO> instances = List.of(
+ InstanceVO.builder().name("production-proxy").type(InstanceType.PROXY).build()
+ );
+ when(instanceRepository.findByTypeAndSearch(InstanceType.PROXY, "prod")).thenReturn(instances);
+
+ List<InstanceVO> result = instanceService.listInstances(InstanceType.PROXY, "prod");
+
+ assertThat(result).hasSize(1);
+ verify(instanceRepository).findByTypeAndSearch(InstanceType.PROXY, "prod");
+ }
+
+ @Test
+ void listInstancesShouldIgnoreBlankSearch() {
+ List<InstanceVO> instances = List.of(InstanceVO.builder().name("inst").build());
+ when(instanceRepository.findAll()).thenReturn(instances);
+
+ List<InstanceVO> result = instanceService.listInstances(null, " ");
+
+ assertThat(result).hasSize(1);
+ verify(instanceRepository).findAll();
+ }
+
+ @Test
+ void createInstanceShouldSetIdAndTimestamps() {
+ InstanceVO input = InstanceVO.builder()
+ .name("new-instance")
+ .endpoint("10.0.1.1:8080")
+ .type(InstanceType.PROXY)
+ .build();
+
+ when(instanceRepository.save(any(InstanceVO.class))).thenAnswer(inv -> inv.getArgument(0));
+
+ InstanceVO result = instanceService.createInstance(input);
+
+ assertThat(result.getId()).isNotBlank();
+ assertThat(result.getCreatedAt()).isNotNull();
+ assertThat(result.getUpdatedAt()).isNotNull();
+ assertThat(result.getName()).isEqualTo("new-instance");
+ verify(instanceRepository).save(any(InstanceVO.class));
+ }
+
+ @Test
+ void createInstanceShouldThrowWhenNameIsNull() {
+ InstanceVO input = InstanceVO.builder()
+ .name(null)
+ .endpoint("10.0.1.1:8080")
+ .build();
+
+ assertThatThrownBy(() -> instanceService.createInstance(input))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("InstanceVO name is required");
+ }
+
+ @Test
+ void createInstanceShouldThrowWhenNameIsBlank() {
+ InstanceVO input = InstanceVO.builder()
+ .name(" ")
+ .endpoint("10.0.1.1:8080")
+ .build();
+
+ assertThatThrownBy(() -> instanceService.createInstance(input))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("InstanceVO name is required");
+ }
+
+ @Test
+ void createInstanceShouldThrowWhenEndpointIsNull() {
+ InstanceVO input = InstanceVO.builder()
+ .name("valid-name")
+ .endpoint(null)
+ .build();
+
+ assertThatThrownBy(() -> instanceService.createInstance(input))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("InstanceVO endpoint is required");
+ }
+
+ @Test
+ void createInstanceShouldThrowWhenEndpointIsBlank() {
+ InstanceVO input = InstanceVO.builder()
+ .name("valid-name")
+ .endpoint(" ")
+ .build();
+
+ assertThatThrownBy(() -> instanceService.createInstance(input))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("InstanceVO endpoint is required");
+ }
+
+ @Test
+ void updateInstanceShouldMergeFieldsOntoExisting() {
+ InstanceVO existing = InstanceVO.builder()
+ .name("old-name")
+ .endpoint("10.0.1.1:8080")
+ .type(InstanceType.PROXY)
+ .remark("old remark")
+ .build();
+ existing.setId("inst-1");
+
+ InstanceVO update = InstanceVO.builder()
+ .name("new-name")
+ .remark("new remark")
+ .build();
+ update.setId("inst-1");
+
+ when(instanceRepository.findById("inst-1")).thenReturn(Optional.of(existing));
+ when(instanceRepository.save(any(InstanceVO.class))).thenAnswer(inv -> inv.getArgument(0));
+
+ InstanceVO result = instanceService.updateInstance(update);
+
+ assertThat(result.getName()).isEqualTo("new-name");
+ assertThat(result.getEndpoint()).isEqualTo("10.0.1.1:8080");
+ assertThat(result.getType()).isEqualTo(InstanceType.PROXY);
+ assertThat(result.getRemark()).isEqualTo("new remark");
+ assertThat(result.getUpdatedAt()).isNotNull();
+ }
+
+ @Test
+ void updateInstanceShouldThrowWhenIdIsNull() {
+ InstanceVO input = InstanceVO.builder().name("test").build();
+ input.setId(null);
+
+ assertThatThrownBy(() -> instanceService.updateInstance(input))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("InstanceVO ID is required");
+ }
+
+ @Test
+ void updateInstanceShouldThrowWhenIdIsBlank() {
+ InstanceVO input = InstanceVO.builder().name("test").build();
+ input.setId(" ");
+
+ assertThatThrownBy(() -> instanceService.updateInstance(input))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("InstanceVO ID is required");
+ }
+
+ @Test
+ void updateInstanceShouldThrowWhenInstanceNotFound() {
+ InstanceVO input = InstanceVO.builder().name("test").build();
+ input.setId("nonexistent");
+
+ when(instanceRepository.findById("nonexistent")).thenReturn(Optional.empty());
+
+ assertThatThrownBy(() -> instanceService.updateInstance(input))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("InstanceVO not found: nonexistent");
+ }
+
+ @Test
+ void deleteInstanceShouldRemoveExistingInstance() {
+ InstanceVO existing = InstanceVO.builder().name("to-delete").build();
+ existing.setId("inst-1");
+
+ when(instanceRepository.findById("inst-1")).thenReturn(Optional.of(existing));
+
+ instanceService.deleteInstance("inst-1");
+
+ verify(instanceRepository).deleteById("inst-1");
+ }
+
+ @Test
+ void deleteInstanceShouldThrowWhenIdIsNull() {
+ assertThatThrownBy(() -> instanceService.deleteInstance(null))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("InstanceVO ID is required");
+ }
+
+ @Test
+ void deleteInstanceShouldThrowWhenIdIsBlank() {
+ assertThatThrownBy(() -> instanceService.deleteInstance(" "))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("InstanceVO ID is required");
+ }
+
+ @Test
+ void deleteInstanceShouldThrowWhenInstanceNotFound() {
+ when(instanceRepository.findById("missing")).thenReturn(Optional.empty());
+
+ assertThatThrownBy(() -> instanceService.deleteInstance("missing"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("InstanceVO not found: missing");
+ }
+}
diff --git a/server/src/test/java/com/rocketmq/studio/instance/acl/AclControllerTest.java b/server/src/test/java/com/rocketmq/studio/instance/acl/AclControllerTest.java
new file mode 100644
index 0000000..1525082
--- /dev/null
+++ b/server/src/test/java/com/rocketmq/studio/instance/acl/AclControllerTest.java
@@ -0,0 +1,140 @@
+/*
+ * 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 com.rocketmq.studio.instance.acl;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
+import org.springframework.boot.test.autoconfigure.web.servlet.WebMvcTest;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.http.MediaType;
+import org.springframework.test.web.servlet.MockMvc;
+
+import java.time.LocalDateTime;
+import java.util.List;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+
+@WebMvcTest(AclController.class)
+@AutoConfigureMockMvc(addFilters = false)
+class AclControllerTest {
+
+ @Autowired
+ private MockMvc mockMvc;
+
+ @MockBean
+ private AclService aclService;
+
+ @Autowired
+ private ObjectMapper objectMapper;
+
+ @Test
+ void listRulesShouldReturnRules() throws Exception {
+ AclRuleVO rule = AclRuleVO.builder()
+ .principal("user1")
+ .resource("topic-1")
+ .resourceType("TOPIC")
+ .decision("ALLOW")
+ .build();
+ rule.setId("rule-1");
+ rule.setCreatedAt(LocalDateTime.of(2026, 1, 1, 0, 0));
+
+ when(aclService.listRules(isNull(), isNull())).thenReturn(List.of(rule));
+
+ mockMvc.perform(get("/api/acl/rules"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.code").value(200))
+ .andExpect(jsonPath("$.data").isArray())
+ .andExpect(jsonPath("$.data[0].id").value("rule-1"))
+ .andExpect(jsonPath("$.data[0].principal").value("user1"))
+ .andExpect(jsonPath("$.data[0].decision").value("ALLOW"));
+ }
+
+ @Test
+ void listRulesShouldPassQueryParams() throws Exception {
+ when(aclService.listRules(eq("cluster-1"), eq("user1"))).thenReturn(List.of());
+
+ mockMvc.perform(get("/api/acl/rules")
+ .param("clusterId", "cluster-1")
+ .param("principal", "user1"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.data").isArray());
+
+ verify(aclService).listRules(eq("cluster-1"), eq("user1"));
+ }
+
+ @Test
+ void createRuleShouldReturnCreatedRule() throws Exception {
+ AclRuleVO input = AclRuleVO.builder()
+ .principal("user1")
+ .resource("topic-1")
+ .resourceType("TOPIC")
+ .decision("ALLOW")
+ .build();
+
+ AclRuleVO created = AclRuleVO.builder()
+ .principal("user1")
+ .resource("topic-1")
+ .resourceType("TOPIC")
+ .decision("ALLOW")
+ .build();
+ created.setId("new-rule-id");
+ created.setCreatedAt(LocalDateTime.of(2026, 7, 8, 12, 0));
+
+ when(aclService.createRule(any(AclRuleVO.class))).thenReturn(created);
+
+ mockMvc.perform(post("/api/acl/rules/create")
+ .contentType(MediaType.APPLICATION_JSON)
+ .content(objectMapper.writeValueAsString(input)))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.code").value(200))
+ .andExpect(jsonPath("$.data.id").value("new-rule-id"))
+ .andExpect(jsonPath("$.data.principal").value("user1"));
+ }
+
+ @Test
+ void listUsersShouldReturnAllUsers() throws Exception {
+ AclUserVO user = AclUserVO.builder()
+ .username("admin")
+ .accessKey("ak123")
+ .secretKey("sk456")
+ .admin(true)
+ .build();
+ user.setId("user-1");
+ user.setCreatedAt(LocalDateTime.of(2026, 1, 1, 0, 0));
+
+ when(aclService.listUsers()).thenReturn(List.of(user));
+
+ mockMvc.perform(get("/api/acl/users"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.code").value(200))
+ .andExpect(jsonPath("$.data").isArray())
+ .andExpect(jsonPath("$.data[0].id").value("user-1"))
+ .andExpect(jsonPath("$.data[0].username").value("admin"))
+ .andExpect(jsonPath("$.data[0].admin").value(true));
+ }
+}
diff --git a/server/src/test/java/com/rocketmq/studio/instance/acl/AclServiceTest.java b/server/src/test/java/com/rocketmq/studio/instance/acl/AclServiceTest.java
new file mode 100644
index 0000000..c24a7a2
--- /dev/null
+++ b/server/src/test/java/com/rocketmq/studio/instance/acl/AclServiceTest.java
@@ -0,0 +1,136 @@
+/*
+ * 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 com.rocketmq.studio.instance.acl;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class AclServiceTest {
+
+ @Mock
+ private AclRepository aclRepository;
+
+ @InjectMocks
+ private AclService aclService;
+
+ @Test
+ void listRulesShouldReturnRulesFromRepository() {
+ List<AclRuleVO> rules = List.of(
+ AclRuleVO.builder().principal("user1").resource("topic-1").decision("ALLOW").build(),
+ AclRuleVO.builder().principal("user2").resource("topic-2").decision("DENY").build()
+ );
+ when(aclRepository.findRules("cluster-1", "user1")).thenReturn(rules);
+
+ List<AclRuleVO> result = aclService.listRules("cluster-1", "user1");
+
+ assertThat(result).hasSize(2);
+ assertThat(result.get(0).getPrincipal()).isEqualTo("user1");
+ verify(aclRepository).findRules("cluster-1", "user1");
+ }
+
+ @Test
+ void listRulesShouldPassNullFilters() {
+ when(aclRepository.findRules(null, null)).thenReturn(List.of());
+
+ List<AclRuleVO> result = aclService.listRules(null, null);
+
+ assertThat(result).isEmpty();
+ verify(aclRepository).findRules(null, null);
+ }
+
+ @Test
+ void createRuleShouldSetIdAndTimestamp() {
+ AclRuleVO input = AclRuleVO.builder()
+ .principal("user1")
+ .resource("topic-1")
+ .resourceType("TOPIC")
+ .decision("ALLOW")
+ .build();
+
+ when(aclRepository.saveRule(any(AclRuleVO.class))).thenAnswer(inv -> inv.getArgument(0));
+
+ AclRuleVO result = aclService.createRule(input);
+
+ assertThat(result.getId()).isNotBlank();
+ assertThat(result.getCreatedAt()).isNotNull();
+ assertThat(result.getPrincipal()).isEqualTo("user1");
+ assertThat(result.getResource()).isEqualTo("topic-1");
+ verify(aclRepository).saveRule(any(AclRuleVO.class));
+ }
+
+ @Test
+ void deleteRuleShouldDelegateToRepository() {
+ aclService.deleteRule("rule-1");
+
+ verify(aclRepository).deleteRule("rule-1");
+ }
+
+ @Test
+ void listUsersShouldReturnAllUsers() {
+ List<AclUserVO> users = List.of(
+ AclUserVO.builder().username("admin").admin(true).build(),
+ AclUserVO.builder().username("reader").admin(false).build()
+ );
+ when(aclRepository.findUsers()).thenReturn(users);
+
+ List<AclUserVO> result = aclService.listUsers();
+
+ assertThat(result).hasSize(2);
+ assertThat(result.get(0).getUsername()).isEqualTo("admin");
+ verify(aclRepository).findUsers();
+ }
+
+ @Test
+ void createUserShouldGenerateKeysAndSetTimestamp() {
+ AclUserVO input = AclUserVO.builder()
+ .username("newuser")
+ .admin(false)
+ .build();
+
+ when(aclRepository.saveUser(any(AclUserVO.class))).thenAnswer(inv -> inv.getArgument(0));
+
+ AclUserVO result = aclService.createUser(input);
+
+ assertThat(result.getId()).isNotBlank();
+ assertThat(result.getAccessKey()).isNotBlank();
+ assertThat(result.getSecretKey()).isNotBlank();
+ assertThat(result.getAccessKey()).doesNotContain("-");
+ assertThat(result.getSecretKey()).doesNotContain("-");
+ assertThat(result.getCreatedAt()).isNotNull();
+ assertThat(result.getUsername()).isEqualTo("newuser");
+ verify(aclRepository).saveUser(any(AclUserVO.class));
+ }
+
+ @Test
+ void deleteUserShouldDelegateToRepository() {
+ aclService.deleteUser("user-1");
+
+ verify(aclRepository).deleteUser("user-1");
+ }
+}
diff --git a/server/src/test/java/com/rocketmq/studio/instance/dlq/DLQControllerTest.java b/server/src/test/java/com/rocketmq/studio/instance/dlq/DLQControllerTest.java
new file mode 100644
index 0000000..3293ecb
--- /dev/null
+++ b/server/src/test/java/com/rocketmq/studio/instance/dlq/DLQControllerTest.java
@@ -0,0 +1,125 @@
+/*
+ * 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 com.rocketmq.studio.instance.dlq;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
+import org.springframework.boot.test.autoconfigure.web.servlet.WebMvcTest;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.http.MediaType;
+import org.springframework.test.web.servlet.MockMvc;
+
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Map;
+
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+
+@WebMvcTest(DLQController.class)
+@AutoConfigureMockMvc(addFilters = false)
+class DLQControllerTest {
+
+ @Autowired
+ private MockMvc mockMvc;
+
+ @MockBean
+ private DLQService dlqService;
+
+ @Autowired
+ private ObjectMapper objectMapper;
+
+ @Test
+ void listDLQGroupsShouldReturnGroups() throws Exception {
+ DLQGroupVO group = DLQGroupVO.builder()
+ .groupName("test-group")
+ .dlqTopic("%DLQ%test-group")
+ .messageCount(10)
+ .lastEnqueueTime(LocalDateTime.of(2026, 7, 8, 10, 0))
+ .retryCount(3)
+ .status("ACTIVE")
+ .build();
+
+ when(dlqService.listDLQGroups(isNull())).thenReturn(List.of(group));
+
+ mockMvc.perform(get("/api/dlq"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.code").value(200))
+ .andExpect(jsonPath("$.data").isArray())
+ .andExpect(jsonPath("$.data[0].groupName").value("test-group"))
+ .andExpect(jsonPath("$.data[0].dlqTopic").value("%DLQ%test-group"))
+ .andExpect(jsonPath("$.data[0].messageCount").value(10));
+ }
+
+ @Test
+ void listDLQGroupsShouldPassClusterId() throws Exception {
+ when(dlqService.listDLQGroups(eq("cluster-1"))).thenReturn(List.of());
+
+ mockMvc.perform(get("/api/dlq")
+ .param("clusterId", "cluster-1"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.data").isArray());
+
+ verify(dlqService).listDLQGroups(eq("cluster-1"));
+ }
+
+ @Test
+ void resendMessagesShouldReturnSuccess() throws Exception {
+ Map<String, Object> body = Map.of(
+ "groupName", "test-group",
+ "startTime", 1000,
+ "endTime", 2000,
+ "targetTopic", "target-topic"
+ );
+
+ mockMvc.perform(post("/api/dlq/resend")
+ .contentType(MediaType.APPLICATION_JSON)
+ .content(objectMapper.writeValueAsString(body)))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.code").value(200))
+ .andExpect(jsonPath("$.message").value("success"));
+
+ verify(dlqService).resendMessages(
+ eq("test-group"), eq(1000L), eq(2000L), eq("target-topic"));
+ }
+
+ @Test
+ void resendMessagesShouldHandleNullTimeRange() throws Exception {
+ Map<String, Object> body = Map.of(
+ "groupName", "test-group",
+ "targetTopic", "target-topic"
+ );
+
+ mockMvc.perform(post("/api/dlq/resend")
+ .contentType(MediaType.APPLICATION_JSON)
+ .content(objectMapper.writeValueAsString(body)))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.code").value(200));
+
+ verify(dlqService).resendMessages(
+ eq("test-group"), isNull(), isNull(), eq("target-topic"));
+ }
+}
diff --git a/server/src/test/java/com/rocketmq/studio/instance/dlq/DLQServiceTest.java b/server/src/test/java/com/rocketmq/studio/instance/dlq/DLQServiceTest.java
new file mode 100644
index 0000000..c1631f7
--- /dev/null
+++ b/server/src/test/java/com/rocketmq/studio/instance/dlq/DLQServiceTest.java
@@ -0,0 +1,90 @@
+/*
+ * 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 com.rocketmq.studio.instance.dlq;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class DLQServiceTest {
+
+ @Mock
+ private DLQProvider dlqProvider;
+
+ @InjectMocks
+ private DLQService dlqService;
+
+ @Test
+ void listDLQGroupsShouldReturnGroupsFromProvider() {
+ List<DLQGroupVO> groups = List.of(
+ DLQGroupVO.builder()
+ .groupName("group-1")
+ .dlqTopic("%DLQ%group-1")
+ .messageCount(10)
+ .status("ACTIVE")
+ .build(),
+ DLQGroupVO.builder()
+ .groupName("group-2")
+ .dlqTopic("%DLQ%group-2")
+ .messageCount(5)
+ .status("ACTIVE")
+ .build()
+ );
+ when(dlqProvider.listDLQGroups("cluster-1")).thenReturn(groups);
+
+ List<DLQGroupVO> result = dlqService.listDLQGroups("cluster-1");
+
+ assertThat(result).hasSize(2);
+ assertThat(result.get(0).getGroupName()).isEqualTo("group-1");
+ assertThat(result.get(0).getDlqTopic()).isEqualTo("%DLQ%group-1");
+ verify(dlqProvider).listDLQGroups("cluster-1");
+ }
+
+ @Test
+ void listDLQGroupsShouldReturnEmptyWhenNone() {
+ when(dlqProvider.listDLQGroups("cluster-2")).thenReturn(List.of());
+
+ List<DLQGroupVO> result = dlqService.listDLQGroups("cluster-2");
+
+ assertThat(result).isEmpty();
+ verify(dlqProvider).listDLQGroups("cluster-2");
+ }
+
+ @Test
+ void resendMessagesShouldDelegateToProvider() {
+ dlqService.resendMessages("group-1", 1000L, 2000L, "target-topic");
+
+ verify(dlqProvider).resendMessages("group-1", 1000L, 2000L, "target-topic");
+ }
+
+ @Test
+ void resendMessagesShouldAcceptNullTimeRange() {
+ dlqService.resendMessages("group-1", null, null, "target-topic");
+
+ verify(dlqProvider).resendMessages("group-1", null, null, "target-topic");
+ }
+}
diff --git a/server/src/test/java/com/rocketmq/studio/instance/topic/MetadataServiceTest.java b/server/src/test/java/com/rocketmq/studio/instance/topic/MetadataServiceTest.java
new file mode 100644
index 0000000..9e45d71
--- /dev/null
+++ b/server/src/test/java/com/rocketmq/studio/instance/topic/MetadataServiceTest.java
@@ -0,0 +1,144 @@
+/*
+ * 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 com.rocketmq.studio.instance.topic;
+
+import com.rocketmq.studio.instance.group.ConsumerGroupVO;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class MetadataServiceTest {
+
+ @Mock
+ private MetadataProvider metadataProvider;
+
+ @Mock
+ private AdminClient adminClient;
+
+ @InjectMocks
+ private MetadataService metadataService;
+
+ @Test
+ void listTopicsShouldReturnTopicsFromProvider() {
+ TopicVO topic = new TopicVO();
+ topic.setName("test-topic");
+ topic.setWriteQueues(8);
+ topic.setReadQueues(8);
+
+ when(metadataProvider.listTopics("cluster-1", "NORMAL", null)).thenReturn(List.of(topic));
+
+ List<TopicVO> result = metadataService.listTopics("cluster-1", "NORMAL", null);
+
+ assertThat(result).hasSize(1);
+ assertThat(result.get(0).getName()).isEqualTo("test-topic");
+ verify(metadataProvider).listTopics("cluster-1", "NORMAL", null);
+ }
+
+ @Test
+ void listTopicsShouldReturnEmptyWhenNone() {
+ when(metadataProvider.listTopics(null, null, "nonexistent")).thenReturn(List.of());
+
+ List<TopicVO> result = metadataService.listTopics(null, null, "nonexistent");
+
+ assertThat(result).isEmpty();
+ }
+
+ @Test
+ void createTopicShouldDelegateToAdminClient() {
+ TopicVO input = new TopicVO();
+ input.setName("new-topic");
+ input.setWriteQueues(8);
+ input.setReadQueues(8);
+
+ TopicVO created = new TopicVO();
+ created.setName("new-topic");
+ created.setWriteQueues(8);
+ created.setReadQueues(8);
+
+ when(adminClient.createTopic(any(TopicVO.class))).thenReturn(created);
+
+ TopicVO result = metadataService.createTopic(input);
+
+ assertThat(result.getName()).isEqualTo("new-topic");
+ verify(adminClient).createTopic(input);
+ }
+
+ @Test
+ void deleteTopicShouldDelegateToAdminClient() {
+ metadataService.deleteTopic("topic-to-delete");
+
+ verify(adminClient).deleteTopic("topic-to-delete");
+ }
+
+ @Test
+ void sendMessageShouldReturnResult() {
+ SendMessageDTO request = SendMessageDTO.builder()
+ .topic("test-topic")
+ .tag("TagA")
+ .body("hello")
+ .build();
+
+ SendMessageVO expectedResult = SendMessageVO.builder()
+ .msgId("msg-001")
+ .sendTime(System.currentTimeMillis())
+ .offsetMsgId("offset-001")
+ .build();
+
+ when(adminClient.sendMessage(request)).thenReturn(expectedResult);
+
+ SendMessageVO result = metadataService.sendMessage(request);
+
+ assertThat(result.getMsgId()).isEqualTo("msg-001");
+ assertThat(result.getOffsetMsgId()).isEqualTo("offset-001");
+ verify(adminClient).sendMessage(request);
+ }
+
+ @Test
+ void listConsumerGroupsShouldReturnGroupsFromProvider() {
+ ConsumerGroupVO group = new ConsumerGroupVO();
+ group.setName("test-group");
+
+ when(metadataProvider.listConsumerGroups("cluster-1", null)).thenReturn(List.of(group));
+
+ List<ConsumerGroupVO> result = metadataService.listConsumerGroups("cluster-1", null);
+
+ assertThat(result).hasSize(1);
+ assertThat(result.get(0).getName()).isEqualTo("test-group");
+ verify(metadataProvider).listConsumerGroups("cluster-1", null);
+ }
+
+ @Test
+ void listConsumerGroupsShouldPassSearchFilter() {
+ when(metadataProvider.listConsumerGroups(null, "order")).thenReturn(List.of());
+
+ List<ConsumerGroupVO> result = metadataService.listConsumerGroups(null, "order");
+
+ assertThat(result).isEmpty();
+ verify(metadataProvider).listConsumerGroups(null, "order");
+ }
+}
diff --git a/server/src/test/java/com/rocketmq/studio/instance/topic/TopicControllerTest.java b/server/src/test/java/com/rocketmq/studio/instance/topic/TopicControllerTest.java
new file mode 100644
index 0000000..7d7ab13
--- /dev/null
+++ b/server/src/test/java/com/rocketmq/studio/instance/topic/TopicControllerTest.java
@@ -0,0 +1,133 @@
+/*
+ * 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 com.rocketmq.studio.instance.topic;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
+import org.springframework.boot.test.autoconfigure.web.servlet.WebMvcTest;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.http.MediaType;
+import org.springframework.test.web.servlet.MockMvc;
+
+import java.util.List;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+
+@WebMvcTest(TopicController.class)
+@AutoConfigureMockMvc(addFilters = false)
+class TopicControllerTest {
+
+ @Autowired
+ private MockMvc mockMvc;
+
+ @MockBean
+ private MetadataService metadataService;
+
+ @Autowired
+ private ObjectMapper objectMapper;
+
+ @Test
+ void listTopicsShouldReturnTopics() throws Exception {
+ TopicVO topic = new TopicVO();
+ topic.setName("test-topic");
+ topic.setWriteQueues(8);
+ topic.setReadQueues(8);
+
+ when(metadataService.listTopics(isNull(), isNull(), isNull())).thenReturn(List.of(topic));
+
+ mockMvc.perform(get("/api/topics"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.code").value(200))
+ .andExpect(jsonPath("$.data").isArray())
+ .andExpect(jsonPath("$.data[0].name").value("test-topic"))
+ .andExpect(jsonPath("$.data[0].writeQueues").value(8));
+ }
+
+ @Test
+ void listTopicsShouldPassQueryParams() throws Exception {
+ when(metadataService.listTopics(eq("cluster-1"), eq("NORMAL"), eq("test")))
+ .thenReturn(List.of());
+
+ mockMvc.perform(get("/api/topics")
+ .param("clusterId", "cluster-1")
+ .param("type", "NORMAL")
+ .param("search", "test"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.data").isArray());
+
+ verify(metadataService).listTopics(eq("cluster-1"), eq("NORMAL"), eq("test"));
+ }
+
+ @Test
+ void createTopicShouldReturnCreatedTopic() throws Exception {
+ TopicVO input = new TopicVO();
+ input.setName("new-topic");
+ input.setWriteQueues(16);
+ input.setReadQueues(16);
+
+ TopicVO created = new TopicVO();
+ created.setName("new-topic");
+ created.setWriteQueues(16);
+ created.setReadQueues(16);
+
+ when(metadataService.createTopic(any(TopicVO.class))).thenReturn(created);
+
+ mockMvc.perform(post("/api/topics/create")
+ .contentType(MediaType.APPLICATION_JSON)
+ .content(objectMapper.writeValueAsString(input)))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.code").value(200))
+ .andExpect(jsonPath("$.data.name").value("new-topic"))
+ .andExpect(jsonPath("$.data.writeQueues").value(16));
+ }
+
+ @Test
+ void sendMessageShouldReturnResult() throws Exception {
+ SendMessageDTO request = SendMessageDTO.builder()
+ .topic("test-topic")
+ .tag("TagA")
+ .body("hello world")
+ .build();
+
+ SendMessageVO result = SendMessageVO.builder()
+ .msgId("msg-001")
+ .sendTime(1720000000000L)
+ .offsetMsgId("offset-001")
+ .build();
+
+ when(metadataService.sendMessage(any(SendMessageDTO.class))).thenReturn(result);
+
+ mockMvc.perform(post("/api/topics/send")
+ .contentType(MediaType.APPLICATION_JSON)
+ .content(objectMapper.writeValueAsString(request)))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.code").value(200))
+ .andExpect(jsonPath("$.data.msgId").value("msg-001"))
+ .andExpect(jsonPath("$.data.offsetMsgId").value("offset-001"));
+ }
+}