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 4781e3ee4 fix(dlq): confine resends to the owning instance and
cluster, and stop lossy exports (#5448)
4781e3ee4 is described below
commit 4781e3ee46c757d009ac7acf0a93f3c300f9fb93
Author: 烤化の初雪 <[email protected]>
AuthorDate: Fri Oct 9 17:47:09 2026 +0800
fix(dlq): confine resends to the owning instance and cluster, and stop
lossy exports (#5448)
Four fixes on the DLQ / message-redelivery path, same author and same files.
- A resend was not gated on ownership of the group and the target topic, so
a request could act on a
resource belonging to another instance; `DLQService` now resolves both
through
`ResourceOwnershipGuard`. (#5448)
- `validateResendTargetTopic` checked the target topic against every
cluster on the NameServer rather
than against the DLQ topic's own cluster. It now takes the DLQ topic and
compares clusters, and a
route lookup that fails falls through instead of turning the "no route
yet" 404 into a 502. (#5451)
- The Excel export wrote message bodies straight into a cell, where POI's
32767-character limit cut
them - possibly mid surrogate pair; bodies are now abbreviated on a
character boundary, with the
raw bytes still available in the `Body Base64` column. (#5424)
- Redelivery accepted a message whose properties had been truncated for
display, i.e. a projection
that cannot prove what the original carried; `requireExactSourceMessage`
now refuses when the
property cap was reached or a user property was abbreviated. (#5427)
`RocketMQDLQProviderTest`, `DLQServiceTest`, `DLQControllerTest`,
`DLQMessageExcelRowTest`,
`MetadataServiceTest`, `MessageRedeliveryToolHandlerTest`: 173 tests green
with the dev profile,
0 checkstyle violations. Rebased onto today's trunk, keeping both the `Body
Base64` column (#5205)
and the abbreviated body.
Folded in #5424, #5427 and #5451.
---
.../studio/common/util/MessagePropertyDisplay.java | 5 +-
.../studio/instance/dlq/DLQMessageExcelRow.java | 22 ++-
.../rocketmq/studio/instance/dlq/DLQService.java | 35 +++-
.../studio/instance/topic/MetadataService.java | 75 +++++++-
.../message/MessageRedeliveryToolHandler.java | 9 +-
.../provider/apache/RocketMQDLQProvider.java | 69 +++++++-
.../instance/dlq/DLQMessageExcelRowTest.java | 17 ++
.../studio/instance/dlq/DLQServiceTest.java | 65 +++++++
.../studio/instance/topic/MetadataServiceTest.java | 189 +++++++++++++++++++++
.../message/MessageRedeliveryToolHandlerTest.java | 51 ++++++
.../provider/apache/RocketMQDLQProviderTest.java | 106 ++++++++++++
11 files changed, 628 insertions(+), 15 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/common/util/MessagePropertyDisplay.java
b/server/src/main/java/org/apache/rocketmq/studio/common/util/MessagePropertyDisplay.java
index 66b8e3f37..d008cdc37 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/common/util/MessagePropertyDisplay.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/common/util/MessagePropertyDisplay.java
@@ -34,6 +34,9 @@ public final class MessagePropertyDisplay {
public static final int MAX_PROPERTIES = 64;
+ /** Marker appended by {@link #abbreviate} to a value it cut at the
code-point cap. */
+ public static final String ABBREVIATION_SUFFIX = "...";
+
/**
* Caps a single property value by code point rather than by UTF-16 {@code
char}. The two differ
* for supplementary characters (an emoji, a CJK extension character),
which occupy two chars:
@@ -86,6 +89,6 @@ public final class MessagePropertyDisplay {
private static String abbreviate(String value) {
// A property value is end-user text, so the cut has to land on a code
point boundary: it is
// always passed through whole or abbreviated with the ellipsis, never
split mid-character.
- return TextBounds.truncate(value, MAX_PROPERTY_VALUE_CODE_POINTS,
"...");
+ return TextBounds.truncate(value, MAX_PROPERTY_VALUE_CODE_POINTS,
ABBREVIATION_SUFFIX);
}
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQMessageExcelRow.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQMessageExcelRow.java
index 121f47a41..c640c5d29 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQMessageExcelRow.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/dlq/DLQMessageExcelRow.java
@@ -30,6 +30,14 @@ import java.time.format.DateTimeFormatter;
@Data
public class DLQMessageExcelRow {
+ /**
+ * POI rejects any cell whose text exceeds this many characters, so one
oversized dead-letter
+ * body would otherwise abort the whole export with a 502. The budget is
counted in UTF-16
+ * chars, not code points, which is why {@code TextBounds} cannot be
reused here.
+ */
+ private static final int MAX_EXCEL_CELL_CHARS = 32_767;
+ private static final String TRUNCATION_SUFFIX = "...[truncated]";
+
private static final DateTimeFormatter STORE_TIME_FORMAT =
DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
@@ -62,8 +70,20 @@ public class DLQMessageExcelRow {
Instant.ofEpochMilli(vo.getStoreTime()),
ZoneId.systemDefault()).format(STORE_TIME_FORMAT));
row.setReconsumeTimes(vo.getReconsumeTimes());
row.setKeys(vo.getKeys());
- row.setBody(vo.getBody());
+ row.setBody(abbreviateBody(vo.getBody()));
row.setBodyBase64(vo.getBody() == null ? vo.getBodyBase64() : null);
return row;
}
+
+ private static String abbreviateBody(String body) {
+ if (body == null || body.length() <= MAX_EXCEL_CELL_CHARS) {
+ return body;
+ }
+ int keep = MAX_EXCEL_CELL_CHARS - TRUNCATION_SUFFIX.length();
+ // Back off onto a code-point boundary so the cut never splits a
surrogate pair.
+ if (Character.isHighSurrogate(body.charAt(keep - 1))) {
+ keep--;
+ }
+ return body.substring(0, keep) + TRUNCATION_SUFFIX;
+ }
}
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 7acd5c3e3..22a118225 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
@@ -19,13 +19,19 @@ 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.instance.InstanceVO;
+import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard;
+import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Kind;
+import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Resource;
import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
+import java.util.ArrayList;
import java.util.List;
+import java.util.function.Supplier;
@Service
@RequiredArgsConstructor
@@ -37,6 +43,7 @@ public class DLQService {
private final DLQProvider dlqProvider;
private final InstanceProviderRegistry providerRegistry;
+ private final ResourceOwnershipGuard ownershipGuard;
public PageResult<DLQGroupVO> listDLQGroups(String instanceId, String
search, int page, int pageSize) {
requireApacheInstance(instanceId);
@@ -61,8 +68,9 @@ public class DLQService {
String normalizedTargetTopic = normalizeOptional(targetTopic);
log.info("Resending DLQ messages: group={}, targetTopic={}",
normalizedGroupName, normalizedTargetTopic);
- return dlqProvider.resendMessages(
- instanceId, normalizedGroupName, startTime, endTime,
normalizedTargetTopic);
+ return withOwnedDlqTargets(instanceId, normalizedGroupName,
normalizedTargetTopic, () ->
+ dlqProvider.resendMessages(
+ instanceId, normalizedGroupName, startTime, endTime,
normalizedTargetTopic));
}
public DLQExportResultVO exportMessages(String instanceId, String
groupName, Long startTime, Long endTime,
@@ -98,8 +106,27 @@ public class DLQService {
String normalizedTargetTopic = normalizeOptional(targetTopic);
log.info("Resending selected DLQ messages: group={}, count={},
targetTopic={}",
normalizedGroupName, normalizedMsgIds.size(),
normalizedTargetTopic);
- return dlqProvider.resendMessages(
- instanceId, normalizedGroupName, normalizedMsgIds,
normalizedTargetTopic);
+ return withOwnedDlqTargets(instanceId, normalizedGroupName,
normalizedTargetTopic, () ->
+ dlqProvider.resendMessages(
+ instanceId, normalizedGroupName, normalizedMsgIds,
normalizedTargetTopic));
+ }
+
+ /**
+ * A resend reads the group's dead-letter topic and publishes back into a
topic, so it is a
+ * group+topic write like the send/redeliver paths - which check exactly
these resources. Without
+ * the check any Apache instance registered against the same cluster could
replay another
+ * instance's dead letters and publish them into a topic it does not own:
the DLQ topic
+ * {@code %DLQ%<group>} belongs to the group's owner, which is what this
guard expresses.
+ */
+ private <T> T withOwnedDlqTargets(String instanceId, String groupName,
String targetTopic,
+ Supplier<T> action) {
+ InstanceVO instance = ownershipGuard.requireInstance(instanceId);
+ List<Resource> resources = new ArrayList<>();
+ resources.add(new Resource(Kind.GROUP, groupName));
+ if (StringUtils.hasText(targetTopic)) {
+ resources.add(ownershipGuard.topicResource(targetTopic));
+ }
+ return ownershipGuard.withOwned(instance, resources, action);
}
public DLQExcelExportResultVO exportExcel(String instanceId, String
groupName, Long startTime, Long endTime,
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
index 1cc1562f9..02321edc6 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
@@ -41,6 +41,7 @@ import
org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
import org.apache.rocketmq.studio.common.domain.enums.SubscriptionMode;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.common.util.CsvUtil;
+import org.apache.rocketmq.studio.common.util.MessagePropertyDisplay;
import org.apache.rocketmq.studio.common.util.SystemTopicFilter;
import org.apache.rocketmq.studio.instance.group.CreateConsumerGroupDTO;
import org.apache.rocketmq.studio.instance.group.ImportConsumerGroupsResultVO;
@@ -323,6 +324,7 @@ public class MetadataService {
if (original == null) {
throw new BusinessException(400, "Message to redeliver must not be
null");
}
+ requireExactSourceMessage(original);
String destination = StringUtils.hasText(targetTopic)
? targetTopic.trim()
: MixAll.getRetryTopic(group);
@@ -341,6 +343,71 @@ public class MetadataService {
return ownershipGuard.withOwned(instance, resources, () ->
sendMessage(request));
}
+ /**
+ * A source loaded by {@link #findMessageForRedelivery} is the message
explorer's display
+ * projection, not the stored record: the body stops at {@code
MAX_BODY_DISPLAY_BYTES}, a binary
+ * body is replaced by Base64 text, and the property map is capped.
Publishing that projection
+ * stores different bytes and silently drops properties, so a lossy source
is refused instead.
+ * <p>The property check fires on the two loss modes the projection can
prove. Reaching
+ * {@link MessagePropertyDisplay#MAX_PROPERTIES} entries means whole
entries were dropped: the
+ * projection cannot say which ones, but the cap itself is the proof, and
since
+ * {@code limitProperties} sorts by key the lowercase user keys are
exactly what is lost. An
+ * abbreviated value on a visible user key is the other mode.
System-reserved keys alone are
+ * never enough — {@link #redeliveryProperties} discards them before the
send, so a map that
+ * stayed under the cap with intact values republishes the same user
properties.
+ * <p>Refuses display projections whose body or user properties cannot be
republished exactly.
+ */
+ public void requireExactSourceMessage(MessageRecordVO original) {
+ if (original.isBodyTruncated()) {
+ throw new BusinessException(409,
+ "Source message body is truncated for display and cannot
be redelivered exactly");
+ }
+ if ("BASE64".equalsIgnoreCase(original.getBodyEncoding())) {
+ throw new BusinessException(409,
+ "Source message body is binary and cannot be redelivered
exactly");
+ }
+ if (original.isPropertiesTruncated() &&
isPropertyDisplayCapReached(original.getProperties())) {
+ throw new BusinessException(409,
+ "Source message properties reached the display cap, so
whole entries were dropped "
+ + "and cannot be redelivered exactly");
+ }
+ if (original.isPropertiesTruncated() &&
hasAbbreviatedUserProperty(original.getProperties())) {
+ throw new BusinessException(409,
+ "Source message properties are truncated for display and
cannot be redelivered exactly");
+ }
+ }
+
+ /**
+ * True when the projection kept {@link
MessagePropertyDisplay#MAX_PROPERTIES} entries, which
+ * happens only if the raw map held more and the overflow was dropped
whole.
+ */
+ private static boolean isPropertyDisplayCapReached(Map<String, String>
properties) {
+ return properties != null && properties.size() >=
MessagePropertyDisplay.MAX_PROPERTIES;
+ }
+
+ /**
+ * True when a <em>user</em> property visible in the display projection
carries an abbreviated
+ * value: republishing it would store a cut value. System-reserved keys
are excluded —
+ * {@link #redeliveryProperties} drops them before the send, abbreviated
or not.
+ */
+ private static boolean hasAbbreviatedUserProperty(Map<String, String>
properties) {
+ if (properties == null || properties.isEmpty()) {
+ return false;
+ }
+ return properties.entrySet().stream()
+ .anyMatch(entry -> !isSystemProperty(entry.getKey())
+ && isAbbreviatedValue(entry.getValue()));
+ }
+
+ /**
+ * Mirrors the display projection's abbreviation: any value whose rendered
form ends in the
+ * {@code ABBREVIATION_SUFFIX} marker was cut, whether it sits on a user
or a system key.
+ */
+ private static boolean isAbbreviatedValue(String value) {
+ return value != null && value.length() >=
MessagePropertyDisplay.ABBREVIATION_SUFFIX.length()
+ && value.endsWith(MessagePropertyDisplay.ABBREVIATION_SUFFIX);
+ }
+
/** Drops the system-reserved keys {@code Message.putUserProperty} would
reject (§15.5.1 KEYS defect). */
private static Map<String, String> redeliveryProperties(Map<String,
String> properties) {
if (properties == null || properties.isEmpty()) {
@@ -365,7 +432,13 @@ public class MetadataService {
|| key.startsWith(MixAll.DLQ_GROUP_TOPIC_PREFIX);
}
- /** Loads the exact source message used by redelivery without exposing or
publishing its body. */
+ /**
+ * Loads the message-explorer record used as the redelivery source without
exposing or
+ * publishing its body. This is the explorer's <em>display
projection</em>, not the stored
+ * record: the body is capped and binary bodies are Base64-text, so
+ * {@link #redeliverMessage(String, String, MessageRecordVO, String)}
refuses lossy sources
+ * instead of republishing them.
+ */
public MessageRecordVO findMessageForRedelivery(String instanceId, String
sourceTopic, String msgId) {
String messageId = requireName(msgId, "message id");
List<MessageRecordVO> matches = messageService.queryMessages(
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageRedeliveryToolHandler.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageRedeliveryToolHandler.java
index 1a6d8e8d4..b9001ae90 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageRedeliveryToolHandler.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageRedeliveryToolHandler.java
@@ -43,7 +43,11 @@ public class MessageRedeliveryToolHandler extends
MutationToolHandler<MessageRed
private static final PlanDescription PLAN_DESCRIPTION = new
PlanDescription(
"redeliver message '%s' for consumer group '%s' in instance '%s'.",
List.of("Publishes a new message containing the source payload and
properties."),
- List.of());
+ List.of("The source is the message explorer's display projection;
a lossy source "
+ + "(truncated or binary body, abbreviated user properties)
is refused "
+ + "before the plan is produced.",
+ "A complete user value that happens to end in '...' is
read as abbreviated "
+ + "while the truncated flag is set, so such a source is
refused too."));
private final MetadataService metadataService;
@@ -62,6 +66,9 @@ public class MessageRedeliveryToolHandler extends
MutationToolHandler<MessageRed
String groupName = requireGroupName(input);
MessageRecordVO source = metadataService.findMessageForRedelivery(
context.instanceId(), input.sourceTopic(), input.msgId());
+ // Same exact-source guard execute() applies, so the plan can never
promise a
+ // publish that the execution stage would refuse.
+ metadataService.requireExactSourceMessage(source);
String destination = resolveDestination(input, groupName);
return PLAN_DESCRIPTION.builder(input.msgId(), groupName,
context.instanceId())
.before(new RedeliverySource(source.getMsgId(),
source.getTopic(), source.getTag(), source.getKey(),
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 0578f23c3..3365a8fb7 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
@@ -29,6 +29,8 @@ import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
import org.apache.rocketmq.remoting.protocol.admin.TopicOffset;
import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
import org.apache.rocketmq.remoting.protocol.body.TopicList;
@@ -71,7 +73,9 @@ import java.util.LinkedHashSet;
import java.util.Locale;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
+import java.util.stream.Collectors;
/**
* Real {@link DLQProvider} backed by the RocketMQ admin API. Lists
dead-letter groups by scanning
@@ -189,12 +193,11 @@ public class RocketMQDLQProvider implements DLQProvider {
if (begin >= end) {
throw new BusinessException(400, "DLQ resend start time must be
before end time");
}
+ String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + groupName;
if (StringUtils.hasText(targetTopic)) {
- validateResendTargetTopic(instanceId, targetTopic);
+ validateResendTargetTopic(instanceId, dlqTopic, targetTopic);
}
- String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + groupName;
-
DeadLetterScanResult scanResult;
try {
scanResult = collectDeadLetters(instanceId, dlqTopic, begin, end,
RESEND_HARD_CAP);
@@ -262,12 +265,11 @@ public class RocketMQDLQProvider implements DLQProvider {
if (selected.isEmpty()) {
throw new BusinessException(400, "At least one valid msgId is
required for selected DLQ resend");
}
+ String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + groupName;
if (StringUtils.hasText(targetTopic)) {
- validateResendTargetTopic(instanceId, targetTopic);
+ validateResendTargetTopic(instanceId, dlqTopic, targetTopic);
}
- String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + groupName;
-
List<MessageExt> deadLetters =
runtimeAdminClientResolver.execute(instanceId, admin -> {
List<MessageExt> resolved = new ArrayList<>(selected.size());
for (String msgId : selected) {
@@ -673,7 +675,7 @@ public class RocketMQDLQProvider implements DLQProvider {
* create new topics on clusters with autoCreateTopicEnable. Restrict it
to valid, existing,
* non-system topics on the selected instance.
*/
- private void validateResendTargetTopic(String instanceId, String
targetTopic) {
+ private void validateResendTargetTopic(String instanceId, String dlqTopic,
String targetTopic) {
TopicValidator.ValidateResult validity =
TopicValidator.validateTopic(targetTopic);
if (!validity.isValid()) {
throw new BusinessException(400, "targetTopic is not a valid
RocketMQ topic name: "
@@ -701,6 +703,59 @@ public class RocketMQDLQProvider implements DLQProvider {
"targetTopic does not exist on the selected instance;
create the topic before resending: "
+ targetTopic);
}
+ requireSameClusterAsDlqTopic(instanceId, dlqTopic, targetTopic);
+ }
+
+ /**
+ * A resend must land in the cluster the dead-letter topic belongs to. The
existence check above
+ * asks the NameServer, and one NameServer can serve several clusters, so
a target topic that only
+ * exists in a sibling cluster passes it - and the producer, which routes
by topic name across the
+ * whole endpoint, would then publish the replayed messages into that
other cluster while the API
+ * and the audit record report success. Sibling write paths (send, offset
reset, direct consume)
+ * reject exactly this with {@code
ApacheWriteTargetResolver.requireTopicRoute}.
+ *
+ * <p>Only a resolvable route can prove the mismatch, so an unresolvable
one is left to the send
+ * itself to fail - in particular a {@code %DLQ%<group>} that has no route
yet must keep reaching
+ * the scan's precise 404 rather than being reported as a route failure.
+ */
+ private void requireSameClusterAsDlqTopic(String instanceId, String
dlqTopic, String targetTopic) {
+ try {
+ runtimeAdminClientResolver.execute(instanceId, admin -> {
+ Set<String> dlqClusters = clustersOfRoute(admin, dlqTopic);
+ Set<String> targetClusters = clustersOfRoute(admin,
targetTopic);
+ if (!dlqClusters.isEmpty() && !targetClusters.isEmpty()
+ && !dlqClusters.containsAll(targetClusters)) {
+ throw new BusinessException(409, "targetTopic resolves to
a different cluster than "
+ + dlqTopic + "; resend rejected: " + targetTopic);
+ }
+ return null;
+ });
+ } catch (BusinessException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new BusinessException(502, "Failed to verify the targetTopic
route: " + e.getMessage());
+ }
+ }
+
+ /** An unresolvable route proves nothing, so a failed lookup contributes
no cluster. */
+ private Set<String> clustersOfRoute(MQAdminExt admin, String topic) {
+ try {
+ return clustersOf(admin.examineTopicRouteInfo(topic));
+ } catch (Exception e) {
+ log.debug("Cannot resolve the route of {} for the DLQ cluster
check: {}", topic, e.getMessage());
+ return Collections.emptySet();
+ }
+ }
+
+ private static Set<String> clustersOf(TopicRouteData route) {
+ if (route == null || route.getBrokerDatas() == null) {
+ return Collections.emptySet();
+ }
+ return route.getBrokerDatas().stream()
+ .filter(Objects::nonNull)
+ .map(BrokerData::getCluster)
+ .filter(StringUtils::hasText)
+ .collect(Collectors.toSet());
}
private String resolveTargetTopic(MessageExt deadLetter, String
targetTopic) {
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQMessageExcelRowTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQMessageExcelRowTest.java
index c821ff7f1..0747a6566 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQMessageExcelRowTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/dlq/DLQMessageExcelRowTest.java
@@ -63,6 +63,23 @@ class DLQMessageExcelRowTest {
assertThat(cells).containsEntry("Body Base64", "wyg=");
}
+ @Test
+ void shouldClampAnOversizedBodyOnACodePointBoundaryTest() {
+ // Asserted on the row rather than on a written sheet: POI sanitises
an unpaired surrogate
+ // on the way out, so only the in-memory cell shows whether the cut
split a pair. The astral
+ // run reaches index 32753, putting a high surrogate exactly at the
32752 cut point.
+ DLQMessageVO message = DLQMessageVO.builder()
+ .msgId("msg-huge")
+ .storeTime(1_700_000_000_000L)
+ .body("\uD835\uDC00".repeat(16_377) + "x".repeat(40_000))
+ .build();
+
+ String cell = DLQMessageExcelRow.from(message).getBody();
+
+ assertThat(cell).hasSize(32_766).endsWith("...[truncated]");
+ assertThat(cell.codePoints()).noneMatch(codePoint -> codePoint >=
0xD800 && codePoint <= 0xDFFF);
+ }
+
private static Map<String, String>
firstDataRowByColumnName(List<DLQMessageVO> messages) {
List<Map<Integer, String>> rows = readBack(messages);
Map<Integer, String> header = rows.get(0);
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 a08a31803..df87b1dd5 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
@@ -19,7 +19,13 @@ 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.common.domain.enums.InstanceVendor;
+import org.apache.rocketmq.studio.instance.InstanceVO;
+import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard;
+import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Kind;
+import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Resource;
import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InjectMocks;
@@ -27,9 +33,17 @@ import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import java.util.List;
+import java.util.function.Supplier;
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.anyList;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
@@ -43,9 +57,25 @@ class DLQServiceTest {
@Mock
private InstanceProviderRegistry providerRegistry;
+ @Mock
+ private ResourceOwnershipGuard ownershipGuard;
+
@InjectMocks
private DLQService dlqService;
+ @BeforeEach
+ void ownEveryResource() {
+ // The ownership check is what the resend tests are not about: let it
pass by default and keep
+ // the mocked provider reachable through the wrapper. Lenient because
the validation-failure
+ // tests never reach it.
+ lenient().when(ownershipGuard.requireInstance(anyString()))
+
.thenReturn(InstanceVO.builder().name("instance-1").vendor(InstanceVendor.APACHE).build());
+ lenient().when(ownershipGuard.topicResource(anyString()))
+ .thenAnswer(invocation -> new Resource(Kind.TOPIC,
invocation.getArgument(0)));
+ lenient().when(ownershipGuard.withOwned(any(), anyList(), any()))
+ .thenAnswer(invocation -> ((Supplier<?>)
invocation.getArgument(2)).get());
+ }
+
@Test
void listDLQGroupsShouldReturnGroupsFromProvider() {
List<DLQGroupVO> groups = List.of(
@@ -89,6 +119,41 @@ class DLQServiceTest {
verify(dlqProvider).resendMessages("instance-1", "group-1", 1000L,
2000L, "target-topic");
}
+ @Test
+ void resendMessagesShouldCheckOwnershipOfTheGroupAndTheTargetTopicTest() {
+ dlqService.resendMessages("instance-1", "group-1", 1000L, 2000L,
"target-topic");
+
+ // The DLQ topic %DLQ%group-1 belongs to the group's owner, and the
destination is a topic
+ // write of its own: the resend must be gated on both.
+ verify(ownershipGuard).withOwned(
+ any(InstanceVO.class),
+ eq(List.of(new Resource(Kind.GROUP, "group-1"), new
Resource(Kind.TOPIC, "target-topic"))),
+ any());
+ }
+
+ @Test
+ void
resendSelectedMessagesShouldCheckOwnershipOfTheGroupOnlyWhenNoTargetIsNamedTest()
{
+ dlqService.resendSelectedMessages("instance-1", "group-1",
List.of("msg-1"), null);
+
+ verify(ownershipGuard).withOwned(
+ any(InstanceVO.class),
+ eq(List.of(new Resource(Kind.GROUP, "group-1"))),
+ any());
+ }
+
+ @Test
+ void resendMessagesShouldNotReachTheProviderWhenOwnershipRefusesTest() {
+ // doThrow: the when(...) form would invoke the mock and run the
lenient answer below.
+ doThrow(new BusinessException(409, "Resource is owned by another
instance"))
+ .when(ownershipGuard).withOwned(any(), anyList(), any());
+
+ assertThatThrownBy(() ->
+ dlqService.resendMessages("instance-1", "group-1", 1000L,
2000L, null))
+ .isInstanceOfSatisfying(BusinessException.class,
+ exception ->
assertThat(exception.getCode()).isEqualTo(409));
+ verify(dlqProvider, never()).resendMessages(anyString(), anyString(),
any(), any(), any());
+ }
+
@Test
void actionsShouldNormalizeIdentifiersBeforeDelegatingTest() {
List<String> msgIds = List.of(" msg-1 ", " msg-2 ");
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
index b94c5db76..64daa4ec7 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
@@ -36,6 +36,7 @@ import
org.apache.rocketmq.studio.common.domain.enums.SubscriptionMode;
import org.apache.rocketmq.studio.common.domain.enums.TopicPerm;
import org.apache.rocketmq.studio.common.domain.enums.TopicType;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.common.util.MessagePropertyDisplay;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
import org.apache.rocketmq.studio.instance.group.CreateConsumerGroupDTO;
import org.apache.rocketmq.studio.instance.group.ImportConsumerGroupsResultVO;
@@ -199,6 +200,194 @@ class MetadataServiceTest {
assertThat(request.getValue().getKey()).isEqualTo("order-1");
}
+ @Test
+ void redeliverMessageShouldRefuseATruncatedSourceBodyTest() {
+ // The message explorer hands out a display projection whose body
stops at 64 KiB; publishing
+ // that projection as the redelivered payload drops the rest of the
source message.
+ MessageRecordVO original = MessageRecordVO.builder()
+ .msgId("msg-original")
+ .topic("orders")
+ .body("x".repeat(65536))
+ .bodyEncoding("UTF-8")
+ .bodyTruncated(true)
+ .build();
+ when(messageService.queryMessages(
+ "instance-a", "orders", "msg-original", null, null, null,
null))
+ .thenReturn(List.of(original));
+
+ assertThatThrownBy(() -> metadataService.redeliverMessage(
+ "instance-a", "group-a", "orders", "msg-original",
"orders-copy"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("truncated")
+ .satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(409));
+
+ verify(apacheProvider, never()).sendMessage(any(SendMessageDTO.class));
+ }
+
+ @Test
+ void redeliverMessageShouldRefuseABase64SourceBodyTest() {
+ // A binary body is projected as Base64 text; republishing that text
stores different bytes.
+ MessageRecordVO original = MessageRecordVO.builder()
+ .msgId("msg-original")
+ .topic("orders")
+ .body("AAECAwQ=")
+ .bodyEncoding("BASE64")
+ .build();
+ when(messageService.queryMessages(
+ "instance-a", "orders", "msg-original", null, null, null,
null))
+ .thenReturn(List.of(original));
+
+ assertThatThrownBy(() -> metadataService.redeliverMessage(
+ "instance-a", "group-a", "orders", "msg-original", null))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("binary")
+ .satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(409));
+
+ verify(apacheProvider, never()).sendMessage(any(SendMessageDTO.class));
+ }
+
+ @Test
+ void redeliverMessageShouldRefuseATruncatedSourcePropertySetTest() {
+ // The flag is true AND a visible user property carries an abbreviated
value: the
+ // user-property set really is incomplete, so the redelivery must be
refused.
+ MessageRecordVO original = MessageRecordVO.builder()
+ .msgId("msg-original")
+ .topic("orders")
+ .body("payload")
+ .properties(Map.of("tenant", "alpha..."))
+ .propertiesTruncated(true)
+ .build();
+ when(messageService.queryMessages(
+ "instance-a", "orders", "msg-original", null, null, null,
null))
+ .thenReturn(List.of(original));
+
+ assertThatThrownBy(() -> metadataService.redeliverMessage(
+ "instance-a", "group-a", "orders", "msg-original", null))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("properties")
+ .satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(409));
+
+ verify(apacheProvider, never()).sendMessage(any(SendMessageDTO.class));
+ }
+
+ @Test
+ void
redeliverMessageShouldAllowATruncatedFlagCausedOnlyBySystemPropertiesTest() {
+ // propertiesTruncated is computed over the raw map including broker
system keys, but
+ // redelivery discards system keys anyway: with no visible user
property abbreviated,
+ // the user-property set is intact and the redelivery must go through.
+ Map<String, String> properties = new HashMap<>();
+ properties.put(MessageConst.PROPERTY_KEYS, "order-1");
+ properties.put(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX,
"uniq");
+ properties.put(MessageConst.PROPERTY_REAL_TOPIC, "orders");
+ properties.put("tenant", "alpha");
+ MessageRecordVO original = MessageRecordVO.builder()
+ .msgId("msg-original")
+ .topic("orders")
+ .body("payload")
+ .properties(properties)
+ .propertiesTruncated(true)
+ .build();
+ when(messageService.queryMessages(
+ "instance-a", "orders", "msg-original", null, null, null,
null))
+ .thenReturn(List.of(original));
+ when(apacheProvider.sendMessage(any(SendMessageDTO.class)))
+ .thenReturn(SendMessageVO.builder().msgId("msg-new").build());
+
+ metadataService.redeliverMessage("instance-a", "group-a", "orders",
"msg-original", null);
+
+ ArgumentCaptor<SendMessageDTO> request =
ArgumentCaptor.forClass(SendMessageDTO.class);
+ verify(apacheProvider).sendMessage(request.capture());
+
assertThat(request.getValue().getProperties()).containsExactlyEntriesOf(Map.of("tenant",
"alpha"));
+ }
+
+ @Test
+ void
redeliverMessageShouldAllowATruncatedFlagWithAnIntactUserPropertySetBelowTheCapTest()
{
+ // MessageConst.STRING_HASH_SET holds 54 system keys, so this
projection stays below the
+ // 64-entry cap; with no abbreviated value the two user properties are
provably intact.
+ Map<String, String> properties = new HashMap<>();
+ for (String key : MessageConst.STRING_HASH_SET) {
+ properties.put(key, "v");
+ }
+ properties.put("tenant", "alpha");
+ properties.put("region", "cn-hangzhou");
+
assertThat(properties).hasSizeLessThan(MessagePropertyDisplay.MAX_PROPERTIES);
+ MessageRecordVO original = MessageRecordVO.builder()
+ .msgId("msg-original")
+ .topic("orders")
+ .body("payload")
+ .properties(properties)
+ .propertiesTruncated(true)
+ .build();
+ when(messageService.queryMessages(
+ "instance-a", "orders", "msg-original", null, null, null,
null))
+ .thenReturn(List.of(original));
+ when(apacheProvider.sendMessage(any(SendMessageDTO.class)))
+ .thenReturn(SendMessageVO.builder().msgId("msg-new").build());
+
+ metadataService.redeliverMessage("instance-a", "group-a", "orders",
"msg-original", null);
+
+ ArgumentCaptor<SendMessageDTO> request =
ArgumentCaptor.forClass(SendMessageDTO.class);
+ verify(apacheProvider).sendMessage(request.capture());
+ assertThat(request.getValue().getProperties())
+ .containsExactlyInAnyOrderEntriesOf(Map.of("tenant", "alpha",
"region", "cn-hangzhou"));
+ }
+
+ @Test
+ void
redeliverMessageShouldRefuseAPropertySetThatReachedTheDisplayCapTest() {
+ // A projection holding exactly MAX_PROPERTIES entries can only come
from a raw map that
+ // had more, and limitProperties dropped the overflow whole - sorted
by key, that overflow
+ // is the lowercase user keys, so the republished message would carry
fewer properties.
+ Map<String, String> properties = new HashMap<>();
+ for (String key : MessageConst.STRING_HASH_SET) {
+ properties.put(key, "v");
+ }
+ for (int i = properties.size(); i <
MessagePropertyDisplay.MAX_PROPERTIES; i++) {
+ properties.put("user" + i, "v");
+ }
+ assertThat(properties).hasSize(MessagePropertyDisplay.MAX_PROPERTIES);
+ MessageRecordVO original = MessageRecordVO.builder()
+ .msgId("msg-original")
+ .topic("orders")
+ .body("payload")
+ .properties(properties)
+ .propertiesTruncated(true)
+ .build();
+ when(messageService.queryMessages(
+ "instance-a", "orders", "msg-original", null, null, null,
null))
+ .thenReturn(List.of(original));
+
+ assertThatThrownBy(() -> metadataService.redeliverMessage(
+ "instance-a", "group-a", "orders", "msg-original", null))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("display cap")
+ .satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(409));
+
+ verify(apacheProvider, never()).sendMessage(any(SendMessageDTO.class));
+ }
+
+ @Test
+ void
redeliverMessageShouldAllowNonSystemPropertiesWhoseValuesEndInEllipsisTest() {
+ // A complete user value can legitimately end in "...": without
propertiesTruncated the
+ // abbreviation marker alone must never refuse a redelivery.
+ MessageRecordVO original = MessageRecordVO.builder()
+ .msgId("msg-original")
+ .topic("orders")
+ .body("payload")
+ .properties(Map.of("tenant", "alpha..."))
+ .build();
+ when(messageService.queryMessages(
+ "instance-a", "orders", "msg-original", null, null, null,
null))
+ .thenReturn(List.of(original));
+ when(apacheProvider.sendMessage(any(SendMessageDTO.class)))
+ .thenReturn(SendMessageVO.builder().msgId("msg-new").build());
+
+ metadataService.redeliverMessage("instance-a", "group-a", "orders",
"msg-original", null);
+
+ ArgumentCaptor<SendMessageDTO> request =
ArgumentCaptor.forClass(SendMessageDTO.class);
+ verify(apacheProvider).sendMessage(request.capture());
+
assertThat(request.getValue().getProperties()).containsExactlyEntriesOf(Map.of("tenant",
"alpha..."));
+ }
+
@Test
void redeliverMessageShouldRejectBlankGroupNameTest() {
assertThatThrownBy(() -> metadataService.redeliverMessage(
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageRedeliveryToolHandlerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageRedeliveryToolHandlerTest.java
index 22238cbb0..22ffce1a2 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageRedeliveryToolHandlerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/handler/message/MessageRedeliveryToolHandlerTest.java
@@ -17,6 +17,7 @@
package org.apache.rocketmq.studio.ops.ai.tool.handler.message;
import org.apache.rocketmq.studio.instance.message.MessageRecordVO;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.instance.topic.MetadataService;
import org.apache.rocketmq.studio.instance.topic.SendMessageVO;
import
org.apache.rocketmq.studio.ops.ai.tool.contract.message.MessageRedeliveryInput;
@@ -36,6 +37,9 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
@@ -87,6 +91,53 @@ class MessageRedeliveryToolHandlerTest {
new MessageRedeliveryOutput("original", "redelivered",
"TopicB"));
}
+ @Test
+ void previewRefusesALossySourceInsteadOfPromisingAnUnexecutablePlanTest() {
+ MetadataService metadata = mock(MetadataService.class);
+ MessageRedeliveryToolHandler handler = new
MessageRedeliveryToolHandler(metadata);
+ ToolExecutionContext execution = context("instance-a");
+ MessageRedeliveryInput input =
+ new MessageRedeliveryInput("instance-a", "group-a",
"original", "orders", null);
+ MessageRecordVO lossySource =
MessageRecordVO.builder().msgId("original").topic("orders")
+ .body("x").bodyTruncated(true).build();
+ when(metadata.findMessageForRedelivery("instance-a", "orders",
"original"))
+ .thenReturn(lossySource);
+ doThrow(new BusinessException(409,
+ "Source message body is truncated for display and cannot be
redelivered exactly"))
+ .when(metadata).requireExactSourceMessage(lossySource);
+
+ assertThatThrownBy(() -> handler.preview(input, execution))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("truncated");
+
+ verify(metadata).requireExactSourceMessage(lossySource);
+ verify(metadata, never()).redeliverMessage(
+ any(), any(), any(MessageRecordVO.class), any());
+ }
+
+ @Test
+ void previewAndExecuteUseTheSameExactSourceGuardTest() {
+ // The guard method called in preview is exactly the one
redeliverMessage runs before
+ // publishing, so a preview that passes cannot be refused at execution
time.
+ MetadataService metadata = mock(MetadataService.class);
+ MessageRedeliveryToolHandler handler = new
MessageRedeliveryToolHandler(metadata);
+ ToolExecutionContext execution = context("instance-a");
+ MessageRedeliveryInput input =
+ new MessageRedeliveryInput("instance-a", "group-a",
"original", "orders", null);
+ MessageRecordVO source =
MessageRecordVO.builder().msgId("original").topic("orders")
+ .body("payload").build();
+ when(metadata.findMessageForRedelivery("instance-a", "orders",
"original"))
+ .thenReturn(source);
+ when(metadata.redeliverMessage("instance-a", "group-a", "orders",
"original", null))
+
.thenReturn(SendMessageVO.builder().msgId("redelivered").build());
+
+ ToolPlan plan = handler.preview(input, execution);
+
+ verify(metadata).requireExactSourceMessage(source);
+ assertThat(plan.summary()).isEqualTo(
+ "redeliver message 'original' for consumer group 'group-a' in
instance 'instance-a'.");
+ }
+
@ParameterizedTest
@NullSource
@ValueSource(strings = {"", " "})
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 8596179c9..d8aaa2630 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
@@ -16,6 +16,7 @@
*/
package org.apache.rocketmq.studio.provider.apache;
+import com.alibaba.excel.EasyExcel;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.PullStatus;
@@ -35,11 +36,13 @@ import
org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.body.TopicList;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
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.DLQExcelExportResultVO;
import org.apache.rocketmq.studio.instance.dlq.DLQExportResultVO;
import org.apache.rocketmq.studio.instance.dlq.DLQGroupVO;
import org.apache.rocketmq.studio.instance.dlq.DLQMessageVO;
@@ -54,6 +57,7 @@ import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
+import java.io.ByteArrayInputStream;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
@@ -294,6 +298,60 @@ class RocketMQDLQProviderTest {
verify(pullConsumer, never()).pull(any(MessageQueue.class),
anyString(), anyLong(), anyInt());
}
+ @Test
+ void resendRejectsATargetTopicThatResolvesToAnotherClusterTest() throws
Exception {
+ String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
+ TopicList existingTargets = new TopicList();
+ existingTargets.setTopicList(Set.of("target-topic"));
+ when(adminExt.fetchAllTopicList()).thenReturn(existingTargets);
+ // One NameServer can serve several clusters, so the target exists on
the endpoint while
+ // belonging to a different cluster than the dead-letter topic it
would be replayed from.
+ // The producer routes by topic name and would publish into that other
cluster silently.
+
when(adminExt.examineTopicRouteInfo(dlqTopic)).thenReturn(routeInCluster("cluster-a"));
+
when(adminExt.examineTopicRouteInfo("target-topic")).thenReturn(routeInCluster("cluster-b"));
+
+ assertThatThrownBy(() -> provider.resendMessages(
+ "instance-a", "group-a", 100L, 200L, "target-topic"))
+ .isInstanceOfSatisfying(BusinessException.class, exception -> {
+ assertThat(exception.getCode()).isEqualTo(409);
+ assertThat(exception.getMessage()).contains("different
cluster");
+ });
+ verify(runtimeAdminClientResolver,
never()).executeProducer(anyString(), any());
+ }
+
+ private static TopicRouteData routeInCluster(String cluster) {
+ BrokerData broker = new BrokerData();
+ broker.setCluster(cluster);
+ broker.setBrokerName("broker-a");
+ broker.setBrokerAddrs(new HashMap<>(Map.of(0L, "10.0.0.1:10911")));
+ TopicRouteData route = new TopicRouteData();
+ route.setBrokerDatas(List.of(broker));
+ return route;
+ }
+
+ @Test
+ void resendLeavesAnUnresolvableDlqRouteToTheScanNotFoundTest() throws
Exception {
+ String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
+ TopicList existingTargets = new TopicList();
+ existingTargets.setTopicList(Set.of("target-topic"));
+ when(adminExt.fetchAllTopicList()).thenReturn(existingTargets);
+ // Only a resolvable route can prove a cluster mismatch, so a
dead-letter topic that has no
+ // route yet must fall through to the scan's precise 404 instead of a
route-verification 502.
+ when(adminExt.examineTopicRouteInfo(dlqTopic)).thenThrow(new
MQClientException(
+ "No topic route info in name server for the topic: " +
dlqTopic, null));
+
when(adminExt.examineTopicRouteInfo("target-topic")).thenReturn(routeInCluster("cluster-b"));
+ when(pullConsumer.fetchSubscribeMessageQueues(dlqTopic))
+ .thenThrow(new MQClientException("Can not find Message Queue
for this topic, " + dlqTopic, null));
+
+ assertThatThrownBy(() -> provider.resendMessages(
+ "instance-a", "group-a", 100L, 200L, "target-topic"))
+ .isInstanceOf(BusinessException.class)
+ .satisfies(error -> assertThat(((BusinessException)
error).getCode()).isEqualTo(404));
+ verify(auditService).record(eq("RESEND_DLQ"), eq("DLQ"),
eq("group-a"), isNull(),
+ contains("dlqTopicMissing=true"), eq("NOT_FOUND"));
+ verify(runtimeAdminClientResolver,
never()).executeProducer(anyString(), any());
+ }
+
@Test
void resendSelectedMessagesRejectsSystemTopicAsTarget() throws Exception {
assertThatThrownBy(() -> provider.resendMessages(
@@ -1110,6 +1168,54 @@ class RocketMQDLQProviderTest {
verify(pullConsumer, never()).searchOffset(queue, Long.MIN_VALUE);
}
+ @Test
+ void
exportExcelTruncatesOversizedBodiesInsteadOfFailingTheWholeExportTest() throws
Exception {
+ String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
+ MessageQueue queue = new MessageQueue(dlqTopic, "broker-a", 0);
+ MessageExt oversized = new MessageExt();
+ oversized.setMsgId("msg-huge");
+ oversized.setTopic(dlqTopic);
+ oversized.setQueueId(0);
+ oversized.setQueueOffset(5L);
+ oversized.setStoreTimestamp(150L);
+ oversized.setKeys("key-huge");
+ // Leading astral characters put a high surrogate exactly at the
truncation boundary,
+ // so the cut must back off onto a code-point boundary.
+ oversized.setBody(("\uD835\uDC00".repeat(16_377) + "x".repeat(40_000))
+ .getBytes(StandardCharsets.UTF_8));
+ MessageExt small = new MessageExt();
+ small.setMsgId("msg-small");
+ small.setTopic(dlqTopic);
+ small.setQueueId(0);
+ small.setQueueOffset(6L);
+ small.setStoreTimestamp(160L);
+ small.setKeys("key-small");
+ small.setBody("hello dlq".getBytes(StandardCharsets.UTF_8));
+ PullResult pullResult = new PullResult(PullStatus.FOUND, 2L, 0L, 0L,
List.of(oversized, small));
+
when(pullConsumer.fetchSubscribeMessageQueues(dlqTopic)).thenReturn(Set.of(queue));
+ // The scan resolves the window with searchOffset(begin) and
searchOffset(end + 1).
+ when(pullConsumer.searchOffset(eq(queue), anyLong())).thenReturn(0L,
2L);
+ when(pullConsumer.pull(eq(queue), eq("*"), eq(0L),
eq(32))).thenReturn(pullResult);
+
+ DLQExcelExportResultVO exported =
+ provider.exportExcel("instance-a", "group-a", 100L, 200L,
null);
+ assertThat(exported.getData()).isNotEmpty();
+
+ List<Map<Integer, String>> rows;
+ try (ByteArrayInputStream input = new
ByteArrayInputStream(exported.getData())) {
+ rows = EasyExcel.read(input).sheet("DLQ").doReadSync();
+ }
+ assertThat(rows).hasSize(2);
+ int bodyColumn = 7; // Message ID, Topic, Queue ID, Offset, Store
Time, Reconsume Times, Keys, Body
+ String oversizedBody = rows.get(0).get(bodyColumn);
+ assertThat(oversizedBody.length()).isLessThanOrEqualTo(32_767);
+ assertThat(oversizedBody).endsWith("...[truncated]");
+ // The cut must not split a surrogate pair into an unpaired half.
+ assertThat(oversizedBody.codePoints())
+ .noneMatch(codePoint -> codePoint >= 0xD800 && codePoint <=
0xDFFF);
+ assertThat(rows.get(1).get(bodyColumn)).isEqualTo("hello dlq");
+ }
+
private void stubExistingTarget(String topic) throws Exception {
TopicList existingTargets = new TopicList();
existingTargets.setTopicList(Set.of(topic));
