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 20f797822 fix(dlq): preserve FIFO queue affinity during replay (#5192)
20f797822 is described below
commit 20f797822b9949c76ee1e8c6c98df852c4ee1fab
Author: zmuxuny <[email protected]>
AuthorDate: Mon Oct 19 03:54:34 2026 -0700
fix(dlq): preserve FIFO queue affinity during replay (#5192)
Verified against `rocketmq-studio` at f81e73a49 before merging:
缺陷仍在（`server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQDLQProvider.java:645`，`resendOne`(:614)
仍一律 `producer.send(message)`，全文无
merge-tree clean against the current trunk. Merged as-is (squash) per the
maintainer review of 2026-10-19.
---
.../provider/apache/RocketMQDLQProvider.java | 8 +-
.../provider/apache/RocketMQDLQProviderTest.java | 153 ++++++++++++++++++++-
2 files changed, 159 insertions(+), 2 deletions(-)
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 12a3a5251..c440a4590 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
@@ -642,7 +642,13 @@ public class RocketMQDLQProvider implements DLQProvider {
}
message.putUserProperty(ORIGIN_MESSAGE_ID_PROPERTY,
deadLetter.getMsgId());
message.putUserProperty(ORIGIN_TOPIC_PROPERTY,
deadLetter.getTopic());
- SendResult sendResult = producer.send(message);
+ // FIFO group metadata alone does not influence the ordinary
producer's queue selection.
+ // Keep one destination queue per group, matching the grouped
message-send path.
+ String messageGroup =
message.getProperty(MessageConst.PROPERTY_SHARDING_KEY);
+ SendResult sendResult = StringUtils.hasText(messageGroup)
+ ? producer.send(message, (queues, msg, group) ->
+ queues.get(Math.floorMod(group.hashCode(),
queues.size())), messageGroup)
+ : producer.send(message);
if (sendResult == null || sendResult.getSendStatus() !=
SendStatus.SEND_OK) {
log.warn("DLQ resend was not accepted: msgId={} topic={}
sendStatus={}",
deadLetter.getMsgId(), destination,
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 6adc6c4ca..02b1ab36a 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
@@ -21,7 +21,9 @@ 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.exception.MQClientException;
+import org.apache.rocketmq.client.impl.producer.TopicPublishInfo;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.MixAll;
@@ -53,9 +55,9 @@ import org.apache.rocketmq.tools.admin.MQAdminExt;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
-import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
+import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
@@ -66,6 +68,7 @@ import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
+import java.util.ArrayList;
import java.util.Base64;
import java.util.HashMap;
import java.util.List;
@@ -1152,6 +1155,154 @@ class RocketMQDLQProviderTest {
.doesNotContainEntry(MessageConst.PROPERTY_REAL_TOPIC,
"system-topic-should-not-copy");
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void resendMessagesKeepsEachFifoGroupOnOneQueueTest(boolean selected)
throws Exception {
+ String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
+ String destination = selected ? "orders" : "replay-orders";
+ List<String> groups = List.of("order-b", "polygenelubricants",
"order-b", "polygenelubricants");
+ List<MessageExt> deadLetters = IntStream.range(0,
groups.size()).mapToObj(index -> {
+ MessageExt deadLetter = new MessageExt();
+ deadLetter.setMsgId("fifo-" + index);
+ deadLetter.setTopic(dlqTopic);
+ deadLetter.setBody(new byte[] {(byte) index});
+ deadLetter.setStoreTimestamp(150L);
+ MessageAccessor.putProperty(deadLetter,
MessageConst.PROPERTY_SHARDING_KEY, groups.get(index));
+ MessageAccessor.putProperty(deadLetter,
MessageConst.PROPERTY_DLQ_ORIGIN_TOPIC, "orders");
+ return deadLetter;
+ }).toList();
+ List<MessageQueue> queues = List.of(new MessageQueue(destination,
"broker-a", 0),
+ new MessageQueue(destination, "broker-a", 1), new
MessageQueue(destination, "broker-a", 2));
+ TopicPublishInfo publishInfo = new TopicPublishInfo();
+ publishInfo.setMessageQueueList(queues);
+ List<MessageQueue> sentQueues = new ArrayList<>();
+ List<Message> sentMessages = new ArrayList<>();
+ SendResult success = new SendResult();
+ success.setSendStatus(SendStatus.SEND_OK);
+ // Exercise the SDK's ordinary round-robin choice if replay takes the
ungrouped path.
+ lenient().when(dlqProducer.send(any(Message.class))).thenAnswer(call
-> {
+ sentMessages.add(call.getArgument(0));
+ sentQueues.add(publishInfo.selectOneMessageQueue());
+ return success;
+ });
+ lenient().when(dlqProducer.send(any(Message.class),
any(MessageQueueSelector.class), any()))
+ .thenAnswer(call -> {
+ Message message = call.getArgument(0);
+ MessageQueueSelector selector = call.getArgument(1);
+ Object group = call.getArgument(2);
+
assertThat(group).isEqualTo(message.getProperty(MessageConst.PROPERTY_SHARDING_KEY));
+ sentMessages.add(message);
+ sentQueues.add(selector.select(queues, message, group));
+ return success;
+ });
+ DLQResendResultVO result;
+ if (selected) {
+ for (MessageExt deadLetter : deadLetters) {
+ when(adminExt.viewMessage(dlqTopic,
deadLetter.getMsgId())).thenReturn(deadLetter);
+ }
+ result = provider.resendMessages("instance-a", "group-a",
+ deadLetters.stream().map(MessageExt::getMsgId).toList(),
null);
+ } else {
+ MessageQueue sourceQueue = new MessageQueue(dlqTopic, "broker-a",
0);
+
when(pullConsumer.fetchSubscribeMessageQueues(dlqTopic)).thenReturn(Set.of(sourceQueue));
+ when(pullConsumer.searchOffset(sourceQueue, 100L)).thenReturn(0L);
+ when(pullConsumer.searchOffset(sourceQueue, 201L)).thenReturn(4L);
+ when(pullConsumer.pull(sourceQueue, "*", 0L, 32))
+ .thenReturn(new PullResult(PullStatus.FOUND, 4L, 0L, 4L,
deadLetters));
+ stubExistingTarget(destination);
+ result = provider.resendMessages("instance-a", "group-a", 100L,
200L, destination);
+ }
+
+ assertThat(result).extracting("matched", "resent", "failed", "outcome")
+ .containsExactly(4, 4, 0, "SUCCESS");
+ assertThat(sentMessages).extracting(message ->
message.getProperty(MessageConst.PROPERTY_SHARDING_KEY))
+ .containsExactlyElementsOf(groups);
+ assertThat(sentQueues).hasSize(4);
+ assertThat(sentQueues.get(0)).isEqualTo(sentQueues.get(2));
+ assertThat(sentQueues.get(1)).isEqualTo(sentQueues.get(3));
+ assertThat(sentQueues.get(0)).isNotEqualTo(sentQueues.get(1));
+ for (int index = 0; index < groups.size(); index++) {
+ assertThat(sentQueues.get(index))
+
.isEqualTo(queues.get(Math.floorMod(groups.get(index).hashCode(),
queues.size())));
+
assertThat(sentMessages.get(index).getTopic()).isEqualTo(destination);
+
assertThat(sentMessages.get(index).getBody()).containsExactly((byte) index);
+ }
+ verify(dlqProducer, never()).send(any(Message.class));
+ }
+
+ @Test
+ void resendMessagesWithoutFifoGroupKeepOrdinarySendTest() throws Exception
{
+ MessageExt deadLetter = fifoDeadLetter("plain-msg", null);
+ SendResult success = new SendResult();
+ success.setSendStatus(SendStatus.SEND_OK);
+ when(adminExt.viewMessage(deadLetter.getTopic(),
deadLetter.getMsgId())).thenReturn(deadLetter);
+ when(dlqProducer.send(any(Message.class))).thenReturn(success);
+
+ DLQResendResultVO result = provider.resendMessages(
+ "instance-a", "group-a", List.of(deadLetter.getMsgId()), null);
+
+ assertThat(result).extracting("resent", "failed",
"outcome").containsExactly(1, 0, "SUCCESS");
+ verify(dlqProducer).send(any(Message.class));
+ verify(dlqProducer, never()).send(any(Message.class),
any(MessageQueueSelector.class), any());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"", " "})
+ void resendMessagesKeepsRejectingBlankGroupPropertiesTest(String group)
throws Exception {
+ MessageExt deadLetter = fifoDeadLetter("blank-group", group);
+ when(adminExt.viewMessage(deadLetter.getTopic(),
deadLetter.getMsgId())).thenReturn(deadLetter);
+
+ DLQResendResultVO result = provider.resendMessages(
+ "instance-a", "group-a", List.of(deadLetter.getMsgId()), null);
+
+ // The SDK's user-property validation already rejects these values
before either send path.
+ assertThat(result).extracting("resent", "failed",
"outcome").containsExactly(0, 1, "FAILED");
+ assertThat(result.getFailures()).singleElement().satisfies(failure ->
+ assertThat(failure.getReason()).contains("property can not be
null or blank"));
+ verifyNoInteractions(dlqProducer);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void failedFifoResendDoesNotFallBackToOrdinarySendTest(boolean
throwsException) throws Exception {
+ MessageExt deadLetter = fifoDeadLetter("failed-fifo", "order-42");
+ when(adminExt.viewMessage(deadLetter.getTopic(),
deadLetter.getMsgId())).thenReturn(deadLetter);
+ if (throwsException) {
+ when(dlqProducer.send(any(Message.class),
any(MessageQueueSelector.class), eq("order-42")))
+ .thenThrow(new IllegalStateException("broker
unavailable"));
+ } else {
+ SendResult rejected = new SendResult();
+ rejected.setSendStatus(SendStatus.FLUSH_DISK_TIMEOUT);
+ when(dlqProducer.send(any(Message.class),
any(MessageQueueSelector.class), eq("order-42")))
+ .thenReturn(rejected);
+ }
+
+ DLQResendResultVO result = provider.resendMessages(
+ "instance-a", "group-a", List.of(deadLetter.getMsgId()), null);
+
+ assertThat(result).extracting("matched", "resent", "failed", "outcome")
+ .containsExactly(1, 0, 1, "FAILED");
+ assertThat(result.getFailures()).singleElement().satisfies(failure -> {
+ assertThat(failure.getMsgId()).isEqualTo("failed-fifo");
+ assertThat(failure.getTargetTopic()).isEqualTo("orders");
+ assertThat(failure.getReason())
+ .contains(throwsException ? "broker unavailable" :
"FLUSH_DISK_TIMEOUT");
+ });
+ verify(dlqProducer, never()).send(any(Message.class));
+ }
+
+ private static MessageExt fifoDeadLetter(String messageId, String group) {
+ MessageExt deadLetter = new MessageExt();
+ deadLetter.setMsgId(messageId);
+ deadLetter.setTopic(MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a");
+ deadLetter.setBody(new byte[] {1});
+ MessageAccessor.putProperty(deadLetter,
MessageConst.PROPERTY_DLQ_ORIGIN_TOPIC, "orders");
+ if (group != null) {
+ MessageAccessor.putProperty(deadLetter,
MessageConst.PROPERTY_SHARDING_KEY, group);
+ }
+ return deadLetter;
+ }
+
@Test
void resendSelectedMessagesLooksUpIdsWithoutAStoreTimeWindow() throws
Exception {
String dlqTopic = MixAll.DLQ_GROUP_TOPIC_PREFIX + "group-a";
