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 529ba3ed5 fix(dlq): bound retained bodies before read conversion
(#5365)
529ba3ed5 is described below
commit 529ba3ed54d1aabdf97f8ccd763356e0281140d0
Author: zmuxuny <[email protected]>
AuthorDate: Mon Oct 19 03:54:48 2026 -0700
fix(dlq): bound retained bodies before read conversion (#5365)
Verified against `rocketmq-studio` at f81e73a49 before merging:
缺陷仍在（`RocketMQDLQProvider.java:341-346` `listMessages` 先全量
`map(this::toExportVO)` 再 `subList`；`scanDeadLetters`(:444) 无字节预算，全文无
`retainedBodyBytes` /
merge-tree clean against the current trunk. Merged as-is (squash) per the
maintainer review of 2026-10-19.
---
docs/api-spec.md | 6 +
.../provider/apache/RocketMQDLQProvider.java | 44 ++-
.../studio/instance/dlq/DLQControllerTest.java | 32 +++
.../apache/RocketMQDLQPayloadBudgetTest.java | 299 +++++++++++++++++++++
4 files changed, 370 insertions(+), 11 deletions(-)
diff --git a/docs/api-spec.md b/docs/api-spec.md
index fa0d30721..3efa974b0 100644
--- a/docs/api-spec.md
+++ b/docs/api-spec.md
@@ -1839,6 +1839,12 @@ POST /api/messages/direct-consume
## 9. 死信队列 Dead Letter Queue
+DLQ 消息列表、JSON 导出和 Excel 导出在扫描时共享每次请求 **10 MiB 原始消息体**的累积上限。
+超过上限会返回 `413` 和提示 `DLQ read exceeds the 10 MiB message body limit; narrow the
time range`，
+不会返回截断的消息体或部分成功的下载。请缩小时间窗口，或在 JSON 导出中减小 `maxCount`。
+Excel 的 `msgIds` 过滤在扫描后执行，因此扫描本身仍受此上限约束。原有条数上限和重发行为不变。
+此限制不包含消息属性、客户端已解码的单个拉取批次或并发请求；它不是整个 JVM 的内存上限。
+
### 9.1 获取 DLQ 列表
```
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 c440a4590..d64ca19c6 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
@@ -89,6 +89,9 @@ public class RocketMQDLQProvider implements DLQProvider {
private static final long ONE_HOUR_MILLIS = 3600_000L;
private static final int RESEND_HARD_CAP = 5000;
+ // Bounds retained raw bodies before read-side UTF-8/Base64 expansion.
This is not a total
+ // heap limit: message metadata and the client's already-decoded pull
batch are separate.
+ private static final long MAX_READ_BODY_BYTES = 10L * 1024 * 1024;
private static final int MAX_PAGE_SIZE = 100;
private static final int MAX_CONSECUTIVE_OFFSET_ILLEGAL = 3;
private static final int MAX_REPORTED_RESEND_FAILURES = 100;
@@ -205,7 +208,8 @@ public class RocketMQDLQProvider implements DLQProvider {
DeadLetterScanResult scanResult;
try {
- scanResult = collectDeadLetters(instanceId, dlqTopic, begin, end,
RESEND_HARD_CAP);
+ scanResult = collectDeadLetters(
+ instanceId, dlqTopic, begin, end, RESEND_HARD_CAP,
Long.MAX_VALUE);
} 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",
@@ -336,14 +340,13 @@ public class RocketMQDLQProvider implements DLQProvider {
if (begin >= end) {
throw new BusinessException(400, "DLQ detail start time must be
before end time");
}
- List<DLQMessageVO> all = collectDeadLetters(instanceId, dlqTopic,
begin, end, RESEND_HARD_CAP)
- .messages().stream()
- .map(this::toExportVO)
- .toList();
+ List<MessageExt> all = collectDeadLetters(
+ instanceId, dlqTopic, begin, end, RESEND_HARD_CAP,
MAX_READ_BODY_BYTES).messages();
long offset = Pagination.pageOffset(page, pageSize);
int from = (int) Math.min(offset, all.size());
int to = (int) Math.min(offset + pageSize, all.size());
- return PageResult.of(all.subList(from, to), all.size(), page,
pageSize);
+ List<DLQMessageVO> items = all.subList(from,
to).stream().map(this::toExportVO).toList();
+ return PageResult.of(items, all.size(), page, pageSize);
}
@Override
@@ -356,7 +359,8 @@ public class RocketMQDLQProvider implements DLQProvider {
String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + groupName;
long end = endTime != null ? endTime : System.currentTimeMillis();
long begin = startTime != null ? startTime : end - ONE_HOUR_MILLIS;
- DeadLetterScanResult scanResult = collectDeadLetters(instanceId,
dlqTopic, begin, end, RESEND_HARD_CAP);
+ DeadLetterScanResult scanResult = collectDeadLetters(
+ instanceId, dlqTopic, begin, end, RESEND_HARD_CAP,
MAX_READ_BODY_BYTES);
Set<String> selected = msgIds == null ? Collections.emptySet()
: new java.util.HashSet<>(msgIds);
List<DLQMessageVO> messages = scanResult.messages().stream()
@@ -389,7 +393,8 @@ public class RocketMQDLQProvider implements DLQProvider {
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(instanceId,
dlqTopic, begin, end, cap);
+ DeadLetterScanResult scanResult = collectDeadLetters(
+ instanceId, dlqTopic, begin, end, cap, MAX_READ_BODY_BYTES);
return DLQExportResultVO.builder()
.messages(scanResult.messages().stream().map(this::toExportVO).toList())
.truncated(scanResult.truncated())
@@ -436,14 +441,15 @@ public class RocketMQDLQProvider implements DLQProvider {
}
private DeadLetterScanResult collectDeadLetters(String instanceId, String
dlqTopic,
- long begin, long end, int
cap) {
+ long begin, long end, int
cap, long bodyByteBudget) {
return runtimeAdminClientResolver.executePullConsumer(instanceId,
- consumer -> scanDeadLetters(consumer, dlqTopic, begin, end,
cap));
+ consumer -> scanDeadLetters(consumer, dlqTopic, begin, end,
cap, bodyByteBudget));
}
private DeadLetterScanResult scanDeadLetters(DefaultMQPullConsumer
consumer, String dlqTopic,
- long begin, long end, int
cap) {
+ long begin, long end, int
cap, long bodyByteBudget) {
List<MessageExt> result = new ArrayList<>();
+ long retainedBodyBytes = 0;
int failedQueueCount = 0;
boolean truncated = false;
try {
@@ -511,6 +517,12 @@ public class RocketMQDLQProvider implements DLQProvider {
for (MessageExt messageExt :
pullResult.getMsgFoundList()) {
if (messageExt.getStoreTimestamp() >= begin
&& messageExt.getStoreTimestamp() <= end) {
+ byte[] body = messageExt.getBody();
+ int bodyBytes = body == null ? 0 : body.length;
+ if (bodyBytes > bodyByteBudget -
retainedBodyBytes) {
+ throw new ReadBodyLimitExceededException();
+ }
+ retainedBodyBytes += bodyBytes;
result.add(messageExt);
if (result.size() >= cap) {
truncated = true;
@@ -519,6 +531,10 @@ public class RocketMQDLQProvider implements DLQProvider {
}
}
}
+ } catch (ReadBodyLimitExceededException e) {
+ // A read limit is fatal for the whole request, not a
failed queue that can
+ // be skipped while returning a partial export as if it
had succeeded.
+ throw e;
} catch (Exception e) {
failedQueueCount++;
log.warn("Failed to scan DLQ queue {} in {}: {}", queue,
dlqTopic, e.getMessage());
@@ -837,6 +853,12 @@ public class RocketMQDLQProvider implements DLQProvider {
}
}
+ private static final class ReadBodyLimitExceededException extends
BusinessException {
+ private ReadBodyLimitExceededException() {
+ super(413, "DLQ read exceeds the 10 MiB message body limit; narrow
the time range");
+ }
+ }
+
private record DeadLetterScanResult(List<MessageExt> messages, int
failedQueueCount, boolean truncated,
boolean topicMissing) {
boolean scanIncomplete() {
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 9a8f8c483..94a252ed8 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
@@ -24,8 +24,11 @@ import org.springframework.context.annotation.Import;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.rocketmq.studio.common.domain.PageResult;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.webmvc.test.autoconfigure.AutoConfigureMockMvc;
import org.springframework.boot.webmvc.test.autoconfigure.WebMvcTest;
@@ -553,6 +556,35 @@ class DLQControllerTest extends WebMvcAuthTestSupport {
verifyNoInteractions(dlqService);
}
+ @ParameterizedTest
+ @ValueSource(strings = {"messages", "export", "export-excel"})
+ void dlqReadShouldReturnPayloadTooLargeWhenBodyBudgetIsExceededTest(String
operation) throws Exception {
+ BusinessException tooLarge = new BusinessException(413,
+ "DLQ read exceeds the 10 MiB message body limit; narrow the
time range");
+ String path;
+ if ("messages".equals(operation)) {
+ path = "/api/dlq/test-group/messages";
+ when(dlqService.listMessages(eq("instance-1"), eq("test-group"),
isNull(), isNull(), eq(1), eq(20)))
+ .thenThrow(tooLarge);
+ } else if ("export".equals(operation)) {
+ path = "/api/dlq/export";
+ when(dlqService.exportMessages(eq("instance-1"), eq("test-group"),
isNull(), isNull(), isNull()))
+ .thenThrow(tooLarge);
+ } else {
+ path = "/api/dlq/export-excel";
+ when(dlqService.exportExcel(eq("instance-1"), eq("test-group"),
isNull(), isNull(), isNull()))
+ .thenThrow(tooLarge);
+ }
+
+ mockMvc.perform(get(path)
+ .param("instanceId", "instance-1")
+ .param("groupName", "test-group"))
+ .andExpect(status().isPayloadTooLarge())
+
.andExpect(header().doesNotExist(HttpHeaders.CONTENT_DISPOSITION))
+ .andExpect(jsonPath("$.code").value(413))
+ .andExpect(jsonPath("$.message").value(tooLarge.getMessage()));
+ }
+
@Test
void exportDLQExcelShouldRejectBracketedSelectionTest() throws Exception {
mockMvc.perform(get("/api/dlq/export-excel")
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQPayloadBudgetTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQPayloadBudgetTest.java
new file mode 100644
index 000000000..f395349ee
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQPayloadBudgetTest.java
@@ -0,0 +1,299 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.provider.apache;
+
+import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
+import org.apache.rocketmq.client.consumer.PullResult;
+import org.apache.rocketmq.client.consumer.PullStatus;
+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.MixAll;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageConst;
+import org.apache.rocketmq.common.message.MessageAccessor;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.studio.cluster.broker.MqClientPool;
+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.DLQExportResultVO;
+import org.apache.rocketmq.studio.instance.dlq.DLQMessageVO;
+import org.apache.rocketmq.studio.instance.dlq.DLQResendResultVO;
+import org.apache.rocketmq.studio.ops.audit.AuditService;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+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.eq;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class RocketMQDLQPayloadBudgetTest {
+
+ private static final String DLQ_TOPIC = MixAll.DLQ_GROUP_TOPIC_PREFIX +
"group-a";
+ private static final int BODY_BUDGET = 10 * 1024 * 1024;
+ private static final int HALF_BUDGET = BODY_BUDGET / 2;
+
+ @Mock
+ private RuntimeAdminClientResolver resolver;
+ @Mock
+ private DefaultMQPullConsumer consumer;
+ @Mock
+ private DefaultMQProducer producer;
+ @Mock
+ private AuditService auditService;
+
+ private RocketMQDLQProvider provider;
+
+ @BeforeEach
+ void setUp() {
+ when(resolver.executePullConsumer(eq("instance-a"),
any())).thenAnswer(invocation -> {
+ MqClientPool.ClientAction<DefaultMQPullConsumer, Object> action =
invocation.getArgument(1);
+ return action.apply(consumer);
+ });
+ provider = new RocketMQDLQProvider(resolver, auditService);
+ }
+
+ @ParameterizedTest
+ @EnumSource(ReadOperation.class)
+ void
readRejectsAggregateBodyOverflowBeforeConvertingMessagesTest(ReadOperation
operation) throws Exception {
+ MessageExt first = spy(message("first", new byte[HALF_BUDGET]));
+ MessageExt second = spy(message("second", new byte[HALF_BUDGET + 1]));
+ stubMessages(List.of(first, second));
+
+ assertReadTooLarge(operation);
+
+ verify(first, never()).getProperties();
+ verify(second, never()).getProperties();
+ verifyNoInteractions(producer, auditService);
+ }
+
+ @Test
+ void readCountsBodiesAcrossPullBatchesTest() throws Exception {
+ MessageQueue queue = new MessageQueue(DLQ_TOPIC, "broker-a", 0);
+
when(consumer.fetchSubscribeMessageQueues(DLQ_TOPIC)).thenReturn(Set.of(queue));
+ stubWindow(queue, 3L);
+ when(consumer.pull(queue, "*", 0L, 32)).thenReturn(found(1L,
+ List.of(message("first", new byte[HALF_BUDGET]))));
+ when(consumer.pull(queue, "*", 1L, 32)).thenReturn(found(2L,
+ List.of(message("second", new byte[HALF_BUDGET + 1]))));
+
+ assertReadTooLarge(ReadOperation.JSON);
+
+ verify(consumer, never()).pull(queue, "*", 2L, 32);
+ }
+
+ @Test
+ void
readCountsBodiesAcrossQueuesAndDoesNotReturnPartialSuccessOnOverflowTest()
throws Exception {
+ MessageQueue first = new MessageQueue(DLQ_TOPIC, "broker-a", 0);
+ MessageQueue unavailable = new MessageQueue(DLQ_TOPIC, "broker-b", 0);
+ MessageQueue last = new MessageQueue(DLQ_TOPIC, "broker-c", 0);
+ when(consumer.fetchSubscribeMessageQueues(DLQ_TOPIC))
+ .thenReturn(new LinkedHashSet<>(List.of(first, unavailable,
last)));
+ stubWindow(first, 1L);
+ when(consumer.searchOffset(unavailable, 100L)).thenThrow(new
BusinessException(502, "Unavailable"));
+ stubWindow(last, 1L);
+ when(consumer.pull(first, "*", 0L, 32)).thenReturn(found(1L,
+ List.of(message("first", new byte[HALF_BUDGET]))));
+ when(consumer.pull(last, "*", 0L, 32)).thenReturn(found(1L,
+ List.of(message("last", new byte[HALF_BUDGET + 1]))));
+
+ assertReadTooLarge(ReadOperation.JSON);
+ }
+
+ @Test
+ void exportPreservesCompleteBodiesAtExactBudgetTest() throws Exception {
+ stubMessages(List.of(message("first", new byte[HALF_BUDGET]),
+ message("second", new byte[HALF_BUDGET])));
+
+ DLQExportResultVO result = provider.exportMessages("instance-a",
"group-a", 100L, 200L, 1000);
+
+ assertThat(result.isTruncated()).isFalse();
+ assertThat(result.getFailedQueueCount()).isZero();
+ assertThat(result.getMessages()).hasSize(2).allSatisfy(message -> {
+ assertThat(message.getBody()).hasSize(HALF_BUDGET);
+ assertThat(message.getBodyBase64()).hasSize(4 * ((HALF_BUDGET + 2)
/ 3));
+ });
+ }
+
+ @Test
+ void nullBodyDoesNotExceedAnExactlyExhaustedBudgetTest() throws Exception {
+ stubMessages(List.of(message("first", new byte[HALF_BUDGET]),
+ message("second", new byte[HALF_BUDGET]), message("empty",
null)));
+
+ PageResult<DLQMessageVO> result = provider.listMessages("instance-a",
"group-a", 100L, 200L, 3, 1);
+
+ assertThat(result.getTotal()).isEqualTo(3);
+ assertThat(result.getItems()).singleElement().satisfies(message -> {
+ assertThat(message.getMsgId()).isEqualTo("empty");
+ assertThat(message.getBody()).isNull();
+ assertThat(message.getBodyBase64()).isNull();
+ });
+ }
+
+ @Test
+ void listConvertsOnlyTheRequestedPageTest() throws Exception {
+ MessageExt first = spy(message("first", new byte[] {1}));
+ MessageExt second = spy(message("second", new byte[] {2}));
+ MessageExt third = spy(message("third", new byte[] {3}));
+ stubMessages(List.of(first, second, third));
+
+ PageResult<DLQMessageVO> result = provider.listMessages("instance-a",
"group-a", 100L, 200L, 2, 1);
+
+ assertThat(result.getTotal()).isEqualTo(3);
+ assertThat(result.getPage()).isEqualTo(2);
+ assertThat(result.getSize()).isEqualTo(1);
+ assertThat(result.getItems()).singleElement().satisfies(message -> {
+ assertThat(message.getMsgId()).isEqualTo("second");
+ assertThat(message.getBodyBase64()).isEqualTo("Ag==");
+ });
+ for (MessageExt outsidePage : List.of(first, third)) {
+ verify(outsidePage).getBody(); // Budget accounting only, without
UTF-8/Base64 conversion.
+ verify(outsidePage, never()).getProperties();
+ verify(outsidePage, never()).getMsgId();
+ }
+ }
+
+ @Test
+ void listBeyondTheLastPageDoesNotConvertMessagesTest() throws Exception {
+ MessageExt message = spy(message("first", new byte[] {1}));
+ stubMessages(List.of(message));
+
+ PageResult<DLQMessageVO> result = provider.listMessages(
+ "instance-a", "group-a", 100L, 200L, Integer.MAX_VALUE, 100);
+
+ assertThat(result.getTotal()).isEqualTo(1);
+ assertThat(result.getItems()).isEmpty();
+ verify(message).getBody();
+ verify(message, never()).getProperties();
+ }
+
+ @Test
+ void readDoesNotChargeBodiesOutsideTheRequestedTimeWindowTest() throws
Exception {
+ MessageExt outsideWindow = message("outside", new byte[BODY_BUDGET +
1]);
+ outsideWindow.setStoreTimestamp(99L);
+ stubMessages(List.of(outsideWindow, message("inside", new byte[]
{1})));
+
+ DLQExportResultVO result = provider.exportMessages("instance-a",
"group-a", 100L, 200L, 1000);
+
+
assertThat(result.getMessages()).extracting(DLQMessageVO::getMsgId).containsExactly("inside");
+ assertThat(result.isTruncated()).isFalse();
+ }
+
+ @Test
+ void exportKeepsTheRequestedCountCapBeforeConsideringFurtherBodiesTest()
throws Exception {
+ stubMessages(List.of(message("first", new byte[] {1}),
+ message("unneeded", new byte[BODY_BUDGET + 1])));
+
+ DLQExportResultVO result = provider.exportMessages("instance-a",
"group-a", 100L, 200L, 1);
+
+
assertThat(result.getMessages()).extracting(DLQMessageVO::getMsgId).containsExactly("first");
+ assertThat(result.getLimit()).isEqualTo(1);
+ assertThat(result.isTruncated()).isTrue();
+ assertThat(result.getFailedQueueCount()).isZero();
+ }
+
+ @Test
+ void resendKeepsBodiesAboveTheReadBudgetUnchangedTest() throws Exception {
+ MessageExt first = message("first", new byte[HALF_BUDGET]);
+ MessageExt second = message("second", new byte[HALF_BUDGET + 1]);
+ MessageAccessor.setProperties(first,
Map.of(MessageConst.PROPERTY_RETRY_TOPIC, "orders"));
+ MessageAccessor.setProperties(second,
Map.of(MessageConst.PROPERTY_RETRY_TOPIC, "orders"));
+ stubMessages(List.of(first, second));
+ when(resolver.executeProducer(eq("instance-a"),
any())).thenAnswer(invocation -> {
+ MqClientPool.ClientAction<DefaultMQProducer, Object> action =
invocation.getArgument(1);
+ return action.apply(producer);
+ });
+ SendResult accepted = new SendResult();
+ accepted.setSendStatus(SendStatus.SEND_OK);
+ when(producer.send(any(Message.class))).thenReturn(accepted);
+
+ DLQResendResultVO result = provider.resendMessages("instance-a",
"group-a", 100L, 200L, null);
+
+ assertThat(result.getMatched()).isEqualTo(2);
+ assertThat(result.getResent()).isEqualTo(2);
+ assertThat(result.getOutcome()).isEqualTo("SUCCESS");
+ ArgumentCaptor<Message> sent = ArgumentCaptor.forClass(Message.class);
+ verify(producer, times(2)).send(sent.capture());
+
assertThat(sent.getAllValues().get(0).getBody()).isSameAs(first.getBody());
+
assertThat(sent.getAllValues().get(1).getBody()).isSameAs(second.getBody());
+ }
+
+ private void assertReadTooLarge(ReadOperation operation) {
+ assertThatThrownBy(() -> {
+ switch (operation) {
+ case LIST -> provider.listMessages("instance-a", "group-a",
100L, 200L, 1, 1);
+ case JSON -> provider.exportMessages("instance-a", "group-a",
100L, 200L, 1000);
+ case EXCEL -> provider.exportExcel("instance-a", "group-a",
100L, 200L, null);
+ case SELECTED_EXCEL -> provider.exportExcel(
+ "instance-a", "group-a", 100L, 200L, List.of("first"));
+ }
+ }).isInstanceOfSatisfying(BusinessException.class, exception -> {
+ assertThat(exception.getCode()).isEqualTo(413);
+ assertThat(exception.getMessage()).contains("10 MiB", "narrow the
time range");
+ });
+ }
+
+ private void stubMessages(List<MessageExt> messages) throws Exception {
+ MessageQueue queue = new MessageQueue(DLQ_TOPIC, "broker-a", 0);
+
when(consumer.fetchSubscribeMessageQueues(DLQ_TOPIC)).thenReturn(Set.of(queue));
+ stubWindow(queue, messages.size());
+ when(consumer.pull(queue, "*", 0L,
32)).thenReturn(found(messages.size(), messages));
+ }
+
+ private void stubWindow(MessageQueue queue, long endOffset) throws
Exception {
+ when(consumer.searchOffset(queue, 100L)).thenReturn(0L);
+ when(consumer.searchOffset(queue, 201L)).thenReturn(endOffset);
+ }
+
+ private static PullResult found(long nextOffset, List<MessageExt>
messages) {
+ return new PullResult(PullStatus.FOUND, nextOffset, 0L, nextOffset,
messages);
+ }
+
+ private static MessageExt message(String msgId, byte[] body) {
+ MessageExt message = new MessageExt();
+ message.setMsgId(msgId);
+ message.setTopic(DLQ_TOPIC);
+ message.setStoreTimestamp(150L);
+ message.setBody(body);
+ return message;
+ }
+
+ private enum ReadOperation {
+ LIST, JSON, EXCEL, SELECTED_EXCEL
+ }
+}
