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 288876e0 feat(tencent): support Topic management on Tencent Cloud
RocketMQ 5.x instances (#1535)
288876e0 is described below
commit 288876e0558d944f222c8399ba4bbd983376185f
Author: Lei Zhiyuan <[email protected]>
AuthorDate: Tue Aug 11 11:51:10 2026 +0800
feat(tencent): support Topic management on Tencent Cloud RocketMQ 5.x
instances (#1535)
* fix: resolve build and startup failures on the studio branch
Fixes several issues that blocked the project from compiling and booting:
- AlertService: import java.util.regex.Pattern used by metric/duration
patterns
- RocketMQMetadataProvider / RocketMQDashboardProvider: import
SystemTopicFilter
- RocketMQDLQProvider: MessageExt#getUserProperties() does not exist in
rocketmq-tools 5.5.0; read properties via MessageExt#getProperties()
- RocketMQMessageProvider: DefaultMQAdminExt#viewMessage(String) was
removed;
resolve the message by offset id to derive the trace window
- GrafanaDashboardService: mark the public constructor @Autowired so Spring
picks it over the package-private two-arg constructor (No default
constructor found at startup)
* feat(tencent): support topic management on Tencent Cloud RocketMQ 5.x
Wire the Tencent Cloud provider (Trocket v20230308 OpenAPI) to the instance
topic CRUD surface:
- listTopics: page through DescribeTopicList, filter by type/name/remark,
and
enrich created/updated times per topic via DescribeTopic (millisecond ts)
- createTopic / updateTopic / deleteTopic via Create/Modify/DeleteTopic
- getTopicConsumers via DescribeTopic subscription data
- countTopics via a cheap list path (no N+1 detail lookups)
- validate types (reject LiteTopic), queue counts and cloud binding
Also harden the topic page date formatter so missing timestamps render as
"-"
instead of 1970-01-01, and move the no-instance reset out of the synchronous
effect body to satisfy react-hooks/set-state-in-effect.
---
.../metrics/grafana/GrafanaDashboardService.java | 2 +
.../rocketmq/studio/ops/alert/AlertService.java | 1 +
.../provider/apache/RocketMQDLQProvider.java | 2 +-
.../provider/apache/RocketMQDashboardProvider.java | 1 +
.../provider/apache/RocketMQMessageProvider.java | 2 +-
.../provider/apache/RocketMQMetadataProvider.java | 1 +
.../provider/tencent/TencentInstanceProvider.java | 298 ++++++++++++++++++++-
.../tencent/TencentInstanceProviderTest.java | 178 +++++++++---
web/src/pages/instance/topic.tsx | 16 +-
9 files changed, 450 insertions(+), 51 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/grafana/GrafanaDashboardService.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/grafana/GrafanaDashboardService.java
index c16f2226..e0c5dfdd 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/grafana/GrafanaDashboardService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/grafana/GrafanaDashboardService.java
@@ -20,6 +20,7 @@ import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.Resource;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.core.io.support.ResourcePatternResolver;
@@ -48,6 +49,7 @@ public class GrafanaDashboardService {
private final ObjectMapper objectMapper;
private final ResourcePatternResolver resourceResolver;
+ @Autowired
public GrafanaDashboardService(ObjectMapper objectMapper) {
this(objectMapper, new PathMatchingResourcePatternResolver());
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertService.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertService.java
index a163bd76..250b73f0 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertService.java
@@ -30,6 +30,7 @@ import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
+import java.util.regex.Pattern;
@Slf4j
@Service
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProvider.java
index c0872f75..2338d4c7 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProvider.java
@@ -270,7 +270,7 @@ public class RocketMQDLQProvider implements DLQProvider {
}
// Copy user properties from the original message, skipping
system-reserved
// keys to avoid conflicts with broker-internal properties.
- Map<String, String> userProperties =
deadLetter.getUserProperties();
+ Map<String, String> userProperties = deadLetter.getProperties();
if (userProperties != null) {
for (Map.Entry<String, String> entry :
userProperties.entrySet()) {
String key = entry.getKey();
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDashboardProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDashboardProvider.java
index d31603d7..cd9fd14d 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDashboardProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDashboardProvider.java
@@ -42,6 +42,7 @@ import
org.apache.rocketmq.studio.ops.dashboard.DashboardDataVO;
import org.apache.rocketmq.studio.ops.dashboard.DashboardProvider;
import org.apache.rocketmq.studio.ops.dashboard.DashboardStatsVO;
import org.apache.rocketmq.studio.common.util.SystemGroupFilter;
+import org.apache.rocketmq.studio.common.util.SystemTopicFilter;
import org.apache.rocketmq.tools.admin.MQAdminExt;
import org.springframework.context.annotation.Primary;
import org.springframework.stereotype.Service;
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java
index 5151c318..c61a174b 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProvider.java
@@ -321,7 +321,7 @@ public class RocketMQMessageProvider implements
MessageProvider {
*/
private long resolveMessageStoreTimestamp(DefaultMQAdminExt adminExt,
String msgId) {
try {
- MessageExt messageExt = adminExt.viewMessage(msgId);
+ MessageExt messageExt = viewMessageByOffsetId(adminExt, msgId);
if (messageExt != null) {
return messageExt.getStoreTimestamp();
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
index da3dea7b..13140550 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
@@ -34,6 +34,7 @@ import org.apache.rocketmq.tools.admin.MQAdminExt;
import org.apache.rocketmq.studio.common.domain.enums.ConsumeType;
import org.apache.rocketmq.studio.common.domain.enums.TopicPerm;
import org.apache.rocketmq.studio.common.util.SystemGroupFilter;
+import org.apache.rocketmq.studio.common.util.SystemTopicFilter;
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;
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
index 62b9f8c0..16a867ea 100644
---
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
@@ -16,7 +16,22 @@
*/
package org.apache.rocketmq.studio.provider.tencent;
+import com.tencentcloudapi.trocket.v20230308.models.CreateTopicRequest;
+import com.tencentcloudapi.trocket.v20230308.models.DeleteTopicRequest;
+import com.tencentcloudapi.trocket.v20230308.models.DescribeTopicListRequest;
+import com.tencentcloudapi.trocket.v20230308.models.DescribeTopicListResponse;
+import com.tencentcloudapi.trocket.v20230308.models.DescribeTopicRequest;
+import com.tencentcloudapi.trocket.v20230308.models.DescribeTopicResponse;
+import com.tencentcloudapi.trocket.v20230308.models.ModifyTopicRequest;
+import com.tencentcloudapi.trocket.v20230308.models.SubscriptionData;
+import com.tencentcloudapi.trocket.v20230308.models.TopicItem;
+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.TopicPerm;
+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.group.SubscriptionEntryVO;
@@ -26,16 +41,35 @@ 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 org.springframework.util.StringUtils;
+import lombok.RequiredArgsConstructor;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.util.ArrayList;
import java.util.List;
+import java.util.Locale;
/**
- * Tencent Cloud TDMQ placeholder: package structure reserved, all operations
unsupported.
+ * Tencent Cloud TDMQ RocketMQ 5.x topic operations backed by Trocket
v20230308 OpenAPI.
+ *
+ * <p>Topic management is supported independently from the other
instance-scoped operations. The
+ * remaining operations intentionally retain the provider's
unsupported-operation behavior until
+ * their corresponding Tencent Cloud APIs are mapped to Studio's common
models.</p>
*/
+@RequiredArgsConstructor
@Component
public class TencentInstanceProvider implements InstanceProvider {
- private static final String NOT_IMPLEMENTED = "Tencent Cloud provider is
not implemented yet";
+ static final int PAGE_SIZE = 100;
+ static final int MAX_PAGES = 100;
+ static final int CONSUMER_PAGE_SIZE = 100;
+ static final int DEFAULT_QUEUE_NUM = 8;
+ private static final String NOT_IMPLEMENTED = "Tencent Cloud operation is
not implemented yet";
+
+ private final TencentClientFactory clientFactory;
+ private final InstanceRepository instanceRepository;
@Override
public InstanceVendor vendor() {
@@ -44,7 +78,7 @@ public class TencentInstanceProvider implements
InstanceProvider {
@Override
public int countTopics(String instanceId) {
- throw new UnsupportedOperationException(NOT_IMPLEMENTED);
+ return listTopics(instanceId, null, null, false).size();
}
@Override
@@ -54,27 +88,143 @@ public class TencentInstanceProvider implements
InstanceProvider {
@Override
public List<TopicVO> listTopics(String instanceId, String type, String
search) {
- throw new UnsupportedOperationException(NOT_IMPLEMENTED);
+ return listTopics(instanceId, type, search, true);
+ }
+
+ private List<TopicVO> listTopics(String instanceId, String type, String
search, boolean enrichTimes) {
+ Context context = resolve(instanceId);
+ List<TopicVO> topics = new ArrayList<>();
+ for (int page = 0; page < MAX_PAGES; page++) {
+ DescribeTopicListRequest request = new DescribeTopicListRequest();
+ request.setInstanceId(context.cloudInstanceId());
+ request.setOffset((long) page * PAGE_SIZE);
+ request.setLimit((long) PAGE_SIZE);
+ DescribeTopicListResponse response =
clientFactory.call(context.credentialId(), context.regionId(),
+ client -> client.DescribeTopicList(request));
+ TopicItem[] data = response == null ? null : response.getData();
+ if (data == null || data.length == 0) {
+ break;
+ }
+ for (TopicItem item : data) {
+ if (item == null) {
+ continue;
+ }
+ TopicVO topic = toTopic(item, instanceId);
+ if (matchesType(type, topic) && matchesSearch(search, topic)) {
+ if (enrichTimes) {
+ enrichTopicTimes(context, topic);
+ }
+ topics.add(topic);
+ }
+ }
+ if (data.length < PAGE_SIZE) {
+ break;
+ }
+ }
+ return topics;
+ }
+
+ /**
+ * DescribeTopicList does not expose creation/update timestamps, so
resolve them per-topic
+ * from DescribeTopic. Kept off the cheap count path to avoid N+1 calls
for instance listings.
+ */
+ private void enrichTopicTimes(Context context, TopicVO topic) {
+ try {
+ DescribeTopicRequest request = new DescribeTopicRequest();
+ request.setInstanceId(context.cloudInstanceId());
+ request.setTopic(topic.getName());
+ DescribeTopicResponse response =
clientFactory.call(context.credentialId(), context.regionId(),
+ client -> client.DescribeTopic(request));
+ if (response == null) {
+ return;
+ }
+ topic.setCreatedAt(toLocalDateTime(response.getCreatedTime()));
+ topic.setUpdatedAt(toLocalDateTime(response.getLastUpdateTime()));
+ } catch (BusinessException ignored) {
+ // A single topic detail lookup failure should not fail the whole
list.
+ }
+ }
+
+ private static LocalDateTime toLocalDateTime(Long epoch) {
+ if (epoch == null || epoch <= 0L) {
+ return null;
+ }
+ // Tencent Cloud returns millisecond timestamps; tolerate second
precision as a fallback.
+ long epochMillis = epoch >= 10_000_000_000L ? epoch : epoch * 1000L;
+ return
Instant.ofEpochMilli(epochMillis).atZone(ZoneId.systemDefault()).toLocalDateTime();
}
@Override
public TopicVO createTopic(String instanceId, TopicVO topic) {
- throw new UnsupportedOperationException(NOT_IMPLEMENTED);
+ Context context = resolve(instanceId);
+ validateCreateTopic(topic);
+ CreateTopicRequest request = new CreateTopicRequest();
+ request.setInstanceId(context.cloudInstanceId());
+ request.setTopic(topic.getName());
+ request.setTopicType(topic.getType().name());
+ request.setQueueNum(queueNum(topic));
+ request.setRemark(topic.getRemark());
+ clientFactory.call(context.credentialId(), context.regionId(), client
-> client.CreateTopic(request));
+ topic.setInstanceId(instanceId);
+ topic.setPerm(defaultPerm(topic.getPerm()));
+ topic.setWriteQueues(queueNum(topic).intValue());
+ topic.setReadQueues(queueNum(topic).intValue());
+ topic.setCreatedAt(LocalDateTime.now());
+ topic.setUpdatedAt(LocalDateTime.now());
+ return topic;
}
@Override
public TopicVO updateTopic(String instanceId, TopicVO topic) {
- throw new UnsupportedOperationException(NOT_IMPLEMENTED);
+ Context context = resolve(instanceId);
+ validateUpdateTopic(topic);
+ ModifyTopicRequest request = new ModifyTopicRequest();
+ request.setInstanceId(context.cloudInstanceId());
+ request.setTopic(topic.getName());
+ request.setRemark(topic.getRemark());
+ if (topic.getWriteQueues() > 0 || topic.getReadQueues() > 0) {
+ request.setQueueNum(queueNum(topic));
+ }
+ clientFactory.call(context.credentialId(), context.regionId(), client
-> client.ModifyTopic(request));
+ topic.setInstanceId(instanceId);
+ topic.setPerm(defaultPerm(topic.getPerm()));
+ topic.setUpdatedAt(LocalDateTime.now());
+ return topic;
}
@Override
public void deleteTopic(String instanceId, String topicName) {
- throw new UnsupportedOperationException(NOT_IMPLEMENTED);
+ Context context = resolve(instanceId);
+ requireTopicName(topicName);
+ DeleteTopicRequest request = new DeleteTopicRequest();
+ request.setInstanceId(context.cloudInstanceId());
+ request.setTopic(topicName);
+ clientFactory.call(context.credentialId(), context.regionId(), client
-> client.DeleteTopic(request));
}
@Override
public List<TopicConsumerVO> getTopicConsumers(String instanceId, String
topicName) {
- throw new UnsupportedOperationException(NOT_IMPLEMENTED);
+ Context context = resolve(instanceId);
+ requireTopicName(topicName);
+ DescribeTopicRequest request = new DescribeTopicRequest();
+ request.setInstanceId(context.cloudInstanceId());
+ request.setTopic(topicName);
+ request.setOffset(0L);
+ request.setLimit((long) CONSUMER_PAGE_SIZE);
+ DescribeTopicResponse response =
clientFactory.call(context.credentialId(), context.regionId(),
+ client -> client.DescribeTopic(request));
+ SubscriptionData[] data = response == null ? null :
response.getSubscriptionData();
+ List<TopicConsumerVO> consumers = new ArrayList<>();
+ if (data == null) {
+ return consumers;
+ }
+ for (SubscriptionData subscription : data) {
+ if (subscription == null) {
+ continue;
+ }
+ consumers.add(toTopicConsumer(subscription));
+ }
+ return consumers;
}
@Override
@@ -117,4 +267,136 @@ public class TencentInstanceProvider implements
InstanceProvider {
public TraceRecordVO getMessageTrace(String instanceId, String msgId) {
throw new UnsupportedOperationException(NOT_IMPLEMENTED);
}
+
+ private Context resolve(String instanceId) {
+ if (!StringUtils.hasText(instanceId)) {
+ throw new BusinessException(400, "instanceId is required");
+ }
+ InstanceVO instance = instanceRepository.findById(instanceId)
+ .orElseThrow(() -> new BusinessException(404, "Instance not
found: " + instanceId));
+ if (!StringUtils.hasText(instance.getCloudInstanceId()) ||
!StringUtils.hasText(instance.getRegionId())
+ || !StringUtils.hasText(instance.getCredentialId())) {
+ throw new BusinessException(400, "Instance " + instanceId + " is
missing Tencent Cloud binding");
+ }
+ return new Context(instance.getCloudInstanceId(),
instance.getRegionId(), instance.getCredentialId());
+ }
+
+ private static TopicVO toTopic(TopicItem item, String instanceId) {
+ TopicVO topic = new TopicVO();
+ topic.setName(item.getTopic());
+ topic.setInstanceId(instanceId);
+ topic.setType(toTopicType(item.getTopicType()));
+ topic.setWriteQueues(toInteger(item.getQueueNum()));
+ topic.setReadQueues(toInteger(item.getQueueNum()));
+ topic.setPerm(TopicPerm.RW);
+ topic.setRemark(item.getRemark());
+ topic.setNamespace(item.getNamespaceV4());
+ topic.setClusterId(item.getClusterIdV4());
+ return topic;
+ }
+
+ private static TopicConsumerVO toTopicConsumer(SubscriptionData
subscription) {
+ String messageModel = subscription.getMessageModel();
+ if (!StringUtils.hasText(messageModel)) {
+ messageModel = subscription.getConsumeType();
+ }
+ return TopicConsumerVO.builder()
+ .group(subscription.getConsumerGroup())
+ .consumeType(toConsumeType(subscription.getConsumeType(),
messageModel))
+ .messageModel(messageModel)
+ .diffTotal(subscription.getConsumerLag() == null ? 0L :
subscription.getConsumerLag())
+ .build();
+ }
+
+ private static TopicType toTopicType(String raw) {
+ if (!StringUtils.hasText(raw)) {
+ return null;
+ }
+ try {
+ return TopicType.valueOf(raw.trim().toUpperCase(Locale.ROOT));
+ } catch (IllegalArgumentException ignored) {
+ return null;
+ }
+ }
+
+ private static ConsumeType toConsumeType(String consumeType, String
messageModel) {
+ String value = StringUtils.hasText(consumeType) ? consumeType :
messageModel;
+ if (!StringUtils.hasText(value)) {
+ return null;
+ }
+ if (value.toUpperCase(Locale.ROOT).contains("BROADCAST")) {
+ return ConsumeType.BROADCASTING;
+ }
+ if (value.toUpperCase(Locale.ROOT).contains("CLUSTER")) {
+ return ConsumeType.CLUSTERING;
+ }
+ return null;
+ }
+
+ private static boolean matchesType(String type, TopicVO topic) {
+ return !StringUtils.hasText(type)
+ || topic.getType() != null &&
topic.getType().name().equalsIgnoreCase(type.trim());
+ }
+
+ private static boolean matchesSearch(String search, TopicVO topic) {
+ if (!StringUtils.hasText(search)) {
+ return true;
+ }
+ String needle = search.trim().toLowerCase(Locale.ROOT);
+ return contains(topic.getName(), needle) ||
contains(topic.getRemark(), needle);
+ }
+
+ private static boolean contains(String value, String needle) {
+ return value != null &&
value.toLowerCase(Locale.ROOT).contains(needle);
+ }
+
+ private static void validateCreateTopic(TopicVO topic) {
+ validateTopicName(topic);
+ if (topic.getType() == null) {
+ throw new BusinessException(400, "Topic type is required");
+ }
+ if (topic.getType() == TopicType.LITE) {
+ throw new BusinessException(400, "Tencent Cloud RocketMQ does not
support LiteTopic");
+ }
+ validateQueueNumber(topic);
+ }
+
+ private static void validateUpdateTopic(TopicVO topic) {
+ validateTopicName(topic);
+ validateQueueNumber(topic);
+ }
+
+ private static void validateTopicName(TopicVO topic) {
+ if (topic == null || !StringUtils.hasText(topic.getName())) {
+ throw new BusinessException(400, "Topic name is required");
+ }
+ }
+
+ private static void validateQueueNumber(TopicVO topic) {
+ if (topic.getWriteQueues() < 0 || topic.getReadQueues() < 0) {
+ throw new BusinessException(400, "Topic queue number must not be
negative");
+ }
+ }
+
+ private static Long queueNum(TopicVO topic) {
+ int queueNum = topic.getWriteQueues() > 0 ? topic.getWriteQueues() :
topic.getReadQueues();
+ return (long) (queueNum > 0 ? queueNum : DEFAULT_QUEUE_NUM);
+ }
+
+ private static TopicPerm defaultPerm(TopicPerm perm) {
+ return perm == null ? TopicPerm.RW : perm;
+ }
+
+ private static int toInteger(Long value) {
+ return value == null ? 0 : Math.toIntExact(value);
+ }
+
+ private static void requireTopicName(String topicName) {
+ if (!StringUtils.hasText(topicName)) {
+ throw new BusinessException(400, "Topic name is required");
+ }
+ }
+
+ private record Context(String cloudInstanceId, String regionId, String
credentialId) {
+ }
}
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
index e3c084d3..a9ed54d7 100644
---
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
@@ -16,54 +16,160 @@
*/
package org.apache.rocketmq.studio.provider.tencent;
+import com.tencentcloudapi.trocket.v20230308.models.CreateTopicRequest;
+import com.tencentcloudapi.trocket.v20230308.models.DescribeTopicListResponse;
+import com.tencentcloudapi.trocket.v20230308.models.DescribeTopicResponse;
+import com.tencentcloudapi.trocket.v20230308.models.ModifyTopicRequest;
+import com.tencentcloudapi.trocket.v20230308.models.SubscriptionData;
+import com.tencentcloudapi.trocket.v20230308.models.TopicItem;
+import com.tencentcloudapi.trocket.v20230308.TrocketClient;
import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
-import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
+import org.apache.rocketmq.studio.common.domain.enums.TopicType;
+import org.apache.rocketmq.studio.instance.InstanceRepository;
+import org.apache.rocketmq.studio.instance.InstanceVO;
+import org.apache.rocketmq.studio.instance.topic.TopicConsumerVO;
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.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.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+@ExtendWith(MockitoExtension.class)
class TencentInstanceProviderTest {
- private final TencentInstanceProvider provider = new
TencentInstanceProvider();
+ private static final String STUDIO_INSTANCE_ID = "inst-1";
+ private static final String CLOUD_INSTANCE_ID = "rmq-abc";
+ private static final String REGION = "ap-chengdu";
+ private static final String CREDENTIAL_ID = "cred-1";
+
+ @Mock
+ private TencentClientFactory clientFactory;
+
+ @Mock
+ private InstanceRepository instanceRepository;
+
+ @Mock
+ private TrocketClient client;
+
+ private TencentInstanceProvider provider;
+
+ @BeforeEach
+ void setUp() {
+ provider = new TencentInstanceProvider(clientFactory,
instanceRepository);
+
when(instanceRepository.findById(STUDIO_INSTANCE_ID)).thenReturn(Optional.of(InstanceVO.builder()
+ .name("tencent-prod")
+ .vendor(InstanceVendor.TENCENT)
+ .cloudInstanceId(CLOUD_INSTANCE_ID)
+ .regionId(REGION)
+ .credentialId(CREDENTIAL_ID)
+ .build()));
+ when(clientFactory.call(anyString(), anyString(),
any())).thenAnswer(invocation -> {
+ TencentClientFactory.TencentCall<Object> action =
invocation.getArgument(2);
+ return action.execute(client);
+ });
+ }
@Test
- void vendorShouldBeTencentTest() {
- assertThat(provider.vendor()).isEqualTo(InstanceVendor.TENCENT);
+ void listTopicsShouldMapAndFilterAndEnrichTimesTest() throws Exception {
+ TopicItem normal = topicItem("orders", "NORMAL", 8L);
+ TopicItem fifo = topicItem("orders-fifo", "FIFO", 4L);
+ DescribeTopicListResponse response = new DescribeTopicListResponse();
+ response.setData(new TopicItem[]{normal, fifo});
+ when(client.DescribeTopicList(any())).thenReturn(response);
+ DescribeTopicResponse detail = new DescribeTopicResponse();
+ detail.setCreatedTime(1600000000000L);
+ detail.setLastUpdateTime(1600000100000L);
+ when(client.DescribeTopic(any())).thenReturn(detail);
+
+ List<TopicVO> topics = provider.listTopics(STUDIO_INSTANCE_ID, "FIFO",
"fifo");
+
+ assertThat(topics).hasSize(1);
+ assertThat(topics.get(0).getName()).isEqualTo("orders-fifo");
+ assertThat(topics.get(0).getType()).isEqualTo(TopicType.FIFO);
+ assertThat(topics.get(0).getWriteQueues()).isEqualTo(4);
+ assertThat(topics.get(0).getReadQueues()).isEqualTo(4);
+
assertThat(topics.get(0).getInstanceId()).isEqualTo(STUDIO_INSTANCE_ID);
+ assertThat(topics.get(0).getCreatedAt())
+ .isEqualTo(java.time.Instant.ofEpochMilli(1600000000000L)
+
.atZone(java.time.ZoneId.systemDefault()).toLocalDateTime());
+ assertThat(topics.get(0).getUpdatedAt())
+ .isEqualTo(java.time.Instant.ofEpochMilli(1600000100000L)
+
.atZone(java.time.ZoneId.systemDefault()).toLocalDateTime());
}
@Test
- void allOperationsShouldThrowUnsupportedTest() {
- assertThatThrownBy(() -> provider.countTopics("inst"))
- .isInstanceOf(UnsupportedOperationException.class);
- assertThatThrownBy(() -> provider.countGroups("inst"))
- .isInstanceOf(UnsupportedOperationException.class);
- 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);
+ void createTopicShouldCallTencentOpenApiTest() throws Exception {
+ when(client.CreateTopic(any())).thenReturn(null);
+ TopicVO topic = new TopicVO();
+ topic.setName("orders");
+ topic.setType(TopicType.NORMAL);
+ topic.setWriteQueues(12);
+ topic.setRemark("business orders");
+
+ TopicVO created = provider.createTopic(STUDIO_INSTANCE_ID, topic);
+
+ ArgumentCaptor<CreateTopicRequest> captor =
ArgumentCaptor.forClass(CreateTopicRequest.class);
+ verify(client).CreateTopic(captor.capture());
+
assertThat(captor.getValue().getInstanceId()).isEqualTo(CLOUD_INSTANCE_ID);
+ assertThat(captor.getValue().getTopic()).isEqualTo("orders");
+ assertThat(captor.getValue().getTopicType()).isEqualTo("NORMAL");
+ assertThat(captor.getValue().getQueueNum()).isEqualTo(12L);
+ assertThat(created.getInstanceId()).isEqualTo(STUDIO_INSTANCE_ID);
+ }
+
+ @Test
+ void updateAndDeleteTopicShouldCallTencentOpenApiTest() throws Exception {
+ when(client.ModifyTopic(any())).thenReturn(null);
+ when(client.DeleteTopic(any())).thenReturn(null);
+ TopicVO topic = new TopicVO();
+ topic.setName("orders");
+ topic.setType(TopicType.NORMAL);
+ topic.setRemark("updated");
+
+ provider.updateTopic(STUDIO_INSTANCE_ID, topic);
+ provider.deleteTopic(STUDIO_INSTANCE_ID, "orders");
+
+ ArgumentCaptor<ModifyTopicRequest> updateCaptor =
ArgumentCaptor.forClass(ModifyTopicRequest.class);
+ verify(client).ModifyTopic(updateCaptor.capture());
+
assertThat(updateCaptor.getValue().getInstanceId()).isEqualTo(CLOUD_INSTANCE_ID);
+ assertThat(updateCaptor.getValue().getTopic()).isEqualTo("orders");
+ assertThat(updateCaptor.getValue().getRemark()).isEqualTo("updated");
+ }
+
+ @Test
+ void getTopicConsumersShouldMapSubscriptionsTest() throws Exception {
+ SubscriptionData subscription = new SubscriptionData();
+ subscription.setConsumerGroup("GID_orders");
+ subscription.setConsumeType("CLUSTERING");
+ subscription.setMessageModel("CLUSTERING");
+ subscription.setConsumerLag(42L);
+ DescribeTopicResponse response = new DescribeTopicResponse();
+ response.setSubscriptionData(new SubscriptionData[]{subscription});
+ when(client.DescribeTopic(any())).thenReturn(response);
+
+ List<TopicConsumerVO> consumers =
provider.getTopicConsumers(STUDIO_INSTANCE_ID, "orders");
+
+ assertThat(consumers).hasSize(1);
+ assertThat(consumers.get(0).getGroup()).isEqualTo("GID_orders");
+ assertThat(consumers.get(0).getDiffTotal()).isEqualTo(42L);
+ }
+
+ private static TopicItem topicItem(String name, String type, long
queueNum) {
+ TopicItem item = new TopicItem();
+ item.setTopic(name);
+ item.setTopicType(type);
+ item.setQueueNum(queueNum);
+ return item;
}
}
diff --git a/web/src/pages/instance/topic.tsx b/web/src/pages/instance/topic.tsx
index 7b574dd5..20411237 100644
--- a/web/src/pages/instance/topic.tsx
+++ b/web/src/pages/instance/topic.tsx
@@ -258,8 +258,10 @@ const parsePropsText = (text: string): Record<string,
string> => {
}
return props;
};
-const formatDateTime = (iso: string): string => {
+const formatDateTime = (iso?: string): string => {
+ if (!iso) return '-';
const d = new Date(iso);
+ if (Number.isNaN(d.getTime())) return '-';
const pad = (n: number) => String(n).padStart(2, '0');
return `${d.getFullYear()}-${pad(d.getMonth() + 1)}-${pad(d.getDate())}
${pad(d.getHours())}:${pad(d.getMinutes())}:${pad(d.getSeconds())}`;
};
@@ -308,10 +310,14 @@ const TopicPage = () => {
useEffect(() => {
if (!selectedInstanceId) {
topicRequestIdRef.current += 1;
- setTopics([]);
- setSelectedRowKeys([]);
- setLoading(false);
- return;
+ const resetTimer = window.setTimeout(() => {
+ setTopics([]);
+ setSelectedRowKeys([]);
+ setLoading(false);
+ }, 0);
+ return () => {
+ window.clearTimeout(resetTimer);
+ };
}
const requestId = ++topicRequestIdRef.current;
const timer = window.setTimeout(() => {
