This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new 4bfd002c feat: add cloud credential store, vendor-aware instances and
instance provider routing (#1147)
4bfd002c is described below
commit 4bfd002c7c94cd08cf6dfee4193a54bd778aa55c
Author: lizhimins <[email protected]>
AuthorDate: Thu Aug 6 20:46:53 2026 +0800
feat: add cloud credential store, vendor-aware instances and instance
provider routing (#1147)
---
deploy/mysql/upgrade-cloud-vendor.sql | 89 +++++++++++
.../rocketmq/studio/auth/AuthInterceptor.java | 7 +-
.../credential/CloudCredentialController.java | 74 ++++++++++
.../CloudCredentialDeleteRequestDTO.java} | 25 ++--
.../credential/CloudCredentialRepository.java} | 20 +--
.../cloud/credential/CloudCredentialService.java | 149 +++++++++++++++++++
.../studio/cloud/credential/CloudCredentialVO.java | 102 +++++++++++++
.../cloud/credential/CreateCloudCredentialDTO.java | 98 +++++++++++++
.../MybatisPlusCloudCredentialRepository.java | 139 ++++++++++++++++++
.../credential/UpdateCloudCredentialDTO.java} | 49 +++++--
.../domain/enums/InstanceVendor.java} | 17 +--
.../common/exception/GlobalExceptionHandler.java | 7 +
.../studio/instance/InstanceRepository.java | 2 +
.../rocketmq/studio/instance/InstanceService.java | 131 +++++++++++++++--
.../rocketmq/studio/instance/InstanceVO.java | 5 +
.../instance/MybatisPlusInstanceRepository.java | 26 ++++
.../rocketmq/studio/instance/dlq/DLQService.java | 13 ++
.../instance/group/ConsumerGroupController.java | 26 ++--
.../instance/group/CreateConsumerGroupDTO.java | 3 +
.../instance/group/DeleteConsumerGroupDTO.java | 2 +
.../instance/group/ResetConsumerOffsetDTO.java | 2 +
.../studio/instance/message/MessageService.java | 10 +-
.../studio/instance/topic/CreateTopicDTO.java | 3 +
.../studio/instance/topic/DeleteTopicDTO.java | 2 +
.../studio/instance/topic/MetadataService.java | 124 ++++++++++++++--
.../studio/instance/topic/SendMessageDTO.java | 2 +
.../studio/instance/topic/TopicController.java | 17 ++-
.../persistence/entity/RmqCloudCredential.java | 108 ++++++++++++++
.../studio/persistence/entity/RmqInstance.java | 8 +
.../mapper/RmqCloudCredentialMapper.java} | 17 +--
.../CloudCatalogProvider.java} | 25 ++--
.../studio/provider/CloudInstanceDetailVO.java | 106 ++++++++++++++
.../studio/provider/CloudInstanceOptionVO.java | 84 +++++++++++
.../CloudRegionVO.java} | 41 +++---
.../rocketmq/studio/provider/InstanceProvider.java | 65 ++++++++
.../studio/provider/InstanceProviderRegistry.java | 74 ++++++++++
.../provider/apache/ApacheInstanceProvider.java | 136 +++++++++++++++++
.../provider/tencent/TencentInstanceProvider.java | 110 ++++++++++++++
server/src/main/resources/db/schema.sql | 17 +++
.../credential/CloudCredentialServiceTest.java | 163 +++++++++++++++++++++
.../studio/instance/InstanceServiceTest.java | 108 ++++++++++++++
.../studio/instance/dlq/DLQServiceTest.java | 4 +
.../group/ConsumerGroupControllerTest.java | 5 +-
.../instance/message/MessageServiceTest.java | 7 +-
.../studio/instance/topic/MetadataServiceTest.java | 4 +
.../studio/instance/topic/TopicControllerTest.java | 8 +-
.../provider/InstanceProviderRegistryTest.java | 109 ++++++++++++++
.../tencent/TencentInstanceProviderTest.java | 65 ++++++++
48 files changed, 2255 insertions(+), 153 deletions(-)
diff --git a/deploy/mysql/upgrade-cloud-vendor.sql
b/deploy/mysql/upgrade-cloud-vendor.sql
new file mode 100644
index 00000000..8286a1bb
--- /dev/null
+++ b/deploy/mysql/upgrade-cloud-vendor.sql
@@ -0,0 +1,89 @@
+-- deploy/mysql/upgrade-cloud-vendor.sql
+-- 存量 MySQL 数据卷增量迁移：rmq_instance 增加厂商维度 + 新增 rmq_cloud_credential 凭据表
+-- 适用：数据卷已初始化、docker-entrypoint-initdb.d 不会再执行的存量部署。
+-- 全新数据卷由 server/src/main/resources/db/schema.sql 直接覆盖，无需本脚本。
+-- 幂等：可重复执行。
+--
+-- 用法（远程容器内执行）：
+-- docker exec -i rocketmq-studio-mysql mysql -uroot -pstudio123
rocketmq_studio < upgrade-cloud-vendor.sql
+
+-- 固定连接编码，防止 mysql 客户端以 latin1 解释 UTF-8 字节导致中文双重编码
+SET NAMES utf8mb4;
+
+SET @schema_name := DATABASE();
+
+-- 1. rmq_instance.vendor
+SET @vendor_column_exists := (
+ SELECT COUNT(*)
+ FROM information_schema.columns
+ WHERE table_schema = @schema_name
+ AND table_name = 'rmq_instance'
+ AND column_name = 'vendor'
+);
+SET @vendor_sql := IF(@vendor_column_exists = 0,
+ "ALTER TABLE rmq_instance ADD COLUMN vendor VARCHAR(32) NOT NULL DEFAULT
'APACHE' COMMENT 'APACHE/ALIYUN/TENCENT' AFTER endpoint",
+ 'SELECT 1');
+PREPARE vendor_statement FROM @vendor_sql;
+EXECUTE vendor_statement;
+DEALLOCATE PREPARE vendor_statement;
+
+-- 2. rmq_instance.cloud_instance_id
+SET @cloud_instance_id_column_exists := (
+ SELECT COUNT(*)
+ FROM information_schema.columns
+ WHERE table_schema = @schema_name
+ AND table_name = 'rmq_instance'
+ AND column_name = 'cloud_instance_id'
+);
+SET @cloud_instance_id_sql := IF(@cloud_instance_id_column_exists = 0,
+ "ALTER TABLE rmq_instance ADD COLUMN cloud_instance_id VARCHAR(128)
COMMENT '云厂商实例 ID（vendor 非 APACHE 时必填）' AFTER vendor",
+ 'SELECT 1');
+PREPARE cloud_instance_id_statement FROM @cloud_instance_id_sql;
+EXECUTE cloud_instance_id_statement;
+DEALLOCATE PREPARE cloud_instance_id_statement;
+
+-- 3. rmq_instance.credential_id
+SET @credential_id_column_exists := (
+ SELECT COUNT(*)
+ FROM information_schema.columns
+ WHERE table_schema = @schema_name
+ AND table_name = 'rmq_instance'
+ AND column_name = 'credential_id'
+);
+SET @credential_id_sql := IF(@credential_id_column_exists = 0,
+ "ALTER TABLE rmq_instance ADD COLUMN credential_id VARCHAR(64) COMMENT '引用
rmq_cloud_credential.id' AFTER cloud_instance_id",
+ 'SELECT 1');
+PREPARE credential_id_statement FROM @credential_id_sql;
+EXECUTE credential_id_statement;
+DEALLOCATE PREPARE credential_id_statement;
+
+-- 4. rmq_instance.region_id
+SET @region_id_column_exists := (
+ SELECT COUNT(*)
+ FROM information_schema.columns
+ WHERE table_schema = @schema_name
+ AND table_name = 'rmq_instance'
+ AND column_name = 'region_id'
+);
+SET @region_id_sql := IF(@region_id_column_exists = 0,
+ "ALTER TABLE rmq_instance ADD COLUMN region_id VARCHAR(64) COMMENT '云
region' AFTER credential_id",
+ 'SELECT 1');
+PREPARE region_id_statement FROM @region_id_sql;
+EXECUTE region_id_statement;
+DEALLOCATE PREPARE region_id_statement;
+
+-- 5. 存量实例兜底回填
+UPDATE rmq_instance SET vendor = 'APACHE' WHERE vendor IS NULL OR vendor = '';
+
+-- 6. 云厂商凭据表（secret_key 为 base64 编码，禁止明文；access_key 明文用于唯一键与打码展示）
+CREATE TABLE IF NOT EXISTS rmq_cloud_credential (
+ id VARCHAR(64) PRIMARY KEY,
+ name VARCHAR(128) NOT NULL COMMENT '凭据显示名',
+ vendor VARCHAR(32) NOT NULL COMMENT 'ALIYUN/TENCENT',
+ access_key VARCHAR(255) NOT NULL,
+ secret_key VARCHAR(512) NOT NULL COMMENT 'base64 编码的 SK',
+ remark VARCHAR(255),
+ created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+ updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+ UNIQUE KEY uk_vendor_access_key (vendor, access_key)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/auth/AuthInterceptor.java
b/server/src/main/java/org/apache/rocketmq/studio/auth/AuthInterceptor.java
index f642319b..293b933c 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/auth/AuthInterceptor.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/auth/AuthInterceptor.java
@@ -98,7 +98,12 @@ public class AuthInterceptor implements HandlerInterceptor {
}
private boolean isAdminOnlyGetPath(String path) {
- return path.startsWith("/api/acl/users/") &&
path.endsWith("/credentials");
+ return isCredentialRevealPath(path, "/api/acl/users/")
+ || isCredentialRevealPath(path, "/api/cloud-credentials/");
+ }
+
+ private boolean isCredentialRevealPath(String path, String prefix) {
+ return path.startsWith(prefix) && path.endsWith("/credentials");
}
private void writeError(HttpServletResponse response, HttpStatus status,
String message)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cloud/credential/CloudCredentialController.java
b/server/src/main/java/org/apache/rocketmq/studio/cloud/credential/CloudCredentialController.java
new file mode 100644
index 00000000..540cf116
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/cloud/credential/CloudCredentialController.java
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.cloud.credential;
+
+import jakarta.validation.Valid;
+import org.apache.rocketmq.studio.common.domain.Result;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+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.RestController;
+
+import java.util.List;
+
+@RestController
+@RequestMapping("/api/cloud-credentials")
+public class CloudCredentialController {
+
+ private final CloudCredentialService credentialService;
+
+ public CloudCredentialController(CloudCredentialService credentialService)
{
+ this.credentialService = credentialService;
+ }
+
+ @GetMapping
+ public Result<List<CloudCredentialVO>> listCredentials() {
+ return Result.ok(credentialService.listMasked());
+ }
+
+ @PostMapping("/create")
+ public Result<CloudCredentialVO> createCredential(
+ @Valid @RequestBody(required = false) CreateCloudCredentialDTO
request) {
+ if (request == null) {
+ throw new BusinessException(400, "Cloud credential request is
required");
+ }
+ return
Result.ok(credentialService.create(request.toCloudCredentialVO()));
+ }
+
+ @PostMapping("/update")
+ public Result<CloudCredentialVO> updateCredential(
+ @Valid @RequestBody(required = false) UpdateCloudCredentialDTO
request) {
+ if (request == null) {
+ throw new BusinessException(400, "Cloud credential request is
required");
+ }
+ return Result.ok(credentialService.update(request));
+ }
+
+ @PostMapping("/delete")
+ public Result<Void> deleteCredential(@Valid @RequestBody
CloudCredentialDeleteRequestDTO request) {
+ credentialService.delete(request.getId());
+ return Result.ok();
+ }
+
+ @GetMapping("/{id}/credentials")
+ public Result<CloudCredentialVO> getCredentialSecrets(@PathVariable String
id) {
+ return Result.ok(credentialService.reveal(id));
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/DeleteTopicDTO.java
b/server/src/main/java/org/apache/rocketmq/studio/cloud/credential/CloudCredentialDeleteRequestDTO.java
similarity index 73%
copy from
server/src/main/java/org/apache/rocketmq/studio/instance/topic/DeleteTopicDTO.java
copy to
server/src/main/java/org/apache/rocketmq/studio/cloud/credential/CloudCredentialDeleteRequestDTO.java
index 50f256c7..d4513327 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/DeleteTopicDTO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cloud/credential/CloudCredentialDeleteRequestDTO.java
@@ -14,19 +14,20 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.rocketmq.studio.instance.topic;
+package org.apache.rocketmq.studio.cloud.credential;
import jakarta.validation.constraints.NotBlank;
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Data;
-import lombok.NoArgsConstructor;
-@Data
-@Builder
-@NoArgsConstructor
-@AllArgsConstructor
-public class DeleteTopicDTO {
- @NotBlank(message = "name is required")
- private String name;
+public class CloudCredentialDeleteRequestDTO {
+
+ @NotBlank(message = "credential id is required")
+ private String id;
+
+ public String getId() {
+ return id;
+ }
+
+ public void setId(String id) {
+ this.id = id;
+ }
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceRepository.java
b/server/src/main/java/org/apache/rocketmq/studio/cloud/credential/CloudCredentialRepository.java
similarity index 64%
copy from
server/src/main/java/org/apache/rocketmq/studio/instance/InstanceRepository.java
copy to
server/src/main/java/org/apache/rocketmq/studio/cloud/credential/CloudCredentialRepository.java
index 53410e6f..94e5d4fd 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceRepository.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cloud/credential/CloudCredentialRepository.java
@@ -14,26 +14,22 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package org.apache.rocketmq.studio.cloud.credential;
-package org.apache.rocketmq.studio.instance;
-
-import org.apache.rocketmq.studio.common.domain.enums.InstanceType;
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
import java.util.List;
import java.util.Optional;
-public interface InstanceRepository {
- List<InstanceVO> findAll();
-
- List<InstanceVO> findByType(InstanceType type);
+public interface CloudCredentialRepository {
- List<InstanceVO> search(String keyword);
+ List<CloudCredentialVO> findAll();
- List<InstanceVO> findByTypeAndSearch(InstanceType type, String keyword);
+ Optional<CloudCredentialVO> findById(String id);
- Optional<InstanceVO> findById(String id);
+ Optional<CloudCredentialVO> findByVendorAndAccessKey(InstanceVendor
vendor, String accessKey);
- InstanceVO save(InstanceVO instance);
+ CloudCredentialVO save(CloudCredentialVO credential);
- void deleteById(String id);
+ boolean deleteById(String id);
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cloud/credential/CloudCredentialService.java
b/server/src/main/java/org/apache/rocketmq/studio/cloud/credential/CloudCredentialService.java
new file mode 100644
index 00000000..c1418344
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/cloud/credential/CloudCredentialService.java
@@ -0,0 +1,149 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.cloud.credential;
+
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.instance.InstanceRepository;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.UUID;
+
+@Service
+public class CloudCredentialService {
+
+ private static final Logger log =
LoggerFactory.getLogger(CloudCredentialService.class);
+
+ private static final int VISIBLE_CREDENTIAL_CHARS = 4;
+ private static final int MIN_PARTIALLY_MASKED_CREDENTIAL_CHARS = 17;
+ private static final String CREDENTIAL_MASK = "****";
+
+ private final CloudCredentialRepository credentialRepository;
+ private final InstanceRepository instanceRepository;
+
+ public CloudCredentialService(CloudCredentialRepository
credentialRepository,
+ InstanceRepository instanceRepository) {
+ this.credentialRepository = credentialRepository;
+ this.instanceRepository = instanceRepository;
+ }
+
+ public List<CloudCredentialVO> listMasked() {
+ log.info("Listing cloud credentials (masked)");
+ return credentialRepository.findAll().stream()
+ .map(this::maskAccessKey)
+ .toList();
+ }
+
+ public CloudCredentialVO create(CloudCredentialVO credential) {
+ if (credential == null) {
+ throw new BusinessException(400, "Cloud credential request is
required");
+ }
+ if (isBlank(credential.getName())) {
+ throw new BusinessException(400, "Cloud credential name is
required");
+ }
+ if (credential.getVendor() == null || credential.getVendor() ==
org.apache.rocketmq.studio.common.domain.enums.InstanceVendor.APACHE) {
+ throw new BusinessException(400, "Cloud credential vendor must be
ALIYUN or TENCENT");
+ }
+ if (isBlank(credential.getAccessKey()) ||
isBlank(credential.getSecretKey())) {
+ throw new BusinessException(400, "Cloud credential accessKey and
secretKey are required");
+ }
+ credentialRepository.findByVendorAndAccessKey(credential.getVendor(),
credential.getAccessKey())
+ .ifPresent(existing -> {
+ throw new BusinessException(400,
+ "Cloud credential already exists for vendor " +
credential.getVendor()
+ + " and accessKey " +
maskCredential(credential.getAccessKey()));
+ });
+ log.info("Creating cloud credential name={}, vendor={}",
credential.getName(), credential.getVendor());
+ credential.setId(UUID.randomUUID().toString());
+ credential.setCreatedAt(LocalDateTime.now());
+ credential.setUpdatedAt(LocalDateTime.now());
+ return maskAccessKey(credentialRepository.save(credential));
+ }
+
+ public CloudCredentialVO update(UpdateCloudCredentialDTO request) {
+ if (request == null || isBlank(request.getId())) {
+ throw new BusinessException(400, "Cloud credential id is
required");
+ }
+ log.info("Updating cloud credential id={}", request.getId());
+ CloudCredentialVO existing =
credentialRepository.findById(request.getId())
+ .orElseThrow(() -> new BusinessException(404, "Cloud
credential not found: " + request.getId()));
+ if (request.getName() != null && request.getName().isBlank()) {
+ throw new BusinessException(400, "Cloud credential name cannot be
blank");
+ }
+ if (request.getName() != null) {
+ existing.setName(request.getName());
+ }
+ if (request.getSecretKey() != null &&
!request.getSecretKey().isBlank()) {
+ existing.setSecretKey(request.getSecretKey());
+ }
+ if (request.getRemark() != null) {
+ existing.setRemark(request.getRemark());
+ }
+ existing.setUpdatedAt(LocalDateTime.now());
+ return maskAccessKey(credentialRepository.save(existing));
+ }
+
+ public void delete(String id) {
+ if (isBlank(id)) {
+ throw new BusinessException(400, "Cloud credential id is
required");
+ }
+ log.info("Deleting cloud credential id={}", id);
+ credentialRepository.findById(id)
+ .orElseThrow(() -> new BusinessException(404, "Cloud
credential not found: " + id));
+ if (instanceRepository.existsByCredentialId(id)) {
+ throw new BusinessException(400, "Cloud credential is referenced
by existing instances");
+ }
+ credentialRepository.deleteById(id);
+ }
+
+ public CloudCredentialVO reveal(String id) {
+ if (isBlank(id)) {
+ throw new BusinessException(400, "Cloud credential id is
required");
+ }
+ return credentialRepository.findById(id)
+ .orElseThrow(() -> new BusinessException(404, "Cloud
credential not found: " + id));
+ }
+
+ private CloudCredentialVO maskAccessKey(CloudCredentialVO credential) {
+ CloudCredentialVO masked = new CloudCredentialVO();
+ masked.setId(credential.getId());
+ masked.setName(credential.getName());
+ masked.setVendor(credential.getVendor());
+ masked.setAccessKey(maskCredential(credential.getAccessKey()));
+ masked.setSecretKey(null);
+ masked.setRemark(credential.getRemark());
+ masked.setCreatedAt(credential.getCreatedAt());
+ masked.setUpdatedAt(credential.getUpdatedAt());
+ return masked;
+ }
+
+ static String maskCredential(String value) {
+ if (value == null || value.length() <
MIN_PARTIALLY_MASKED_CREDENTIAL_CHARS) {
+ return CREDENTIAL_MASK;
+ }
+ return value.substring(0, VISIBLE_CREDENTIAL_CHARS)
+ + CREDENTIAL_MASK
+ + value.substring(value.length() - VISIBLE_CREDENTIAL_CHARS);
+ }
+
+ private static boolean isBlank(String value) {
+ return value == null || value.isBlank();
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cloud/credential/CloudCredentialVO.java
b/server/src/main/java/org/apache/rocketmq/studio/cloud/credential/CloudCredentialVO.java
new file mode 100644
index 00000000..275f29b2
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/cloud/credential/CloudCredentialVO.java
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.cloud.credential;
+
+import org.apache.rocketmq.studio.common.domain.BaseEntity;
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
+
+import java.util.Objects;
+
+public class CloudCredentialVO extends BaseEntity {
+
+ private String name;
+ private InstanceVendor vendor;
+ private String accessKey;
+ private String secretKey;
+ private String remark;
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public InstanceVendor getVendor() {
+ return vendor;
+ }
+
+ public void setVendor(InstanceVendor vendor) {
+ this.vendor = vendor;
+ }
+
+ public String getAccessKey() {
+ return accessKey;
+ }
+
+ public void setAccessKey(String accessKey) {
+ this.accessKey = accessKey;
+ }
+
+ public String getSecretKey() {
+ return secretKey;
+ }
+
+ public void setSecretKey(String secretKey) {
+ this.secretKey = secretKey;
+ }
+
+ public String getRemark() {
+ return remark;
+ }
+
+ public void setRemark(String remark) {
+ this.remark = remark;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof CloudCredentialVO)) {
+ return false;
+ }
+ CloudCredentialVO that = (CloudCredentialVO) o;
+ return Objects.equals(getId(), that.getId())
+ && Objects.equals(name, that.name)
+ && vendor == that.vendor
+ && Objects.equals(accessKey, that.accessKey)
+ && Objects.equals(secretKey, that.secretKey)
+ && Objects.equals(remark, that.remark)
+ && Objects.equals(getCreatedAt(), that.getCreatedAt())
+ && Objects.equals(getUpdatedAt(), that.getUpdatedAt());
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(getId(), name, vendor, accessKey, secretKey,
remark,
+ getCreatedAt(), getUpdatedAt());
+ }
+
+ @Override
+ public String toString() {
+ return "CloudCredentialVO{id=" + getId() + ", name=" + name + ",
vendor=" + vendor
+ + ", accessKey=" + accessKey + ", secretKey=****}";
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cloud/credential/CreateCloudCredentialDTO.java
b/server/src/main/java/org/apache/rocketmq/studio/cloud/credential/CreateCloudCredentialDTO.java
new file mode 100644
index 00000000..79856bb8
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/cloud/credential/CreateCloudCredentialDTO.java
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.cloud.credential;
+
+import jakarta.validation.constraints.NotBlank;
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
+
+public class CreateCloudCredentialDTO {
+
+ @NotBlank(message = "credential name is required")
+ private String name;
+
+ @NotBlank(message = "credential vendor is required")
+ private String vendor;
+
+ @NotBlank(message = "credential accessKey is required")
+ private String accessKey;
+
+ @NotBlank(message = "credential secretKey is required")
+ private String secretKey;
+
+ private String remark;
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public String getVendor() {
+ return vendor;
+ }
+
+ public void setVendor(String vendor) {
+ this.vendor = vendor;
+ }
+
+ public String getAccessKey() {
+ return accessKey;
+ }
+
+ public void setAccessKey(String accessKey) {
+ this.accessKey = accessKey;
+ }
+
+ public String getSecretKey() {
+ return secretKey;
+ }
+
+ public void setSecretKey(String secretKey) {
+ this.secretKey = secretKey;
+ }
+
+ public String getRemark() {
+ return remark;
+ }
+
+ public void setRemark(String remark) {
+ this.remark = remark;
+ }
+
+ public CloudCredentialVO toCloudCredentialVO() {
+ CloudCredentialVO vo = new CloudCredentialVO();
+ vo.setName(name);
+ vo.setVendor(parseVendor(vendor));
+ vo.setAccessKey(accessKey);
+ vo.setSecretKey(secretKey);
+ vo.setRemark(remark);
+ return vo;
+ }
+
+ static InstanceVendor parseVendor(String vendor) {
+ if (vendor == null) {
+ return null;
+ }
+ try {
+ return InstanceVendor.valueOf(vendor.trim().toUpperCase());
+ } catch (IllegalArgumentException ex) {
+ return null;
+ }
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cloud/credential/MybatisPlusCloudCredentialRepository.java
b/server/src/main/java/org/apache/rocketmq/studio/cloud/credential/MybatisPlusCloudCredentialRepository.java
new file mode 100644
index 00000000..c83b7352
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/cloud/credential/MybatisPlusCloudCredentialRepository.java
@@ -0,0 +1,139 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.cloud.credential;
+
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
+import org.apache.rocketmq.studio.persistence.entity.RmqCloudCredential;
+import org.apache.rocketmq.studio.persistence.mapper.RmqCloudCredentialMapper;
+import org.springframework.stereotype.Repository;
+
+import java.nio.charset.StandardCharsets;
+import java.time.LocalDateTime;
+import java.util.Base64;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/**
+ * MySQL-backed cloud credential repository. Secret keys are stored
base64-encoded in
+ * {@code rmq_cloud_credential.secret_key} and decoded when read; plain text
is never persisted.
+ */
+@Repository
+public class MybatisPlusCloudCredentialRepository implements
CloudCredentialRepository {
+
+ private final RmqCloudCredentialMapper credentialMapper;
+
+ public MybatisPlusCloudCredentialRepository(RmqCloudCredentialMapper
credentialMapper) {
+ this.credentialMapper = credentialMapper;
+ }
+
+ @Override
+ public List<CloudCredentialVO> findAll() {
+ return credentialMapper.selectList(
+ new
QueryWrapper<RmqCloudCredential>().orderByAsc("id")).stream()
+ .map(MybatisPlusCloudCredentialRepository::toVO)
+ .collect(Collectors.toList());
+ }
+
+ @Override
+ public Optional<CloudCredentialVO> findById(String id) {
+ return Optional.ofNullable(credentialMapper.selectById(id))
+ .map(MybatisPlusCloudCredentialRepository::toVO);
+ }
+
+ @Override
+ public Optional<CloudCredentialVO> findByVendorAndAccessKey(InstanceVendor
vendor, String accessKey) {
+ RmqCloudCredential entity = credentialMapper.selectOne(
+ new QueryWrapper<RmqCloudCredential>()
+ .eq("vendor", vendor.name())
+ .eq("access_key", accessKey)
+ .last("LIMIT 1"));
+ return
Optional.ofNullable(entity).map(MybatisPlusCloudCredentialRepository::toVO);
+ }
+
+ @Override
+ public CloudCredentialVO save(CloudCredentialVO credential) {
+ RmqCloudCredential entity = toEntity(credential);
+ if (credentialMapper.selectById(entity.getId()) != null) {
+ credentialMapper.updateById(entity);
+ } else {
+ credentialMapper.insert(entity);
+ }
+ return credential;
+ }
+
+ @Override
+ public boolean deleteById(String id) {
+ return credentialMapper.deleteById(id) > 0;
+ }
+
+ // ── Mapping ────────────────────────────────────────────────────
+
+ private static CloudCredentialVO toVO(RmqCloudCredential entity) {
+ CloudCredentialVO vo = new CloudCredentialVO();
+ vo.setId(entity.getId());
+ vo.setName(entity.getName());
+ vo.setVendor(parseVendor(entity.getVendor()));
+ vo.setAccessKey(entity.getAccessKey());
+ vo.setSecretKey(decodeBase64(entity.getSecretKey()));
+ vo.setRemark(entity.getRemark());
+ vo.setCreatedAt(entity.getCreatedAt());
+ vo.setUpdatedAt(entity.getUpdatedAt());
+ return vo;
+ }
+
+ private static RmqCloudCredential toEntity(CloudCredentialVO vo) {
+ RmqCloudCredential entity = new RmqCloudCredential();
+ entity.setId(vo.getId());
+ entity.setName(vo.getName());
+ entity.setVendor(vo.getVendor() == null ? null :
vo.getVendor().name());
+ entity.setAccessKey(vo.getAccessKey());
+ entity.setSecretKey(encodeBase64(vo.getSecretKey()));
+ entity.setRemark(vo.getRemark());
+ entity.setCreatedAt(vo.getCreatedAt());
+ entity.setUpdatedAt(vo.getUpdatedAt() == null ? LocalDateTime.now() :
vo.getUpdatedAt());
+ return entity;
+ }
+
+ private static InstanceVendor parseVendor(String vendor) {
+ try {
+ return InstanceVendor.valueOf(vendor);
+ } catch (IllegalArgumentException | NullPointerException ex) {
+ return null;
+ }
+ }
+
+ static String encodeBase64(String plainText) {
+ if (plainText == null) {
+ return null;
+ }
+ return
Base64.getEncoder().encodeToString(plainText.getBytes(StandardCharsets.UTF_8));
+ }
+
+ static String decodeBase64(String stored) {
+ if (stored == null) {
+ return null;
+ }
+ try {
+ return new String(Base64.getDecoder().decode(stored),
StandardCharsets.UTF_8);
+ } catch (IllegalArgumentException ex) {
+ // tolerate legacy values that were stored without encoding
+ return stored;
+ }
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqInstance.java
b/server/src/main/java/org/apache/rocketmq/studio/cloud/credential/UpdateCloudCredentialDTO.java
similarity index 53%
copy from
server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqInstance.java
copy to
server/src/main/java/org/apache/rocketmq/studio/cloud/credential/UpdateCloudCredentialDTO.java
index 7c4cdbfd..067c58f6 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqInstance.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cloud/credential/UpdateCloudCredentialDTO.java
@@ -14,31 +14,50 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.rocketmq.studio.persistence.entity;
+package org.apache.rocketmq.studio.cloud.credential;
-import com.baomidou.mybatisplus.annotation.IdType;
-import com.baomidou.mybatisplus.annotation.TableId;
-import com.baomidou.mybatisplus.annotation.TableName;
-import lombok.Data;
+import jakarta.validation.constraints.NotBlank;
-import java.time.LocalDateTime;
+public class UpdateCloudCredentialDTO {
-@Data
-@TableName("rmq_instance")
-public class RmqInstance {
-
- @TableId(type = IdType.INPUT)
+ @NotBlank(message = "credential id is required")
private String id;
private String name;
+ private String secretKey;
+
private String remark;
- private String type;
+ public String getId() {
+ return id;
+ }
+
+ public void setId(String id) {
+ this.id = id;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public String getSecretKey() {
+ return secretKey;
+ }
- private String endpoint;
+ public void setSecretKey(String secretKey) {
+ this.secretKey = secretKey;
+ }
- private LocalDateTime createdAt;
+ public String getRemark() {
+ return remark;
+ }
- private LocalDateTime updatedAt;
+ public void setRemark(String remark) {
+ this.remark = remark;
+ }
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/DeleteTopicDTO.java
b/server/src/main/java/org/apache/rocketmq/studio/common/domain/enums/InstanceVendor.java
similarity index 68%
copy from
server/src/main/java/org/apache/rocketmq/studio/instance/topic/DeleteTopicDTO.java
copy to
server/src/main/java/org/apache/rocketmq/studio/common/domain/enums/InstanceVendor.java
index 50f256c7..01ede90b 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/DeleteTopicDTO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/common/domain/enums/InstanceVendor.java
@@ -14,19 +14,8 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.rocketmq.studio.instance.topic;
+package org.apache.rocketmq.studio.common.domain.enums;
-import jakarta.validation.constraints.NotBlank;
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Data;
-import lombok.NoArgsConstructor;
-
-@Data
-@Builder
-@NoArgsConstructor
-@AllArgsConstructor
-public class DeleteTopicDTO {
- @NotBlank(message = "name is required")
- private String name;
+public enum InstanceVendor {
+ APACHE, ALIYUN, TENCENT
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/common/exception/GlobalExceptionHandler.java
b/server/src/main/java/org/apache/rocketmq/studio/common/exception/GlobalExceptionHandler.java
index ca390d77..98297fc5 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/common/exception/GlobalExceptionHandler.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/common/exception/GlobalExceptionHandler.java
@@ -43,6 +43,13 @@ public class GlobalExceptionHandler {
.body(Result.error(ex.getCode(), ex.getMessage()));
}
+ @ExceptionHandler(UnsupportedOperationException.class)
+ @ResponseStatus(HttpStatus.NOT_IMPLEMENTED)
+ public Result<?>
handleUnsupportedOperationException(UnsupportedOperationException ex) {
+ log.warn("Unsupported operation: {}", ex.getMessage());
+ return Result.error(HttpStatus.NOT_IMPLEMENTED.value(),
ex.getMessage());
+ }
+
@ExceptionHandler(PrometheusException.class)
public ResponseEntity<Result<?>>
handlePrometheusException(PrometheusException ex) {
log.warn("Prometheus exception: status={}, message={}",
ex.getStatusCode(), ex.getMessage());
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceRepository.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceRepository.java
index 53410e6f..f6c41f4d 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceRepository.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceRepository.java
@@ -36,4 +36,6 @@ public interface InstanceRepository {
InstanceVO save(InstanceVO instance);
void deleteById(String id);
+
+ boolean existsByCredentialId(String credentialId);
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceService.java
index b95b48f2..18b33e9e 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceService.java
@@ -17,8 +17,13 @@
package org.apache.rocketmq.studio.instance;
+import org.apache.rocketmq.studio.cloud.credential.CloudCredentialRepository;
+import org.apache.rocketmq.studio.cloud.credential.CloudCredentialVO;
import org.apache.rocketmq.studio.common.domain.enums.InstanceType;
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.provider.CloudInstanceDetailVO;
+import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
@@ -33,36 +38,127 @@ import java.util.UUID;
public class InstanceService {
private final InstanceRepository instanceRepository;
+ private final CloudCredentialRepository cloudCredentialRepository;
+ private final InstanceProviderRegistry providerRegistry;
public List<InstanceVO> listInstances(InstanceType type, String search) {
log.debug("Listing instances, type={}, search={}", type, search);
String normalizedSearch = search == null || search.isBlank() ? null :
search.trim();
+ List<InstanceVO> instances;
if (type != null && normalizedSearch != null) {
- return instanceRepository.findByTypeAndSearch(type,
normalizedSearch);
+ instances = instanceRepository.findByTypeAndSearch(type,
normalizedSearch);
} else if (type != null) {
- return instanceRepository.findByType(type);
+ instances = instanceRepository.findByType(type);
} else if (normalizedSearch != null) {
- return instanceRepository.search(normalizedSearch);
+ instances = instanceRepository.search(normalizedSearch);
+ } else {
+ instances = instanceRepository.findAll();
+ }
+ instances.forEach(this::fillCloudCounts);
+ return instances;
+ }
+
+ /**
+ * Cloud instances keep topics/groups on the vendor side, so the local
table counts are
+ * always zero; pull live counts through the vendor provider instead.
+ */
+ private void fillCloudCounts(InstanceVO instance) {
+ if (instance.getVendor() == null || instance.getVendor() ==
InstanceVendor.APACHE) {
+ return;
+ }
+ try {
+ org.apache.rocketmq.studio.provider.InstanceProvider provider =
+ providerRegistry.forVendor(instance.getVendor());
+ instance.setTopicCount(provider.listTopics(instance.getId(), null,
null).size());
+
instance.setConsumerGroupCount(provider.listConsumerGroups(instance.getId(),
null).size());
+ } catch (RuntimeException ex) {
+ log.warn("Failed to load cloud resource counts for instance {}:
{}",
+ instance.getId(), ex.getMessage());
}
- return instanceRepository.findAll();
}
public InstanceVO createInstance(InstanceVO instance) {
requireInstance(instance);
- log.info("Creating instance: {}", instance.getName());
+ InstanceVendor vendor = instance.getVendor() == null ?
InstanceVendor.APACHE : instance.getVendor();
+ log.info("Creating instance: name={}, vendor={}", instance.getName(),
vendor);
+
+ switch (vendor) {
+ case APACHE -> createApacheInstance(instance);
+ case ALIYUN -> createAliyunInstance(instance);
+ case TENCENT -> throw new BusinessException(501, "Tencent Cloud
instance is not supported yet");
+ }
+
+ instance.setId(UUID.randomUUID().toString());
+ instance.setCreatedAt(LocalDateTime.now());
+ instance.setUpdatedAt(LocalDateTime.now());
+ return instanceRepository.save(instance);
+ }
+ private void createApacheInstance(InstanceVO instance) {
+ instance.setVendor(InstanceVendor.APACHE);
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);
+ /**
+ * Commercial instances are never created manually: the user picks a
stored credential and
+ * one of the cloud instances returned by the vendor catalog; endpoint is
resolved from the
+ * cloud instance detail (VPC endpoint preferred).
+ */
+ private void createAliyunInstance(InstanceVO instance) {
+ instance.setVendor(InstanceVendor.ALIYUN);
+ if (instance.getEndpoint() != null &&
!instance.getEndpoint().isBlank()) {
+ throw new BusinessException(400, "Commercial instances must be
selected from the cloud catalog, endpoint cannot be set manually");
+ }
+ if (isBlank(instance.getCredentialId()) ||
isBlank(instance.getCloudInstanceId())
+ || isBlank(instance.getRegionId())) {
+ throw new BusinessException(400, "credentialId, cloudInstanceId
and regionId are required for Aliyun instances");
+ }
+ CloudCredentialVO credential =
cloudCredentialRepository.findById(instance.getCredentialId())
+ .orElseThrow(() -> new BusinessException(404, "Cloud
credential not found: " + instance.getCredentialId()));
+ if (credential.getVendor() != InstanceVendor.ALIYUN) {
+ throw new BusinessException(400, "Cloud credential vendor does not
match ALIYUN");
+ }
+ CloudInstanceDetailVO detail =
providerRegistry.catalogFor(InstanceVendor.ALIYUN)
+ .getCloudInstance(instance.getCredentialId(),
instance.getRegionId(), instance.getCloudInstanceId());
+ if (instance.getName() == null || instance.getName().isBlank()) {
+ instance.setName(detail.getInstanceName() != null &&
!detail.getInstanceName().isBlank()
+ ? detail.getInstanceName() : detail.getInstanceId());
+ }
+ instance.setType(InstanceType.PROXY);
+ instance.setEndpoint(resolveEndpoint(detail));
+ }
+
+ private String resolveEndpoint(CloudInstanceDetailVO detail) {
+ if (detail.getEndpoints() == null || detail.getEndpoints().isEmpty()) {
+ throw new BusinessException(502, "Cloud instance has no endpoint:
" + detail.getInstanceId());
+ }
+ return detail.getEndpoints().stream()
+ .filter(endpoint -> endpoint.getEndpointUrl() != null &&
!endpoint.getEndpointUrl().isBlank())
+ .sorted((a, b) ->
Integer.compare(endpointPriority(a.getEndpointType()),
endpointPriority(b.getEndpointType())))
+ .map(CloudInstanceDetailVO.CloudEndpoint::getEndpointUrl)
+ .findFirst()
+ .orElseThrow(() -> new BusinessException(502, "Cloud instance
has no usable endpoint: " + detail.getInstanceId()));
+ }
+
+ private int endpointPriority(String endpointType) {
+ if (endpointType == null) {
+ return 2;
+ }
+ return switch (endpointType.toUpperCase()) {
+ case "TCP_VPC" -> 0;
+ case "TCP_INTERNET" -> 1;
+ default -> 2;
+ };
+ }
+
+ private static boolean isBlank(String value) {
+ return value == null || value.isBlank();
}
public InstanceVO updateInstance(InstanceVO instance) {
@@ -84,14 +180,17 @@ public class InstanceService {
}
InstanceVO updated = copyOf(existing);
+ boolean cloudInstance = existing.getVendor() != null &&
existing.getVendor() != InstanceVendor.APACHE;
if (instance.getName() != null) {
updated.setName(instance.getName());
}
- if (instance.getType() != null) {
- updated.setType(instance.getType());
- }
- if (instance.getEndpoint() != null) {
- updated.setEndpoint(instance.getEndpoint());
+ if (!cloudInstance) {
+ if (instance.getType() != null) {
+ updated.setType(instance.getType());
+ }
+ if (instance.getEndpoint() != null) {
+ updated.setEndpoint(instance.getEndpoint());
+ }
}
if (instance.getRemark() != null) {
updated.setRemark(instance.getRemark());
@@ -125,6 +224,10 @@ public class InstanceService {
.remark(instance.getRemark())
.type(instance.getType())
.endpoint(instance.getEndpoint())
+ .vendor(instance.getVendor() == null ? InstanceVendor.APACHE :
instance.getVendor())
+ .cloudInstanceId(instance.getCloudInstanceId())
+ .credentialId(instance.getCredentialId())
+ .regionId(instance.getRegionId())
.topicCount(instance.getTopicCount())
.consumerGroupCount(instance.getConsumerGroupCount())
.build();
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceVO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceVO.java
index 6332ba80..bc86d347 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceVO.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceVO.java
@@ -19,6 +19,7 @@ package org.apache.rocketmq.studio.instance;
import org.apache.rocketmq.studio.common.domain.BaseEntity;
import org.apache.rocketmq.studio.common.domain.enums.InstanceType;
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
@@ -35,6 +36,10 @@ public class InstanceVO extends BaseEntity {
private String remark;
private InstanceType type;
private String endpoint;
+ private InstanceVendor vendor;
+ private String cloudInstanceId;
+ private String credentialId;
+ private String regionId;
private int topicCount;
private int consumerGroupCount;
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/MybatisPlusInstanceRepository.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/MybatisPlusInstanceRepository.java
index e2f600e7..e903af4f 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/MybatisPlusInstanceRepository.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/MybatisPlusInstanceRepository.java
@@ -19,6 +19,7 @@ package org.apache.rocketmq.studio.instance;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import org.apache.rocketmq.studio.common.domain.enums.InstanceType;
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
import org.apache.rocketmq.studio.persistence.entity.RmqGroup;
import org.apache.rocketmq.studio.persistence.entity.RmqInstance;
import org.apache.rocketmq.studio.persistence.entity.RmqTopic;
@@ -114,6 +115,15 @@ public class MybatisPlusInstanceRepository implements
InstanceRepository {
instanceMapper.deleteById(id);
}
+ @Override
+ public boolean existsByCredentialId(String credentialId) {
+ if (credentialId == null || credentialId.isBlank()) {
+ return false;
+ }
+ return instanceMapper.selectCount(
+ new QueryWrapper<RmqInstance>().eq("credential_id",
credentialId)) > 0;
+ }
+
private List<InstanceVO> withCounts(List<RmqInstance> entities) {
if (entities.isEmpty()) {
return List.of();
@@ -149,6 +159,10 @@ public class MybatisPlusInstanceRepository implements
InstanceRepository {
.remark(entity.getRemark())
.type(parseType(entity.getType()))
.endpoint(entity.getEndpoint())
+ .vendor(parseVendor(entity.getVendor()))
+ .cloudInstanceId(entity.getCloudInstanceId())
+ .credentialId(entity.getCredentialId())
+ .regionId(entity.getRegionId())
.topicCount(topicCounts.getOrDefault(entity.getId(),
0L).intValue())
.consumerGroupCount(groupCounts.getOrDefault(entity.getId(),
0L).intValue())
.build();
@@ -166,6 +180,14 @@ public class MybatisPlusInstanceRepository implements
InstanceRepository {
}
}
+ private InstanceVendor parseVendor(String vendor) {
+ try {
+ return InstanceVendor.valueOf(vendor);
+ } catch (IllegalArgumentException | NullPointerException ex) {
+ return InstanceVendor.APACHE;
+ }
+ }
+
private RmqInstance toEntity(InstanceVO vo) {
RmqInstance entity = new RmqInstance();
entity.setId(vo.getId());
@@ -173,6 +195,10 @@ public class MybatisPlusInstanceRepository implements
InstanceRepository {
entity.setRemark(vo.getRemark());
entity.setType(vo.getType() == null ? null : vo.getType().name());
entity.setEndpoint(vo.getEndpoint());
+ entity.setVendor(vo.getVendor() == null ? InstanceVendor.APACHE.name()
: vo.getVendor().name());
+ entity.setCloudInstanceId(vo.getCloudInstanceId());
+ entity.setCredentialId(vo.getCredentialId());
+ entity.setRegionId(vo.getRegionId());
entity.setCreatedAt(vo.getCreatedAt());
entity.setUpdatedAt(vo.getUpdatedAt());
return entity;
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQService.java
index a5a6baaf..674ac107 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQService.java
@@ -16,7 +16,9 @@
*/
package org.apache.rocketmq.studio.instance.dlq;
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpStatus;
@@ -31,19 +33,30 @@ import java.util.List;
public class DLQService {
private final DLQProvider dlqProvider;
+ private final InstanceProviderRegistry providerRegistry;
public List<DLQGroupVO> listDLQGroups(String instanceId) {
+ requireApacheInstance(instanceId);
log.info("Listing DLQ groups for instance: {}", instanceId);
return dlqProvider.listDLQGroups(instanceId);
}
public DLQResendResultVO resendMessages(String instanceId, String
groupName, Long startTime, Long endTime,
String targetTopic) {
+ requireApacheInstance(instanceId);
validateResendRequest(groupName, startTime, endTime);
log.info("Resending DLQ messages: group={}, targetTopic={}",
groupName, targetTopic);
return dlqProvider.resendMessages(instanceId, groupName, startTime,
endTime, targetTopic);
}
+ private void requireApacheInstance(String instanceId) {
+ providerRegistry.byInstanceId(instanceId).ifPresent(provider -> {
+ if (provider.vendor() != InstanceVendor.APACHE) {
+ throw new BusinessException(501, "DLQ operations are not
supported for cloud instances");
+ }
+ });
+ }
+
private void validateResendRequest(String groupName, Long startTime, Long
endTime) {
if (!StringUtils.hasText(groupName)) {
throw new BusinessException(HttpStatus.BAD_REQUEST.value(),
"groupName is required");
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupController.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupController.java
index a9a2715b..4d7ddf70 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupController.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupController.java
@@ -41,24 +41,31 @@ public class ConsumerGroupController {
@GetMapping
public Result<List<ConsumerGroupVO>> listConsumerGroups(
+ @RequestParam(required = false) String instanceId,
@RequestParam(required = false) String clusterId,
@RequestParam(required = false) String search) {
- return Result.ok(metadataService.listConsumerGroups(clusterId,
search));
+ return Result.ok(metadataService.listConsumerGroups(instanceId,
clusterId, search));
}
@GetMapping("/{name}")
- public Result<ConsumerGroupVO> getConsumerGroup(@PathVariable String name)
{
- return Result.ok(metadataService.getConsumerGroup(name));
+ public Result<ConsumerGroupVO> getConsumerGroup(
+ @PathVariable String name,
+ @RequestParam(required = false) String instanceId) {
+ return Result.ok(metadataService.getConsumerGroup(instanceId, name));
}
@GetMapping("/{name}/progress")
- public Result<List<QueueProgressVO>> getGroupProgress(@PathVariable String
name) {
- return Result.ok(metadataService.getGroupProgress(name));
+ public Result<List<QueueProgressVO>> getGroupProgress(
+ @PathVariable String name,
+ @RequestParam(required = false) String instanceId) {
+ return Result.ok(metadataService.getGroupProgress(instanceId, name));
}
@GetMapping("/{name}/subscriptions")
- public Result<List<SubscriptionEntryVO>>
getGroupSubscriptions(@PathVariable String name) {
- return Result.ok(metadataService.getGroupSubscriptions(name));
+ public Result<List<SubscriptionEntryVO>> getGroupSubscriptions(
+ @PathVariable String name,
+ @RequestParam(required = false) String instanceId) {
+ return Result.ok(metadataService.getGroupSubscriptions(instanceId,
name));
}
@GetMapping("/{name}/instances/{clientId}/stack")
@@ -75,13 +82,14 @@ public class ConsumerGroupController {
@PostMapping("/delete")
public Result<Void> deleteConsumerGroup(@Valid @RequestBody
DeleteConsumerGroupDTO request) {
- metadataService.deleteConsumerGroup(request.getName());
+ metadataService.deleteConsumerGroup(request.getInstanceId(),
request.getName());
return Result.ok();
}
@PostMapping("/reset-offset")
public Result<Void> resetOffset(@Valid @RequestBody ResetConsumerOffsetDTO
request) {
- metadataService.resetOffset(request.getName(), request.getTimestamp(),
request.getTopic());
+ metadataService.resetOffset(request.getInstanceId(), request.getName(),
+ request.getTimestamp(), request.getTopic());
return Result.ok();
}
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/CreateConsumerGroupDTO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/CreateConsumerGroupDTO.java
index 8175a889..5ebde988 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/CreateConsumerGroupDTO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/CreateConsumerGroupDTO.java
@@ -37,8 +37,11 @@ public class CreateConsumerGroupDTO {
@PositiveOrZero(message = "delaySeconds must be zero or positive")
private Integer delaySeconds;
+ private String instanceId;
+
public ConsumerGroupVO toConsumerGroupVO() {
ConsumerGroupVO group = new ConsumerGroupVO();
+ group.setInstanceId(instanceId);
group.setName(name);
group.setNamespace(namespace);
group.setClusterId(clusterId);
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/DeleteConsumerGroupDTO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/DeleteConsumerGroupDTO.java
index 69cc2ec2..5a3e5caa 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/DeleteConsumerGroupDTO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/DeleteConsumerGroupDTO.java
@@ -29,4 +29,6 @@ import lombok.NoArgsConstructor;
public class DeleteConsumerGroupDTO {
@NotBlank(message = "name is required")
private String name;
+
+ private String instanceId;
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/ResetConsumerOffsetDTO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ResetConsumerOffsetDTO.java
index b7f5265c..64111393 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/group/ResetConsumerOffsetDTO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/group/ResetConsumerOffsetDTO.java
@@ -37,4 +37,6 @@ public class ResetConsumerOffsetDTO {
private Long timestamp;
private String topic;
+
+ private String instanceId;
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageService.java
index 58412c06..21d0a4e5 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageService.java
@@ -19,6 +19,7 @@ package org.apache.rocketmq.studio.instance.message;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
import org.springframework.stereotype.Service;
import java.util.List;
@@ -31,17 +32,22 @@ public class MessageService {
private static final long MAX_TOPIC_QUERY_WINDOW_MILLIS = 7L * 24 * 60 *
60 * 1000;
private final MessageProvider messageProvider;
+ private final InstanceProviderRegistry providerRegistry;
public List<MessageRecordVO> queryMessages(
String instanceId, String topic, String msgId, String tag, String
key, Long startTime, Long endTime) {
validateTopicQueryWindow(topic, msgId, key, startTime, endTime);
log.info("Querying messages: topic={}, msgId={}, tag={}, key={}",
topic, msgId, tag, key);
- return messageProvider.queryMessages(instanceId, topic, msgId, tag,
key, startTime, endTime);
+ return providerRegistry.byInstanceId(instanceId)
+ .map(provider -> provider.queryMessages(instanceId, topic,
msgId, tag, key, startTime, endTime))
+ .orElseGet(() -> messageProvider.queryMessages(instanceId,
topic, msgId, tag, key, startTime, endTime));
}
public TraceRecordVO getMessageTrace(String instanceId, String msgId) {
log.info("Getting message trace: msgId={}", msgId);
- return messageProvider.getMessageTrace(instanceId, msgId);
+ return providerRegistry.byInstanceId(instanceId)
+ .map(provider -> provider.getMessageTrace(instanceId, msgId))
+ .orElseGet(() -> messageProvider.getMessageTrace(instanceId,
msgId));
}
private void validateTopicQueryWindow(String topic, String msgId, String
key, Long startTime, Long endTime) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/CreateTopicDTO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/CreateTopicDTO.java
index 2f0146d5..efae80d6 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/CreateTopicDTO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/CreateTopicDTO.java
@@ -36,8 +36,11 @@ public class CreateTopicDTO {
private TopicPerm perm;
private String remark;
+ private String instanceId;
+
public TopicVO toTopicVO() {
TopicVO topic = new TopicVO();
+ topic.setInstanceId(instanceId);
topic.setName(name);
topic.setNamespace(namespace);
topic.setClusterId(clusterId);
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/DeleteTopicDTO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/DeleteTopicDTO.java
index 50f256c7..746e887c 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/DeleteTopicDTO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/DeleteTopicDTO.java
@@ -29,4 +29,6 @@ import lombok.NoArgsConstructor;
public class DeleteTopicDTO {
@NotBlank(message = "name is required")
private String name;
+
+ private String instanceId;
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
index 4a829d5e..94093daf 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
@@ -16,16 +16,20 @@
*/
package org.apache.rocketmq.studio.instance.topic;
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
import org.apache.rocketmq.studio.instance.group.QueueProgressVO;
import org.apache.rocketmq.studio.instance.group.SubscriptionEntryVO;
+import org.apache.rocketmq.studio.provider.InstanceProvider;
+import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import java.util.List;
+import java.util.Optional;
@Slf4j
@Service
@@ -34,47 +38,87 @@ public class MetadataService {
private final MetadataProvider metadataProvider;
private final AdminClient adminClient;
+ private final InstanceProviderRegistry providerRegistry;
// ── TopicVO ───────────────────────────────────────────────────────
public List<TopicVO> listTopics(String clusterId, String type, String
search) {
- return metadataProvider.listTopics(
- normalizeFilter(clusterId),
- normalizeFilter(type),
- normalizeFilter(search));
+ return listTopics(null, clusterId, type, search);
+ }
+
+ public List<TopicVO> listTopics(String instanceId, String clusterId,
String type, String search) {
+ return providerFor(instanceId)
+ .map(provider -> provider.listTopics(instanceId,
normalizeFilter(type), normalizeFilter(search)))
+ .orElseGet(() -> metadataProvider.listTopics(
+ normalizeFilter(clusterId),
+ normalizeFilter(type),
+ normalizeFilter(search)));
}
public TopicVO createTopic(TopicVO topic) {
requireTopic(topic);
- return adminClient.createTopic(topic);
+ return providerFor(topic.getInstanceId())
+ .map(provider -> provider.createTopic(topic.getInstanceId(),
topic))
+ .orElseGet(() -> adminClient.createTopic(topic));
}
public TopicVO updateTopic(TopicVO topic) {
requireTopic(topic);
- return adminClient.updateTopic(topic);
+ return providerFor(topic.getInstanceId())
+ .map(provider -> provider.updateTopic(topic.getInstanceId(),
topic))
+ .orElseGet(() -> adminClient.updateTopic(topic));
}
public void deleteTopic(String name) {
- adminClient.deleteTopic(name);
+ deleteTopic(null, name);
+ }
+
+ public void deleteTopic(String instanceId, String name) {
+ Optional<InstanceProvider> provider = providerFor(instanceId);
+ if (provider.isPresent()) {
+ provider.get().deleteTopic(instanceId, name);
+ } else {
+ adminClient.deleteTopic(name);
+ }
}
public List<BrokerRouteVO> getTopicRoutes(String name) {
+ return getTopicRoutes(null, name);
+ }
+
+ public List<BrokerRouteVO> getTopicRoutes(String instanceId, String name) {
+ Optional<InstanceProvider> provider = providerFor(instanceId);
+ if (provider.isPresent() && provider.get().vendor() !=
InstanceVendor.APACHE) {
+ // broker routing does not apply to serverless cloud instances
+ return List.of();
+ }
return metadataProvider.getTopicRoutes(name);
}
public List<TopicConsumerVO> getTopicConsumers(String name) {
- return metadataProvider.getTopicConsumers(name);
+ return getTopicConsumers(null, name);
+ }
+
+ public List<TopicConsumerVO> getTopicConsumers(String instanceId, String
name) {
+ return providerFor(instanceId)
+ .map(provider -> provider.getTopicConsumers(instanceId, name))
+ .orElseGet(() -> metadataProvider.getTopicConsumers(name));
}
public SendMessageVO sendMessage(SendMessageDTO request) {
requireSendMessageRequest(request);
+ providerFor(request.getInstanceId()).ifPresent(provider -> {
+ if (provider.vendor() != InstanceVendor.APACHE) {
+ throw new BusinessException(501, "Sending messages is not
supported for cloud instances");
+ }
+ });
return adminClient.sendMessage(request);
}
@@ -82,37 +126,85 @@ public class MetadataService {
public List<ConsumerGroupVO> listConsumerGroups(String clusterId, String
search) {
- return metadataProvider.listConsumerGroups(normalizeFilter(clusterId),
normalizeFilter(search));
+ return listConsumerGroups(null, clusterId, search);
+ }
+
+ public List<ConsumerGroupVO> listConsumerGroups(String instanceId, String
clusterId, String search) {
+ return providerFor(instanceId)
+ .map(provider -> provider.listConsumerGroups(instanceId,
normalizeFilter(search)))
+ .orElseGet(() -> metadataProvider.listConsumerGroups(
+ normalizeFilter(clusterId), normalizeFilter(search)));
}
public ConsumerGroupVO getConsumerGroup(String name) {
+ return getConsumerGroup(null, name);
+ }
+
+ public ConsumerGroupVO getConsumerGroup(String instanceId, String name) {
+ providerFor(instanceId).ifPresent(provider -> {
+ if (provider.vendor() != InstanceVendor.APACHE) {
+ throw new BusinessException(501, "Consumer group detail is not
supported for cloud instances");
+ }
+ });
return adminClient.getConsumerGroup(name);
}
public List<QueueProgressVO> getGroupProgress(String name) {
- return metadataProvider.getGroupProgress(name);
+ return getGroupProgress(null, name);
+ }
+
+ public List<QueueProgressVO> getGroupProgress(String instanceId, String
name) {
+ return providerFor(instanceId)
+ .map(provider -> provider.getGroupProgress(instanceId, name))
+ .orElseGet(() -> metadataProvider.getGroupProgress(name));
}
public List<SubscriptionEntryVO> getGroupSubscriptions(String name) {
- return metadataProvider.getGroupSubscriptions(name);
+ return getGroupSubscriptions(null, name);
+ }
+
+ public List<SubscriptionEntryVO> getGroupSubscriptions(String instanceId,
String name) {
+ return providerFor(instanceId)
+ .map(provider -> provider.getGroupSubscriptions(instanceId,
name))
+ .orElseGet(() -> metadataProvider.getGroupSubscriptions(name));
}
public ConsumerGroupVO createConsumerGroup(ConsumerGroupVO group) {
- return adminClient.createConsumerGroup(group);
+ return providerFor(group == null ? null : group.getInstanceId())
+ .map(provider ->
provider.createConsumerGroup(group.getInstanceId(), group))
+ .orElseGet(() -> adminClient.createConsumerGroup(group));
}
public void deleteConsumerGroup(String name) {
- adminClient.deleteConsumerGroup(name);
+ deleteConsumerGroup(null, name);
+ }
+
+ public void deleteConsumerGroup(String instanceId, String name) {
+ Optional<InstanceProvider> provider = providerFor(instanceId);
+ if (provider.isPresent()) {
+ provider.get().deleteConsumerGroup(instanceId, name);
+ } else {
+ adminClient.deleteConsumerGroup(name);
+ }
}
public void resetOffset(String name, long timestamp, String topic) {
- adminClient.resetOffset(name, timestamp, topic);
+ resetOffset(null, name, timestamp, topic);
+ }
+
+ public void resetOffset(String instanceId, String name, long timestamp,
String topic) {
+ Optional<InstanceProvider> provider = providerFor(instanceId);
+ if (provider.isPresent()) {
+ provider.get().resetOffset(instanceId, name, timestamp, topic);
+ } else {
+ adminClient.resetOffset(name, timestamp, topic);
+ }
}
// ── NamespaceVO ───────────────────────────────────────────────────
@@ -122,6 +214,10 @@ public class MetadataService {
throw new BusinessException(501, "Namespace discovery is not
implemented by the current metadata provider");
}
+ private Optional<InstanceProvider> providerFor(String instanceId) {
+ return providerRegistry.byInstanceId(instanceId);
+ }
+
private String normalizeFilter(String value) {
return value == null || value.isBlank() ? null : value.trim();
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/SendMessageDTO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/SendMessageDTO.java
index 375a72fe..21f343a6 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/SendMessageDTO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/SendMessageDTO.java
@@ -29,6 +29,8 @@ import java.util.Map;
@NoArgsConstructor
@AllArgsConstructor
public class SendMessageDTO {
+ private String instanceId;
+
@NotBlank(message = "topic is required")
private String topic;
private String tag;
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/TopicController.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/TopicController.java
index e1309334..6caeda88 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/TopicController.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/TopicController.java
@@ -39,10 +39,11 @@ public class TopicController {
@GetMapping
public Result<List<TopicVO>> listTopics(
+ @RequestParam(required = false) String instanceId,
@RequestParam(required = false) String clusterId,
@RequestParam(required = false) String type,
@RequestParam(required = false) String search) {
- return Result.ok(metadataService.listTopics(clusterId, type, search));
+ return Result.ok(metadataService.listTopics(instanceId, clusterId,
type, search));
}
@PostMapping("/create")
@@ -60,18 +61,22 @@ public class TopicController {
@PostMapping("/delete")
public Result<Void> deleteTopic(@Valid @RequestBody(required = false)
DeleteTopicDTO request) {
requireDeleteTopicRequest(request);
- metadataService.deleteTopic(request.getName());
+ metadataService.deleteTopic(request.getInstanceId(),
request.getName());
return Result.ok();
}
@GetMapping("/{name}/routes")
- public Result<List<BrokerRouteVO>> getTopicRoutes(@PathVariable String
name) {
- return Result.ok(metadataService.getTopicRoutes(name));
+ public Result<List<BrokerRouteVO>> getTopicRoutes(
+ @PathVariable String name,
+ @RequestParam(required = false) String instanceId) {
+ return Result.ok(metadataService.getTopicRoutes(instanceId, name));
}
@GetMapping("/{name}/consumers")
- public Result<List<TopicConsumerVO>> getTopicConsumers(@PathVariable
String name) {
- return Result.ok(metadataService.getTopicConsumers(name));
+ public Result<List<TopicConsumerVO>> getTopicConsumers(
+ @PathVariable String name,
+ @RequestParam(required = false) String instanceId) {
+ return Result.ok(metadataService.getTopicConsumers(instanceId, name));
}
@PostMapping("/send")
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqCloudCredential.java
b/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqCloudCredential.java
new file mode 100644
index 00000000..155e79f5
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqCloudCredential.java
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.persistence.entity;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+
+import java.time.LocalDateTime;
+
+@TableName("rmq_cloud_credential")
+public class RmqCloudCredential {
+
+ @TableId(type = IdType.INPUT)
+ private String id;
+
+ private String name;
+
+ private String vendor;
+
+ private String accessKey;
+
+ private String secretKey;
+
+ private String remark;
+
+ private LocalDateTime createdAt;
+
+ private LocalDateTime updatedAt;
+
+ public String getId() {
+ return id;
+ }
+
+ public void setId(String id) {
+ this.id = id;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public String getVendor() {
+ return vendor;
+ }
+
+ public void setVendor(String vendor) {
+ this.vendor = vendor;
+ }
+
+ public String getAccessKey() {
+ return accessKey;
+ }
+
+ public void setAccessKey(String accessKey) {
+ this.accessKey = accessKey;
+ }
+
+ public String getSecretKey() {
+ return secretKey;
+ }
+
+ public void setSecretKey(String secretKey) {
+ this.secretKey = secretKey;
+ }
+
+ public String getRemark() {
+ return remark;
+ }
+
+ public void setRemark(String remark) {
+ this.remark = remark;
+ }
+
+ public LocalDateTime getCreatedAt() {
+ return createdAt;
+ }
+
+ public void setCreatedAt(LocalDateTime createdAt) {
+ this.createdAt = createdAt;
+ }
+
+ public LocalDateTime getUpdatedAt() {
+ return updatedAt;
+ }
+
+ public void setUpdatedAt(LocalDateTime updatedAt) {
+ this.updatedAt = updatedAt;
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqInstance.java
b/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqInstance.java
index 7c4cdbfd..c172e113 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqInstance.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqInstance.java
@@ -38,6 +38,14 @@ public class RmqInstance {
private String endpoint;
+ private String vendor;
+
+ private String cloudInstanceId;
+
+ private String credentialId;
+
+ private String regionId;
+
private LocalDateTime createdAt;
private LocalDateTime updatedAt;
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/DeleteTopicDTO.java
b/server/src/main/java/org/apache/rocketmq/studio/persistence/mapper/RmqCloudCredentialMapper.java
similarity index 69%
copy from
server/src/main/java/org/apache/rocketmq/studio/instance/topic/DeleteTopicDTO.java
copy to
server/src/main/java/org/apache/rocketmq/studio/persistence/mapper/RmqCloudCredentialMapper.java
index 50f256c7..07571149 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/DeleteTopicDTO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/persistence/mapper/RmqCloudCredentialMapper.java
@@ -14,19 +14,10 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.rocketmq.studio.instance.topic;
+package org.apache.rocketmq.studio.persistence.mapper;
-import jakarta.validation.constraints.NotBlank;
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Data;
-import lombok.NoArgsConstructor;
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import org.apache.rocketmq.studio.persistence.entity.RmqCloudCredential;
-@Data
-@Builder
-@NoArgsConstructor
-@AllArgsConstructor
-public class DeleteTopicDTO {
- @NotBlank(message = "name is required")
- private String name;
+public interface RmqCloudCredentialMapper extends
BaseMapper<RmqCloudCredential> {
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceRepository.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/CloudCatalogProvider.java
similarity index 56%
copy from
server/src/main/java/org/apache/rocketmq/studio/instance/InstanceRepository.java
copy to
server/src/main/java/org/apache/rocketmq/studio/provider/CloudCatalogProvider.java
index 53410e6f..eb9584cb 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/InstanceRepository.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/CloudCatalogProvider.java
@@ -14,26 +14,23 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package org.apache.rocketmq.studio.provider;
-package org.apache.rocketmq.studio.instance;
-
-import org.apache.rocketmq.studio.common.domain.enums.InstanceType;
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
import java.util.List;
-import java.util.Optional;
-
-public interface InstanceRepository {
- List<InstanceVO> findAll();
- List<InstanceVO> findByType(InstanceType type);
-
- List<InstanceVO> search(String keyword);
+/**
+ * Cloud catalog discovery SPI: list regions / cloud instances with a stored
credential.
+ * Commercial instances are never created manually; users pick one from this
catalog.
+ */
+public interface CloudCatalogProvider {
- List<InstanceVO> findByTypeAndSearch(InstanceType type, String keyword);
+ InstanceVendor vendor();
- Optional<InstanceVO> findById(String id);
+ List<CloudRegionVO> listRegions(String credentialId);
- InstanceVO save(InstanceVO instance);
+ List<CloudInstanceOptionVO> listCloudInstances(String credentialId, String
regionId, String search);
- void deleteById(String id);
+ CloudInstanceDetailVO getCloudInstance(String credentialId, String
regionId, String cloudInstanceId);
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/CloudInstanceDetailVO.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/CloudInstanceDetailVO.java
new file mode 100644
index 00000000..ef736b36
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/CloudInstanceDetailVO.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.provider;
+
+import java.util.List;
+
+public class CloudInstanceDetailVO {
+
+ private String instanceId;
+ private String instanceName;
+ private String status;
+ private String regionId;
+ private String remark;
+ private List<CloudEndpoint> endpoints;
+
+ public String getInstanceId() {
+ return instanceId;
+ }
+
+ public void setInstanceId(String instanceId) {
+ this.instanceId = instanceId;
+ }
+
+ public String getInstanceName() {
+ return instanceName;
+ }
+
+ public void setInstanceName(String instanceName) {
+ this.instanceName = instanceName;
+ }
+
+ public String getStatus() {
+ return status;
+ }
+
+ public void setStatus(String status) {
+ this.status = status;
+ }
+
+ public String getRegionId() {
+ return regionId;
+ }
+
+ public void setRegionId(String regionId) {
+ this.regionId = regionId;
+ }
+
+ public String getRemark() {
+ return remark;
+ }
+
+ public void setRemark(String remark) {
+ this.remark = remark;
+ }
+
+ public List<CloudEndpoint> getEndpoints() {
+ return endpoints;
+ }
+
+ public void setEndpoints(List<CloudEndpoint> endpoints) {
+ this.endpoints = endpoints;
+ }
+
+ public static class CloudEndpoint {
+ private String endpointType;
+ private String endpointUrl;
+
+ public CloudEndpoint() {
+ }
+
+ public CloudEndpoint(String endpointType, String endpointUrl) {
+ this.endpointType = endpointType;
+ this.endpointUrl = endpointUrl;
+ }
+
+ public String getEndpointType() {
+ return endpointType;
+ }
+
+ public void setEndpointType(String endpointType) {
+ this.endpointType = endpointType;
+ }
+
+ public String getEndpointUrl() {
+ return endpointUrl;
+ }
+
+ public void setEndpointUrl(String endpointUrl) {
+ this.endpointUrl = endpointUrl;
+ }
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/CloudInstanceOptionVO.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/CloudInstanceOptionVO.java
new file mode 100644
index 00000000..e3d1da28
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/CloudInstanceOptionVO.java
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.provider;
+
+public class CloudInstanceOptionVO {
+
+ private String instanceId;
+ private String instanceName;
+ private String status;
+ private String regionId;
+ private Integer topicCount;
+ private Integer groupCount;
+ private String remark;
+
+ public String getInstanceId() {
+ return instanceId;
+ }
+
+ public void setInstanceId(String instanceId) {
+ this.instanceId = instanceId;
+ }
+
+ public String getInstanceName() {
+ return instanceName;
+ }
+
+ public void setInstanceName(String instanceName) {
+ this.instanceName = instanceName;
+ }
+
+ public String getStatus() {
+ return status;
+ }
+
+ public void setStatus(String status) {
+ this.status = status;
+ }
+
+ public String getRegionId() {
+ return regionId;
+ }
+
+ public void setRegionId(String regionId) {
+ this.regionId = regionId;
+ }
+
+ public Integer getTopicCount() {
+ return topicCount;
+ }
+
+ public void setTopicCount(Integer topicCount) {
+ this.topicCount = topicCount;
+ }
+
+ public Integer getGroupCount() {
+ return groupCount;
+ }
+
+ public void setGroupCount(Integer groupCount) {
+ this.groupCount = groupCount;
+ }
+
+ public String getRemark() {
+ return remark;
+ }
+
+ public void setRemark(String remark) {
+ this.remark = remark;
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqInstance.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/CloudRegionVO.java
similarity index 56%
copy from
server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqInstance.java
copy to
server/src/main/java/org/apache/rocketmq/studio/provider/CloudRegionVO.java
index 7c4cdbfd..2a068a4b 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqInstance.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/CloudRegionVO.java
@@ -14,31 +14,34 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.rocketmq.studio.persistence.entity;
+package org.apache.rocketmq.studio.provider;
-import com.baomidou.mybatisplus.annotation.IdType;
-import com.baomidou.mybatisplus.annotation.TableId;
-import com.baomidou.mybatisplus.annotation.TableName;
-import lombok.Data;
+public class CloudRegionVO {
-import java.time.LocalDateTime;
+ private String regionId;
+ private String regionName;
-@Data
-@TableName("rmq_instance")
-public class RmqInstance {
+ public CloudRegionVO() {
+ }
- @TableId(type = IdType.INPUT)
- private String id;
+ public CloudRegionVO(String regionId, String regionName) {
+ this.regionId = regionId;
+ this.regionName = regionName;
+ }
- private String name;
+ public String getRegionId() {
+ return regionId;
+ }
- private String remark;
+ public void setRegionId(String regionId) {
+ this.regionId = regionId;
+ }
- private String type;
+ public String getRegionName() {
+ return regionName;
+ }
- private String endpoint;
-
- private LocalDateTime createdAt;
-
- private LocalDateTime updatedAt;
+ public void setRegionName(String regionName) {
+ this.regionName = regionName;
+ }
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProvider.java
new file mode 100644
index 00000000..65a9a945
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProvider.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.provider;
+
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
+import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
+import org.apache.rocketmq.studio.instance.group.QueueProgressVO;
+import org.apache.rocketmq.studio.instance.group.SubscriptionEntryVO;
+import org.apache.rocketmq.studio.instance.message.MessageRecordVO;
+import org.apache.rocketmq.studio.instance.message.TraceRecordVO;
+import org.apache.rocketmq.studio.instance.topic.TopicConsumerVO;
+import org.apache.rocketmq.studio.instance.topic.TopicVO;
+
+import java.util.List;
+
+/**
+ * Unified instance-scoped operations SPI. Every method takes the Studio
instance id as its
+ * first argument; implementations resolve the target cluster / cloud instance
themselves.
+ * Unsupported operations throw {@link UnsupportedOperationException} (mapped
to HTTP 501).
+ */
+public interface InstanceProvider {
+
+ InstanceVendor vendor();
+
+ List<TopicVO> listTopics(String instanceId, String type, String search);
+
+ TopicVO createTopic(String instanceId, TopicVO topic);
+
+ TopicVO updateTopic(String instanceId, TopicVO topic);
+
+ void deleteTopic(String instanceId, String topicName);
+
+ List<TopicConsumerVO> getTopicConsumers(String instanceId, String
topicName);
+
+ List<ConsumerGroupVO> listConsumerGroups(String instanceId, String search);
+
+ ConsumerGroupVO createConsumerGroup(String instanceId, ConsumerGroupVO
group);
+
+ void deleteConsumerGroup(String instanceId, String groupName);
+
+ List<QueueProgressVO> getGroupProgress(String instanceId, String
groupName);
+
+ List<SubscriptionEntryVO> getGroupSubscriptions(String instanceId, String
groupName);
+
+ void resetOffset(String instanceId, String groupName, long timestamp,
String topic);
+
+ List<MessageRecordVO> queryMessages(String instanceId, String topic,
String msgId,
+ String tag, String key, Long
startTime, Long endTime);
+
+ TraceRecordVO getMessageTrace(String instanceId, String msgId);
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProviderRegistry.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProviderRegistry.java
new file mode 100644
index 00000000..998a81a6
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/InstanceProviderRegistry.java
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.provider;
+
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.instance.InstanceRepository;
+import org.apache.rocketmq.studio.instance.InstanceVO;
+import org.springframework.stereotype.Component;
+
+import java.util.EnumMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+@Component
+public class InstanceProviderRegistry {
+
+ private final Map<InstanceVendor, InstanceProvider> providers = new
EnumMap<>(InstanceVendor.class);
+ private final Map<InstanceVendor, CloudCatalogProvider> catalogs = new
EnumMap<>(InstanceVendor.class);
+ private final InstanceRepository instanceRepository;
+
+ public InstanceProviderRegistry(List<InstanceProvider> providerList,
+ List<CloudCatalogProvider> catalogList,
+ InstanceRepository instanceRepository) {
+ providerList.forEach(provider -> providers.put(provider.vendor(),
provider));
+ catalogList.forEach(catalog -> catalogs.put(catalog.vendor(),
catalog));
+ this.instanceRepository = instanceRepository;
+ }
+
+ public InstanceProvider forVendor(InstanceVendor vendor) {
+ InstanceProvider provider = providers.get(vendor);
+ if (provider == null) {
+ throw new BusinessException(501, "No instance provider registered
for vendor " + vendor);
+ }
+ return provider;
+ }
+
+ /**
+ * Resolves the provider for the given Studio instance id. Returns empty
for a blank id
+ * so callers can fall back to the legacy global behavior.
+ */
+ public Optional<InstanceProvider> byInstanceId(String instanceId) {
+ if (instanceId == null || instanceId.isBlank()) {
+ return Optional.empty();
+ }
+ InstanceVO instance = instanceRepository.findById(instanceId)
+ .orElseThrow(() -> new BusinessException(404, "Instance not
found: " + instanceId));
+ InstanceVendor vendor = instance.getVendor() == null ?
InstanceVendor.APACHE : instance.getVendor();
+ return Optional.of(forVendor(vendor));
+ }
+
+ public CloudCatalogProvider catalogFor(InstanceVendor vendor) {
+ CloudCatalogProvider catalog = catalogs.get(vendor);
+ if (catalog == null) {
+ throw new BusinessException(501, "No cloud catalog provider
registered for vendor " + vendor);
+ }
+ return catalog;
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.java
new file mode 100644
index 00000000..1d351546
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ApacheInstanceProvider.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 org.apache.rocketmq.studio.provider.apache;
+
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
+import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
+import org.apache.rocketmq.studio.instance.group.QueueProgressVO;
+import org.apache.rocketmq.studio.instance.group.SubscriptionEntryVO;
+import org.apache.rocketmq.studio.instance.message.MessageProvider;
+import org.apache.rocketmq.studio.instance.message.MessageRecordVO;
+import org.apache.rocketmq.studio.instance.message.TraceRecordVO;
+import org.apache.rocketmq.studio.instance.topic.AdminClient;
+import org.apache.rocketmq.studio.instance.topic.MetadataProvider;
+import org.apache.rocketmq.studio.instance.topic.TopicConsumerVO;
+import org.apache.rocketmq.studio.instance.topic.TopicVO;
+import org.apache.rocketmq.studio.provider.InstanceProvider;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * Open-source Apache RocketMQ implementation: pure delegation to the existing
admin-client
+ * beans, keeping behavior identical to the pre-provider code path.
+ */
+@Component
+public class ApacheInstanceProvider implements InstanceProvider {
+
+ private final MetadataProvider metadataProvider;
+ private final AdminClient adminClient;
+ private final MessageProvider messageProvider;
+
+ public ApacheInstanceProvider(MetadataProvider metadataProvider,
+ AdminClient adminClient,
+ MessageProvider messageProvider) {
+ this.metadataProvider = metadataProvider;
+ this.adminClient = adminClient;
+ this.messageProvider = messageProvider;
+ }
+
+ @Override
+ public InstanceVendor vendor() {
+ return InstanceVendor.APACHE;
+ }
+
+ @Override
+ public List<TopicVO> listTopics(String instanceId, String type, String
search) {
+ return metadataProvider.listTopics(null, type, search).stream()
+ .filter(topic -> matchesInstance(topic.getInstanceId(),
instanceId))
+ .toList();
+ }
+
+ @Override
+ public TopicVO createTopic(String instanceId, TopicVO topic) {
+ return adminClient.createTopic(topic);
+ }
+
+ @Override
+ public TopicVO updateTopic(String instanceId, TopicVO topic) {
+ return adminClient.updateTopic(topic);
+ }
+
+ @Override
+ public void deleteTopic(String instanceId, String topicName) {
+ adminClient.deleteTopic(topicName);
+ }
+
+ @Override
+ public List<TopicConsumerVO> getTopicConsumers(String instanceId, String
topicName) {
+ return metadataProvider.getTopicConsumers(topicName);
+ }
+
+ @Override
+ public List<ConsumerGroupVO> listConsumerGroups(String instanceId, String
search) {
+ return metadataProvider.listConsumerGroups(null, search).stream()
+ .filter(group -> matchesInstance(group.getInstanceId(),
instanceId))
+ .toList();
+ }
+
+ @Override
+ public ConsumerGroupVO createConsumerGroup(String instanceId,
ConsumerGroupVO group) {
+ return adminClient.createConsumerGroup(group);
+ }
+
+ @Override
+ public void deleteConsumerGroup(String instanceId, String groupName) {
+ adminClient.deleteConsumerGroup(groupName);
+ }
+
+ @Override
+ public List<QueueProgressVO> getGroupProgress(String instanceId, String
groupName) {
+ return metadataProvider.getGroupProgress(groupName);
+ }
+
+ @Override
+ public List<SubscriptionEntryVO> getGroupSubscriptions(String instanceId,
String groupName) {
+ return metadataProvider.getGroupSubscriptions(groupName);
+ }
+
+ @Override
+ public void resetOffset(String instanceId, String groupName, long
timestamp, String topic) {
+ adminClient.resetOffset(groupName, timestamp, topic);
+ }
+
+ @Override
+ public List<MessageRecordVO> queryMessages(String instanceId, String
topic, String msgId,
+ String tag, String key, Long
startTime, Long endTime) {
+ return messageProvider.queryMessages(instanceId, topic, msgId, tag,
key, startTime, endTime);
+ }
+
+ @Override
+ public TraceRecordVO getMessageTrace(String instanceId, String msgId) {
+ return messageProvider.getMessageTrace(instanceId, msgId);
+ }
+
+ private boolean matchesInstance(String topicInstanceId, String instanceId)
{
+ if (instanceId == null || instanceId.isBlank()) {
+ return true;
+ }
+ return Objects.equals(topicInstanceId, instanceId);
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProvider.java
new file mode 100644
index 00000000..ca4df5f4
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProvider.java
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.provider.tencent;
+
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
+import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
+import org.apache.rocketmq.studio.instance.group.QueueProgressVO;
+import org.apache.rocketmq.studio.instance.group.SubscriptionEntryVO;
+import org.apache.rocketmq.studio.instance.message.MessageRecordVO;
+import org.apache.rocketmq.studio.instance.message.TraceRecordVO;
+import org.apache.rocketmq.studio.instance.topic.TopicConsumerVO;
+import org.apache.rocketmq.studio.instance.topic.TopicVO;
+import org.apache.rocketmq.studio.provider.InstanceProvider;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+
+/**
+ * Tencent Cloud TDMQ placeholder: package structure reserved, all operations
unsupported.
+ */
+@Component
+public class TencentInstanceProvider implements InstanceProvider {
+
+ private static final String NOT_IMPLEMENTED = "Tencent Cloud provider is
not implemented yet";
+
+ @Override
+ public InstanceVendor vendor() {
+ return InstanceVendor.TENCENT;
+ }
+
+ @Override
+ public List<TopicVO> listTopics(String instanceId, String type, String
search) {
+ throw new UnsupportedOperationException(NOT_IMPLEMENTED);
+ }
+
+ @Override
+ public TopicVO createTopic(String instanceId, TopicVO topic) {
+ throw new UnsupportedOperationException(NOT_IMPLEMENTED);
+ }
+
+ @Override
+ public TopicVO updateTopic(String instanceId, TopicVO topic) {
+ throw new UnsupportedOperationException(NOT_IMPLEMENTED);
+ }
+
+ @Override
+ public void deleteTopic(String instanceId, String topicName) {
+ throw new UnsupportedOperationException(NOT_IMPLEMENTED);
+ }
+
+ @Override
+ public List<TopicConsumerVO> getTopicConsumers(String instanceId, String
topicName) {
+ throw new UnsupportedOperationException(NOT_IMPLEMENTED);
+ }
+
+ @Override
+ public List<ConsumerGroupVO> listConsumerGroups(String instanceId, String
search) {
+ throw new UnsupportedOperationException(NOT_IMPLEMENTED);
+ }
+
+ @Override
+ public ConsumerGroupVO createConsumerGroup(String instanceId,
ConsumerGroupVO group) {
+ throw new UnsupportedOperationException(NOT_IMPLEMENTED);
+ }
+
+ @Override
+ public void deleteConsumerGroup(String instanceId, String groupName) {
+ throw new UnsupportedOperationException(NOT_IMPLEMENTED);
+ }
+
+ @Override
+ public List<QueueProgressVO> getGroupProgress(String instanceId, String
groupName) {
+ throw new UnsupportedOperationException(NOT_IMPLEMENTED);
+ }
+
+ @Override
+ public List<SubscriptionEntryVO> getGroupSubscriptions(String instanceId,
String groupName) {
+ throw new UnsupportedOperationException(NOT_IMPLEMENTED);
+ }
+
+ @Override
+ public void resetOffset(String instanceId, String groupName, long
timestamp, String topic) {
+ throw new UnsupportedOperationException(NOT_IMPLEMENTED);
+ }
+
+ @Override
+ public List<MessageRecordVO> queryMessages(String instanceId, String
topic, String msgId,
+ String tag, String key, Long
startTime, Long endTime) {
+ throw new UnsupportedOperationException(NOT_IMPLEMENTED);
+ }
+
+ @Override
+ public TraceRecordVO getMessageTrace(String instanceId, String msgId) {
+ throw new UnsupportedOperationException(NOT_IMPLEMENTED);
+ }
+}
diff --git a/server/src/main/resources/db/schema.sql
b/server/src/main/resources/db/schema.sql
index 8475f8cd..53aed259 100644
--- a/server/src/main/resources/db/schema.sql
+++ b/server/src/main/resources/db/schema.sql
@@ -25,6 +25,10 @@ CREATE TABLE IF NOT EXISTS rmq_instance (
remark VARCHAR(255),
type VARCHAR(32) NOT NULL COMMENT 'PROXY/DIRECT',
endpoint VARCHAR(512) NOT NULL,
+ vendor VARCHAR(32) NOT NULL DEFAULT 'APACHE' COMMENT 'APACHE/ALIYUN/TENCENT',
+ cloud_instance_id VARCHAR(128) COMMENT '云厂商实例 ID（vendor 非 APACHE 时必填）',
+ credential_id VARCHAR(64) COMMENT '引用 rmq_cloud_credential.id',
+ region_id VARCHAR(64) COMMENT '云 region',
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
@@ -209,6 +213,19 @@ CREATE TABLE IF NOT EXISTS rmq_system_alert (
INDEX idx_acknowledged (acknowledged)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+-- 15. 云厂商凭据（secret_key 为 base64 编码，禁止明文；access_key 明文用于唯一键与打码展示）
+CREATE TABLE IF NOT EXISTS rmq_cloud_credential (
+ id VARCHAR(64) PRIMARY KEY,
+ name VARCHAR(128) NOT NULL COMMENT '凭据显示名',
+ vendor VARCHAR(32) NOT NULL COMMENT 'ALIYUN/TENCENT',
+ access_key VARCHAR(255) NOT NULL,
+ secret_key VARCHAR(512) NOT NULL COMMENT 'base64 编码的 SK',
+ remark VARCHAR(255),
+ created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+ updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+ UNIQUE KEY uk_vendor_access_key (vendor, access_key)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
-- ============================================================
-- 样例数据（幂等）：instance / topic / group 列表以本库为准，创建时写库、读取时读库。
-- 实例管理页默认 5 个实例：2 个 DIRECT（instance-direct-1/2）+ 3 个
PROXY（instance-proxy-1/2/3）。
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cloud/credential/CloudCredentialServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cloud/credential/CloudCredentialServiceTest.java
new file mode 100644
index 00000000..5bb98e59
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/cloud/credential/CloudCredentialServiceTest.java
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.cloud.credential;
+
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.instance.InstanceRepository;
+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.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class CloudCredentialServiceTest {
+
+ @Mock
+ private CloudCredentialRepository credentialRepository;
+
+ @Mock
+ private InstanceRepository instanceRepository;
+
+ @InjectMocks
+ private CloudCredentialService service;
+
+ @Test
+ void listShouldMaskAccessKeyAndHideSecretTest() {
+ CloudCredentialVO stored = new CloudCredentialVO();
+ stored.setId("cred-1");
+ stored.setName("aliyun-test");
+ stored.setVendor(InstanceVendor.ALIYUN);
+ stored.setAccessKey("LTAI5tUnitTestKey000000001");
+ stored.setSecretKey("secret-value");
+ when(credentialRepository.findAll()).thenReturn(List.of(stored));
+
+ List<CloudCredentialVO> result = service.listMasked();
+
+ assertThat(result).hasSize(1);
+ assertThat(result.get(0).getAccessKey()).isEqualTo("LTAI****0001");
+ assertThat(result.get(0).getSecretKey()).isNull();
+ }
+
+ @Test
+ void listShouldFullyMaskShortAccessKeyTest() {
+ CloudCredentialVO stored = new CloudCredentialVO();
+ stored.setVendor(InstanceVendor.ALIYUN);
+ stored.setAccessKey("short-ak");
+ when(credentialRepository.findAll()).thenReturn(List.of(stored));
+
+
assertThat(service.listMasked().get(0).getAccessKey()).isEqualTo("****");
+ }
+
+ @Test
+ void createShouldRejectDuplicateVendorAndAccessKeyTest() {
+ CloudCredentialVO request = new CloudCredentialVO();
+ request.setName("dup");
+ request.setVendor(InstanceVendor.ALIYUN);
+ request.setAccessKey("LTAI5tDupKey000000000001");
+ request.setSecretKey("sk");
+
when(credentialRepository.findByVendorAndAccessKey(InstanceVendor.ALIYUN,
"LTAI5tDupKey000000000001"))
+ .thenReturn(Optional.of(new CloudCredentialVO()));
+
+ assertThatThrownBy(() -> service.create(request))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("already exists");
+ verify(credentialRepository, never()).save(any());
+ }
+
+ @Test
+ void createShouldRejectApacheVendorTest() {
+ CloudCredentialVO request = new CloudCredentialVO();
+ request.setName("bad");
+ request.setVendor(InstanceVendor.APACHE);
+ request.setAccessKey("LTAI5tBadVendor000000001");
+ request.setSecretKey("sk");
+
+ assertThatThrownBy(() -> service.create(request))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("ALIYUN or TENCENT");
+ }
+
+ @Test
+ void createShouldAssignIdAndMaskResultTest() {
+ CloudCredentialVO request = new CloudCredentialVO();
+ request.setName("ok");
+ request.setVendor(InstanceVendor.ALIYUN);
+ request.setAccessKey("LTAI5tGoodKey00000000001");
+ request.setSecretKey("sk-value");
+ when(credentialRepository.findByVendorAndAccessKey(any(),
any())).thenReturn(Optional.empty());
+ when(credentialRepository.save(any(CloudCredentialVO.class)))
+ .thenAnswer(invocation -> invocation.getArgument(0));
+
+ CloudCredentialVO created = service.create(request);
+
+ assertThat(created.getId()).isNotBlank();
+ assertThat(created.getAccessKey()).isEqualTo("LTAI****0001");
+ assertThat(created.getSecretKey()).isNull();
+ }
+
+ @Test
+ void deleteShouldRejectWhenReferencedByInstanceTest() {
+ CloudCredentialVO stored = new CloudCredentialVO();
+ stored.setId("cred-1");
+
when(credentialRepository.findById("cred-1")).thenReturn(Optional.of(stored));
+
when(instanceRepository.existsByCredentialId("cred-1")).thenReturn(true);
+
+ assertThatThrownBy(() -> service.delete("cred-1"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("referenced");
+ verify(credentialRepository, never()).deleteById(any());
+ }
+
+ @Test
+ void revealShouldReturnUnmaskedCredentialTest() {
+ CloudCredentialVO stored = new CloudCredentialVO();
+ stored.setId("cred-1");
+ stored.setVendor(InstanceVendor.ALIYUN);
+ stored.setAccessKey("LTAI5tRevealKey000000001");
+ stored.setSecretKey("plain-secret");
+
when(credentialRepository.findById("cred-1")).thenReturn(Optional.of(stored));
+
+ CloudCredentialVO revealed = service.reveal("cred-1");
+
+
assertThat(revealed.getAccessKey()).isEqualTo("LTAI5tRevealKey000000001");
+ assertThat(revealed.getSecretKey()).isEqualTo("plain-secret");
+ }
+
+ @Test
+ void repositoryShouldBase64EncodeSecretTest() {
+ String encoded =
MybatisPlusCloudCredentialRepository.encodeBase64("plain-secret");
+ assertThat(encoded).isNotEqualTo("plain-secret");
+
assertThat(MybatisPlusCloudCredentialRepository.decodeBase64(encoded)).isEqualTo("plain-secret");
+ }
+
+ @Test
+ void repositoryShouldTolerateLegacyPlainSecretTest() {
+ assertThat(MybatisPlusCloudCredentialRepository.decodeBase64("not
base64 !!!")).isEqualTo("not base64 !!!");
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/InstanceServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/InstanceServiceTest.java
index 64cbc05d..096339c9 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/InstanceServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/InstanceServiceTest.java
@@ -17,8 +17,14 @@
package org.apache.rocketmq.studio.instance;
+import org.apache.rocketmq.studio.cloud.credential.CloudCredentialRepository;
+import org.apache.rocketmq.studio.cloud.credential.CloudCredentialVO;
import org.apache.rocketmq.studio.common.domain.enums.InstanceType;
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.provider.CloudCatalogProvider;
+import org.apache.rocketmq.studio.provider.CloudInstanceDetailVO;
+import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InjectMocks;
@@ -43,6 +49,12 @@ class InstanceServiceTest {
@Mock
private InstanceRepository instanceRepository;
+ @Mock
+ private CloudCredentialRepository cloudCredentialRepository;
+
+ @Mock
+ private InstanceProviderRegistry providerRegistry;
+
@InjectMocks
private InstanceService instanceService;
@@ -418,4 +430,100 @@ class InstanceServiceTest {
.isInstanceOf(BusinessException.class)
.hasMessage("InstanceVO not found: missing");
}
+
+ @Test
+ void createInstanceShouldDefaultToApacheVendorTest() {
+ InstanceVO instance =
InstanceVO.builder().name("inst").endpoint("10.0.0.1:8080").build();
+
when(instanceRepository.save(any(InstanceVO.class))).thenAnswer(invocation ->
invocation.getArgument(0));
+
+ InstanceVO created = instanceService.createInstance(instance);
+
+ assertThat(created.getVendor()).isEqualTo(InstanceVendor.APACHE);
+ verifyNoInteractions(cloudCredentialRepository, providerRegistry);
+ }
+
+ @Test
+ void createInstanceShouldRejectManualEndpointForAliyunTest() {
+ InstanceVO instance = InstanceVO.builder()
+ .vendor(InstanceVendor.ALIYUN)
+ .endpoint("rmq-xxx.cn-hangzhou.rmq.aliyuncs.com:8080")
+ .build();
+
+ assertThatThrownBy(() -> instanceService.createInstance(instance))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("cannot be set manually");
+ verify(instanceRepository, never()).save(any(InstanceVO.class));
+ }
+
+ @Test
+ void createInstanceShouldRequireCloudFieldsForAliyunTest() {
+ InstanceVO instance =
InstanceVO.builder().vendor(InstanceVendor.ALIYUN).build();
+
+ assertThatThrownBy(() -> instanceService.createInstance(instance))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("credentialId");
+ }
+
+ @Test
+ void createInstanceShouldResolveAliyunEndpointFromCatalogTest() {
+ InstanceVO instance = InstanceVO.builder()
+ .vendor(InstanceVendor.ALIYUN)
+ .credentialId("cred-1")
+ .cloudInstanceId("rmq-cn-xxx")
+ .regionId("cn-hangzhou")
+ .build();
+ CloudCredentialVO credential = new CloudCredentialVO();
+ credential.setId("cred-1");
+ credential.setVendor(InstanceVendor.ALIYUN);
+
when(cloudCredentialRepository.findById("cred-1")).thenReturn(Optional.of(credential));
+ CloudCatalogProvider catalog =
org.mockito.Mockito.mock(CloudCatalogProvider.class);
+ CloudInstanceDetailVO detail = new CloudInstanceDetailVO();
+ detail.setInstanceId("rmq-cn-xxx");
+ detail.setInstanceName("prod-mq");
+ detail.setEndpoints(List.of(
+ new CloudInstanceDetailVO.CloudEndpoint("TCP_INTERNET",
"public:8080"),
+ new CloudInstanceDetailVO.CloudEndpoint("TCP_VPC",
"vpc:8080")));
+
when(providerRegistry.catalogFor(InstanceVendor.ALIYUN)).thenReturn(catalog);
+ when(catalog.getCloudInstance("cred-1", "cn-hangzhou",
"rmq-cn-xxx")).thenReturn(detail);
+
when(instanceRepository.save(any(InstanceVO.class))).thenAnswer(invocation ->
invocation.getArgument(0));
+
+ InstanceVO created = instanceService.createInstance(instance);
+
+ assertThat(created.getName()).isEqualTo("prod-mq");
+ assertThat(created.getEndpoint()).isEqualTo("vpc:8080");
+ assertThat(created.getType()).isEqualTo(InstanceType.PROXY);
+ }
+
+ @Test
+ void createInstanceShouldRejectTencentVendorTest() {
+ InstanceVO instance =
InstanceVO.builder().vendor(InstanceVendor.TENCENT).build();
+
+ assertThatThrownBy(() -> instanceService.createInstance(instance))
+ .isInstanceOf(BusinessException.class)
+ .satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(501));
+ }
+
+ @Test
+ void updateInstanceShouldKeepCloudFieldsImmutableTest() {
+ InstanceVO existing = InstanceVO.builder()
+ .name("aliyun-inst")
+ .vendor(InstanceVendor.ALIYUN)
+ .cloudInstanceId("rmq-cn-xxx")
+ .credentialId("cred-1")
+ .regionId("cn-hangzhou")
+ .type(InstanceType.PROXY)
+ .endpoint("vpc:8080")
+ .build();
+ existing.setId("inst-1");
+
when(instanceRepository.findById("inst-1")).thenReturn(Optional.of(existing));
+
when(instanceRepository.save(any(InstanceVO.class))).thenAnswer(invocation ->
invocation.getArgument(0));
+
+ InstanceVO request =
InstanceVO.builder().endpoint("hacked:8080").remark("updated").build();
+ request.setId("inst-1");
+ InstanceVO updated = instanceService.updateInstance(request);
+
+ assertThat(updated.getEndpoint()).isEqualTo("vpc:8080");
+ assertThat(updated.getRemark()).isEqualTo("updated");
+ assertThat(updated.getCloudInstanceId()).isEqualTo("rmq-cn-xxx");
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQServiceTest.java
index 2cb65468..8f33cefc 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQServiceTest.java
@@ -17,6 +17,7 @@
package org.apache.rocketmq.studio.instance.dlq;
+import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InjectMocks;
@@ -37,6 +38,9 @@ class DLQServiceTest {
@Mock
private DLQProvider dlqProvider;
+ @Mock
+ private InstanceProviderRegistry providerRegistry;
+
@InjectMocks
private DLQService dlqService;
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupControllerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupControllerTest.java
index 82ad24da..09b9f718 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupControllerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/group/ConsumerGroupControllerTest.java
@@ -35,6 +35,7 @@ import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.Mockito.when;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
@@ -169,7 +170,7 @@ class ConsumerGroupControllerTest {
.andExpect(jsonPath("$.code").value(200))
.andExpect(jsonPath("$.message").value("success"));
- verify(metadataService).resetOffset(eq("cg-orders"),
eq(1784246400000L), eq("orders"));
+ verify(metadataService).resetOffset(isNull(), eq("cg-orders"),
eq(1784246400000L), eq("orders"));
}
@Test
@@ -181,7 +182,7 @@ class ConsumerGroupControllerTest {
.andExpect(jsonPath("$.code").value(200))
.andExpect(jsonPath("$.message").value("success"));
- verify(metadataService).deleteConsumerGroup("cg-orders");
+ verify(metadataService).deleteConsumerGroup(isNull(), eq("cg-orders"));
}
@Test
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageServiceTest.java
index 2c8de5a3..52128c8f 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageServiceTest.java
@@ -11,6 +11,7 @@
package org.apache.rocketmq.studio.instance.message;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -22,7 +23,8 @@ class MessageServiceTest {
@Test
void rejectsReversedTopicQueryWindowBeforeCallingProvider() {
MessageProvider provider = mock(MessageProvider.class);
- MessageService service = new MessageService(provider);
+ InstanceProviderRegistry registry =
mock(InstanceProviderRegistry.class);
+ MessageService service = new MessageService(provider, registry);
assertThatThrownBy(() -> service.queryMessages("instance-a", "TopicA",
null, null, null, 200L, 100L))
.isInstanceOf(BusinessException.class)
@@ -34,7 +36,8 @@ class MessageServiceTest {
@Test
void rejectsTopicQueryWindowLongerThanSevenDaysBeforeCallingProvider() {
MessageProvider provider = mock(MessageProvider.class);
- MessageService service = new MessageService(provider);
+ InstanceProviderRegistry registry =
mock(InstanceProviderRegistry.class);
+ MessageService service = new MessageService(provider, registry);
assertThatThrownBy(() -> service.queryMessages("instance-a", "TopicA",
null, null, null, 0L,
8L * 24 * 60 * 60 * 1000))
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
index a7c0d7d3..7ed2013b 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
@@ -19,6 +19,7 @@ package org.apache.rocketmq.studio.instance.topic;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
+import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InjectMocks;
@@ -43,6 +44,9 @@ class MetadataServiceTest {
@Mock
private AdminClient adminClient;
+ @Mock
+ private InstanceProviderRegistry providerRegistry;
+
@InjectMocks
private MetadataService metadataService;
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/TopicControllerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/TopicControllerTest.java
index 06753721..6da4af95 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/TopicControllerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/TopicControllerTest.java
@@ -62,7 +62,7 @@ class TopicControllerTest {
topic.setWriteQueues(8);
topic.setReadQueues(8);
- when(metadataService.listTopics(isNull(), isNull(),
isNull())).thenReturn(List.of(topic));
+ when(metadataService.listTopics(isNull(), isNull(), isNull(),
isNull())).thenReturn(List.of(topic));
mockMvc.perform(get("/api/topics"))
.andExpect(status().isOk())
@@ -74,7 +74,7 @@ class TopicControllerTest {
@Test
void listTopicsShouldPassQueryParams() throws Exception {
- when(metadataService.listTopics(eq("cluster-1"), eq("NORMAL"),
eq("test")))
+ when(metadataService.listTopics(isNull(), eq("cluster-1"),
eq("NORMAL"), eq("test")))
.thenReturn(List.of());
mockMvc.perform(get("/api/topics")
@@ -84,7 +84,7 @@ class TopicControllerTest {
.andExpect(status().isOk())
.andExpect(jsonPath("$.data").isArray());
- verify(metadataService).listTopics(eq("cluster-1"), eq("NORMAL"),
eq("test"));
+ verify(metadataService).listTopics(isNull(), eq("cluster-1"),
eq("NORMAL"), eq("test"));
}
@Test
@@ -230,7 +230,7 @@ class TopicControllerTest {
.andExpect(jsonPath("$.code").value(200))
.andExpect(jsonPath("$.message").value("success"));
- verify(metadataService).deleteTopic("test-topic");
+ verify(metadataService).deleteTopic(isNull(), eq("test-topic"));
}
@Test
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/InstanceProviderRegistryTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/InstanceProviderRegistryTest.java
new file mode 100644
index 00000000..633c5f3e
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/InstanceProviderRegistryTest.java
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.provider;
+
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.instance.InstanceRepository;
+import org.apache.rocketmq.studio.instance.InstanceVO;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+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.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class InstanceProviderRegistryTest {
+
+ @Mock
+ private InstanceRepository instanceRepository;
+
+ private InstanceProviderRegistry registry;
+ private InstanceProvider apacheProvider;
+ private InstanceProvider aliyunProvider;
+
+ @BeforeEach
+ void setUp() {
+ apacheProvider = stubProvider(InstanceVendor.APACHE);
+ aliyunProvider = stubProvider(InstanceVendor.ALIYUN);
+ registry = new InstanceProviderRegistry(List.of(apacheProvider,
aliyunProvider), List.of(), instanceRepository);
+ }
+
+ @Test
+ void forVendorShouldReturnRegisteredProviderTest() {
+
assertThat(registry.forVendor(InstanceVendor.ALIYUN)).isSameAs(aliyunProvider);
+ }
+
+ @Test
+ void forVendorShouldThrowWhenMissingTest() {
+ assertThatThrownBy(() -> registry.forVendor(InstanceVendor.TENCENT))
+ .isInstanceOf(BusinessException.class)
+ .satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(501));
+ }
+
+ @Test
+ void byInstanceIdShouldReturnEmptyForBlankIdTest() {
+ assertThat(registry.byInstanceId(null)).isEmpty();
+ assertThat(registry.byInstanceId(" ")).isEmpty();
+ }
+
+ @Test
+ void byInstanceIdShouldThrowWhenInstanceMissingTest() {
+
when(instanceRepository.findById("missing")).thenReturn(Optional.empty());
+
+ assertThatThrownBy(() -> registry.byInstanceId("missing"))
+ .isInstanceOf(BusinessException.class)
+ .satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(404));
+ }
+
+ @Test
+ void byInstanceIdShouldResolveVendorProviderTest() {
+ InstanceVO instance =
InstanceVO.builder().vendor(InstanceVendor.ALIYUN).build();
+
when(instanceRepository.findById("inst-1")).thenReturn(Optional.of(instance));
+
+
assertThat(registry.byInstanceId("inst-1")).containsSame(aliyunProvider);
+ }
+
+ @Test
+ void byInstanceIdShouldDefaultToApacheWhenVendorNullTest() {
+ InstanceVO instance = InstanceVO.builder().build();
+
when(instanceRepository.findById("inst-2")).thenReturn(Optional.of(instance));
+
+
assertThat(registry.byInstanceId("inst-2")).containsSame(apacheProvider);
+ }
+
+ @Test
+ void catalogForShouldThrowWhenNoCatalogRegisteredTest() {
+ assertThatThrownBy(() -> registry.catalogFor(InstanceVendor.ALIYUN))
+ .isInstanceOf(BusinessException.class)
+ .satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(501));
+ }
+
+ private InstanceProvider stubProvider(InstanceVendor vendor) {
+ InstanceProvider provider = mock(InstanceProvider.class);
+ when(provider.vendor()).thenReturn(vendor);
+ return provider;
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProviderTest.java
new file mode 100644
index 00000000..b7660bac
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProviderTest.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.provider.tencent;
+
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
+import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
+import org.apache.rocketmq.studio.instance.topic.TopicVO;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class TencentInstanceProviderTest {
+
+ private final TencentInstanceProvider provider = new
TencentInstanceProvider();
+
+ @Test
+ void vendorShouldBeTencentTest() {
+ assertThat(provider.vendor()).isEqualTo(InstanceVendor.TENCENT);
+ }
+
+ @Test
+ void allOperationsShouldThrowUnsupportedTest() {
+ assertThatThrownBy(() -> provider.listTopics("inst", null, null))
+ .isInstanceOf(UnsupportedOperationException.class);
+ assertThatThrownBy(() -> provider.createTopic("inst", new TopicVO()))
+ .isInstanceOf(UnsupportedOperationException.class);
+ assertThatThrownBy(() -> provider.updateTopic("inst", new TopicVO()))
+ .isInstanceOf(UnsupportedOperationException.class);
+ assertThatThrownBy(() -> provider.deleteTopic("inst", "topic"))
+ .isInstanceOf(UnsupportedOperationException.class);
+ assertThatThrownBy(() -> provider.getTopicConsumers("inst", "topic"))
+ .isInstanceOf(UnsupportedOperationException.class);
+ assertThatThrownBy(() -> provider.listConsumerGroups("inst", null))
+ .isInstanceOf(UnsupportedOperationException.class);
+ assertThatThrownBy(() -> provider.createConsumerGroup("inst", new
ConsumerGroupVO()))
+ .isInstanceOf(UnsupportedOperationException.class);
+ assertThatThrownBy(() -> provider.deleteConsumerGroup("inst", "group"))
+ .isInstanceOf(UnsupportedOperationException.class);
+ assertThatThrownBy(() -> provider.getGroupProgress("inst", "group"))
+ .isInstanceOf(UnsupportedOperationException.class);
+ assertThatThrownBy(() -> provider.getGroupSubscriptions("inst",
"group"))
+ .isInstanceOf(UnsupportedOperationException.class);
+ assertThatThrownBy(() -> provider.resetOffset("inst", "group", 1L,
"topic"))
+ .isInstanceOf(UnsupportedOperationException.class);
+ assertThatThrownBy(() -> provider.queryMessages("inst", "topic", null,
null, null, null, null))
+ .isInstanceOf(UnsupportedOperationException.class);
+ assertThatThrownBy(() -> provider.getMessageTrace("inst", "msg"))
+ .isInstanceOf(UnsupportedOperationException.class);
+ }
+}
