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 05b1577c8 fix(provider): instance ACL hooks, topic type propagation,
DLQ pagination and bounded message queries (#2308)
05b1577c8 is described below
commit 05b1577c81b699f1fccaafcfe63f20512b0a4a3d
Author: aias00 <[email protected]>
AuthorDate: Tue Aug 18 21:03:03 2026 +0800
fix(provider): instance ACL hooks, topic type propagation, DLQ pagination
and bounded message queries (#2308)
* fix: preserve instance ACL credentials across short-lived runtime clients
Short-lived admin clients created outside MqAdminExtFactory (cluster
discovery, NameServer config diff, DLQ scan/resend/export, topic message
queries and probe sends) previously connected anonymously, failing against
ACL-enabled instances. Add RuntimeAdminClientResolver.resolveCredentialHook
and wire the selected instance's admin credential hook through every
runtime client constructor.
Adapted from PR #2308 (consolidates the five credential commits) to the
registry-driven ClusterProvider structure: instance-scoped discovery and
refresh route through the resolver while the nameserver-registry
discoverClustersAt path stays anonymous.
* fix: propagate topic message type when creating or updating topics
Create/update topic now writes the +message.type topic attribute to the
broker so the declared type (NORMAL/FIFO/DELAY/TRANSACTION) is enforced,
and only defaults the persisted type to NORMAL for brand-new records.
Re-creating or updating a topic without an explicit type keeps the
existing type. Adapted from PR #2308.
* feat: paginate DLQ group discovery
Support server-side search and pagination for DLQ group listing so
instances with many consumer groups no longer pay the full scan cost
on every page load. The frontend drives page, page size, and search
through the paginated envelope instead of filtering a full list.
* fix: bound topic message query scans
Reject topic-scope message queries spanning more than seven days and
cap per-queue pull attempts so an oversized range cannot stall the
admin client or hammer the brokers.
---
.../cluster/broker/RuntimeAdminClientResolver.java | 17 ++++-
.../nameserver/NameServerConfigDiffService.java | 40 ++++++----
.../studio/instance/dlq/DLQController.java | 8 +-
.../rocketmq/studio/instance/dlq/DLQProvider.java | 4 +
.../studio/instance/dlq/DLQProviderStub.java | 8 ++
.../rocketmq/studio/instance/dlq/DLQService.java | 12 +++
.../provider/apache/RocketMQAdminClientImpl.java | 29 ++++++-
.../provider/apache/RocketMQClusterProvider.java | 18 ++++-
.../provider/apache/RocketMQDLQProvider.java | 56 +++++++++-----
.../provider/apache/RocketMQMessageProvider.java | 28 +++++--
.../broker/RuntimeAdminClientResolverTest.java | 36 +++++++++
.../NameServerConfigDiffServiceTest.java | 17 ++++-
.../studio/instance/dlq/DLQControllerTest.java | 20 +++--
.../studio/instance/dlq/DLQProviderStubTest.java | 9 +++
.../studio/instance/dlq/DLQServiceTest.java | 27 +++++++
.../apache/RocketMQAdminClientImplTest.java | 57 +++++++++++++-
.../apache/RocketMQClusterProviderTest.java | 72 ++++++++++++++++++
.../provider/apache/RocketMQDLQProviderTest.java | 88 ++++++++++++++++++++++
.../apache/RocketMQMessageProviderTest.java | 45 +++++++++++
web/src/api/message.ts | 13 +++-
web/src/pages/instance/__tests__/DLQPage.test.tsx | 43 ++++++-----
web/src/pages/instance/dlq.tsx | 45 ++++++-----
web/src/services/messageService.test.ts | 21 +++++-
web/src/services/messageService.ts | 24 +++++-
24 files changed, 634 insertions(+), 103 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/RuntimeAdminClientResolver.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/RuntimeAdminClientResolver.java
index dee110c11..37cdf1c75 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/RuntimeAdminClientResolver.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/RuntimeAdminClientResolver.java
@@ -41,6 +41,15 @@ public class RuntimeAdminClientResolver {
return instance.getEndpoint().trim();
}
+ /**
+ * Resolves the ACL hook for short-lived runtime clients that cannot be
created by
+ * {@link MqAdminExtFactory}, such as a {@code DefaultMQPullConsumer}.
+ */
+ public RPCHook resolveCredentialHook(String instanceId) {
+ InstanceVO instance =
requireApacheInstance(resolveInstance(instanceId));
+ return resolveCredential(credentialRef(instance));
+ }
+
public <T> T execute(String instanceId, MqAdminExtFactory.AdminAction<T>
action) {
return execute(resolveInstance(instanceId), action);
}
@@ -50,12 +59,16 @@ public class RuntimeAdminClientResolver {
if (instance == null || !StringUtils.hasText(instance.getEndpoint())) {
throw new BusinessException(400, "Instance endpoint is required");
}
- String credentialRef =
StringUtils.hasText(instance.getAdminCredentialRef())
- ? instance.getAdminCredentialRef().trim() : null;
+ String credentialRef = credentialRef(instance);
return adminFactory.execute(instance.getEndpoint().trim(),
resolveCredential(credentialRef),
credentialRef, action);
}
+ private String credentialRef(InstanceVO instance) {
+ return StringUtils.hasText(instance.getAdminCredentialRef())
+ ? instance.getAdminCredentialRef().trim() : null;
+ }
+
private RPCHook resolveCredential(String credentialRef) {
if (!StringUtils.hasText(credentialRef)) {
return null;
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/nameserver/NameServerConfigDiffService.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/nameserver/NameServerConfigDiffService.java
index e05028c1f..cf27e7876 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/nameserver/NameServerConfigDiffService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/nameserver/NameServerConfigDiffService.java
@@ -22,7 +22,9 @@ import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
import org.apache.rocketmq.studio.cluster.broker.ClusterService;
import org.apache.rocketmq.studio.cluster.broker.ClusterVO;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
+import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.tools.admin.MQAdminExt;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
@@ -70,19 +72,24 @@ public class NameServerConfigDiffService {
private final ClusterService clusterService;
private final MqAdminExtFactory adminFactory;
+ private final RuntimeAdminClientResolver runtimeAdminClientResolver;
public NameServerConfigDiffVO compare(String clusterId) {
String normalizedClusterId = requireClusterId(clusterId);
- return compare(normalizedClusterId,
clusterService.getCluster(normalizedClusterId));
+ return compare(normalizedClusterId,
clusterService.getCluster(normalizedClusterId), null);
}
public NameServerConfigDiffVO compare(String clusterId, String instanceId)
{
String normalizedClusterId = requireClusterId(clusterId);
+ String normalizedInstanceId = normalizeInstanceId(instanceId);
return compare(normalizedClusterId,
- clusterService.getCluster(normalizedClusterId,
normalizeInstanceId(instanceId)));
+ clusterService.getCluster(normalizedClusterId,
normalizedInstanceId), normalizedInstanceId);
}
- private NameServerConfigDiffVO compare(String normalizedClusterId,
ClusterVO cluster) {
+ private NameServerConfigDiffVO compare(
+ String normalizedClusterId,
+ ClusterVO cluster,
+ String instanceId) {
List<String> addresses = collectNameServerAddresses(cluster);
if (addresses.isEmpty()) {
throw new BusinessException(409,
@@ -95,7 +102,7 @@ public class NameServerConfigDiffService {
for (String address : addresses) {
try {
- Properties config = readConfig(connectionEndpoint, address);
+ Properties config = readConfig(instanceId, connectionEndpoint,
address);
reachableConfigs.put(address, config);
nodes.add(NameServerConfigDiffVO.NodeStatusVO.builder()
.address(address)
@@ -124,16 +131,21 @@ public class NameServerConfigDiffService {
.build();
}
- private Properties readConfig(String connectionEndpoint, String address) {
- return adminFactory.execute(connectionEndpoint, null, admin -> {
- Map<String, Properties> configs =
admin.getNameServerConfig(List.of(address));
- Properties config = configs == null ? null : configs.get(address);
- if (config == null) {
- throw new BusinessException(502,
- "NameServer returned no configuration: " + address);
- }
- return config;
- });
+ private Properties readConfig(String instanceId, String
connectionEndpoint, String address) {
+ if (instanceId != null) {
+ return runtimeAdminClientResolver.execute(instanceId, admin ->
readConfig(admin, address));
+ }
+ return adminFactory.execute(connectionEndpoint, null, admin ->
readConfig(admin, address));
+ }
+
+ private Properties readConfig(MQAdminExt admin, String address) throws
Exception {
+ Map<String, Properties> configs =
admin.getNameServerConfig(List.of(address));
+ Properties config = configs == null ? null : configs.get(address);
+ if (config == null) {
+ throw new BusinessException(502,
+ "NameServer returned no configuration: " + address);
+ }
+ return config;
}
private List<NameServerConfigDiffVO.ConfigDifferenceVO> findDifferences(
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQController.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQController.java
index 6487cba54..7ac1b33cd 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQController.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQController.java
@@ -18,6 +18,7 @@ package org.apache.rocketmq.studio.instance.dlq;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.rocketmq.studio.common.domain.PageResult;
import org.apache.rocketmq.studio.common.domain.Result;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import jakarta.validation.Valid;
@@ -43,8 +44,11 @@ public class DLQController {
private final ObjectMapper objectMapper;
@GetMapping
- public Result<List<DLQGroupVO>> listDLQGroups(@RequestParam String
instanceId) {
- return Result.ok(dlqService.listDLQGroups(instanceId));
+ public Result<PageResult<DLQGroupVO>> listDLQGroups(@RequestParam String
instanceId,
+ @RequestParam(required = false) String search,
+ @RequestParam(defaultValue = "1") int page,
+ @RequestParam(defaultValue = "20") int pageSize) {
+ return Result.ok(dlqService.listDLQGroups(instanceId, search, page,
pageSize));
}
@PostMapping("/resend")
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQProvider.java
index 4f1c7dbd8..974fdd83d 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQProvider.java
@@ -18,9 +18,13 @@ package org.apache.rocketmq.studio.instance.dlq;
import java.util.List;
+import org.apache.rocketmq.studio.common.domain.PageResult;
public interface DLQProvider {
List<DLQGroupVO> listDLQGroups(String instanceId);
+
+ PageResult<DLQGroupVO> listDLQGroups(String instanceId, String search, int
page, int pageSize);
+
DLQResendResultVO resendMessages(String instanceId, String groupName, Long
startTime, Long endTime,
String targetTopic);
List<DLQMessageVO> exportMessages(String instanceId, String groupName,
Long startTime, Long endTime,
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQProviderStub.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQProviderStub.java
index bb9400c52..6d57d83e4 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQProviderStub.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQProviderStub.java
@@ -17,6 +17,7 @@
package org.apache.rocketmq.studio.instance.dlq;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.common.domain.PageResult;
import lombok.extern.slf4j.Slf4j;
import java.util.List;
@@ -35,6 +36,13 @@ public class DLQProviderStub implements DLQProvider {
throw unsupported();
}
+ @Override
+ public PageResult<DLQGroupVO> listDLQGroups(String instanceId, String
search, int page, int pageSize) {
+ log.warn("DLQProviderStub.listDLQGroups(paged) called but no real DLQ
provider is configured. "
+ + "instanceId={}, page={}, pageSize={}", instanceId, page,
pageSize);
+ throw unsupported();
+ }
+
@Override
public DLQResendResultVO resendMessages(String instanceId, String
groupName, Long startTime, Long endTime,
String targetTopic) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQService.java
index 6447e0c5f..0a95afae7 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQService.java
@@ -18,6 +18,7 @@ package org.apache.rocketmq.studio.instance.dlq;
import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.common.domain.PageResult;
import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
@@ -31,9 +32,20 @@ import java.util.List;
@Slf4j
public class DLQService {
+ private static final int MAX_PAGE_SIZE = 100;
+
private final DLQProvider dlqProvider;
private final InstanceProviderRegistry providerRegistry;
+ public PageResult<DLQGroupVO> listDLQGroups(String instanceId, String
search, int page, int pageSize) {
+ requireApacheInstance(instanceId);
+ if (page < 1 || pageSize < 1 || pageSize > MAX_PAGE_SIZE) {
+ throw new BusinessException(400, "Invalid page or pageSize");
+ }
+ return dlqProvider.listDLQGroups(instanceId,
+ StringUtils.hasText(search) ? search.trim() : null, page,
pageSize);
+ }
+
public List<DLQGroupVO> listDLQGroups(String instanceId) {
requireApacheInstance(instanceId);
log.info("Listing DLQ groups for instance: {}", instanceId);
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
index 3e18d765e..5d9698684 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
@@ -22,7 +22,9 @@ import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.TopicConfig;
+import org.apache.rocketmq.common.TopicAttributes;
import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
@@ -31,6 +33,7 @@ import
org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.common.domain.enums.TopicPerm;
+import org.apache.rocketmq.studio.common.domain.enums.TopicType;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
import org.apache.rocketmq.studio.instance.topic.SendMessageDTO;
import org.apache.rocketmq.studio.instance.topic.SendMessageVO;
@@ -158,6 +161,7 @@ public class RocketMQAdminClientImpl implements AdminClient
{
topicConfig.setWriteQueueNums(writeQueues);
topicConfig.setReadQueueNums(readQueues);
topicConfig.setPerm(toRocketMQPerm(effectivePerm));
+ applyTopicType(topicConfig, topic.getType());
for (String addr : brokerAddrs) {
admin.createAndUpdateTopicConfig(addr, topicConfig);
@@ -179,7 +183,11 @@ public class RocketMQAdminClientImpl implements
AdminClient {
if (topic.getInstanceId() != null) {
entity.setInstanceId(topic.getInstanceId());
}
- entity.setTopicType(topic.getType() != null ?
topic.getType().name() : "NORMAL");
+ if (topic.getType() != null) {
+ entity.setTopicType(topic.getType().name());
+ } else if (isNew) {
+ entity.setTopicType(TopicType.NORMAL.name());
+ }
entity.setReadQueueNums(readQueues);
entity.setWriteQueueNums(writeQueues);
entity.setPerm(topicConfig.getPerm());
@@ -248,6 +256,7 @@ public class RocketMQAdminClientImpl implements AdminClient
{
topicConfig.setWriteQueueNums(writeQueues);
topicConfig.setReadQueueNums(readQueues);
topicConfig.setPerm(toRocketMQPerm(effectivePerm));
+ applyTopicType(topicConfig, topic.getType());
for (String addr : brokerAddrs) {
admin.createAndUpdateTopicConfig(addr, topicConfig);
@@ -285,6 +294,15 @@ public class RocketMQAdminClientImpl implements
AdminClient {
});
}
+ private void applyTopicType(TopicConfig topicConfig, TopicType topicType) {
+ if (topicType == null) {
+ return;
+ }
+ topicConfig.setAttributes(Map.of(
+ "+" + TopicAttributes.TOPIC_MESSAGE_TYPE_ATTRIBUTE.getName(),
+ topicType.name()));
+ }
+
@Override
public void deleteTopic(String instanceId, String name) {
String namesrvAddr = namesrvAddr(instanceId);
@@ -340,8 +358,9 @@ public class RocketMQAdminClientImpl implements AdminClient
{
}
String namesrvAddr = namesrvAddr(request.getInstanceId());
+ RPCHook credentialHook = credentialHook(request.getInstanceId());
- DefaultMQProducer producer = new
DefaultMQProducer(nextMessageSenderGroup());
+ DefaultMQProducer producer = new
DefaultMQProducer(nextMessageSenderGroup(), credentialHook);
producer.setNamesrvAddr(namesrvAddr);
producer.setSendMsgTimeout(5000);
@@ -571,6 +590,12 @@ public class RocketMQAdminClientImpl implements
AdminClient {
: namesrvAddr();
}
+ private RPCHook credentialHook(String instanceId) {
+ return StringUtils.hasText(instanceId)
+ ? runtimeAdminClientResolver.resolveCredentialHook(instanceId)
+ : null;
+ }
+
private <T> T executeForInstance(String instanceId,
MqAdminExtFactory.AdminAction<T> action) {
if (StringUtils.hasText(instanceId)) {
return runtimeAdminClientResolver.execute(instanceId, action);
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProvider.java
index 898263243..fd6771135 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProvider.java
@@ -83,16 +83,20 @@ public class RocketMQClusterProvider implements
ClusterProvider {
log.debug("NameServer address not configured, returning empty
cluster list");
return Collections.emptyList();
}
- return discoverClustersAt(namesrvAddr);
+ return discoverClustersAt(namesrvAddr, instanceId);
}
@Override
public List<ClusterVO> discoverClustersAt(String namesrvAddr) {
+ return discoverClustersAt(namesrvAddr, null);
+ }
+
+ private List<ClusterVO> discoverClustersAt(String namesrvAddr, String
instanceId) {
if (!StringUtils.hasText(namesrvAddr)) {
return Collections.emptyList();
}
try {
- return adminFactory.execute(namesrvAddr, null, admin -> {
+ return executeAdmin(instanceId, namesrvAddr, admin -> {
ClusterInfo clusterInfo = admin.examineBrokerClusterInfo();
if (clusterInfo == null || clusterInfo.getClusterAddrTable()
== null) {
return Collections.<ClusterVO>emptyList();
@@ -138,7 +142,7 @@ public class RocketMQClusterProvider implements
ClusterProvider {
}
try {
- return adminFactory.execute(namesrvAddr, null, admin -> {
+ return executeAdmin(instanceId, namesrvAddr, admin -> {
ClusterInfo clusterInfo = admin.examineBrokerClusterInfo();
if (clusterInfo == null || clusterInfo.getClusterAddrTable()
== null) {
return null;
@@ -308,6 +312,14 @@ public class RocketMQClusterProvider implements
ClusterProvider {
return properties.getNamesrvAddr();
}
+ private <T> T executeAdmin(String instanceId, String namesrvAddr,
+ MqAdminExtFactory.AdminAction<T> action) {
+ if (StringUtils.hasText(instanceId)) {
+ return runtimeAdminClientResolver.execute(instanceId, action);
+ }
+ return adminFactory.execute(namesrvAddr, null, action);
+ }
+
private List<ProxyVO> discoverProxiesViaHeartbeatSyncer(MQAdminExt admin) {
try {
ConsumerConnection connection =
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 e7d44c57b..217c7633c 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
@@ -27,11 +27,14 @@ import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.protocol.admin.TopicOffset;
import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
import org.apache.rocketmq.remoting.protocol.body.TopicList;
import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
+import org.apache.rocketmq.studio.common.domain.PageResult;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.common.util.Pagination;
import org.apache.rocketmq.studio.instance.dlq.DLQGroupVO;
import org.apache.rocketmq.studio.instance.dlq.DLQMessageVO;
import org.apache.rocketmq.studio.instance.dlq.DLQProvider;
@@ -51,6 +54,7 @@ import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Base64;
import java.util.Collections;
+import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -76,15 +80,21 @@ public class RocketMQDLQProvider implements DLQProvider {
@Override
public List<DLQGroupVO> listDLQGroups(String instanceId) {
- return runtimeAdminClientResolver.execute(instanceId,
this::listDLQGroups);
+ return listDLQGroups(instanceId, null, 1,
Integer.MAX_VALUE).getItems();
}
- private List<DLQGroupVO> listDLQGroups(MQAdminExt adminExt) throws
Exception {
- Set<String> topics;
+ @Override
+ public PageResult<DLQGroupVO> listDLQGroups(String instanceId, String
search, int page, int pageSize) {
+ return runtimeAdminClientResolver.execute(instanceId,
+ admin -> listDLQGroups(admin, search, page, pageSize));
+ }
+
+ private PageResult<DLQGroupVO> listDLQGroups(MQAdminExt adminExt, String
search, int page, int pageSize)
+ throws Exception {
TopicList topicList = adminExt.fetchAllTopicList();
- topics = topicList == null ? Collections.emptySet() :
topicList.getTopicList();
+ Set<String> topics = topicList == null ? Collections.emptySet() :
topicList.getTopicList();
- List<DLQGroupVO> groups = new ArrayList<>();
+ List<String> dlqTopics = new ArrayList<>();
for (String topic : topics) {
if (topic == null ||
!topic.startsWith(MixAll.DLQ_GROUP_TOPIC_PREFIX)) {
continue;
@@ -93,9 +103,19 @@ public class RocketMQDLQProvider implements DLQProvider {
if (!StringUtils.hasText(groupName)) {
continue;
}
- groups.add(buildDLQGroup(adminExt, groupName, topic));
+ if (search == null || groupName.contains(search) ||
topic.contains(search)) {
+ dlqTopics.add(topic);
+ }
}
- return groups;
+ dlqTopics.sort(Comparator.naturalOrder());
+ long offset = Pagination.pageOffset(page, pageSize);
+ int from = (int) Math.min(offset, dlqTopics.size());
+ int to = (int) Math.min(offset + pageSize, dlqTopics.size());
+ List<DLQGroupVO> groups = dlqTopics.subList(from, to).stream()
+ .map(topic -> buildDLQGroup(adminExt,
+
topic.substring(MixAll.DLQ_GROUP_TOPIC_PREFIX.length()), topic))
+ .toList();
+ return PageResult.of(groups, dlqTopics.size(), page, pageSize);
}
private DLQGroupVO buildDLQGroup(MQAdminExt adminExt, String groupName,
String dlqTopic) {
@@ -151,11 +171,12 @@ public class RocketMQDLQProvider implements DLQProvider {
}
String endpoint =
runtimeAdminClientResolver.resolveEndpoint(instanceId);
+ RPCHook credentialHook =
runtimeAdminClientResolver.resolveCredentialHook(instanceId);
String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + groupName;
DeadLetterScanResult scanResult;
try {
- scanResult = collectDeadLetters(endpoint, dlqTopic, begin, end,
RESEND_HARD_CAP);
+ scanResult = collectDeadLetters(endpoint, credentialHook,
dlqTopic, begin, end, RESEND_HARD_CAP);
} catch (BusinessException e) {
String detail = String.format("instanceId=%s, group=%s,
dlqTopic=%s, targetTopic=%s, "
+ "matched=0, resent=0, failed=0,
scanIncomplete=true, scanFailedQueues=all",
@@ -168,7 +189,7 @@ public class RocketMQDLQProvider implements DLQProvider {
int resent = 0;
int failed = 0;
if (!deadLetters.isEmpty()) {
- DefaultMQProducer producer = newProducer(endpoint);
+ DefaultMQProducer producer = newProducer(endpoint, credentialHook);
try {
producer.start();
for (MessageExt deadLetter : deadLetters) {
@@ -207,11 +228,12 @@ public class RocketMQDLQProvider implements DLQProvider {
public List<DLQMessageVO> exportMessages(String instanceId, String
groupName, Long startTime, Long endTime,
Integer maxCount) {
String endpoint =
runtimeAdminClientResolver.resolveEndpoint(instanceId);
+ RPCHook credentialHook =
runtimeAdminClientResolver.resolveCredentialHook(instanceId);
String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + groupName;
long end = endTime != null ? endTime : System.currentTimeMillis();
long begin = startTime != null ? startTime : end - ONE_HOUR_MILLIS;
int cap = maxCount == null || maxCount <= 0 ? RESEND_HARD_CAP :
Math.min(maxCount, RESEND_HARD_CAP);
- DeadLetterScanResult scanResult = collectDeadLetters(endpoint,
dlqTopic, begin, end, cap);
+ DeadLetterScanResult scanResult = collectDeadLetters(endpoint,
credentialHook, dlqTopic, begin, end, cap);
return scanResult.messages().stream().map(this::toExportVO).toList();
}
@@ -240,9 +262,9 @@ public class RocketMQDLQProvider implements DLQProvider {
}
}
- private DeadLetterScanResult collectDeadLetters(String endpoint, String
dlqTopic, long begin, long end,
- int cap) {
- DefaultMQPullConsumer consumer = newPullConsumer(endpoint);
+ private DeadLetterScanResult collectDeadLetters(String endpoint, RPCHook
credentialHook, String dlqTopic,
+ long begin, long end, int
cap) {
+ DefaultMQPullConsumer consumer = newPullConsumer(endpoint,
credentialHook);
List<MessageExt> result = new ArrayList<>();
int failedQueueCount = 0;
try {
@@ -393,15 +415,15 @@ public class RocketMQDLQProvider implements DLQProvider {
return null;
}
- private DefaultMQPullConsumer newPullConsumer(String endpoint) {
- DefaultMQPullConsumer consumer = new
DefaultMQPullConsumer("studio-dlq-query-group");
+ private DefaultMQPullConsumer newPullConsumer(String endpoint, RPCHook
credentialHook) {
+ DefaultMQPullConsumer consumer = new
DefaultMQPullConsumer("studio-dlq-query-group", credentialHook);
consumer.setInstanceName(ShortLivedClientName.next("studio-dlq-query"));
consumer.setNamesrvAddr(endpoint);
return consumer;
}
- private DefaultMQProducer newProducer(String endpoint) {
- DefaultMQProducer producer = new
DefaultMQProducer(nextResendProducerGroup());
+ private DefaultMQProducer newProducer(String endpoint, RPCHook
credentialHook) {
+ DefaultMQProducer producer = new
DefaultMQProducer(nextResendProducerGroup(), credentialHook);
producer.setInstanceName(ShortLivedClientName.next("studio-dlq-resend"));
producer.setRetryTimesWhenSendFailed(2);
producer.setNamesrvAddr(endpoint);
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 524559972..38846bfe5 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
@@ -25,6 +25,7 @@ import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageId;
import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.common.domain.enums.DeliveryStatus;
@@ -82,6 +83,8 @@ public class RocketMQMessageProvider implements
MessageProvider {
private static final long VIEW_MESSAGE_TIMEOUT_MILLIS = 3000L;
private static final long ONE_HOUR_MILLIS = 3600_000L;
private static final long ONE_DAY_MILLIS = 24 * ONE_HOUR_MILLIS;
+ private static final long MAX_TOPIC_QUERY_WINDOW_MILLIS = 7 *
ONE_DAY_MILLIS;
+ private static final int MAX_PULLS_PER_QUEUE = 1_000;
private static final int MAX_CONSECUTIVE_OFFSET_ILLEGAL = 3;
private static final Comparator<MessageRecordVO> TOPIC_QUERY_ORDER =
Comparator
.comparingLong(MessageRecordVO::getStoreTime)
@@ -94,12 +97,14 @@ public class RocketMQMessageProvider implements
MessageProvider {
public List<MessageRecordVO> queryMessages(String instanceId, String
topic, String msgId, String tag, String key,
Long startTime, Long endTime) {
String endpoint =
runtimeAdminClientResolver.resolveEndpoint(instanceId);
+ RPCHook credentialHook =
runtimeAdminClientResolver.resolveCredentialHook(instanceId);
return runtimeAdminClientResolver.execute(instanceId,
adminExt -> queryMessages(instanceId, (DefaultMQAdminExt)
adminExt, endpoint,
- topic, msgId, tag, key, startTime, endTime));
+ credentialHook, topic, msgId, tag, key, startTime,
endTime));
}
private List<MessageRecordVO> queryMessages(String instanceId,
DefaultMQAdminExt adminExt, String endpoint,
+ RPCHook credentialHook,
String topic, String msgId,
String tag, String key,
Long startTime, Long endTime)
{
@@ -118,8 +123,11 @@ public class RocketMQMessageProvider implements
MessageProvider {
queryType = "KEY";
result = queryByKey(adminExt, topic, key, tag, begin, end);
} else if (StringUtils.hasText(topic)) {
+ if (begin >= 0 && end >= 0 && end - begin >
MAX_TOPIC_QUERY_WINDOW_MILLIS) {
+ throw new BusinessException(400, "Topic message query time
range must not exceed 7 days");
+ }
queryType = "TOPIC";
- result = queryByTopic(endpoint, topic, tag, begin, end,
DEFAULT_TOPIC_LIMIT);
+ result = queryByTopic(endpoint, credentialHook, topic, tag, begin,
end, DEFAULT_TOPIC_LIMIT);
} else {
log.warn("queryMessages requires at least one of msgId/topic,
returning empty list");
return Collections.emptyList();
@@ -194,8 +202,9 @@ public class RocketMQMessageProvider implements
MessageProvider {
* Scan a topic within a time range using a short-lived pull consumer,
mirroring the approach
* used by the RocketMQ dashboard for time-range topic queries.
*/
- private List<MessageRecordVO> queryByTopic(String endpoint, String topic,
String tag, long begin, long end, int limit) {
- DefaultMQPullConsumer consumer = newPullConsumer("studio-msg-query",
endpoint);
+ private List<MessageRecordVO> queryByTopic(String endpoint, RPCHook
credentialHook, String topic, String tag,
+ long begin, long end, int
limit) {
+ DefaultMQPullConsumer consumer = newPullConsumer("studio-msg-query",
endpoint, credentialHook);
int resultLimit = Math.min(limit, TOPIC_QUERY_HARD_CAP);
PriorityQueue<MessageRecordVO> newestMessages = new
PriorityQueue<>(TOPIC_QUERY_ORDER);
try {
@@ -208,7 +217,12 @@ public class RocketMQMessageProvider implements
MessageProvider {
long minOffset = consumer.searchOffset(queue, begin);
long maxOffset = consumer.searchOffset(queue, end);
int consecutiveIllegalOffsets = 0;
+ int pullAttempts = 0;
for (long offset = minOffset; offset <= maxOffset; ) {
+ if (++pullAttempts > MAX_PULLS_PER_QUEUE) {
+ throw new BusinessException(400,
+ "Topic message query exceeded the per-queue
pull budget; narrow the time range");
+ }
PullResult pullResult = consumer.pull(queue, "*", offset,
32);
if (pullResult == null) {
log.warn("Stop topic query for {} because queue {}
returned no pull result", topic, queue);
@@ -253,6 +267,8 @@ public class RocketMQMessageProvider implements
MessageProvider {
}
}
}
+ } catch (BusinessException exception) {
+ throw exception;
} catch (Exception e) {
log.warn("queryByTopic(topic={}) failed: {}", topic,
e.getMessage());
throw new BusinessException(502, "Failed to query messages by
topic: " + e.getMessage());
@@ -596,8 +612,8 @@ public class RocketMQMessageProvider implements
MessageProvider {
return tag.equals(messageExt.getTags());
}
- private DefaultMQPullConsumer newPullConsumer(String groupPrefix, String
endpoint) {
- DefaultMQPullConsumer consumer = new DefaultMQPullConsumer(groupPrefix
+ "-group");
+ private DefaultMQPullConsumer newPullConsumer(String groupPrefix, String
endpoint, RPCHook credentialHook) {
+ DefaultMQPullConsumer consumer = new DefaultMQPullConsumer(groupPrefix
+ "-group", credentialHook);
consumer.setInstanceName(ShortLivedClientName.next(groupPrefix));
consumer.setNamesrvAddr(endpoint);
return consumer;
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/RuntimeAdminClientResolverTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/RuntimeAdminClientResolverTest.java
index 6a6bc5ad2..f54831898 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/RuntimeAdminClientResolverTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/RuntimeAdminClientResolverTest.java
@@ -141,6 +141,42 @@ class RuntimeAdminClientResolverTest {
assertThat(resolvedHook.getSessionCredentials().getSecretKey()).isEqualTo("admin-sk");
}
+ @Test
+ void resolvesCredentialHookForShortLivedRuntimeClients() {
+ InstanceVO instance = InstanceVO.builder()
+ .endpoint("namesrv-b:9876")
+ .adminCredentialRef("production-admin")
+ .build();
+ instance.setId(4L);
+ MqAdminProperties properties = new MqAdminProperties();
+ MqAdminProperties.Credential credential = new
MqAdminProperties.Credential();
+ credential.setAccessKey("admin-ak");
+ credential.setSecretKey("admin-sk");
+ properties.getCredentials().put("production-admin", credential);
+
when(instanceRepository.findByIdentifier("instance-b")).thenReturn(Optional.of(instance));
+ RuntimeAdminClientResolver resolver = new
RuntimeAdminClientResolver(instanceRepository, adminFactory,
+ properties);
+
+ org.apache.rocketmq.acl.common.AclClientRPCHook hook =
+ (org.apache.rocketmq.acl.common.AclClientRPCHook)
resolver.resolveCredentialHook("instance-b");
+
+
assertThat(hook.getSessionCredentials().getAccessKey()).isEqualTo("admin-ak");
+
assertThat(hook.getSessionCredentials().getSecretKey()).isEqualTo("admin-sk");
+ verifyNoInteractions(adminFactory);
+ }
+
+ @Test
+ void
returnsNoCredentialHookWhenTheSelectedInstanceHasNoCredentialReference() {
+ InstanceVO instance =
InstanceVO.builder().endpoint("namesrv-b:9876").build();
+ instance.setId(5L);
+
when(instanceRepository.findByIdentifier("instance-b")).thenReturn(Optional.of(instance));
+ RuntimeAdminClientResolver resolver = new
RuntimeAdminClientResolver(instanceRepository, adminFactory,
+ new MqAdminProperties());
+
+ assertThat(resolver.resolveCredentialHook("instance-b")).isNull();
+ verifyNoInteractions(adminFactory);
+ }
+
@Test
void rejectsUnknownOrIncompleteCredentialReferencesBeforeNetworkCalls() {
MqAdminProperties properties = new MqAdminProperties();
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/nameserver/NameServerConfigDiffServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/nameserver/NameServerConfigDiffServiceTest.java
index c660782b5..62344bd1d 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/nameserver/NameServerConfigDiffServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/nameserver/NameServerConfigDiffServiceTest.java
@@ -19,6 +19,7 @@ package org.apache.rocketmq.studio.cluster.nameserver;
import org.apache.rocketmq.studio.cluster.broker.ClusterService;
import org.apache.rocketmq.studio.cluster.broker.ClusterVO;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
+import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.tools.admin.MQAdminExt;
import org.junit.jupiter.api.BeforeEach;
@@ -36,7 +37,10 @@ import static
org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.assertj.core.api.Assertions.tuple;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.when;
import static org.mockito.Mockito.verify;
@@ -49,6 +53,9 @@ class NameServerConfigDiffServiceTest {
@Mock
private MqAdminExtFactory adminFactory;
+ @Mock
+ private RuntimeAdminClientResolver runtimeAdminClientResolver;
+
@Mock
private MQAdminExt admin;
@@ -56,7 +63,8 @@ class NameServerConfigDiffServiceTest {
@BeforeEach
void setUp() {
- service = new NameServerConfigDiffService(clusterService,
adminFactory);
+ service = new NameServerConfigDiffService(
+ clusterService, adminFactory, runtimeAdminClientResolver);
}
private void stubAdminFactory() {
@@ -111,10 +119,13 @@ class NameServerConfigDiffServiceTest {
@Test
void compareShouldResolveClusterThroughSelectedInstance() throws Exception
{
- stubAdminFactory();
when(clusterService.getCluster("cluster-a",
"instance-a")).thenReturn(cluster(
"ns-a:9876;ns-b:9876",
List.of(nameServer("ns-a:9876"), nameServer("ns-b:9876"))));
+ when(runtimeAdminClientResolver.execute(eq("instance-a"),
any())).thenAnswer(invocation -> {
+ MqAdminExtFactory.AdminAction<Object> action =
invocation.getArgument(1);
+ return action.apply(admin);
+ });
when(admin.getNameServerConfig(List.of("ns-a:9876")))
.thenReturn(Map.of("ns-a:9876", properties("listenPort",
"9876")));
when(admin.getNameServerConfig(List.of("ns-b:9876")))
@@ -124,6 +135,8 @@ class NameServerConfigDiffServiceTest {
assertThat(result.isComplete()).isTrue();
verify(clusterService).getCluster("cluster-a", "instance-a");
+ verify(runtimeAdminClientResolver, times(2)).execute(eq("instance-a"),
any());
+ verify(adminFactory, never()).execute(anyString(), isNull(), any());
}
@Test
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQControllerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQControllerTest.java
index a6063bd1c..c87cfd56d 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQControllerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQControllerTest.java
@@ -18,6 +18,7 @@
package org.apache.rocketmq.studio.instance.dlq;
import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.rocketmq.studio.common.domain.PageResult;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import
org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
@@ -67,27 +68,30 @@ class DLQControllerTest {
.status("ACTIVE")
.build();
-
when(dlqService.listDLQGroups("instance-1")).thenReturn(List.of(group));
+ when(dlqService.listDLQGroups("instance-1", null, 1, 20))
+ .thenReturn(PageResult.of(List.of(group), 1, 1, 20));
mockMvc.perform(get("/api/dlq").param("instanceId", "instance-1"))
.andExpect(status().isOk())
.andExpect(jsonPath("$.code").value(200))
- .andExpect(jsonPath("$.data").isArray())
- .andExpect(jsonPath("$.data[0].groupName").value("test-group"))
-
.andExpect(jsonPath("$.data[0].dlqTopic").value("%DLQ%test-group"))
- .andExpect(jsonPath("$.data[0].messageCount").value(10));
+ .andExpect(jsonPath("$.data.items").isArray())
+
.andExpect(jsonPath("$.data.items[0].groupName").value("test-group"))
+
.andExpect(jsonPath("$.data.items[0].dlqTopic").value("%DLQ%test-group"))
+ .andExpect(jsonPath("$.data.items[0].messageCount").value(10))
+ .andExpect(jsonPath("$.data.total").value(1));
}
@Test
void listDLQGroupsShouldPassInstanceId() throws Exception {
- when(dlqService.listDLQGroups(eq("instance-1"))).thenReturn(List.of());
+ when(dlqService.listDLQGroups(eq("instance-1"), isNull(), eq(1),
eq(20)))
+ .thenReturn(PageResult.empty(1, 20));
mockMvc.perform(get("/api/dlq")
.param("instanceId", "instance-1"))
.andExpect(status().isOk())
- .andExpect(jsonPath("$.data").isArray());
+ .andExpect(jsonPath("$.data.items").isArray());
- verify(dlqService).listDLQGroups(eq("instance-1"));
+ verify(dlqService).listDLQGroups(eq("instance-1"), isNull(), eq(1),
eq(20));
}
@Test
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQProviderStubTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQProviderStubTest.java
index 7b3483532..65041e43c 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQProviderStubTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQProviderStubTest.java
@@ -35,6 +35,15 @@ class DLQProviderStubTest {
.isEqualTo(501);
}
+ @Test
+ void pagedListDLQGroupsShouldFailExplicitlyWhenRealProviderIsMissing() {
+ assertThatThrownBy(() -> provider.listDLQGroups("instance-1",
"search", 1, 20))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("DLQ provider is not configured")
+ .extracting("code")
+ .isEqualTo(501);
+ }
+
@Test
void resendMessagesShouldFailExplicitlyWhenRealProviderIsMissing() {
assertThatThrownBy(() -> provider.resendMessages("instance-1",
"group-1", 1000L, 2000L, "target-topic"))
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQServiceTest.java
index 7936cc444..b80bdd16b 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQServiceTest.java
@@ -18,6 +18,7 @@
package org.apache.rocketmq.studio.instance.dlq;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.common.domain.PageResult;
import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
@@ -160,4 +161,30 @@ class DLQServiceTest {
verifyNoInteractions(dlqProvider);
}
+
+ @Test
+ void listDLQGroupsShouldDelegatePagedQueryWithTrimmedSearch() {
+ PageResult<DLQGroupVO> page = PageResult.of(List.of(), 0, 2, 50);
+ when(dlqProvider.listDLQGroups("instance-1", "order", 2,
50)).thenReturn(page);
+
+ PageResult<DLQGroupVO> result = dlqService.listDLQGroups("instance-1",
" order ", 2, 50);
+
+ assertThat(result).isSameAs(page);
+ verify(dlqProvider).listDLQGroups("instance-1", "order", 2, 50);
+ }
+
+ @Test
+ void listDLQGroupsShouldRejectInvalidPagination() {
+ assertThatThrownBy(() -> dlqService.listDLQGroups("instance-1", null,
0, 20))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Invalid page or pageSize");
+ assertThatThrownBy(() -> dlqService.listDLQGroups("instance-1", null,
1, 0))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Invalid page or pageSize");
+ assertThatThrownBy(() -> dlqService.listDLQGroups("instance-1", null,
1, 101))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Invalid page or pageSize");
+
+ verifyNoInteractions(dlqProvider);
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
index aa984d2c7..5ffb181ad 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
@@ -21,6 +21,7 @@ import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
@@ -47,6 +48,7 @@ import org.mockito.MockedConstruction;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -61,6 +63,7 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.doThrow;
@@ -288,6 +291,45 @@ class RocketMQAdminClientImplTest {
verify(topicMapper).updateById(existing);
}
+ @Test
+ void topicWritesSendMessageTypeAttributeToBroker() throws Exception {
+ TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new
MybatisConfiguration(), ""), RmqTopic.class);
+
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfoWithMaster());
+ when(topicMapper.selectOne(any())).thenReturn(null);
+ doNothing().when(adminExt).createAndUpdateTopicConfig(anyString(),
any(TopicConfig.class));
+
+ TopicVO topic = new TopicVO();
+ topic.setName("orders");
+ topic.setType(TopicType.FIFO);
+
+ adminClient.createTopic(topic);
+ adminClient.updateTopic(topic);
+
+ ArgumentCaptor<TopicConfig> captor =
ArgumentCaptor.forClass(TopicConfig.class);
+ verify(adminExt, times(2)).createAndUpdateTopicConfig(anyString(),
captor.capture());
+ assertThat(captor.getAllValues()).allSatisfy(config ->
+
assertThat(config.getAttributes()).containsEntry("+message.type",
TopicType.FIFO.name()));
+ }
+
+ @Test
+ void updateTopicWithoutTypePreservesExistingType() throws Exception {
+ TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new
MybatisConfiguration(), ""), RmqTopic.class);
+ RmqTopic existing = new RmqTopic();
+ existing.setTopicType(TopicType.TRANSACTION.name());
+
when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfoWithMaster());
+ when(topicMapper.selectOne(any())).thenReturn(existing);
+
+ TopicVO topic = new TopicVO();
+ topic.setName("orders");
+
+ adminClient.updateTopic(topic);
+
+ ArgumentCaptor<TopicConfig> captor =
ArgumentCaptor.forClass(TopicConfig.class);
+ verify(adminExt).createAndUpdateTopicConfig(anyString(),
captor.capture());
+
assertThat(captor.getValue().getAttributes()).doesNotContainKey("+message.type");
+
assertThat(existing.getTopicType()).isEqualTo(TopicType.TRANSACTION.name());
+ }
+
@Test
void updateTopicPreservesQueueCountsWhenNotSpecified() throws Exception {
TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new
MybatisConfiguration(), ""), RmqTopic.class);
@@ -541,8 +583,10 @@ class RocketMQAdminClientImplTest {
@Test
void sendMessageShouldAllowBodyAtMaximumSize() throws Exception {
when(properties.getNamesrvAddr()).thenReturn("10.0.0.1:9876");
+ List<List<?>> constructorArguments = new ArrayList<>();
try (MockedConstruction<DefaultMQProducer> mockedProducers =
mockConstruction(DefaultMQProducer.class, (producer,
context) -> {
+ constructorArguments.add(context.arguments());
doNothing().when(producer).start();
SendResult sendResult = new SendResult();
sendResult.setSendStatus(SendStatus.SEND_OK);
@@ -562,14 +606,21 @@ class RocketMQAdminClientImplTest {
ArgumentCaptor<Message> messageCaptor =
ArgumentCaptor.forClass(Message.class);
verify(producer).send(messageCaptor.capture());
assertThat(messageCaptor.getValue().getBody()).hasSize(4 * 1024 *
1024);
+ assertThat(constructorArguments).singleElement();
+ assertThat(constructorArguments.get(0)).hasSize(2);
+ assertThat(constructorArguments.get(0).get(1)).isNull();
}
}
@Test
- void sendMessageUsesSelectedInstanceEndpoint() throws Exception {
+ void sendMessageUsesSelectedInstanceEndpointAndCredentialHook() throws
Exception {
+ RPCHook credentialHook = mock(RPCHook.class);
+ List<List<?>> constructorArguments = new ArrayList<>();
when(runtimeAdminClientResolver.resolveEndpoint("instance-a")).thenReturn("10.0.0.2:9876");
+
when(runtimeAdminClientResolver.resolveCredentialHook("instance-a")).thenReturn(credentialHook);
try (MockedConstruction<DefaultMQProducer> mockedProducers =
mockConstruction(DefaultMQProducer.class, (producer,
context) -> {
+ constructorArguments.add(context.arguments());
doNothing().when(producer).start();
SendResult sendResult = new SendResult();
sendResult.setSendStatus(SendStatus.SEND_OK);
@@ -587,7 +638,11 @@ class RocketMQAdminClientImplTest {
DefaultMQProducer producer =
mockedProducers.constructed().getFirst();
verify(producer).setNamesrvAddr("10.0.0.2:9876");
+ assertThat(constructorArguments).singleElement();
+ assertThat(constructorArguments.get(0)).hasSize(2);
+
assertThat(constructorArguments.get(0).get(1)).isSameAs(credentialHook);
}
+ verify(runtimeAdminClientResolver).resolveCredentialHook("instance-a");
}
@Test
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProviderTest.java
index ac170cd6b..1c3e12c14 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProviderTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.rocketmq.studio.provider.apache;
+import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.body.Connection;
import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
@@ -23,21 +24,32 @@ import org.apache.rocketmq.remoting.protocol.body.KVTable;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.studio.cluster.broker.ClusterVO;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
+import org.apache.rocketmq.studio.cluster.broker.MqAdminProperties;
import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
import org.apache.rocketmq.studio.common.domain.enums.ClusterStatus;
+import org.apache.rocketmq.studio.common.domain.enums.InstanceType;
+import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.instance.InstanceRepository;
+import org.apache.rocketmq.studio.instance.InstanceVO;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.junit.jupiter.api.Test;
import java.util.HashMap;
import java.util.List;
+import java.util.Optional;
import java.util.Set;
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.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isA;
+import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
class RocketMQClusterProviderTest {
@@ -176,6 +188,66 @@ class RocketMQClusterProviderTest {
return new RocketMQClusterProvider(adminFactory, properties,
mock(RuntimeAdminClientResolver.class));
}
+ @Test
+ void discoverClustersShouldUseSelectedInstanceAdminCredential() throws
Exception {
+ DefaultMQAdminExt adminExt = mock(DefaultMQAdminExt.class);
+ MqAdminExtFactory adminFactory = mock(MqAdminExtFactory.class);
+ RocketMQClusterProvider provider =
newAuthenticatedInstanceProvider(adminFactory, adminExt);
+ when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfo());
+
+ List<ClusterVO> clusters = provider.discoverClusters("instance-a");
+
+ assertThat(clusters).singleElement()
+ .extracting(ClusterVO::getName)
+ .isEqualTo("DefaultCluster");
+ verify(adminFactory).execute(eq("10.0.0.2:9876"),
isA(AclClientRPCHook.class),
+ eq("cluster-admin"), any());
+ verify(adminFactory, never()).execute(eq("10.0.0.2:9876"), isNull(),
any());
+ }
+
+ @Test
+ void refreshClusterDetailShouldUseSelectedInstanceAdminCredential() throws
Exception {
+ DefaultMQAdminExt adminExt = mock(DefaultMQAdminExt.class);
+ MqAdminExtFactory adminFactory = mock(MqAdminExtFactory.class);
+ RocketMQClusterProvider provider =
newAuthenticatedInstanceProvider(adminFactory, adminExt);
+ when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfo());
+
+ ClusterVO cluster = provider.refreshClusterDetail("DefaultCluster",
"instance-a");
+
+ assertThat(cluster).isNotNull();
+ assertThat(cluster.getName()).isEqualTo("DefaultCluster");
+ verify(adminFactory).execute(eq("10.0.0.2:9876"),
isA(AclClientRPCHook.class),
+ eq("cluster-admin"), any());
+ verify(adminFactory, never()).execute(eq("10.0.0.2:9876"), isNull(),
any());
+ }
+
+ private RocketMQClusterProvider
newAuthenticatedInstanceProvider(MqAdminExtFactory adminFactory,
+
DefaultMQAdminExt adminExt) {
+ InstanceRepository instanceRepository = mock(InstanceRepository.class);
+ InstanceVO instance = InstanceVO.builder()
+ .name("Authenticated instance")
+ .vendor(InstanceVendor.APACHE)
+ .type(InstanceType.DIRECT)
+ .endpoint("10.0.0.2:9876")
+ .adminCredentialRef("cluster-admin")
+ .build();
+ instance.setId(1L);
+
when(instanceRepository.findByIdentifier("instance-a")).thenReturn(Optional.of(instance));
+ MqAdminProperties adminProperties = new MqAdminProperties();
+ MqAdminProperties.Credential credential = new
MqAdminProperties.Credential();
+ credential.setAccessKey("admin-ak");
+ credential.setSecretKey("admin-sk");
+ adminProperties.getCredentials().put("cluster-admin", credential);
+ RuntimeAdminClientResolver resolver =
+ new RuntimeAdminClientResolver(instanceRepository,
adminFactory, adminProperties);
+ when(adminFactory.execute(eq("10.0.0.2:9876"), any(),
eq("cluster-admin"), any()))
+ .thenAnswer(invocation ->
invocation.<MqAdminExtFactory.AdminAction<Object>>getArgument(3)
+ .apply(adminExt));
+ RocketMQProperties properties = new RocketMQProperties();
+ properties.setNamesrvAddr("10.0.0.1:9876");
+ return new RocketMQClusterProvider(adminFactory, properties, resolver);
+ }
+
@Test
void discoverClustersShouldExposeNameServerFailures() throws Exception {
DefaultMQAdminExt adminExt = mock(DefaultMQAdminExt.class);
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProviderTest.java
index 7609fa665..0035e97c7 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProviderTest.java
@@ -27,10 +27,12 @@ import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
import org.apache.rocketmq.remoting.protocol.body.TopicList;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
+import org.apache.rocketmq.studio.common.domain.PageResult;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.instance.dlq.DLQGroupVO;
import org.apache.rocketmq.studio.instance.dlq.DLQMessageVO;
@@ -46,6 +48,7 @@ import org.mockito.MockedConstruction;
import org.mockito.junit.jupiter.MockitoExtension;
import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
import java.util.Base64;
import java.util.List;
import java.util.Set;
@@ -62,6 +65,7 @@ import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockConstruction;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
@@ -125,6 +129,37 @@ class RocketMQDLQProviderTest {
});
}
+ @Test
+ void listDLQGroupsShouldPageAndFilterGroupsTest() throws Exception {
+ TopicList topicList = new TopicList();
+ topicList.setTopicList(Set.of(
+ MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-c",
+ MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a",
+ MixAll.DLQ_GROUP_TOPIC_PREFIX + "order-b",
+ "normal-topic"));
+ when(adminExt.fetchAllTopicList()).thenReturn(topicList);
+ when(adminExt.examineTopicStats(anyString())).thenReturn(new
TopicStatsTable());
+
+ PageResult<DLQGroupVO> firstPage =
provider.listDLQGroups("instance-a", null, 1, 2);
+
+ assertThat(firstPage.getTotal()).isEqualTo(3);
+ assertThat(firstPage.getPage()).isEqualTo(1);
+ assertThat(firstPage.getSize()).isEqualTo(2);
+ assertThat(firstPage.getItems()).extracting(DLQGroupVO::getGroupName)
+ .containsExactly("group-a", "group-c");
+
+ PageResult<DLQGroupVO> secondPage =
provider.listDLQGroups("instance-a", null, 2, 2);
+
+ assertThat(secondPage.getItems()).extracting(DLQGroupVO::getGroupName)
+ .containsExactly("order-b");
+
+ PageResult<DLQGroupVO> filtered = provider.listDLQGroups("instance-a",
"order", 1, 20);
+
+ assertThat(filtered.getTotal()).isEqualTo(1);
+ assertThat(filtered.getItems()).extracting(DLQGroupVO::getGroupName)
+ .containsExactly("order-b");
+ }
+
@Test
void resendMessagesShouldRejectInvertedTimeRangeBeforeCreatingConsumers() {
@@ -194,8 +229,10 @@ class RocketMQDLQProviderTest {
@Test
void resendMessagesDoesNotPullWhenDlqQueueSetIsNull() throws Exception {
String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
+ List<List<?>> consumerConstructorArguments = new ArrayList<>();
try (MockedConstruction<DefaultMQPullConsumer> mockedConsumers =
mockConstruction(DefaultMQPullConsumer.class, (consumer,
context) -> {
+ consumerConstructorArguments.add(context.arguments());
doNothing().when(consumer).start();
when(consumer.fetchSubscribeMessageQueues(dlqTopic)).thenReturn(null);
doNothing().when(consumer).shutdown();
@@ -205,6 +242,9 @@ class RocketMQDLQProviderTest {
provider.resendMessages("instance-a", "group-a", 100L, 200L,
"target-topic");
assertThat(mockedConsumers.constructed()).hasSize(1);
+ assertThat(consumerConstructorArguments).singleElement();
+ assertThat(consumerConstructorArguments.get(0)).hasSize(2);
+ assertThat(consumerConstructorArguments.get(0).get(1)).isNull();
DefaultMQPullConsumer consumer =
mockedConsumers.constructed().get(0);
verify(consumer).setNamesrvAddr("namesrv-a:9876");
verify(consumer).start();
@@ -219,6 +259,54 @@ class RocketMQDLQProviderTest {
contains("matched=0, resent=0, failed=0"),
eq("NO_MESSAGES"));
verify(runtimeAdminClientResolver).resolveEndpoint("instance-a");
+ verify(runtimeAdminClientResolver).resolveCredentialHook("instance-a");
+ }
+
+ @Test
+ void resendMessagesUsesSelectedInstanceCredentialHookForScanAndResend()
throws Exception {
+ String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
+ MessageQueue queue = new MessageQueue(dlqTopic, "broker-a", 0);
+ MessageExt deadLetter = new MessageExt();
+ deadLetter.setMsgId("acl-dlq-message");
+ deadLetter.setTopic(dlqTopic);
+ deadLetter.setBody(new byte[] {1});
+ deadLetter.setStoreTimestamp(150L);
+ PullResult pullResult = new PullResult(PullStatus.FOUND, 1L, 0L, 0L,
List.of(deadLetter));
+ SendResult sendResult = new SendResult();
+ sendResult.setSendStatus(SendStatus.SEND_OK);
+ RPCHook credentialHook = mock(RPCHook.class);
+ List<List<?>> consumerConstructorArguments = new ArrayList<>();
+ List<List<?>> producerConstructorArguments = new ArrayList<>();
+
when(runtimeAdminClientResolver.resolveCredentialHook("instance-a")).thenReturn(credentialHook);
+
+ try (MockedConstruction<DefaultMQPullConsumer> mockedConsumers =
+ mockConstruction(DefaultMQPullConsumer.class, (consumer,
context) -> {
+ consumerConstructorArguments.add(context.arguments());
+ doNothing().when(consumer).start();
+
when(consumer.fetchSubscribeMessageQueues(dlqTopic)).thenReturn(Set.of(queue));
+ when(consumer.searchOffset(queue,
100L)).thenReturn(0L);
+ when(consumer.searchOffset(queue,
200L)).thenReturn(0L);
+ when(consumer.pull(queue, "*", 0L,
32)).thenReturn(pullResult);
+ doNothing().when(consumer).shutdown();
+ });
+ MockedConstruction<DefaultMQProducer> mockedProducers =
+ mockConstruction(DefaultMQProducer.class, (producer,
context) -> {
+ producerConstructorArguments.add(context.arguments());
+ doNothing().when(producer).start();
+
when(producer.send(any(Message.class))).thenReturn(sendResult);
+ doNothing().when(producer).shutdown();
+ })) {
+ provider.resendMessages("instance-a", "group-a", 100L, 200L,
"target-topic");
+
+ assertThat(mockedConsumers.constructed()).singleElement();
+ assertThat(mockedProducers.constructed()).singleElement();
+ }
+
+ assertThat(consumerConstructorArguments).singleElement();
+
assertThat(consumerConstructorArguments.get(0).get(1)).isSameAs(credentialHook);
+ assertThat(producerConstructorArguments).singleElement();
+
assertThat(producerConstructorArguments.get(0).get(1)).isSameAs(credentialHook);
+ verify(runtimeAdminClientResolver).resolveCredentialHook("instance-a");
}
@Test
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java
index c2a1a8248..3d8dae308 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMessageProviderTest.java
@@ -27,6 +27,7 @@ import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageId;
import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
import org.apache.rocketmq.studio.common.exception.BusinessException;
@@ -48,6 +49,7 @@ import org.mockito.junit.jupiter.MockitoExtension;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
@@ -97,8 +99,10 @@ class RocketMQMessageProviderTest {
@Test
void queryByTopicReturnsEmptyListWhenQueueSetIsNull() throws Exception {
+ List<List<?>> constructorArguments = new ArrayList<>();
try (MockedConstruction<DefaultMQPullConsumer> mockedConsumers =
mockConstruction(DefaultMQPullConsumer.class, (consumer,
context) -> {
+ constructorArguments.add(context.arguments());
doNothing().when(consumer).start();
when(consumer.fetchSubscribeMessageQueues("TopicA")).thenReturn(null);
doNothing().when(consumer).shutdown();
@@ -108,6 +112,10 @@ class RocketMQMessageProviderTest {
assertThat(messages).isEmpty();
assertThat(mockedConsumers.constructed()).hasSize(1);
DefaultMQPullConsumer consumer =
mockedConsumers.constructed().get(0);
+ assertThat(constructorArguments).singleElement();
+ assertThat(constructorArguments.get(0)).hasSize(2);
+
assertThat(constructorArguments.get(0).get(0)).isEqualTo("studio-msg-query-group");
+ assertThat(constructorArguments.get(0).get(1)).isNull();
verify(consumer).setNamesrvAddr("namesrv-a:9876");
verify(consumer).start();
verify(consumer).fetchSubscribeMessageQueues("TopicA");
@@ -115,11 +123,36 @@ class RocketMQMessageProviderTest {
verify(consumer).shutdown();
}
verify(runtimeAdminClientResolver).resolveEndpoint("instance-a");
+ verify(runtimeAdminClientResolver).resolveCredentialHook("instance-a");
verify(runtimeAdminClientResolver).execute(eq("instance-a"), any());
verify(queryHistoryService).recordMessageQuery("instance-a", "TOPIC",
"TopicA", null, null, null,
100L, 200L, 0);
}
+ @Test
+ void queryByTopicUsesSelectedInstanceCredentialHook() throws Exception {
+ RPCHook credentialHook = mock(RPCHook.class);
+ List<List<?>> constructorArguments = new ArrayList<>();
+
when(runtimeAdminClientResolver.resolveCredentialHook("instance-a")).thenReturn(credentialHook);
+
+ try (MockedConstruction<DefaultMQPullConsumer> mockedConsumers =
+ mockConstruction(DefaultMQPullConsumer.class, (consumer,
context) -> {
+ constructorArguments.add(context.arguments());
+ doNothing().when(consumer).start();
+
when(consumer.fetchSubscribeMessageQueues("TopicA")).thenReturn(Set.of());
+ doNothing().when(consumer).shutdown();
+ })) {
+ provider.queryMessages("instance-a", "TopicA", null, null, null,
100L, 200L);
+
+ assertThat(mockedConsumers.constructed()).singleElement();
+ assertThat(constructorArguments).singleElement();
+ assertThat(constructorArguments.get(0)).hasSize(2);
+
assertThat(constructorArguments.get(0).get(0)).isEqualTo("studio-msg-query-group");
+
assertThat(constructorArguments.get(0).get(1)).isSameAs(credentialHook);
+ }
+ verify(runtimeAdminClientResolver).resolveCredentialHook("instance-a");
+ }
+
@Test
void queryMessagesShouldRejectInvertedTimeRangeBeforeAdminLookup() throws
Exception {
assertThatThrownBy(() -> provider.queryMessages(
@@ -150,6 +183,18 @@ class RocketMQMessageProviderTest {
anyString(), anyString(), anyString(), any(), any(), anyInt());
}
+ @Test
+ void queryMessagesShouldRejectTopicRangesLongerThanSevenDays() throws
Exception {
+ assertThatThrownBy(() -> provider.queryMessages(
+ "instance-a", "TopicA", null, null, null, 0L, 8L * 24 * 60 *
60 * 1000))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Topic message query time range must not exceed 7
days")
+ .satisfies(error -> assertThat(((BusinessException)
error).getCode()).isEqualTo(400));
+
+ verify(queryHistoryService, never()).recordMessageQuery(anyString(),
anyString(), anyString(),
+ anyString(), anyString(), anyString(), any(), any(), anyInt());
+ }
+
@Test
void queryByKeySurfacesAdminFailure() throws Exception {
when(adminExt.queryMessage("TopicA", "order-1", 64, 100L, 200L))
diff --git a/web/src/api/message.ts b/web/src/api/message.ts
index 023283a1a..9bc1cfa12 100644
--- a/web/src/api/message.ts
+++ b/web/src/api/message.ts
@@ -65,6 +65,13 @@ export interface DLQGroup {
statsAvailable?: boolean;
}
+export interface DLQGroupPage {
+ items: DLQGroup[];
+ total: number;
+ page: number;
+ size: number;
+}
+
export interface DLQResendResult {
matched: number;
resent: number;
@@ -92,8 +99,10 @@ export async function getMessageTrace(msgId: string,
instanceId?: string, topic?
}
// ─── DLQ ────────────────────────────────────────────────────────
-export async function listDLQGroups(instanceId: string) {
- const res = await client.get<{ data: DLQGroup[] }>('/dlq', { params: {
instanceId } });
+export async function listDLQGroups(instanceId: string, search?: string, page
= 1, pageSize = 20) {
+ const res = await client.get<{ data: DLQGroupPage }>('/dlq', {
+ params: { instanceId, search, page, pageSize },
+ });
return res.data.data;
}
diff --git a/web/src/pages/instance/__tests__/DLQPage.test.tsx
b/web/src/pages/instance/__tests__/DLQPage.test.tsx
index 060944897..a6a247513 100644
--- a/web/src/pages/instance/__tests__/DLQPage.test.tsx
+++ b/web/src/pages/instance/__tests__/DLQPage.test.tsx
@@ -21,7 +21,7 @@ import userEvent from '@testing-library/user-event';
import type React from 'react';
import { MemoryRouter } from 'react-router-dom';
import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from
'vitest';
-import type { DLQGroup, DLQResendResult } from '../../../api/message';
+import type { DLQGroup, DLQGroupPage, DLQResendResult } from
'../../../api/message';
import { LangProvider } from '../../../i18n/LangContext';
import * as messageService from '../../../services/messageService';
import DLQPage from '../dlq';
@@ -83,6 +83,13 @@ const secondDlqGroup: DLQGroup = {
status: 'ACTIVE',
};
+const pageOf = (items: DLQGroup[]): DLQGroupPage => ({
+ items,
+ total: items.length,
+ page: 1,
+ size: 20,
+});
+
const renderWithProviders = (ui: React.ReactElement) =>
render(
<App>
@@ -125,7 +132,7 @@ describe('DLQ page', () => {
value: revokeObjectURL,
});
clickSpy = vi.spyOn(HTMLAnchorElement.prototype,
'click').mockImplementation(() => {});
- vi.mocked(messageService.listDLQGroups).mockResolvedValue([dlqGroup]);
+
vi.mocked(messageService.listDLQGroups).mockResolvedValue(pageOf([dlqGroup]));
});
afterEach(() => {
@@ -138,7 +145,7 @@ describe('DLQ page', () => {
expect(await screen.findByText('cg-order')).toBeInTheDocument();
expect(screen.getByText('%DLQ%cg-order')).toBeInTheDocument();
- expect(messageService.listDLQGroups).toHaveBeenCalledWith('instance-1');
+ expect(messageService.listDLQGroups).toHaveBeenCalledWith('instance-1',
undefined, 1, 20);
});
it('surfaces unavailable DLQ provider errors when loading groups', async ()
=> {
@@ -151,7 +158,7 @@ describe('DLQ page', () => {
});
it('does not present unavailable DLQ statistics as an empty queue', async ()
=> {
-
vi.mocked(messageService.listDLQGroups).mockResolvedValue([unavailableDlqGroup]);
+
vi.mocked(messageService.listDLQGroups).mockResolvedValue(pageOf([unavailableDlqGroup]));
renderWithProviders(<DLQPage />);
const row = (await screen.findByText('cg-order')).closest('tr');
@@ -164,10 +171,10 @@ describe('DLQ page', () => {
});
it('sorts DLQ rows with missing enqueue timestamps', async () => {
- vi.mocked(messageService.listDLQGroups).mockResolvedValue([
+ vi.mocked(messageService.listDLQGroups).mockResolvedValue(pageOf([
dlqGroup,
{ ...secondDlqGroup, lastEnqueueTime: null },
- ]);
+ ]));
const user = userEvent.setup();
renderWithProviders(<DLQPage />);
@@ -218,7 +225,7 @@ describe('DLQ page', () => {
});
it('exports summaries for the selected groups in one CSV file', async () => {
- vi.mocked(messageService.listDLQGroups).mockResolvedValue([dlqGroup,
secondDlqGroup]);
+
vi.mocked(messageService.listDLQGroups).mockResolvedValue(pageOf([dlqGroup,
secondDlqGroup]));
const user = userEvent.setup();
renderWithProviders(<DLQPage />);
@@ -243,13 +250,13 @@ describe('DLQ page', () => {
});
it('neutralizes formulas hidden behind a leading line feed in CSV summary
exports', async () => {
- vi.mocked(messageService.listDLQGroups).mockResolvedValue([
+ vi.mocked(messageService.listDLQGroups).mockResolvedValue(pageOf([
{
...dlqGroup,
groupName: '\n=1+1',
dlqTopic: '%DLQ%formula',
},
- ]);
+ ]));
const user = userEvent.setup();
renderWithProviders(<DLQPage />);
@@ -264,8 +271,8 @@ describe('DLQ page', () => {
it('clears a selected group when refreshed data shows no dead-letter
messages', async () => {
vi.mocked(messageService.listDLQGroups)
- .mockResolvedValueOnce([dlqGroup])
- .mockResolvedValueOnce([{ ...dlqGroup, messageCount: 0 }]);
+ .mockResolvedValueOnce(pageOf([dlqGroup]))
+ .mockResolvedValueOnce(pageOf([{ ...dlqGroup, messageCount: 0 }]));
const user = userEvent.setup();
renderWithProviders(<DLQPage />);
@@ -331,12 +338,12 @@ describe('DLQ page', () => {
});
it('clears retry state before loading groups for a newly selected instance',
async () => {
- let resolveSecondInstance!: (groups: DLQGroup[]) => void;
+ let resolveSecondInstance!: (page: DLQGroupPage) => void;
vi.mocked(messageService.listDLQGroups)
- .mockResolvedValueOnce([dlqGroup])
+ .mockResolvedValueOnce(pageOf([dlqGroup]))
.mockImplementationOnce(
() =>
- new Promise<DLQGroup[]>((resolve) => {
+ new Promise<DLQGroupPage>((resolve) => {
resolveSecondInstance = resolve;
}),
);
@@ -355,13 +362,13 @@ describe('DLQ page', () => {
);
await waitFor(() => {
-
expect(messageService.listDLQGroups).toHaveBeenLastCalledWith('instance-2');
+
expect(messageService.listDLQGroups).toHaveBeenLastCalledWith('instance-2',
undefined, 1, 20);
});
await waitFor(() => {
expect(screen.queryByRole('row', { name: /cg-order/
})).not.toBeInTheDocument();
});
- resolveSecondInstance([secondDlqGroup]);
+ resolveSecondInstance(pageOf([secondDlqGroup]));
expect(await screen.findByText('-cg-"payment"')).toBeInTheDocument();
});
@@ -374,8 +381,8 @@ describe('DLQ page', () => {
}),
);
vi.mocked(messageService.listDLQGroups)
- .mockResolvedValueOnce([dlqGroup])
- .mockResolvedValueOnce([secondDlqGroup]);
+ .mockResolvedValueOnce(pageOf([dlqGroup]))
+ .mockResolvedValueOnce(pageOf([secondDlqGroup]));
const user = userEvent.setup();
renderWithProviders(<DLQPage />);
diff --git a/web/src/pages/instance/dlq.tsx b/web/src/pages/instance/dlq.tsx
index 56c9e17c5..b670b36b7 100644
--- a/web/src/pages/instance/dlq.tsx
+++ b/web/src/pages/instance/dlq.tsx
@@ -97,6 +97,9 @@ const DLQPage = () => {
const { t } = useLang();
const { selectedInstanceId, selectInstance, instanceOptions } =
useInstanceFilter();
const [groups, setGroups] = useState<DLQGroup[]>([]);
+ const [total, setTotal] = useState(0);
+ const [page, setPage] = useState(1);
+ const [pageSize, setPageSize] = useState(20);
const [loading, setLoading] = useState(true);
const [refreshKey, setRefreshKey] = useState(0);
const [search, setSearch] = useState('');
@@ -139,6 +142,8 @@ const DLQPage = () => {
if (prevScopeKey !== scopeKey) {
setPrevScopeKey(scopeKey);
setGroups([]);
+ setTotal(0);
+ setPage(1);
setSelectedGroupNames([]);
setDetailGroup(null);
setRetryModalOpen(false);
@@ -162,13 +167,14 @@ const DLQPage = () => {
};
}
- void listDLQGroups(selectedInstanceId)
- .then((nextGroups) => {
+ void listDLQGroups(selectedInstanceId, search || undefined, page, pageSize)
+ .then((result) => {
if (!cancelled) {
- setGroups(nextGroups);
+ setGroups(result.items);
+ setTotal(result.total);
setLoadError(null);
const availableGroups = new Set(
- nextGroups.filter((group) => group.messageCount > 0).map((group)
=> group.groupName),
+ result.items.filter((group) => group.messageCount > 0).map((group)
=> group.groupName),
);
setSelectedGroupNames((selected) =>
selected.filter((groupName) => availableGroups.has(groupName)),
@@ -185,17 +191,7 @@ const DLQPage = () => {
return () => {
cancelled = true;
};
- }, [refreshKey, selectedInstanceId]);
-
- /* ─── Filtering ─── */
- const filtered = useMemo(() => {
- if (!search) return groups;
- return groups.filter(
- (g) =>
- g.groupName.toLowerCase().includes(search.toLowerCase()) ||
- g.dlqTopic.toLowerCase().includes(search.toLowerCase()),
- );
- }, [groups, search]);
+ }, [refreshKey, selectedInstanceId, search, page, pageSize]);
const selectedGroups = useMemo(() => {
const selected = new Set(selectedGroupNames);
@@ -407,7 +403,10 @@ const DLQPage = () => {
allowClear
value={search}
onChange={(e) => setSearch(e.target.value)}
- onSearch={setSearch}
+ onSearch={(value) => {
+ setSearch(value);
+ setPage(1);
+ }}
style={{ width: 320 }}
prefix={<MagnifyingGlass size={14} color="#9CA3AF" />}
/>
@@ -446,7 +445,7 @@ const DLQPage = () => {
<Card styles={{ body: { padding: 0 } }}>
<Table
columns={columns}
- dataSource={filtered}
+ dataSource={groups}
rowKey="groupName"
loading={loading}
rowSelection={{
@@ -458,9 +457,17 @@ const DLQPage = () => {
}),
}}
pagination={{
- pageSize: 20,
+ current: page,
+ pageSize,
+ total,
showSizeChanger: true,
- showTotal: (total) => `共 ${total} 个 Group`,
+ pageSizeOptions: [20, 50, 100],
+ showTotal: (totalCount) => `共 ${totalCount} 个 Group`,
+ onChange: (nextPage, nextPageSize) => {
+ setPage(nextPage);
+ setPageSize(nextPageSize);
+ setSelectedGroupNames([]);
+ },
}}
size="small"
/>
diff --git a/web/src/services/messageService.test.ts
b/web/src/services/messageService.test.ts
index b3f12704b..7521764ab 100644
--- a/web/src/services/messageService.test.ts
+++ b/web/src/services/messageService.test.ts
@@ -66,12 +66,25 @@ describe('message service mock data', () => {
it('returns copied DLQ group rows', async () => {
const first = await listDLQGroups('instance-1');
- expect(first[0].groupName).toBe('cg-order-processor');
+ expect(first.items[0].groupName).toBe('cg-order-processor');
+ expect(first.total).toBeGreaterThanOrEqual(1);
- first[0].groupName = 'mutated-group';
+ first.items[0].groupName = 'mutated-group';
const second = await listDLQGroups('instance-1');
- expect(second[0].groupName).toBe('cg-order-processor');
- expect(second[0]).not.toBe(first[0]);
+ expect(second.items[0].groupName).toBe('cg-order-processor');
+ expect(second.items[0]).not.toBe(first.items[0]);
+ });
+
+ it('filters and pages mock DLQ groups', async () => {
+ const filtered = await listDLQGroups('instance-1', 'order', 1, 20);
+ expect(filtered.items.every((group) =>
group.groupName.includes('order'))).toBe(true);
+ expect(filtered.total).toBe(filtered.items.length);
+ expect(filtered.page).toBe(1);
+ expect(filtered.size).toBe(20);
+
+ const all = await listDLQGroups('instance-1', undefined, 1, 1);
+ expect(all.items).toHaveLength(1);
+ expect(all.total).toBeGreaterThanOrEqual(1);
});
});
diff --git a/web/src/services/messageService.ts
b/web/src/services/messageService.ts
index 72c060fb4..a738c1281 100644
--- a/web/src/services/messageService.ts
+++ b/web/src/services/messageService.ts
@@ -6,6 +6,7 @@ import type {
MessageRecord,
TraceRecord,
DLQGroup,
+ DLQGroupPage,
DLQResendResult,
} from '../api/message';
import { mockMessages, mockMessageTraces } from '../mock/messages';
@@ -60,9 +61,26 @@ export async function getMessageTrace(
return messageApi.getMessageTrace(msgId, instanceId, topic);
}
-export async function listDLQGroups(instanceId: string): Promise<DLQGroup[]> {
- if (isMockMode()) return (mockDLQGroups as unknown as
DLQGroup[]).map(cloneDLQGroup);
- return messageApi.listDLQGroups(instanceId);
+export async function listDLQGroups(
+ instanceId: string,
+ search?: string,
+ page = 1,
+ pageSize = 20,
+): Promise<DLQGroupPage> {
+ if (isMockMode()) {
+ const groups = (mockDLQGroups as unknown as DLQGroup[]).filter(
+ (group) =>
+ !search || group.groupName.includes(search) ||
group.dlqTopic.includes(search),
+ );
+ const from = Math.min((page - 1) * pageSize, groups.length);
+ return {
+ items: groups.slice(from, from + pageSize).map(cloneDLQGroup),
+ total: groups.length,
+ page,
+ size: pageSize,
+ };
+ }
+ return messageApi.listDLQGroups(instanceId, search, page, pageSize);
}
export async function resendDLQ(data: {
