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 ea0915623 fix(dlq): validate the resend target topic before
dispatching messages (#2835)
ea0915623 is described below
commit ea0915623d235d4832bb0184614e38fdba926a50
Author: 烤化の初雪 <[email protected]>
AuthorDate: Fri Sep 4 12:04:36 2026 +0800
fix(dlq): validate the resend target topic before dispatching messages
(#2835)
Signed-off-by: unbridled-41
<[email protected]>
Co-authored-by: unbridled-41
<[email protected]>
---
.../provider/apache/RocketMQDLQProvider.java | 44 +++++++++
.../provider/apache/RocketMQDLQProviderTest.java | 110 +++++++++++++++++++++
2 files changed, 154 insertions(+)
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 e6dd8f42a..acb5922ed 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,6 +27,7 @@ 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.common.topic.TopicValidator;
import org.apache.rocketmq.remoting.protocol.admin.TopicOffset;
import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
import org.apache.rocketmq.remoting.protocol.body.TopicList;
@@ -34,6 +35,7 @@ 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.common.util.SystemTopicFilter;
import org.apache.rocketmq.studio.instance.dlq.DLQExcelExportResultVO;
import org.apache.rocketmq.studio.instance.dlq.DLQExportResultVO;
import org.apache.rocketmq.studio.instance.dlq.DLQGroupVO;
@@ -178,6 +180,9 @@ public class RocketMQDLQProvider implements DLQProvider {
if (begin >= end) {
throw new BusinessException(400, "DLQ resend start time must be
before end time");
}
+ if (StringUtils.hasText(targetTopic)) {
+ validateResendTargetTopic(instanceId, targetTopic);
+ }
String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + groupName;
@@ -251,6 +256,9 @@ public class RocketMQDLQProvider implements DLQProvider {
if (selected.isEmpty()) {
throw new BusinessException(400, "At least one valid msgId is
required for selected DLQ resend");
}
+ if (StringUtils.hasText(targetTopic)) {
+ validateResendTargetTopic(instanceId, targetTopic);
+ }
String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + groupName;
@@ -548,6 +556,42 @@ public class RocketMQDLQProvider implements DLQProvider {
}
}
+ /**
+ * An explicit resend target is a powerful override: without validation it
can feed dead
+ * letters back into their own DLQ or retry topic, poison broker system
topics, or silently
+ * 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) {
+ TopicValidator.ValidateResult validity =
TopicValidator.validateTopic(targetTopic);
+ if (!validity.isValid()) {
+ throw new BusinessException(400, "targetTopic is not a valid
RocketMQ topic name: "
+ + targetTopic);
+ }
+ if (SystemTopicFilter.isSystem(targetTopic)) {
+ throw new BusinessException(400,
+ "targetTopic must not be a RocketMQ system, retry or DLQ
topic: " + targetTopic);
+ }
+ boolean exists;
+ try {
+ exists =
Boolean.TRUE.equals(runtimeAdminClientResolver.execute(instanceId, admin -> {
+ TopicList topics = admin.fetchAllTopicList();
+ return topics != null && topics.getTopicList() != null
+ && topics.getTopicList().contains(targetTopic);
+ }));
+ } catch (BusinessException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new BusinessException(502, "Failed to verify targetTopic on
the selected instance: "
+ + e.getMessage());
+ }
+ if (!exists) {
+ throw new BusinessException(400,
+ "targetTopic does not exist on the selected instance;
create the topic before resending: "
+ + targetTopic);
+ }
+ }
+
private String resolveTargetTopic(MessageExt deadLetter, String
targetTopic) {
if (StringUtils.hasText(targetTopic)) {
return targetTopic;
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 12dc1a026..a842d271f 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
@@ -236,10 +236,87 @@ class RocketMQDLQProviderTest {
verify(auditService, never()).record(anyString(), anyString(),
anyString(), anyString());
}
+ @Test
+ void resendMessagesRejectsRetryAndDlqTopicsAsTarget() throws Exception {
+ assertThatThrownBy(() -> provider.resendMessages(
+ "instance-a", "group-a", 100L, 200L, "%DLQ%group-a"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("must not be a RocketMQ system, retry or
DLQ topic")
+ .satisfies(error -> assertThat(((BusinessException)
error).getCode()).isEqualTo(400));
+
+ verify(runtimeAdminClientResolver,
never()).executeProducer(anyString(), any());
+ verify(adminExt, never()).fetchAllTopicList();
+ }
+
+ @Test
+ void resendMessagesRejectsSystemTopicsAsTarget() {
+ assertThatThrownBy(() -> provider.resendMessages(
+ "instance-a", "group-a", 100L, 200L, "RMQ_SYS_TRACE_TOPIC"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("must not be a RocketMQ system, retry or
DLQ topic");
+
+ verify(runtimeAdminClientResolver,
never()).executeProducer(anyString(), any());
+ }
+
+ @Test
+ void resendMessagesRejectsInvalidTargetTopicName() throws Exception {
+ assertThatThrownBy(() -> provider.resendMessages(
+ "instance-a", "group-a", 100L, 200L, "not a valid topic"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("not a valid RocketMQ topic name");
+
+ verify(runtimeAdminClientResolver,
never()).executeProducer(anyString(), any());
+ verify(adminExt, never()).fetchAllTopicList();
+ }
+
+ @Test
+ void resendMessagesRejectsTargetTopicMissingFromInstance() throws
Exception {
+ TopicList otherTopics = new TopicList();
+ otherTopics.setTopicList(Set.of("unrelated-topic"));
+ when(adminExt.fetchAllTopicList()).thenReturn(otherTopics);
+
+ assertThatThrownBy(() -> provider.resendMessages(
+ "instance-a", "group-a", 100L, 200L, "target-topic"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("targetTopic does not exist on the
selected instance");
+
+ verify(runtimeAdminClientResolver,
never()).executeProducer(anyString(), any());
+ verify(pullConsumer, never()).pull(any(MessageQueue.class),
anyString(), anyLong(), anyInt());
+ }
+
+ @Test
+ void resendSelectedMessagesRejectsSystemTopicAsTarget() throws Exception {
+ assertThatThrownBy(() -> provider.resendMessages(
+ "instance-a", "group-a", List.of("msg-1"), "%DLQ%group-a"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("must not be a RocketMQ system, retry or
DLQ topic");
+
+ verify(runtimeAdminClientResolver,
never()).executeProducer(anyString(), any());
+ verify(adminExt, never()).fetchAllTopicList();
+ verify(pullConsumer, never()).pull(any(MessageQueue.class),
anyString(), anyLong(), anyInt());
+ }
+
+ @Test
+ void resendMessagesFailsGracefullyWhenTopicListCannotBeRead() throws
Exception {
+ when(adminExt.fetchAllTopicList()).thenThrow(new
IllegalStateException("nameserver unreachable"));
+
+ assertThatThrownBy(() -> provider.resendMessages(
+ "instance-a", "group-a", 100L, 200L, "target-topic"))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("Failed to verify targetTopic on the
selected instance")
+ .satisfies(error -> assertThat(((BusinessException)
error).getCode()).isEqualTo(502));
+
+ verify(runtimeAdminClientResolver,
never()).executeProducer(anyString(), any());
+ verify(pullConsumer, never()).pull(any(MessageQueue.class),
anyString(), anyLong(), anyInt());
+ }
+
@Test
void
resendMessagesShouldNormalizeGroupNameBeforeBuildingDlqTopicAndAuditing()
throws Exception {
String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
when(pullConsumer.fetchSubscribeMessageQueues(dlqTopic)).thenReturn(null);
+ TopicList existingTargets = new TopicList();
+ existingTargets.setTopicList(Set.of("target-topic"));
+ when(adminExt.fetchAllTopicList()).thenReturn(existingTargets);
provider.resendMessages("instance-a", " group-a ", 100L, 200L,
"target-topic");
verify(runtimeAdminClientResolver).executePullConsumer(eq("instance-a"), any());
@@ -253,6 +330,9 @@ class RocketMQDLQProviderTest {
void resendMessagesDoesNotPullWhenDlqQueueSetIsNull() throws Exception {
String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
when(pullConsumer.fetchSubscribeMessageQueues(dlqTopic)).thenReturn(null);
+ TopicList existingTargets = new TopicList();
+ existingTargets.setTopicList(Set.of("target-topic"));
+ when(adminExt.fetchAllTopicList()).thenReturn(existingTargets);
provider.resendMessages("instance-a", "group-a", 100L, 200L,
"target-topic");
verify(runtimeAdminClientResolver).executePullConsumer(eq("instance-a"), any());
@@ -285,6 +365,9 @@ class RocketMQDLQProviderTest {
when(pullConsumer.searchOffset(queue, 200L)).thenReturn(0L);
when(pullConsumer.pull(queue, "*", 0L, 32)).thenReturn(pullResult);
when(dlqProducer.send(any(Message.class))).thenReturn(sendResult);
+ TopicList existingTargets = new TopicList();
+ existingTargets.setTopicList(Set.of("target-topic"));
+ when(adminExt.fetchAllTopicList()).thenReturn(existingTargets);
provider.resendMessages("instance-a", "group-a", 100L, 200L,
"target-topic");
verify(runtimeAdminClientResolver).executePullConsumer(eq("instance-a"), any());
@@ -296,6 +379,9 @@ class RocketMQDLQProviderTest {
String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
when(pullConsumer.fetchSubscribeMessageQueues(dlqTopic))
.thenThrow(new IllegalStateException("broker unavailable"));
+ TopicList existingTargets = new TopicList();
+ existingTargets.setTopicList(Set.of("target-topic"));
+ when(adminExt.fetchAllTopicList()).thenReturn(existingTargets);
assertThatThrownBy(() -> provider.resendMessages("instance-a",
"group-a", 100L, 200L, "target-topic"))
.isInstanceOf(BusinessException.class)
.hasMessageContaining("Failed to scan DLQ topic " + dlqTopic)
@@ -322,6 +408,9 @@ class RocketMQDLQProviderTest {
.thenThrow(new IllegalStateException("broker unavailable"));
when(pullConsumer.searchOffset(eq(emptyQueue),
anyLong())).thenReturn(0L);
when(pullConsumer.pull(eq(emptyQueue), eq("*"), eq(0L),
eq(32))).thenReturn(emptyResult);
+ TopicList existingTargets = new TopicList();
+ existingTargets.setTopicList(Set.of("target-topic"));
+ when(adminExt.fetchAllTopicList()).thenReturn(existingTargets);
assertThat(provider.resendMessages("instance-a", "group-a", 100L,
200L, "target-topic"))
.extracting("matched", "resent", "failed", "outcome",
"scanIncomplete", "failedQueueCount")
.containsExactly(0, 0, 0, "PARTIAL", true, 1);
@@ -345,6 +434,9 @@ class RocketMQDLQProviderTest {
when(pullConsumer.fetchSubscribeMessageQueues(dlqTopic)).thenReturn(Set.of(queue));
when(pullConsumer.searchOffset(eq(queue), anyLong())).thenReturn(10L);
when(pullConsumer.pull(eq(queue), eq("*"), eq(10L),
eq(32))).thenReturn(stalledResult);
+ TopicList existingTargets = new TopicList();
+ existingTargets.setTopicList(Set.of("target-topic"));
+ when(adminExt.fetchAllTopicList()).thenReturn(existingTargets);
provider.resendMessages("instance-a", "group-a", 100L, 200L,
"target-topic");
verify(pullConsumer, times(1)).pull(queue, "*", 10L, 32);
@@ -373,6 +465,9 @@ class RocketMQDLQProviderTest {
when(pullConsumer.pull(queue, "*", 20L,
32)).thenReturn(foundAfterCorrection);
when(pullConsumer.pull(queue, "*", 40L, 32)).thenReturn(endOfQueue);
when(dlqProducer.send(any(Message.class))).thenReturn(sendResult);
+ TopicList existingTargets = new TopicList();
+ existingTargets.setTopicList(Set.of("orders"));
+ when(adminExt.fetchAllTopicList()).thenReturn(existingTargets);
assertThat(provider.resendMessages("instance-a", "group-a", 100L,
200L, "orders"))
.extracting("matched", "resent", "failed", "outcome")
.containsExactly(1, 1, 0, "SUCCESS");
@@ -409,6 +504,9 @@ class RocketMQDLQProviderTest {
when(pullConsumer.searchOffset(queue, 200L)).thenReturn(0L);
when(pullConsumer.pull(queue, "*", 0L, 32)).thenReturn(pullResult);
when(dlqProducer.send(any(Message.class))).thenReturn(sendResult);
+ TopicList existingTargets = new TopicList();
+ existingTargets.setTopicList(Set.of("target-topic"));
+ when(adminExt.fetchAllTopicList()).thenReturn(existingTargets);
assertThat(provider.resendMessages("instance-a", "group-a", 100L,
200L, "target-topic"))
.extracting("matched", "resent", "failed", "outcome")
.containsExactly(1, 0, 1, "FAILED");
@@ -446,6 +544,9 @@ class RocketMQDLQProviderTest {
when(pullConsumer.searchOffset(queue, 200L)).thenReturn(0L);
when(pullConsumer.pull(queue, "*", 0L, 32)).thenReturn(pullResult);
when(dlqProducer.send(any(Message.class))).thenReturn(sendResult);
+ TopicList existingTargets = new TopicList();
+ existingTargets.setTopicList(Set.of("target-topic"));
+ when(adminExt.fetchAllTopicList()).thenReturn(existingTargets);
assertThat(provider.resendMessages("instance-a", "group-a", 100L,
200L, "target-topic"))
.extracting("matched", "resent", "failed", "outcome")
.containsExactly(1, 1, 0, "SUCCESS");
@@ -474,6 +575,9 @@ class RocketMQDLQProviderTest {
when(adminExt.viewMessage(dlqTopic,
"old-msg")).thenReturn(oldDeadLetter);
when(dlqProducer.send(any(Message.class))).thenReturn(sendResult);
+ TopicList existingTargets = new TopicList();
+ existingTargets.setTopicList(Set.of("target-topic"));
+ when(adminExt.fetchAllTopicList()).thenReturn(existingTargets);
assertThat(provider.resendMessages(
"instance-a", "group-a", List.of("old-msg"), "target-topic"))
@@ -498,6 +602,9 @@ class RocketMQDLQProviderTest {
.thenThrow(new IllegalStateException("message not found"));
when(adminExt.viewMessage(dlqTopic, "found-msg")).thenReturn(found);
when(dlqProducer.send(any(Message.class))).thenReturn(sendResult);
+ TopicList existingTargets = new TopicList();
+ existingTargets.setTopicList(Set.of("target-topic"));
+ when(adminExt.fetchAllTopicList()).thenReturn(existingTargets);
assertThat(provider.resendMessages(
"instance-a", "group-a", List.of("missing-msg", "found-msg"),
"target-topic"))
@@ -531,6 +638,9 @@ class RocketMQDLQProviderTest {
when(pullConsumer.searchOffset(queue, 200L)).thenReturn(5001L);
when(pullConsumer.pull(queue, "*", 0L, 32)).thenReturn(pullResult);
when(dlqProducer.send(any(Message.class))).thenReturn(sendResult);
+ TopicList existingTargets = new TopicList();
+ existingTargets.setTopicList(Set.of("target-topic"));
+ when(adminExt.fetchAllTopicList()).thenReturn(existingTargets);
assertThat(provider.resendMessages("instance-a", "group-a", 100L,
200L, "target-topic"))
.extracting("matched", "resent", "failed", "outcome",
"scanIncomplete", "failedQueueCount")
.containsExactly(5000, 5000, 0, "PARTIAL", true, 0);
