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 e68fb3d8 feat: implement aliyun rocketmq openapi provider with cloud
catalog discovery (#1148)
e68fb3d8 is described below
commit e68fb3d80a196757a0c31defc19f222318eaac00
Author: lizhimins <[email protected]>
AuthorDate: Thu Aug 6 20:47:36 2026 +0800
feat: implement aliyun rocketmq openapi provider with cloud catalog
discovery (#1148)
---
.../provider/alibaba/AliyunCatalogController.java | 53 +++
.../provider/alibaba/AliyunCatalogService.java | 155 ++++++++
.../provider/alibaba/AliyunClientFactory.java | 186 +++++++++
.../studio/provider/alibaba/AliyunConverters.java | 341 ++++++++++++++++
.../provider/alibaba/AliyunInstanceProvider.java | 435 ++++++++++++++++++++
.../provider/alibaba/AliyunCatalogServiceTest.java | 213 ++++++++++
.../provider/alibaba/AliyunClientFactoryTest.java | 188 +++++++++
.../alibaba/AliyunInstanceProviderTest.java | 440 +++++++++++++++++++++
8 files changed, 2011 insertions(+)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunCatalogController.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunCatalogController.java
new file mode 100644
index 00000000..25828c94
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunCatalogController.java
@@ -0,0 +1,53 @@
+/*
+ * 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.alibaba;
+
+import org.apache.rocketmq.studio.common.domain.Result;
+import org.apache.rocketmq.studio.provider.CloudInstanceOptionVO;
+import org.apache.rocketmq.studio.provider.CloudRegionVO;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.List;
+
+/**
+ * REST endpoints for browsing the Aliyun cloud catalog with a stored
credential.
+ */
+@RestController
+@RequestMapping("/api/cloud/aliyun")
+public class AliyunCatalogController {
+
+ private final AliyunCatalogService catalogService;
+
+ public AliyunCatalogController(AliyunCatalogService catalogService) {
+ this.catalogService = catalogService;
+ }
+
+ @GetMapping("/regions")
+ public Result<List<CloudRegionVO>> listRegions(@RequestParam String
credentialId) {
+ return Result.ok(catalogService.listRegions(credentialId));
+ }
+
+ @GetMapping("/instances")
+ public Result<List<CloudInstanceOptionVO>> listInstances(@RequestParam
String credentialId,
+ @RequestParam
String regionId,
+
@RequestParam(required = false) String search) {
+ return Result.ok(catalogService.listCloudInstances(credentialId,
regionId, search));
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunCatalogService.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunCatalogService.java
new file mode 100644
index 00000000..cb1d6311
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunCatalogService.java
@@ -0,0 +1,155 @@
+/*
+ * 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.alibaba;
+
+import com.aliyun.sdk.service.rocketmq20220801.models.GetInstanceRequest;
+import com.aliyun.sdk.service.rocketmq20220801.models.GetInstanceResponse;
+import com.aliyun.sdk.service.rocketmq20220801.models.GetInstanceResponseBody;
+import com.aliyun.sdk.service.rocketmq20220801.models.ListInstancesRequest;
+import com.aliyun.sdk.service.rocketmq20220801.models.ListInstancesResponse;
+import
com.aliyun.sdk.service.rocketmq20220801.models.ListInstancesResponseBody;
+import com.aliyun.sdk.service.rocketmq20220801.models.ListRegionsRequest;
+import com.aliyun.sdk.service.rocketmq20220801.models.ListRegionsResponse;
+import com.aliyun.sdk.service.rocketmq20220801.models.ListRegionsResponseBody;
+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.CloudInstanceOptionVO;
+import org.apache.rocketmq.studio.provider.CloudRegionVO;
+import org.springframework.stereotype.Component;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Locale;
+
+/**
+ * Aliyun cloud catalog: regions and commercial RocketMQ 5.x instances visible
to a credential.
+ */
+@Component
+public class AliyunCatalogService implements CloudCatalogProvider {
+
+ /**
+ * Region used to reach the control plane for region-independent
operations such as ListRegions.
+ */
+ static final String DEFAULT_REGION = "cn-hangzhou";
+
+ private final AliyunClientFactory clientFactory;
+
+ public AliyunCatalogService(AliyunClientFactory clientFactory) {
+ this.clientFactory = clientFactory;
+ }
+
+ @Override
+ public InstanceVendor vendor() {
+ return InstanceVendor.ALIYUN;
+ }
+
+ @Override
+ public List<CloudRegionVO> listRegions(String credentialId) {
+ requireNonBlank(credentialId, "credentialId");
+ ListRegionsResponse response = clientFactory.call(credentialId,
DEFAULT_REGION,
+ client ->
client.listRegions(ListRegionsRequest.builder().build()));
+ ListRegionsResponseBody body = response == null ? null :
response.getBody();
+ List<ListRegionsResponseBody.Data> data = body == null ? null :
body.getData();
+ List<CloudRegionVO> regions = new ArrayList<>();
+ if (data == null) {
+ return regions;
+ }
+ for (ListRegionsResponseBody.Data item : data) {
+ if (Boolean.TRUE.equals(item.getSupportRocketmqV5())) {
+ regions.add(AliyunConverters.toRegionVO(item));
+ }
+ }
+ regions.sort(Comparator.comparing(
+ CloudRegionVO::getRegionId,
Comparator.nullsLast(Comparator.naturalOrder())));
+ return regions;
+ }
+
+ @Override
+ public List<CloudInstanceOptionVO> listCloudInstances(String credentialId,
String regionId, String search) {
+ requireNonBlank(credentialId, "credentialId");
+ requireNonBlank(regionId, "regionId");
+ List<ListInstancesResponseBody.List> all =
fetchAllInstances(credentialId, regionId);
+ List<CloudInstanceOptionVO> options = new ArrayList<>();
+ for (ListInstancesResponseBody.List item : all) {
+ CloudInstanceOptionVO vo =
AliyunConverters.toInstanceOptionVO(item);
+ if (matchesSearch(search, vo)) {
+ options.add(vo);
+ }
+ }
+ return options;
+ }
+
+ @Override
+ public CloudInstanceDetailVO getCloudInstance(String credentialId, String
regionId, String cloudInstanceId) {
+ requireNonBlank(credentialId, "credentialId");
+ requireNonBlank(regionId, "regionId");
+ requireNonBlank(cloudInstanceId, "cloudInstanceId");
+ GetInstanceRequest request =
GetInstanceRequest.builder().instanceId(cloudInstanceId).build();
+ GetInstanceResponse response = clientFactory.call(credentialId,
regionId,
+ client -> client.getInstance(request));
+ GetInstanceResponseBody body = response == null ? null :
response.getBody();
+ GetInstanceResponseBody.Data data = body == null ? null :
body.getData();
+ if (data == null) {
+ throw new BusinessException(404, "Aliyun instance not found: " +
cloudInstanceId);
+ }
+ return AliyunConverters.toInstanceDetailVO(data);
+ }
+
+ private List<ListInstancesResponseBody.List> fetchAllInstances(String
credentialId, String regionId) {
+ List<ListInstancesResponseBody.List> all = new ArrayList<>();
+ for (int page = 1; page <= AliyunConverters.MAX_PAGES; page++) {
+ ListInstancesRequest request = ListInstancesRequest.builder()
+ .pageNumber(page)
+ .pageSize(AliyunConverters.PAGE_SIZE)
+ .build();
+ ListInstancesResponse response = clientFactory.call(credentialId,
regionId,
+ client -> client.listInstances(request));
+ ListInstancesResponseBody body = response == null ? null :
response.getBody();
+ ListInstancesResponseBody.Data data = body == null ? null :
body.getData();
+ List<ListInstancesResponseBody.List> list = data == null ? null :
data.getList();
+ if (list == null || list.isEmpty()) {
+ break;
+ }
+ all.addAll(list);
+ if (list.size() < AliyunConverters.PAGE_SIZE) {
+ break;
+ }
+ }
+ return all;
+ }
+
+ private static boolean matchesSearch(String search, CloudInstanceOptionVO
vo) {
+ if (search == null || search.isBlank()) {
+ return true;
+ }
+ String needle = search.toLowerCase(Locale.ROOT);
+ boolean idMatches = vo.getInstanceId() != null
+ &&
vo.getInstanceId().toLowerCase(Locale.ROOT).contains(needle);
+ boolean nameMatches = vo.getInstanceName() != null
+ &&
vo.getInstanceName().toLowerCase(Locale.ROOT).contains(needle);
+ return idMatches || nameMatches;
+ }
+
+ private static void requireNonBlank(String value, String name) {
+ if (value == null || value.isBlank()) {
+ throw new BusinessException(400, name + " is required");
+ }
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunClientFactory.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunClientFactory.java
new file mode 100644
index 00000000..214a56a8
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunClientFactory.java
@@ -0,0 +1,186 @@
+/*
+ * 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.alibaba;
+
+import com.aliyun.auth.credentials.Credential;
+import com.aliyun.auth.credentials.provider.StaticCredentialProvider;
+import com.aliyun.sdk.gateway.pop.exception.PopClientException;
+import com.aliyun.sdk.gateway.pop.exception.PopServerException;
+import com.aliyun.sdk.service.rocketmq20220801.AsyncClient;
+import darabonba.core.client.ClientOverrideConfiguration;
+import darabonba.core.exception.ClientException;
+import darabonba.core.exception.ServerException;
+import jakarta.annotation.PreDestroy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.rocketmq.studio.cloud.credential.CloudCredentialRepository;
+import org.apache.rocketmq.studio.cloud.credential.CloudCredentialVO;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.springframework.stereotype.Component;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.function.Function;
+
+/**
+ * Builds and caches Aliyun RocketMQ 5.x OpenAPI async clients per
credential#region, and
+ * converts SDK failures into {@link BusinessException} with meaningful
HTTP-style codes.
+ */
+@Component
+public class AliyunClientFactory {
+
+ private static final Logger log =
LoggerFactory.getLogger(AliyunClientFactory.class);
+
+ static final Duration CONNECT_TIMEOUT = Duration.ofSeconds(10);
+ static final Duration RESPONSE_TIMEOUT = Duration.ofSeconds(20);
+ static final long DEFAULT_CALL_TIMEOUT_SECONDS = 30L;
+
+ private final CloudCredentialRepository credentialRepository;
+ private final Map<String, AsyncClient> clients = new ConcurrentHashMap<>();
+ private long callTimeoutSeconds = DEFAULT_CALL_TIMEOUT_SECONDS;
+
+ public AliyunClientFactory(CloudCredentialRepository credentialRepository)
{
+ this.credentialRepository = credentialRepository;
+ }
+
+ public AsyncClient client(String credentialId, String region) {
+ String key = cacheKey(credentialId, region);
+ return clients.computeIfAbsent(key, ignored ->
createClient(credentialId, region));
+ }
+
+ /**
+ * Executes an SDK call with a bounded wait and unified exception mapping.
+ */
+ public <T> T call(String credentialId, String region,
Function<AsyncClient, CompletableFuture<T>> action) {
+ AsyncClient client = client(credentialId, region);
+ CompletableFuture<T> future;
+ try {
+ future = action.apply(client);
+ } catch (RuntimeException ex) {
+ throw mapToBusinessException(ex);
+ }
+ try {
+ return future.get(callTimeoutSeconds, TimeUnit.SECONDS);
+ } catch (TimeoutException ex) {
+ throw new BusinessException(504,
+ "Aliyun OpenAPI request timed out after " +
callTimeoutSeconds + " seconds");
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ throw new BusinessException(502, "Aliyun OpenAPI request was
interrupted");
+ } catch (ExecutionException ex) {
+ throw mapToBusinessException(ex.getCause() == null ? ex :
ex.getCause());
+ }
+ }
+
+ @PreDestroy
+ public void close() {
+ clients.values().forEach(AsyncClient::close);
+ clients.clear();
+ }
+
+ void setCallTimeoutSeconds(long callTimeoutSeconds) {
+ this.callTimeoutSeconds = callTimeoutSeconds;
+ }
+
+ static String cacheKey(String credentialId, String region) {
+ return credentialId + "#" + region;
+ }
+
+ static String endpointFor(String region) {
+ return "rocketmq." + region + ".aliyuncs.com";
+ }
+
+ private AsyncClient createClient(String credentialId, String region) {
+ CloudCredentialVO credential =
credentialRepository.findById(credentialId)
+ .orElseThrow(() -> new BusinessException(404, "Cloud
credential not found: " + credentialId));
+ StaticCredentialProvider credentialProvider =
StaticCredentialProvider.create(
+ Credential.builder()
+ .accessKeyId(credential.getAccessKey())
+ .accessKeySecret(credential.getSecretKey())
+ .build());
+ return AsyncClient.builder()
+ .region(region)
+ .credentialsProvider(credentialProvider)
+ .overrideConfiguration(ClientOverrideConfiguration.create()
+ .setEndpointOverride(endpointFor(region))
+ .setConnectTimeout(CONNECT_TIMEOUT)
+ .setResponseTimeout(RESPONSE_TIMEOUT))
+ .build();
+ }
+
+ static BusinessException mapToBusinessException(Throwable raw) {
+ Throwable cause = raw;
+ while ((cause instanceof CompletionException || cause instanceof
ExecutionException)
+ && cause.getCause() != null) {
+ cause = cause.getCause();
+ }
+ if (cause instanceof BusinessException) {
+ return (BusinessException) cause;
+ }
+ Integer statusCode = null;
+ String errCode = null;
+ String message = cause.getMessage();
+ if (cause instanceof PopServerException) {
+ PopServerException ex = (PopServerException) cause;
+ statusCode = ex.getStatusCode();
+ errCode = ex.getErrCode();
+ message = firstNonBlank(ex.getErrMessage(), message);
+ } else if (cause instanceof PopClientException) {
+ PopClientException ex = (PopClientException) cause;
+ statusCode = ex.getStatusCode();
+ errCode = ex.getErrCode();
+ message = firstNonBlank(ex.getErrMessage(), message);
+ } else if (cause instanceof ServerException) {
+ statusCode = ((ServerException) cause).getStatusCode();
+ } else if (cause instanceof ClientException) {
+ statusCode = ((ClientException) cause).getStatusCode();
+ }
+ int status = statusCode == null ? 0 : statusCode;
+ log.warn("Aliyun OpenAPI failure: status={}, errCode={}, message={}",
status, errCode, message);
+ if (status == 401 || "InvalidAccessKeyId".equals(errCode) ||
"SignatureDoesNotMatch".equals(errCode)) {
+ return new BusinessException(401, "Cloud credential is invalid");
+ }
+ if (status == 403) {
+ return new BusinessException(403, defaultIfBlank(message, "Aliyun
OpenAPI access denied"));
+ }
+ if (status == 404 || errCode != null && errCode.contains("NotFound")) {
+ return new BusinessException(404, defaultIfBlank(message, "Aliyun
resource not found"));
+ }
+ String detail = message != null ? message : errCode != null ? errCode
: cause.getClass().getSimpleName();
+ return new BusinessException(502, "Aliyun OpenAPI error: " + detail);
+ }
+
+ private static String firstNonBlank(String first, String second) {
+ if (first != null && !first.isBlank()) {
+ return first;
+ }
+ return second;
+ }
+
+ private static String defaultIfBlank(String value, String fallback) {
+ if (value == null || value.isBlank()) {
+ return fallback;
+ }
+ return value;
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConverters.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConverters.java
new file mode 100644
index 00000000..8e501d0b
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConverters.java
@@ -0,0 +1,341 @@
+/*
+ * 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.alibaba;
+
+import com.aliyun.sdk.service.rocketmq20220801.models.DataTopicLagMapValue;
+import
com.aliyun.sdk.service.rocketmq20220801.models.GetConsumerGroupLagResponseBody;
+import com.aliyun.sdk.service.rocketmq20220801.models.GetInstanceResponseBody;
+import com.aliyun.sdk.service.rocketmq20220801.models.GetTraceResponseBody;
+import
com.aliyun.sdk.service.rocketmq20220801.models.ListConsumerGroupSubscriptionsResponseBody;
+import
com.aliyun.sdk.service.rocketmq20220801.models.ListConsumerGroupsResponseBody;
+import
com.aliyun.sdk.service.rocketmq20220801.models.ListInstancesResponseBody;
+import com.aliyun.sdk.service.rocketmq20220801.models.ListMessagesResponseBody;
+import com.aliyun.sdk.service.rocketmq20220801.models.ListRegionsResponseBody;
+import
com.aliyun.sdk.service.rocketmq20220801.models.ListTopicSubscriptionsResponseBody;
+import com.aliyun.sdk.service.rocketmq20220801.models.ListTopicsResponseBody;
+import org.apache.rocketmq.studio.common.domain.enums.ConsumeType;
+import org.apache.rocketmq.studio.common.domain.enums.TopicType;
+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.TraceNodeVO;
+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.CloudInstanceDetailVO;
+import org.apache.rocketmq.studio.provider.CloudInstanceOptionVO;
+import org.apache.rocketmq.studio.provider.CloudRegionVO;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.CharacterCodingException;
+import java.nio.charset.CodingErrorAction;
+import java.nio.charset.StandardCharsets;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+/**
+ * Static converters between Aliyun RocketMQ OpenAPI SDK models and Studio VOs.
+ */
+final class AliyunConverters {
+
+ static final int PAGE_SIZE = 100;
+ static final int MAX_PAGES = 5;
+ static final int MESSAGE_PAGE_SIZE = 20;
+ static final int MESSAGE_MAX_PAGES = 5;
+
+ private static final DateTimeFormatter TIME_FORMATTER =
DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+
+ private AliyunConverters() {
+ }
+
+ static CloudRegionVO toRegionVO(ListRegionsResponseBody.Data data) {
+ return new CloudRegionVO(data.getRegionId(), data.getRegionName());
+ }
+
+ static CloudInstanceOptionVO
toInstanceOptionVO(ListInstancesResponseBody.List data) {
+ CloudInstanceOptionVO vo = new CloudInstanceOptionVO();
+ vo.setInstanceId(data.getInstanceId());
+ vo.setInstanceName(data.getInstanceName());
+ vo.setStatus(data.getStatus());
+ vo.setRegionId(data.getRegionId());
+ vo.setTopicCount(toInteger(data.getTopicCount()));
+ vo.setGroupCount(toInteger(data.getGroupCount()));
+ vo.setRemark(data.getRemark());
+ return vo;
+ }
+
+ static CloudInstanceDetailVO
toInstanceDetailVO(GetInstanceResponseBody.Data data) {
+ CloudInstanceDetailVO vo = new CloudInstanceDetailVO();
+ vo.setInstanceId(data.getInstanceId());
+ vo.setInstanceName(data.getInstanceName());
+ vo.setStatus(data.getStatus());
+ vo.setRegionId(data.getRegionId());
+ vo.setRemark(data.getRemark());
+ List<CloudInstanceDetailVO.CloudEndpoint> endpoints = new
ArrayList<>();
+ if (data.getNetworkInfo() != null &&
data.getNetworkInfo().getEndpoints() != null) {
+ for (GetInstanceResponseBody.Endpoints endpoint :
data.getNetworkInfo().getEndpoints()) {
+ endpoints.add(new CloudInstanceDetailVO.CloudEndpoint(
+ endpoint.getEndpointType(),
endpoint.getEndpointUrl()));
+ }
+ }
+ vo.setEndpoints(endpoints);
+ return vo;
+ }
+
+ static TopicVO toTopicVO(ListTopicsResponseBody.List data, String
studioInstanceId) {
+ TopicVO vo = new TopicVO();
+ vo.setName(data.getTopicName());
+ vo.setInstanceId(studioInstanceId);
+ vo.setType(toTopicType(data.getMessageType()));
+ vo.setRemark(data.getRemark());
+ vo.setCreatedAt(parseDateTime(data.getCreateTime()));
+ vo.setUpdatedAt(parseDateTime(data.getUpdateTime()));
+ vo.setWriteQueues(0);
+ vo.setReadQueues(0);
+ return vo;
+ }
+
+ static TopicType toTopicType(String messageType) {
+ if (messageType == null) {
+ return null;
+ }
+ switch (messageType.toUpperCase(Locale.ROOT)) {
+ case "NORMAL":
+ return TopicType.NORMAL;
+ case "FIFO":
+ return TopicType.FIFO;
+ case "DELAY":
+ return TopicType.DELAY;
+ case "TRANSACTION":
+ return TopicType.TRANSACTION;
+ default:
+ return null;
+ }
+ }
+
+ static TopicConsumerVO
toTopicConsumerVO(ListTopicSubscriptionsResponseBody.Data data) {
+ return TopicConsumerVO.builder()
+ .group(data.getConsumerGroupId())
+ .consumeType(toConsumeType(data.getMessageModel()))
+ .messageModel(data.getMessageModel())
+ .build();
+ }
+
+ static ConsumerGroupVO
toConsumerGroupVO(ListConsumerGroupsResponseBody.List data, String
studioInstanceId) {
+ ConsumerGroupVO vo = new ConsumerGroupVO();
+ vo.setName(data.getConsumerGroupId());
+ vo.setInstanceId(studioInstanceId);
+ vo.setConsumeType(toConsumeType(data.getMessageModel()));
+ vo.setCreatedAt(parseDateTime(data.getCreateTime()));
+ vo.setUpdatedAt(parseDateTime(data.getUpdateTime()));
+ return vo;
+ }
+
+ static ConsumeType toConsumeType(String messageModel) {
+ if (messageModel == null) {
+ return null;
+ }
+ if ("Clustering".equalsIgnoreCase(messageModel)) {
+ return ConsumeType.CLUSTERING;
+ }
+ if ("Broadcasting".equalsIgnoreCase(messageModel)) {
+ return ConsumeType.BROADCASTING;
+ }
+ return null;
+ }
+
+ static List<QueueProgressVO>
toQueueProgressRows(GetConsumerGroupLagResponseBody.Data data) {
+ List<QueueProgressVO> rows = new ArrayList<>();
+ Map<String, DataTopicLagMapValue> topicLagMap = data.getTopicLagMap();
+ if (topicLagMap != null) {
+ for (Map.Entry<String, DataTopicLagMapValue> entry :
topicLagMap.entrySet()) {
+ long ready = entry.getValue() == null ||
entry.getValue().getReadyCount() == null
+ ? 0L : entry.getValue().getReadyCount();
+ rows.add(QueueProgressVO.builder()
+ .broker("topic:" + entry.getKey())
+ .queueId(0)
+ .brokerOffset(0L)
+ .consumerOffset(0L)
+ .diffTotal(ready)
+ .build());
+ }
+ }
+ GetConsumerGroupLagResponseBody.TotalLag totalLag = data.getTotalLag();
+ if (totalLag != null && totalLag.getReadyCount() != null) {
+ rows.add(QueueProgressVO.builder()
+ .broker("total")
+ .queueId(0)
+ .brokerOffset(0L)
+ .consumerOffset(0L)
+ .diffTotal(totalLag.getReadyCount())
+ .build());
+ }
+ return rows;
+ }
+
+ static SubscriptionEntryVO
toSubscriptionEntry(ListConsumerGroupSubscriptionsResponseBody.Data data) {
+ return SubscriptionEntryVO.builder()
+ .topic(data.getTopicName())
+ .expression(data.getFilterExpression())
+ .type(data.getFilterExpressionType())
+ .consistency(data.getConsistency() == null ? null :
String.valueOf(data.getConsistency()))
+ .build();
+ }
+
+ static MessageRecordVO toMessageRecord(ListMessagesResponseBody.List data)
{
+ String rawBody = data.getBody();
+ String decodedBody = tryBase64Decode(rawBody);
+ MessageRecordVO.MessageRecordVOBuilder builder =
MessageRecordVO.builder()
+ .msgId(data.getMessageId())
+ .topic(data.getTopicName())
+ .tag(data.getMessageTag())
+ .key(data.getMessageKeys() == null ? null : String.join(" ",
data.getMessageKeys()))
+ .bornHost(data.getBornHost())
+ .storeHost(data.getStoreHost())
+ .storeTime(parseTimeMillis(data.getStoreTime()))
+ .properties(data.getUserProperties())
+ .size(data.getBodySize() == null ? 0 : data.getBodySize());
+ if (decodedBody != null) {
+ builder.body(decodedBody).bodyEncoding("UTF-8");
+ } else {
+ builder.body(rawBody).bodyEncoding("TEXT");
+ }
+ return builder.build();
+ }
+
+ static TraceRecordVO toTraceRecord(GetTraceResponseBody.Data data) {
+ List<TraceNodeVO> nodes = new ArrayList<>();
+ if (data.getProducerInfo() != null &&
data.getProducerInfo().getRecords() != null) {
+ for (GetTraceResponseBody.ProducerInfoRecords record :
data.getProducerInfo().getRecords()) {
+ nodes.add(TraceNodeVO.builder()
+ .title("Producer")
+ .timestamp(parseTimeMillis(record.getProduceTime()))
+ .status(record.getProduceStatus())
+ .costTime(record.getProduceDuration() == null ? 0L :
record.getProduceDuration())
+ .description(joinParts(", ", record.getClientHost(),
record.getMessageSource()))
+ .build());
+ }
+ }
+ if (data.getBrokerInfo() != null &&
data.getBrokerInfo().getOperations() != null) {
+ for (GetTraceResponseBody.Operations operation :
data.getBrokerInfo().getOperations()) {
+ nodes.add(TraceNodeVO.builder()
+ .title("Broker " + operation.getOperateType())
+ .timestamp(parseTimeMillis(operation.getOperateTime()))
+ .build());
+ }
+ }
+ if (data.getConsumerInfos() != null) {
+ for (GetTraceResponseBody.ConsumerInfos consumerInfo :
data.getConsumerInfos()) {
+ if (consumerInfo.getRecords() == null ||
consumerInfo.getRecords().isEmpty()) {
+ nodes.add(TraceNodeVO.builder()
+ .title("Consumer " +
consumerInfo.getConsumerGroupId())
+ .status(consumerInfo.getConsumeStatus())
+ .build());
+ continue;
+ }
+ for (GetTraceResponseBody.Records record :
consumerInfo.getRecords()) {
+ String operateTime = null;
+ if (record.getOperations() != null &&
!record.getOperations().isEmpty()) {
+ operateTime =
record.getOperations().get(0).getOperateTime();
+ }
+ nodes.add(TraceNodeVO.builder()
+ .title("Consumer " +
consumerInfo.getConsumerGroupId())
+ .timestamp(parseTimeMillis(operateTime))
+ .status(record.getConsumeStatus())
+ .description(joinParts(", ",
record.getClientHost(), record.getUserName()))
+ .build());
+ }
+ }
+ }
+ return TraceRecordVO.builder().nodes(nodes).build();
+ }
+
+ static java.time.LocalDateTime parseDateTime(String value) {
+ if (value == null || value.isBlank()) {
+ return null;
+ }
+ try {
+ return java.time.LocalDateTime.parse(value, TIME_FORMATTER);
+ } catch (RuntimeException ex) {
+ return null;
+ }
+ }
+
+ static long parseTimeMillis(String value) {
+ if (value == null || value.isBlank()) {
+ return 0L;
+ }
+ try {
+ LocalDateTime dateTime = LocalDateTime.parse(value,
TIME_FORMATTER);
+ return
dateTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
+ } catch (RuntimeException ignored) {
+ return 0L;
+ }
+ }
+
+ static String formatTimeMillis(long epochMillis) {
+ return TIME_FORMATTER.format(
+ LocalDateTime.ofInstant(Instant.ofEpochMilli(epochMillis),
ZoneId.systemDefault()));
+ }
+
+ static String tryBase64Decode(String raw) {
+ if (raw == null || raw.isBlank()) {
+ return null;
+ }
+ byte[] bytes;
+ try {
+ bytes = Base64.getDecoder().decode(raw);
+ } catch (IllegalArgumentException ignored) {
+ return null;
+ }
+ try {
+ return StandardCharsets.UTF_8.newDecoder()
+ .onMalformedInput(CodingErrorAction.REPORT)
+ .onUnmappableCharacter(CodingErrorAction.REPORT)
+ .decode(ByteBuffer.wrap(bytes))
+ .toString();
+ } catch (CharacterCodingException ignored) {
+ return null;
+ }
+ }
+
+ private static String joinParts(String separator, String... parts) {
+ StringBuilder sb = new StringBuilder();
+ for (String part : parts) {
+ if (part == null || part.isBlank()) {
+ continue;
+ }
+ if (sb.length() > 0) {
+ sb.append(separator);
+ }
+ sb.append(part);
+ }
+ return sb.length() == 0 ? null : sb.toString();
+ }
+
+ private static Integer toInteger(Long value) {
+ return value == null ? null : value.intValue();
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProvider.java
new file mode 100644
index 00000000..04fb8788
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProvider.java
@@ -0,0 +1,435 @@
+/*
+ * 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.alibaba;
+
+import
com.aliyun.sdk.service.rocketmq20220801.models.CreateConsumerGroupRequest;
+import com.aliyun.sdk.service.rocketmq20220801.models.CreateTopicRequest;
+import
com.aliyun.sdk.service.rocketmq20220801.models.DeleteConsumerGroupRequest;
+import com.aliyun.sdk.service.rocketmq20220801.models.DeleteTopicRequest;
+import
com.aliyun.sdk.service.rocketmq20220801.models.GetConsumerGroupLagRequest;
+import
com.aliyun.sdk.service.rocketmq20220801.models.GetConsumerGroupLagResponse;
+import
com.aliyun.sdk.service.rocketmq20220801.models.GetConsumerGroupLagResponseBody;
+import com.aliyun.sdk.service.rocketmq20220801.models.GetTraceRequest;
+import com.aliyun.sdk.service.rocketmq20220801.models.GetTraceResponse;
+import com.aliyun.sdk.service.rocketmq20220801.models.GetTraceResponseBody;
+import
com.aliyun.sdk.service.rocketmq20220801.models.ListConsumerGroupSubscriptionsRequest;
+import
com.aliyun.sdk.service.rocketmq20220801.models.ListConsumerGroupSubscriptionsResponse;
+import
com.aliyun.sdk.service.rocketmq20220801.models.ListConsumerGroupSubscriptionsResponseBody;
+import
com.aliyun.sdk.service.rocketmq20220801.models.ListConsumerGroupsRequest;
+import
com.aliyun.sdk.service.rocketmq20220801.models.ListConsumerGroupsResponse;
+import
com.aliyun.sdk.service.rocketmq20220801.models.ListConsumerGroupsResponseBody;
+import com.aliyun.sdk.service.rocketmq20220801.models.ListMessagesRequest;
+import com.aliyun.sdk.service.rocketmq20220801.models.ListMessagesResponse;
+import com.aliyun.sdk.service.rocketmq20220801.models.ListMessagesResponseBody;
+import
com.aliyun.sdk.service.rocketmq20220801.models.ListTopicSubscriptionsRequest;
+import
com.aliyun.sdk.service.rocketmq20220801.models.ListTopicSubscriptionsResponse;
+import
com.aliyun.sdk.service.rocketmq20220801.models.ListTopicSubscriptionsResponseBody;
+import com.aliyun.sdk.service.rocketmq20220801.models.ListTopicsRequest;
+import com.aliyun.sdk.service.rocketmq20220801.models.ListTopicsResponse;
+import com.aliyun.sdk.service.rocketmq20220801.models.ListTopicsResponseBody;
+import
com.aliyun.sdk.service.rocketmq20220801.models.ResetConsumeOffsetRequest;
+import com.aliyun.sdk.service.rocketmq20220801.models.UpdateTopicRequest;
+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.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.ArrayList;
+import java.util.List;
+
+/**
+ * Aliyun RocketMQ 5.x implementation of the instance-scoped operations SPI,
backed by the
+ * OpenAPI async SDK through {@link AliyunClientFactory}.
+ */
+@Component
+public class AliyunInstanceProvider implements InstanceProvider {
+
+ private static final String DEFAULT_DELIVERY_ORDER_TYPE = "Concurrently";
+ private static final String ORDERLY_DELIVERY_ORDER_TYPE = "Orderly";
+ private static final String DEFAULT_RETRY_POLICY = "DefaultRetryPolicy";
+ private static final String FIXED_RETRY_POLICY = "FixedRetryPolicy";
+ private static final int DEFAULT_MAX_RETRY_TIMES = 16;
+ private static final int DEFAULT_FIXED_RETRY_INTERVAL_SECONDS = 10;
+ private static final String RESET_TYPE_SPECIFIED_TIME = "SPECIFIED_TIME";
+ private static final String RESET_TYPE_LATEST_OFFSET = "LATEST_OFFSET";
+
+ private final AliyunClientFactory clientFactory;
+ private final InstanceRepository instanceRepository;
+
+ public AliyunInstanceProvider(AliyunClientFactory clientFactory,
InstanceRepository instanceRepository) {
+ this.clientFactory = clientFactory;
+ this.instanceRepository = instanceRepository;
+ }
+
+ @Override
+ public InstanceVendor vendor() {
+ return InstanceVendor.ALIYUN;
+ }
+
+ @Override
+ public List<TopicVO> listTopics(String instanceId, String type, String
search) {
+ Context ctx = resolve(instanceId);
+ List<ListTopicsResponseBody.List> all = new ArrayList<>();
+ for (int page = 1; page <= AliyunConverters.MAX_PAGES; page++) {
+ ListTopicsRequest.Builder builder = ListTopicsRequest.builder()
+ .instanceId(ctx.cloudInstanceId())
+ .pageNumber(page)
+ .pageSize(AliyunConverters.PAGE_SIZE);
+ if (!isBlank(search)) {
+ builder.filter(search);
+ }
+ ListTopicsRequest request = builder.build();
+ ListTopicsResponse response =
clientFactory.call(ctx.credentialId(), ctx.regionId(),
+ client -> client.listTopics(request));
+ ListTopicsResponseBody body = response == null ? null :
response.getBody();
+ ListTopicsResponseBody.Data data = body == null ? null :
body.getData();
+ List<ListTopicsResponseBody.List> list = data == null ? null :
data.getList();
+ if (list == null || list.isEmpty()) {
+ break;
+ }
+ all.addAll(list);
+ if (list.size() < AliyunConverters.PAGE_SIZE) {
+ break;
+ }
+ }
+ List<TopicVO> topics = new ArrayList<>();
+ for (ListTopicsResponseBody.List item : all) {
+ TopicVO vo = AliyunConverters.toTopicVO(item, instanceId);
+ if (matchesType(type, vo)) {
+ topics.add(vo);
+ }
+ }
+ return topics;
+ }
+
+ @Override
+ public TopicVO createTopic(String instanceId, TopicVO topic) {
+ Context ctx = resolve(instanceId);
+ if (topic == null || isBlank(topic.getName())) {
+ throw new BusinessException(400, "Topic name is required");
+ }
+ if (topic.getType() == null) {
+ throw new BusinessException(400, "Topic type is required");
+ }
+ CreateTopicRequest request = CreateTopicRequest.builder()
+ .instanceId(ctx.cloudInstanceId())
+ .topicName(topic.getName())
+ .messageType(topic.getType().name())
+ .remark(topic.getRemark())
+ .build();
+ clientFactory.call(ctx.credentialId(), ctx.regionId(), client ->
client.createTopic(request));
+ topic.setInstanceId(instanceId);
+ topic.setCreatedAt(java.time.LocalDateTime.now());
+ topic.setUpdatedAt(java.time.LocalDateTime.now());
+ return topic;
+ }
+
+ @Override
+ public TopicVO updateTopic(String instanceId, TopicVO topic) {
+ Context ctx = resolve(instanceId);
+ if (topic == null || isBlank(topic.getName())) {
+ throw new BusinessException(400, "Topic name is required");
+ }
+ UpdateTopicRequest request = UpdateTopicRequest.builder()
+ .instanceId(ctx.cloudInstanceId())
+ .topicName(topic.getName())
+ .remark(topic.getRemark())
+ .build();
+ clientFactory.call(ctx.credentialId(), ctx.regionId(), client ->
client.updateTopic(request));
+ topic.setInstanceId(instanceId);
+ return topic;
+ }
+
+ @Override
+ public void deleteTopic(String instanceId, String topicName) {
+ Context ctx = resolve(instanceId);
+ DeleteTopicRequest request = DeleteTopicRequest.builder()
+ .instanceId(ctx.cloudInstanceId())
+ .topicName(topicName)
+ .build();
+ clientFactory.call(ctx.credentialId(), ctx.regionId(), client ->
client.deleteTopic(request));
+ }
+
+ @Override
+ public List<TopicConsumerVO> getTopicConsumers(String instanceId, String
topicName) {
+ Context ctx = resolve(instanceId);
+ ListTopicSubscriptionsRequest request =
ListTopicSubscriptionsRequest.builder()
+ .instanceId(ctx.cloudInstanceId())
+ .topicName(topicName)
+ .build();
+ ListTopicSubscriptionsResponse response =
clientFactory.call(ctx.credentialId(), ctx.regionId(),
+ client -> client.listTopicSubscriptions(request));
+ ListTopicSubscriptionsResponseBody body = response == null ? null :
response.getBody();
+ List<ListTopicSubscriptionsResponseBody.Data> data = body == null ?
null : body.getData();
+ List<TopicConsumerVO> consumers = new ArrayList<>();
+ if (data == null) {
+ return consumers;
+ }
+ for (ListTopicSubscriptionsResponseBody.Data item : data) {
+ consumers.add(AliyunConverters.toTopicConsumerVO(item));
+ }
+ return consumers;
+ }
+
+ @Override
+ public List<ConsumerGroupVO> listConsumerGroups(String instanceId, String
search) {
+ Context ctx = resolve(instanceId);
+ List<ListConsumerGroupsResponseBody.List> all = new ArrayList<>();
+ for (int page = 1; page <= AliyunConverters.MAX_PAGES; page++) {
+ ListConsumerGroupsRequest.Builder builder =
ListConsumerGroupsRequest.builder()
+ .instanceId(ctx.cloudInstanceId())
+ .pageNumber(page)
+ .pageSize(AliyunConverters.PAGE_SIZE);
+ if (!isBlank(search)) {
+ builder.filter(search);
+ }
+ ListConsumerGroupsRequest request = builder.build();
+ ListConsumerGroupsResponse response =
clientFactory.call(ctx.credentialId(), ctx.regionId(),
+ client -> client.listConsumerGroups(request));
+ ListConsumerGroupsResponseBody body = response == null ? null :
response.getBody();
+ ListConsumerGroupsResponseBody.Data data = body == null ? null :
body.getData();
+ List<ListConsumerGroupsResponseBody.List> list = data == null ?
null : data.getList();
+ if (list == null || list.isEmpty()) {
+ break;
+ }
+ all.addAll(list);
+ if (list.size() < AliyunConverters.PAGE_SIZE) {
+ break;
+ }
+ }
+ List<ConsumerGroupVO> groups = new ArrayList<>();
+ for (ListConsumerGroupsResponseBody.List item : all) {
+ groups.add(AliyunConverters.toConsumerGroupVO(item, instanceId));
+ }
+ return groups;
+ }
+
+ @Override
+ public ConsumerGroupVO createConsumerGroup(String instanceId,
ConsumerGroupVO group) {
+ Context ctx = resolve(instanceId);
+ if (group == null || isBlank(group.getName())) {
+ throw new BusinessException(400, "Consumer group name is
required");
+ }
+ String deliveryOrderType =
normalizeDeliveryOrderType(group.getDeliveryOrderType());
+ int maxRetryTimes = group.getRetryMaxTimes() > 0 ?
group.getRetryMaxTimes() : DEFAULT_MAX_RETRY_TIMES;
+ CreateConsumerGroupRequest.ConsumeRetryPolicy.Builder retryPolicy =
+ CreateConsumerGroupRequest.ConsumeRetryPolicy.builder()
+ .maxRetryTimes(maxRetryTimes);
+ if (ORDERLY_DELIVERY_ORDER_TYPE.equals(deliveryOrderType)) {
+ // ordered groups reject DefaultRetryPolicy and require a fixed
interval
+ retryPolicy.retryPolicy(FIXED_RETRY_POLICY)
+
.fixedIntervalRetryTime(DEFAULT_FIXED_RETRY_INTERVAL_SECONDS);
+ } else {
+ retryPolicy.retryPolicy(DEFAULT_RETRY_POLICY);
+ }
+ CreateConsumerGroupRequest request =
CreateConsumerGroupRequest.builder()
+ .instanceId(ctx.cloudInstanceId())
+ .consumerGroupId(group.getName())
+ .deliveryOrderType(deliveryOrderType)
+ .consumeRetryPolicy(retryPolicy.build())
+ .build();
+ clientFactory.call(ctx.credentialId(), ctx.regionId(), client ->
client.createConsumerGroup(request));
+ group.setInstanceId(instanceId);
+ group.setDeliveryOrderType(deliveryOrderType);
+ group.setRetryMaxTimes(maxRetryTimes);
+ group.setSubscribedTopics(java.util.List.of());
+ group.setCreatedAt(java.time.LocalDateTime.now());
+ group.setUpdatedAt(java.time.LocalDateTime.now());
+ return group;
+ }
+
+ /**
+ * OpenAPI accepts Concurrently/Orderly; tolerate FIFO/ordered spellings
from the UI.
+ */
+ static String normalizeDeliveryOrderType(String raw) {
+ if (raw == null || raw.isBlank()) {
+ return DEFAULT_DELIVERY_ORDER_TYPE;
+ }
+ String value = raw.trim();
+ if ("FIFO".equalsIgnoreCase(value) ||
"ORDERLY".equalsIgnoreCase(value)) {
+ return ORDERLY_DELIVERY_ORDER_TYPE;
+ }
+ return DEFAULT_DELIVERY_ORDER_TYPE;
+ }
+
+ @Override
+ public void deleteConsumerGroup(String instanceId, String groupName) {
+ Context ctx = resolve(instanceId);
+ DeleteConsumerGroupRequest request =
DeleteConsumerGroupRequest.builder()
+ .instanceId(ctx.cloudInstanceId())
+ .consumerGroupId(groupName)
+ .build();
+ clientFactory.call(ctx.credentialId(), ctx.regionId(), client ->
client.deleteConsumerGroup(request));
+ }
+
+ @Override
+ public List<QueueProgressVO> getGroupProgress(String instanceId, String
groupName) {
+ Context ctx = resolve(instanceId);
+ GetConsumerGroupLagRequest request =
GetConsumerGroupLagRequest.builder()
+ .instanceId(ctx.cloudInstanceId())
+ .consumerGroupId(groupName)
+ .build();
+ GetConsumerGroupLagResponse response =
clientFactory.call(ctx.credentialId(), ctx.regionId(),
+ client -> client.getConsumerGroupLag(request));
+ GetConsumerGroupLagResponseBody body = response == null ? null :
response.getBody();
+ GetConsumerGroupLagResponseBody.Data data = body == null ? null :
body.getData();
+ if (data == null) {
+ return new ArrayList<>();
+ }
+ return AliyunConverters.toQueueProgressRows(data);
+ }
+
+ @Override
+ public List<SubscriptionEntryVO> getGroupSubscriptions(String instanceId,
String groupName) {
+ Context ctx = resolve(instanceId);
+ ListConsumerGroupSubscriptionsRequest request =
ListConsumerGroupSubscriptionsRequest.builder()
+ .instanceId(ctx.cloudInstanceId())
+ .consumerGroupId(groupName)
+ .build();
+ ListConsumerGroupSubscriptionsResponse response =
clientFactory.call(ctx.credentialId(), ctx.regionId(),
+ client -> client.listConsumerGroupSubscriptions(request));
+ ListConsumerGroupSubscriptionsResponseBody body = response == null ?
null : response.getBody();
+ List<ListConsumerGroupSubscriptionsResponseBody.Data> data = body ==
null ? null : body.getData();
+ List<SubscriptionEntryVO> subscriptions = new ArrayList<>();
+ if (data == null) {
+ return subscriptions;
+ }
+ for (ListConsumerGroupSubscriptionsResponseBody.Data item : data) {
+ subscriptions.add(AliyunConverters.toSubscriptionEntry(item));
+ }
+ return subscriptions;
+ }
+
+ @Override
+ public void resetOffset(String instanceId, String groupName, long
timestamp, String topic) {
+ Context ctx = resolve(instanceId);
+ ResetConsumeOffsetRequest.Builder builder =
ResetConsumeOffsetRequest.builder()
+ .instanceId(ctx.cloudInstanceId())
+ .consumerGroupId(groupName);
+ if (!isBlank(topic)) {
+ builder.topicName(topic);
+ }
+ if (timestamp > 0L) {
+ builder.resetType(RESET_TYPE_SPECIFIED_TIME)
+ .resetTime(AliyunConverters.formatTimeMillis(timestamp));
+ } else {
+ builder.resetType(RESET_TYPE_LATEST_OFFSET);
+ }
+ ResetConsumeOffsetRequest request = builder.build();
+ clientFactory.call(ctx.credentialId(), ctx.regionId(), client ->
client.resetConsumeOffset(request));
+ }
+
+ @Override
+ public List<MessageRecordVO> queryMessages(String instanceId, String
topic, String msgId,
+ String tag, String key, Long
startTime, Long endTime) {
+ Context ctx = resolve(instanceId);
+ List<MessageRecordVO> records = new ArrayList<>();
+ for (int page = 1; page <= AliyunConverters.MESSAGE_MAX_PAGES; page++)
{
+ ListMessagesRequest.Builder builder = ListMessagesRequest.builder()
+ .instanceId(ctx.cloudInstanceId())
+ .pageNumber(page)
+ .pageSize(AliyunConverters.MESSAGE_PAGE_SIZE);
+ if (!isBlank(topic)) {
+ builder.topicName(topic);
+ }
+ if (!isBlank(msgId)) {
+ builder.messageId(msgId);
+ }
+ if (!isBlank(key)) {
+ builder.messageKey(key);
+ }
+ if (startTime != null) {
+
builder.startTime(AliyunConverters.formatTimeMillis(startTime));
+ }
+ if (endTime != null) {
+ builder.endTime(AliyunConverters.formatTimeMillis(endTime));
+ }
+ ListMessagesRequest request = builder.build();
+ ListMessagesResponse response =
clientFactory.call(ctx.credentialId(), ctx.regionId(),
+ client -> client.listMessages(request));
+ ListMessagesResponseBody body = response == null ? null :
response.getBody();
+ ListMessagesResponseBody.Data data = body == null ? null :
body.getData();
+ List<ListMessagesResponseBody.List> list = data == null ? null :
data.getList();
+ if (list == null || list.isEmpty()) {
+ break;
+ }
+ for (ListMessagesResponseBody.List item : list) {
+ MessageRecordVO vo = AliyunConverters.toMessageRecord(item);
+ if (isBlank(tag) || tag.equals(vo.getTag())) {
+ records.add(vo);
+ }
+ }
+ if (list.size() < AliyunConverters.MESSAGE_PAGE_SIZE) {
+ break;
+ }
+ }
+ return records;
+ }
+
+ @Override
+ public TraceRecordVO getMessageTrace(String instanceId, String msgId) {
+ Context ctx = resolve(instanceId);
+ GetTraceRequest request = GetTraceRequest.builder()
+ .instanceId(ctx.cloudInstanceId())
+ .messageId(msgId)
+ .build();
+ GetTraceResponse response = clientFactory.call(ctx.credentialId(),
ctx.regionId(),
+ client -> client.getTrace(request));
+ GetTraceResponseBody body = response == null ? null :
response.getBody();
+ GetTraceResponseBody.Data data = body == null ? null : body.getData();
+ if (data == null) {
+ throw new BusinessException(404, "Message trace not found: " +
msgId);
+ }
+ return AliyunConverters.toTraceRecord(data);
+ }
+
+ private Context resolve(String instanceId) {
+ if (isBlank(instanceId)) {
+ throw new BusinessException(400, "instanceId is required");
+ }
+ InstanceVO instance = instanceRepository.findById(instanceId)
+ .orElseThrow(() -> new BusinessException(404, "Instance not
found: " + instanceId));
+ if (isBlank(instance.getCloudInstanceId()) ||
isBlank(instance.getRegionId())
+ || isBlank(instance.getCredentialId())) {
+ throw new BusinessException(400, "Instance " + instanceId + " is
missing Aliyun cloud binding");
+ }
+ return new Context(instance.getCloudInstanceId(),
instance.getRegionId(), instance.getCredentialId());
+ }
+
+ private static boolean matchesType(String type, TopicVO vo) {
+ if (isBlank(type)) {
+ return true;
+ }
+ return vo.getType() != null &&
vo.getType().name().equalsIgnoreCase(type.trim());
+ }
+
+ private static boolean isBlank(String value) {
+ return value == null || value.isBlank();
+ }
+
+ private record Context(String cloudInstanceId, String regionId, String
credentialId) {
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunCatalogServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunCatalogServiceTest.java
new file mode 100644
index 00000000..cacd8388
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunCatalogServiceTest.java
@@ -0,0 +1,213 @@
+/*
+ * 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.alibaba;
+
+import com.aliyun.sdk.service.rocketmq20220801.models.GetInstanceResponse;
+import com.aliyun.sdk.service.rocketmq20220801.models.GetInstanceResponseBody;
+import com.aliyun.sdk.service.rocketmq20220801.models.ListInstancesResponse;
+import
com.aliyun.sdk.service.rocketmq20220801.models.ListInstancesResponseBody;
+import com.aliyun.sdk.service.rocketmq20220801.models.ListRegionsResponse;
+import com.aliyun.sdk.service.rocketmq20220801.models.ListRegionsResponseBody;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.provider.CloudInstanceDetailVO;
+import org.apache.rocketmq.studio.provider.CloudInstanceOptionVO;
+import org.apache.rocketmq.studio.provider.CloudRegionVO;
+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.ArrayList;
+import java.util.List;
+
+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.ArgumentMatchers.eq;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class AliyunCatalogServiceTest {
+
+ private static final String CREDENTIAL_ID = "cred-1";
+ private static final String REGION = "cn-hangzhou";
+
+ @Mock
+ private AliyunClientFactory clientFactory;
+
+ private AliyunCatalogService service;
+
+ @BeforeEach
+ void setUp() {
+ service = new AliyunCatalogService(clientFactory);
+ }
+
+ @Test
+ void listRegionsShouldKeepOnlyRocketmqV5RegionsTest() {
+ ListRegionsResponse response = ListRegionsResponse.create().toBuilder()
+ .statusCode(200)
+ .body(ListRegionsResponseBody.builder()
+ .data(List.of(
+ ListRegionsResponseBody.Data.builder()
+
.regionId("cn-shanghai").regionName("shanghai")
+ .supportRocketmqV5(true).build(),
+ ListRegionsResponseBody.Data.builder()
+
.regionId("cn-hangzhou").regionName("hangzhou")
+ .supportRocketmqV5(true).build(),
+ ListRegionsResponseBody.Data.builder()
+
.regionId("cn-beijing").regionName("beijing")
+ .supportRocketmqV5(true).build(),
+ ListRegionsResponseBody.Data.builder()
+
.regionId("cn-legacy").regionName("legacy")
+ .supportRocketmqV5(false).build()))
+ .build())
+ .build();
+ when(clientFactory.call(eq(CREDENTIAL_ID),
eq(AliyunCatalogService.DEFAULT_REGION), any()))
+ .thenReturn(response);
+
+ List<CloudRegionVO> regions = service.listRegions(CREDENTIAL_ID);
+
+ assertThat(regions).hasSize(3);
+ assertThat(regions).extracting(CloudRegionVO::getRegionId)
+ .containsExactly("cn-beijing", "cn-hangzhou", "cn-shanghai");
+ assertThat(regions.get(1).getRegionName()).isEqualTo("hangzhou");
+ }
+
+ @Test
+ void listCloudInstancesShouldAggregatePagesTest() {
+ ListInstancesResponse firstPage =
instancesResponse(instanceRows(AliyunConverters.PAGE_SIZE, 0));
+ ListInstancesResponse secondPage = instancesResponse(instanceRows(3,
AliyunConverters.PAGE_SIZE));
+ when(clientFactory.call(eq(CREDENTIAL_ID), eq(REGION), any()))
+ .thenReturn(firstPage, secondPage);
+
+ List<CloudInstanceOptionVO> options =
service.listCloudInstances(CREDENTIAL_ID, REGION, null);
+
+ assertThat(options).hasSize(AliyunConverters.PAGE_SIZE + 3);
+ verify(clientFactory, times(2)).call(eq(CREDENTIAL_ID), eq(REGION),
any());
+ }
+
+ @Test
+ void listCloudInstancesShouldStopAtMaxPagesTest() {
+ ListInstancesResponse fullPage =
instancesResponse(instanceRows(AliyunConverters.PAGE_SIZE, 0));
+ when(clientFactory.call(eq(CREDENTIAL_ID), eq(REGION),
any())).thenReturn(fullPage);
+
+ List<CloudInstanceOptionVO> options =
service.listCloudInstances(CREDENTIAL_ID, REGION, null);
+
+ assertThat(options).hasSize(AliyunConverters.PAGE_SIZE *
AliyunConverters.MAX_PAGES);
+ verify(clientFactory,
times(AliyunConverters.MAX_PAGES)).call(eq(CREDENTIAL_ID), eq(REGION), any());
+ }
+
+ @Test
+ void listCloudInstancesShouldApplySearchFilterTest() {
+ List<ListInstancesResponseBody.List> rows = new ArrayList<>();
+ rows.add(instanceRow("rmq-prod-001", "Production"));
+ rows.add(instanceRow("rmq-test-002", "Staging"));
+ when(clientFactory.call(eq(CREDENTIAL_ID), eq(REGION), any()))
+ .thenReturn(instancesResponse(rows));
+
+ List<CloudInstanceOptionVO> byId =
service.listCloudInstances(CREDENTIAL_ID, REGION, "TEST-002");
+ List<CloudInstanceOptionVO> byName =
service.listCloudInstances(CREDENTIAL_ID, REGION, "staging");
+
+ assertThat(byId).hasSize(1);
+ assertThat(byId.get(0).getInstanceId()).isEqualTo("rmq-test-002");
+ assertThat(byName).hasSize(1);
+ assertThat(byName.get(0).getInstanceName()).isEqualTo("Staging");
+ }
+
+ @Test
+ void listCloudInstancesShouldRequireRegionTest() {
+ assertThatThrownBy(() -> service.listCloudInstances(CREDENTIAL_ID, "
", null))
+ .isInstanceOf(BusinessException.class)
+ .extracting("code")
+ .isEqualTo(400);
+ }
+
+ @Test
+ void getCloudInstanceShouldMapEndpointsTest() {
+ GetInstanceResponse response = GetInstanceResponse.create().toBuilder()
+ .statusCode(200)
+ .body(GetInstanceResponseBody.builder()
+ .data(GetInstanceResponseBody.Data.builder()
+ .instanceId("rmq-cn-001")
+ .instanceName("prod")
+ .status("RUNNING")
+ .regionId(REGION)
+ .remark("prod instance")
+
.networkInfo(GetInstanceResponseBody.NetworkInfo.builder()
+ .endpoints(List.of(
+
GetInstanceResponseBody.Endpoints.builder()
+
.endpointType("TCP_INTERNET")
+
.endpointUrl("rmq-cn-001.rmq.aliyuncs.com:8080")
+ .build(),
+
GetInstanceResponseBody.Endpoints.builder()
+
.endpointType("TCP_VPC")
+
.endpointUrl("rmq-cn-001-vpc.rmq.aliyuncs.com:8080")
+ .build()))
+ .build())
+ .build())
+ .build())
+ .build();
+ when(clientFactory.call(eq(CREDENTIAL_ID), eq(REGION),
any())).thenReturn(response);
+
+ CloudInstanceDetailVO detail = service.getCloudInstance(CREDENTIAL_ID,
REGION, "rmq-cn-001");
+
+ assertThat(detail.getInstanceId()).isEqualTo("rmq-cn-001");
+ assertThat(detail.getStatus()).isEqualTo("RUNNING");
+ assertThat(detail.getRemark()).isEqualTo("prod instance");
+ assertThat(detail.getEndpoints()).hasSize(2);
+
assertThat(detail.getEndpoints().get(0).getEndpointType()).isEqualTo("TCP_INTERNET");
+ assertThat(detail.getEndpoints().get(1).getEndpointUrl())
+ .isEqualTo("rmq-cn-001-vpc.rmq.aliyuncs.com:8080");
+ }
+
+ private static List<ListInstancesResponseBody.List> instanceRows(int
count, int idOffset) {
+ List<ListInstancesResponseBody.List> rows = new ArrayList<>();
+ for (int i = 0; i < count; i++) {
+ rows.add(instanceRow(String.format("rmq-instance-%04d", idOffset +
i), "instance-" + i));
+ }
+ return rows;
+ }
+
+ private static ListInstancesResponseBody.List instanceRow(String
instanceId, String instanceName) {
+ return ListInstancesResponseBody.List.builder()
+ .instanceId(instanceId)
+ .instanceName(instanceName)
+ .status("RUNNING")
+ .regionId("cn-hangzhou")
+ .topicCount(3L)
+ .groupCount(2L)
+ .remark("remark-" + instanceId)
+ .build();
+ }
+
+ private static ListInstancesResponse
instancesResponse(List<ListInstancesResponseBody.List> rows) {
+ return ListInstancesResponse.create().toBuilder()
+ .statusCode(200)
+ .body(ListInstancesResponseBody.builder()
+ .data(ListInstancesResponseBody.Data.builder()
+ .list(rows)
+ .pageNumber(1L)
+ .pageSize((long) AliyunConverters.PAGE_SIZE)
+ .totalCount((long) rows.size())
+ .build())
+ .build())
+ .build();
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunClientFactoryTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunClientFactoryTest.java
new file mode 100644
index 00000000..766072a4
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunClientFactoryTest.java
@@ -0,0 +1,188 @@
+/*
+ * 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.alibaba;
+
+import com.aliyun.sdk.gateway.pop.exception.PopServerException;
+import com.aliyun.sdk.service.rocketmq20220801.AsyncClient;
+import com.aliyun.sdk.service.rocketmq20220801.models.ListRegionsRequest;
+import com.aliyun.sdk.service.rocketmq20220801.models.ListRegionsResponse;
+import com.aliyun.sdk.service.rocketmq20220801.models.ListRegionsResponseBody;
+import org.apache.rocketmq.studio.cloud.credential.CloudCredentialRepository;
+import org.apache.rocketmq.studio.cloud.credential.CloudCredentialVO;
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+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.Mockito;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+
+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.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class AliyunClientFactoryTest {
+
+ private static final String CREDENTIAL_ID = "cred-1";
+ private static final String REGION = "cn-hangzhou";
+
+ @Mock
+ private CloudCredentialRepository credentialRepository;
+
+ @Mock
+ private AsyncClient asyncClient;
+
+ private AliyunClientFactory factory;
+
+ @BeforeEach
+ void setUp() {
+ factory = new AliyunClientFactory(credentialRepository);
+ }
+
+ @Test
+ void endpointShouldFollowRegionTest() {
+ assertThat(AliyunClientFactory.endpointFor("cn-hangzhou"))
+ .isEqualTo("rocketmq.cn-hangzhou.aliyuncs.com");
+ assertThat(AliyunClientFactory.endpointFor("ap-southeast-1"))
+ .isEqualTo("rocketmq.ap-southeast-1.aliyuncs.com");
+ }
+
+ @Test
+ void clientShouldCachePerCredentialAndRegionTest() {
+
when(credentialRepository.findById(CREDENTIAL_ID)).thenReturn(Optional.of(credential()));
+
+ AsyncClient first = factory.client(CREDENTIAL_ID, REGION);
+ AsyncClient second = factory.client(CREDENTIAL_ID, REGION);
+ AsyncClient otherRegion = factory.client(CREDENTIAL_ID, "cn-beijing");
+
+ assertThat(first).isSameAs(second);
+ assertThat(otherRegion).isNotSameAs(first);
+ factory.close();
+ }
+
+ @Test
+ void clientShouldThrow404WhenCredentialMissingTest() {
+
when(credentialRepository.findById("missing")).thenReturn(Optional.empty());
+
+ assertThatThrownBy(() -> factory.client("missing", REGION))
+ .isInstanceOf(BusinessException.class)
+ .extracting("code")
+ .isEqualTo(404);
+ }
+
+ @Test
+ void callShouldThrow504OnTimeoutTest() {
+ factory.setCallTimeoutSeconds(1L);
+ AliyunClientFactory spy = Mockito.spy(factory);
+ doReturn(asyncClient).when(spy).client(anyString(), anyString());
+
+ assertThatThrownBy(() -> spy.call(CREDENTIAL_ID, REGION,
+ client -> new CompletableFuture<>()))
+ .isInstanceOf(BusinessException.class)
+ .extracting("code")
+ .isEqualTo(504);
+ }
+
+ @Test
+ void callShouldMapServer404ToBusinessExceptionTest() {
+ AliyunClientFactory spy = Mockito.spy(factory);
+ doReturn(asyncClient).when(spy).client(anyString(), anyString());
+ PopServerException error = new PopServerException("instance not
found");
+ error.setStatusCode(404);
+ error.setErrCode("Instance.NotFound");
+ when(asyncClient.listRegions(any(ListRegionsRequest.class)))
+ .thenReturn(CompletableFuture.failedFuture(error));
+
+ assertThatThrownBy(() -> spy.call(CREDENTIAL_ID, REGION, client ->
client.listRegions(
+ ListRegionsRequest.builder().build())))
+ .isInstanceOf(BusinessException.class)
+ .satisfies(ex -> {
+ BusinessException business = (BusinessException) ex;
+ assertThat(business.getCode()).isEqualTo(404);
+ assertThat(business.getMessage()).contains("instance not
found");
+ });
+ }
+
+ @Test
+ void callShouldMapInvalidAccessKeyTo401Test() {
+ AliyunClientFactory spy = Mockito.spy(factory);
+ doReturn(asyncClient).when(spy).client(anyString(), anyString());
+ PopServerException error = new PopServerException("bad key");
+ error.setStatusCode(403);
+ error.setErrCode("InvalidAccessKeyId");
+ when(asyncClient.listRegions(any(ListRegionsRequest.class)))
+ .thenReturn(CompletableFuture.failedFuture(error));
+
+ assertThatThrownBy(() -> spy.call(CREDENTIAL_ID, REGION, client ->
client.listRegions(
+ ListRegionsRequest.builder().build())))
+ .isInstanceOf(BusinessException.class)
+ .extracting("code")
+ .isEqualTo(401);
+ }
+
+ @Test
+ void callShouldMapGenericFailureTo502Test() {
+ AliyunClientFactory spy = Mockito.spy(factory);
+ doReturn(asyncClient).when(spy).client(anyString(), anyString());
+ when(asyncClient.listRegions(any(ListRegionsRequest.class)))
+ .thenReturn(CompletableFuture.failedFuture(new
IllegalStateException("boom")));
+
+ assertThatThrownBy(() -> spy.call(CREDENTIAL_ID, REGION, client ->
client.listRegions(
+ ListRegionsRequest.builder().build())))
+ .isInstanceOf(BusinessException.class)
+ .satisfies(ex -> {
+ BusinessException business = (BusinessException) ex;
+ assertThat(business.getCode()).isEqualTo(502);
+ assertThat(business.getMessage()).contains("boom");
+ });
+ }
+
+ @Test
+ void callShouldReturnSuccessfulBodyTest() {
+ AliyunClientFactory spy = Mockito.spy(factory);
+ doReturn(asyncClient).when(spy).client(anyString(), anyString());
+ ListRegionsResponse response = ListRegionsResponse.create().toBuilder()
+ .statusCode(200)
+ .body(ListRegionsResponseBody.builder().success(true).build())
+ .build();
+ when(asyncClient.listRegions(any(ListRegionsRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(response));
+
+ ListRegionsResponse result = spy.call(CREDENTIAL_ID, REGION,
+ client ->
client.listRegions(ListRegionsRequest.builder().build()));
+
+ assertThat(result.getBody().getSuccess()).isTrue();
+ }
+
+ private CloudCredentialVO credential() {
+ CloudCredentialVO credential = new CloudCredentialVO();
+ credential.setId(CREDENTIAL_ID);
+ credential.setName("unit-test");
+ credential.setVendor(InstanceVendor.ALIYUN);
+ credential.setAccessKey("LTAI5tUnitTestKey000000001");
+ credential.setSecretKey("UnitTestSecret000000000000000001");
+ return credential;
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProviderTest.java
new file mode 100644
index 00000000..330f1bcd
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunInstanceProviderTest.java
@@ -0,0 +1,440 @@
+/*
+ * 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.alibaba;
+
+import com.aliyun.sdk.service.rocketmq20220801.AsyncClient;
+import
com.aliyun.sdk.service.rocketmq20220801.models.CreateConsumerGroupRequest;
+import
com.aliyun.sdk.service.rocketmq20220801.models.CreateConsumerGroupResponse;
+import
com.aliyun.sdk.service.rocketmq20220801.models.CreateConsumerGroupResponseBody;
+import com.aliyun.sdk.service.rocketmq20220801.models.DataTopicLagMapValue;
+import
com.aliyun.sdk.service.rocketmq20220801.models.GetConsumerGroupLagResponse;
+import
com.aliyun.sdk.service.rocketmq20220801.models.GetConsumerGroupLagResponseBody;
+import com.aliyun.sdk.service.rocketmq20220801.models.GetTraceResponse;
+import com.aliyun.sdk.service.rocketmq20220801.models.GetTraceResponseBody;
+import
com.aliyun.sdk.service.rocketmq20220801.models.ListConsumerGroupsResponse;
+import
com.aliyun.sdk.service.rocketmq20220801.models.ListConsumerGroupsResponseBody;
+import com.aliyun.sdk.service.rocketmq20220801.models.ListMessagesResponse;
+import com.aliyun.sdk.service.rocketmq20220801.models.ListMessagesResponseBody;
+import com.aliyun.sdk.service.rocketmq20220801.models.ListTopicsRequest;
+import com.aliyun.sdk.service.rocketmq20220801.models.ListTopicsResponse;
+import com.aliyun.sdk.service.rocketmq20220801.models.ListTopicsResponseBody;
+import
com.aliyun.sdk.service.rocketmq20220801.models.ResetConsumeOffsetRequest;
+import
com.aliyun.sdk.service.rocketmq20220801.models.ResetConsumeOffsetResponse;
+import
com.aliyun.sdk.service.rocketmq20220801.models.ResetConsumeOffsetResponseBody;
+import org.apache.rocketmq.studio.common.domain.enums.ConsumeType;
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
+import org.apache.rocketmq.studio.common.domain.enums.TopicType;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.instance.InstanceRepository;
+import org.apache.rocketmq.studio.instance.InstanceVO;
+import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
+import org.apache.rocketmq.studio.instance.group.QueueProgressVO;
+import org.apache.rocketmq.studio.instance.message.MessageRecordVO;
+import org.apache.rocketmq.studio.instance.message.TraceNodeVO;
+import org.apache.rocketmq.studio.instance.message.TraceRecordVO;
+import org.apache.rocketmq.studio.instance.topic.TopicVO;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
+
+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.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class AliyunInstanceProviderTest {
+
+ private static final String STUDIO_INSTANCE_ID = "inst-1";
+ private static final String CLOUD_INSTANCE_ID = "rmq-cn-001";
+ private static final String REGION = "cn-hangzhou";
+ private static final String CREDENTIAL_ID = "cred-1";
+
+ @Mock
+ private AliyunClientFactory clientFactory;
+
+ @Mock
+ private InstanceRepository instanceRepository;
+
+ @Mock
+ private AsyncClient asyncClient;
+
+ private AliyunInstanceProvider provider;
+
+ @BeforeEach
+ void setUp() {
+ provider = new AliyunInstanceProvider(clientFactory,
instanceRepository);
+ }
+
+ @Test
+ void listTopicsShouldMapMessageTypeAndFilterTest() {
+ stubInstance();
+ stubCallThrough();
+ ListTopicsResponse response = topicsResponse(
+ topicRow("topic-normal", "NORMAL"),
+ topicRow("topic-fifo", "FIFO"),
+ topicRow("topic-mystery", "MYSTERY"));
+ when(asyncClient.listTopics(any(ListTopicsRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(response));
+
+ List<TopicVO> all = provider.listTopics(STUDIO_INSTANCE_ID, null,
null);
+
+ assertThat(all).hasSize(3);
+ assertThat(all.get(0).getName()).isEqualTo("topic-normal");
+ assertThat(all.get(0).getType()).isEqualTo(TopicType.NORMAL);
+ assertThat(all.get(0).getInstanceId()).isEqualTo(STUDIO_INSTANCE_ID);
+ assertThat(all.get(0).getWriteQueues()).isZero();
+ assertThat(all.get(0).getReadQueues()).isZero();
+ assertThat(all.get(0).getRemark()).isEqualTo("remark-topic-normal");
+ assertThat(all.get(2).getType()).isNull();
+
+ List<TopicVO> fifos = provider.listTopics(STUDIO_INSTANCE_ID, "FIFO",
null);
+
+ assertThat(fifos).hasSize(1);
+ assertThat(fifos.get(0).getType()).isEqualTo(TopicType.FIFO);
+ }
+
+ @Test
+ void listConsumerGroupsShouldMapGroupIdTest() {
+ stubInstance();
+ stubCallThrough();
+ ListConsumerGroupsResponse response =
ListConsumerGroupsResponse.create().toBuilder()
+ .statusCode(200)
+ .body(ListConsumerGroupsResponseBody.builder()
+ .data(ListConsumerGroupsResponseBody.Data.builder()
+
.list(List.of(ListConsumerGroupsResponseBody.List.builder()
+ .consumerGroupId("GID_test")
+ .messageModel("Clustering")
+ .status("RUNNING")
+ .remark("test group")
+ .build()))
+ .pageNumber(1L)
+ .pageSize(100L)
+ .totalCount(1L)
+ .build())
+ .build())
+ .build();
+ when(asyncClient.listConsumerGroups(any()))
+ .thenReturn(CompletableFuture.completedFuture(response));
+
+ List<ConsumerGroupVO> groups =
provider.listConsumerGroups(STUDIO_INSTANCE_ID, null);
+
+ assertThat(groups).hasSize(1);
+ assertThat(groups.get(0).getName()).isEqualTo("GID_test");
+
assertThat(groups.get(0).getInstanceId()).isEqualTo(STUDIO_INSTANCE_ID);
+
assertThat(groups.get(0).getConsumeType()).isEqualTo(ConsumeType.CLUSTERING);
+ }
+
+ @Test
+ void getGroupProgressShouldMapLagRowsTest() {
+ stubInstance();
+ stubCallThrough();
+ GetConsumerGroupLagResponse response =
GetConsumerGroupLagResponse.create().toBuilder()
+ .statusCode(200)
+ .body(GetConsumerGroupLagResponseBody.builder()
+ .data(GetConsumerGroupLagResponseBody.Data.builder()
+ .consumerGroupId("GID_test")
+ .topicLagMap(Map.of("topic-a",
+
DataTopicLagMapValue.builder().readyCount(42L).build()))
+
.totalLag(GetConsumerGroupLagResponseBody.TotalLag.builder()
+ .readyCount(100L)
+ .build())
+ .build())
+ .build())
+ .build();
+ when(asyncClient.getConsumerGroupLag(any()))
+ .thenReturn(CompletableFuture.completedFuture(response));
+
+ List<QueueProgressVO> rows =
provider.getGroupProgress(STUDIO_INSTANCE_ID, "GID_test");
+
+ assertThat(rows).hasSize(2);
+ QueueProgressVO topicRow = rows.stream()
+ .filter(row -> "topic:topic-a".equals(row.getBroker()))
+ .findFirst()
+ .orElseThrow();
+ assertThat(topicRow.getDiffTotal()).isEqualTo(42L);
+ QueueProgressVO totalRow = rows.stream()
+ .filter(row -> "total".equals(row.getBroker()))
+ .findFirst()
+ .orElseThrow();
+ assertThat(totalRow.getDiffTotal()).isEqualTo(100L);
+ }
+
+ @Test
+ void queryMessagesShouldMapFieldsAndDecodeBase64BodyTest() {
+ stubInstance();
+ stubCallThrough();
+ String encodedBody =
Base64.getEncoder().encodeToString("hello".getBytes(StandardCharsets.UTF_8));
+ ListMessagesResponse response =
ListMessagesResponse.create().toBuilder()
+ .statusCode(200)
+ .body(ListMessagesResponseBody.builder()
+ .data(ListMessagesResponseBody.Data.builder()
+ .list(List.of(
+ ListMessagesResponseBody.List.builder()
+ .messageId("msg-1")
+ .topicName("topic-a")
+ .messageTag("tagA")
+ .messageKeys(List.of("k1",
"k2"))
+ .body(encodedBody)
+ .bodySize(5)
+ .bornHost("10.0.0.1")
+ .storeHost("10.0.0.2")
+ .storeTime("2023-03-22
12:17:08")
+ .userProperties(Map.of("a",
"b"))
+ .build(),
+ ListMessagesResponseBody.List.builder()
+ .messageId("msg-2")
+ .topicName("topic-a")
+ .messageTag("tagB")
+ .body("{}")
+ .bodySize(2)
+ .build()))
+ .pageNumber(1L)
+ .pageSize(20L)
+ .totalCount(2L)
+ .build())
+ .build())
+ .build();
+ when(asyncClient.listMessages(any()))
+ .thenReturn(CompletableFuture.completedFuture(response));
+
+ List<MessageRecordVO> records =
provider.queryMessages(STUDIO_INSTANCE_ID, "topic-a", null,
+ null, null, null, null);
+
+ assertThat(records).hasSize(2);
+ MessageRecordVO first = records.get(0);
+ assertThat(first.getMsgId()).isEqualTo("msg-1");
+ assertThat(first.getTag()).isEqualTo("tagA");
+ assertThat(first.getKey()).isEqualTo("k1 k2");
+ assertThat(first.getBody()).isEqualTo("hello");
+ assertThat(first.getBodyEncoding()).isEqualTo("UTF-8");
+ assertThat(first.getStoreTime())
+ .isEqualTo(AliyunConverters.parseTimeMillis("2023-03-22
12:17:08"));
+ assertThat(first.getBornHost()).isEqualTo("10.0.0.1");
+ assertThat(first.getProperties()).containsEntry("a", "b");
+ MessageRecordVO second = records.get(1);
+ assertThat(second.getBody()).isEqualTo("{}");
+ assertThat(second.getBodyEncoding()).isEqualTo("TEXT");
+
+ List<MessageRecordVO> filtered =
provider.queryMessages(STUDIO_INSTANCE_ID, "topic-a", null,
+ "tagB", null, null, null);
+
+ assertThat(filtered).hasSize(1);
+ assertThat(filtered.get(0).getMsgId()).isEqualTo("msg-2");
+ }
+
+ @Test
+ void createConsumerGroupShouldApplyDefaultsTest() {
+ stubInstance();
+ stubCallThrough();
+ when(asyncClient.createConsumerGroup(any()))
+
.thenReturn(CompletableFuture.completedFuture(CreateConsumerGroupResponse.create()
+ .toBuilder()
+ .statusCode(200)
+
.body(CreateConsumerGroupResponseBody.builder().data(true).build())
+ .build()));
+ ConsumerGroupVO group = new ConsumerGroupVO();
+ group.setName("GID_new");
+
+ ConsumerGroupVO created =
provider.createConsumerGroup(STUDIO_INSTANCE_ID, group);
+
+ ArgumentCaptor<CreateConsumerGroupRequest> captor =
+ ArgumentCaptor.forClass(CreateConsumerGroupRequest.class);
+ verify(asyncClient).createConsumerGroup(captor.capture());
+ CreateConsumerGroupRequest request = captor.getValue();
+ assertThat(request.getInstanceId()).isEqualTo(CLOUD_INSTANCE_ID);
+ assertThat(request.getConsumerGroupId()).isEqualTo("GID_new");
+ assertThat(request.getDeliveryOrderType()).isEqualTo("Concurrently");
+
assertThat(request.getConsumeRetryPolicy().getRetryPolicy()).isEqualTo("DefaultRetryPolicy");
+
assertThat(request.getConsumeRetryPolicy().getMaxRetryTimes()).isEqualTo(16);
+ assertThat(created.getInstanceId()).isEqualTo(STUDIO_INSTANCE_ID);
+ assertThat(created.getDeliveryOrderType()).isEqualTo("Concurrently");
+ assertThat(created.getRetryMaxTimes()).isEqualTo(16);
+ }
+
+ @Test
+ void resetOffsetShouldUseSpecifiedTimeTest() {
+ stubInstance();
+ stubCallThrough();
+ when(asyncClient.resetConsumeOffset(any()))
+
.thenReturn(CompletableFuture.completedFuture(ResetConsumeOffsetResponse.create()
+ .toBuilder()
+ .statusCode(200)
+
.body(ResetConsumeOffsetResponseBody.builder().success(true).build())
+ .build()));
+ long timestamp = 1679458628000L;
+
+ provider.resetOffset(STUDIO_INSTANCE_ID, "GID_test", timestamp,
"topic-a");
+
+ ArgumentCaptor<ResetConsumeOffsetRequest> captor =
+ ArgumentCaptor.forClass(ResetConsumeOffsetRequest.class);
+ verify(asyncClient).resetConsumeOffset(captor.capture());
+ ResetConsumeOffsetRequest request = captor.getValue();
+ assertThat(request.getConsumerGroupId()).isEqualTo("GID_test");
+ assertThat(request.getTopicName()).isEqualTo("topic-a");
+ assertThat(request.getResetType()).isEqualTo("SPECIFIED_TIME");
+
assertThat(request.getResetTime()).isEqualTo(AliyunConverters.formatTimeMillis(timestamp));
+ }
+
+ @Test
+ void getMessageTraceShouldMapNodesTest() {
+ stubInstance();
+ stubCallThrough();
+ GetTraceResponse response = GetTraceResponse.create().toBuilder()
+ .statusCode(200)
+ .body(GetTraceResponseBody.builder()
+ .data(GetTraceResponseBody.Data.builder()
+
.producerInfo(GetTraceResponseBody.ProducerInfo.builder()
+
.records(List.of(GetTraceResponseBody.ProducerInfoRecords.builder()
+ .produceTime("2023-03-22
12:17:08")
+ .produceStatus("SEND_OK")
+ .produceDuration(12L)
+ .clientHost("10.0.0.1")
+ .messageSource("SDK")
+ .build()))
+ .build())
+
.brokerInfo(GetTraceResponseBody.BrokerInfo.builder()
+
.operations(List.of(GetTraceResponseBody.Operations.builder()
+ .operateType("store")
+ .operateTime("2023-03-22
12:17:09")
+ .build()))
+ .build())
+
.consumerInfos(List.of(GetTraceResponseBody.ConsumerInfos.builder()
+ .consumerGroupId("GID_test")
+
.records(List.of(GetTraceResponseBody.Records.builder()
+ .consumeStatus("CONSUME_OK")
+ .clientHost("10.0.0.3")
+ .operations(List.of(
+
GetTraceResponseBody.RecordsOperations.builder()
+
.operateType("pull")
+
.operateTime("2023-03-22 12:17:10")
+ .build()))
+ .build()))
+ .build()))
+ .build())
+ .build())
+ .build();
+
when(asyncClient.getTrace(any())).thenReturn(CompletableFuture.completedFuture(response));
+
+ TraceRecordVO trace = provider.getMessageTrace(STUDIO_INSTANCE_ID,
"msg-1");
+
+ assertThat(trace.getNodes()).hasSize(3);
+ TraceNodeVO producer = trace.getNodes().get(0);
+ assertThat(producer.getTitle()).isEqualTo("Producer");
+ assertThat(producer.getStatus()).isEqualTo("SEND_OK");
+ assertThat(producer.getCostTime()).isEqualTo(12L);
+ assertThat(producer.getTimestamp())
+ .isEqualTo(AliyunConverters.parseTimeMillis("2023-03-22
12:17:08"));
+ assertThat(trace.getNodes().get(1).getTitle()).isEqualTo("Broker
store");
+ TraceNodeVO consumer = trace.getNodes().get(2);
+ assertThat(consumer.getTitle()).isEqualTo("Consumer GID_test");
+ assertThat(consumer.getStatus()).isEqualTo("CONSUME_OK");
+ }
+
+ @Test
+ void mappedBusinessExceptionShouldPropagateTest() {
+ stubInstance();
+ when(clientFactory.call(anyString(), anyString(), any()))
+ .thenThrow(new BusinessException(404, "Aliyun resource not
found"));
+
+ assertThatThrownBy(() -> provider.listTopics(STUDIO_INSTANCE_ID, null,
null))
+ .isInstanceOf(BusinessException.class)
+ .extracting("code")
+ .isEqualTo(404);
+ }
+
+ @Test
+ void resolveShouldRejectMissingCloudBindingTest() {
+ InstanceVO instance = InstanceVO.builder()
+ .name("incomplete")
+ .vendor(InstanceVendor.ALIYUN)
+ .cloudInstanceId(CLOUD_INSTANCE_ID)
+ .regionId(REGION)
+ .build();
+
when(instanceRepository.findById(STUDIO_INSTANCE_ID)).thenReturn(Optional.of(instance));
+
+ assertThatThrownBy(() -> provider.listTopics(STUDIO_INSTANCE_ID, null,
null))
+ .isInstanceOf(BusinessException.class)
+ .extracting("code")
+ .isEqualTo(400);
+ }
+
+ private void stubInstance() {
+ InstanceVO instance = InstanceVO.builder()
+ .name("aliyun-prod")
+ .vendor(InstanceVendor.ALIYUN)
+ .cloudInstanceId(CLOUD_INSTANCE_ID)
+ .regionId(REGION)
+ .credentialId(CREDENTIAL_ID)
+ .build();
+
when(instanceRepository.findById(STUDIO_INSTANCE_ID)).thenReturn(Optional.of(instance));
+ }
+
+ private void stubCallThrough() {
+ when(clientFactory.call(anyString(), anyString(),
any())).thenAnswer(invocation -> {
+ Function<AsyncClient, CompletableFuture<Object>> action =
invocation.getArgument(2);
+ return action.apply(asyncClient).join();
+ });
+ }
+
+ private static ListTopicsResponse
topicsResponse(ListTopicsResponseBody.List... rows) {
+ return ListTopicsResponse.create().toBuilder()
+ .statusCode(200)
+ .body(ListTopicsResponseBody.builder()
+ .data(ListTopicsResponseBody.Data.builder()
+ .list(List.of(rows))
+ .pageNumber(1L)
+ .pageSize(100L)
+ .totalCount((long) rows.length)
+ .build())
+ .build())
+ .build();
+ }
+
+ private static ListTopicsResponseBody.List topicRow(String name, String
messageType) {
+ return ListTopicsResponseBody.List.builder()
+ .topicName(name)
+ .messageType(messageType)
+ .remark("remark-" + name)
+ .build();
+ }
+
+ @Test
+ void normalizeDeliveryOrderTypeShouldMapFifoToOrderlyTest() {
+ org.junit.jupiter.api.Assertions.assertEquals("Orderly",
+ AliyunInstanceProvider.normalizeDeliveryOrderType("FIFO"));
+ org.junit.jupiter.api.Assertions.assertEquals("Orderly",
+ AliyunInstanceProvider.normalizeDeliveryOrderType("orderly"));
+ org.junit.jupiter.api.Assertions.assertEquals("Concurrently",
+ AliyunInstanceProvider.normalizeDeliveryOrderType(null));
+ org.junit.jupiter.api.Assertions.assertEquals("Concurrently",
+
AliyunInstanceProvider.normalizeDeliveryOrderType("Concurrently"));
+ }
+}
