[ISSUE #10405] Route orderly retry messages to topic brokers

client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageOrderlyService.java

@@ -380,7 +380,7 @@ public boolean sendMessageBack(final MessageExt msg) {
         try {
             // max reconsume times exceeded then send to dead letter queue.
             Message newMsg = new Message(MixAll.getRetryTopic(this.defaultMQPushConsumer.getConsumerGroup()), msg.getBody());
-            MessageAccessor.setProperties(newMsg, msg.getProperties());
+            MessageAccessor.setProperties(newMsg, MessageAccessor.deepCopyProperties(msg.getProperties()));
             String originMsgId = MessageAccessor.getOriginMessageId(msg);
             MessageAccessor.setOriginMessageId(newMsg, UtilAll.isBlank(originMsgId) ? msg.getMsgId() : originMsgId);
             newMsg.setFlag(msg.getFlag());
@@ -390,7 +390,7 @@ public boolean sendMessageBack(final MessageExt msg) {
             MessageAccessor.clearProperty(newMsg, MessageConst.PROPERTY_TRANSACTION_PREPARED);
             newMsg.setDelayTimeLevel(3 + msg.getReconsumeTimes());
 
-            this.defaultMQPushConsumerImpl.getmQClientFactory().getDefaultMQProducer().send(newMsg);
+            OrderlyRetryMessageSender.send(this.defaultMQPushConsumerImpl, msg, newMsg);
             return true;
         } catch (Exception e) {
             log.error("sendMessageBack exception, group: " + this.consumerGroup + " msg: " + msg.toString(), e);

client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessagePopOrderlyService.java

@@ -295,7 +295,7 @@ public boolean sendMessageBack(final MessageExt msg) {
         try {
             // max reconsume times exceeded then send to dead letter queue.
             Message newMsg = new Message(MixAll.getRetryTopic(this.defaultMQPushConsumer.getConsumerGroup()), msg.getBody());
-            MessageAccessor.setProperties(newMsg, msg.getProperties());
+            MessageAccessor.setProperties(newMsg, MessageAccessor.deepCopyProperties(msg.getProperties()));
             String originMsgId = MessageAccessor.getOriginMessageId(msg);
             MessageAccessor.setOriginMessageId(newMsg, UtilAll.isBlank(originMsgId) ? msg.getMsgId() : originMsgId);
             newMsg.setFlag(msg.getFlag());
@@ -304,7 +304,7 @@ public boolean sendMessageBack(final MessageExt msg) {
             MessageAccessor.setMaxReconsumeTimes(newMsg, String.valueOf(getMaxReconsumeTimes()));
             newMsg.setDelayTimeLevel(3 + msg.getReconsumeTimes());
 
-            this.defaultMQPushConsumerImpl.getmQClientFactory().getDefaultMQProducer().send(newMsg);
+            OrderlyRetryMessageSender.send(this.defaultMQPushConsumerImpl, msg, newMsg);
             return true;
         } catch (Exception e) {
             log.error("sendMessageBack exception, group: " + this.consumerGroup + " msg: " + msg.toString(), e);

client/src/main/java/org/apache/rocketmq/client/impl/consumer/OrderlyRetryMessageSender.java

@@ -0,0 +1,109 @@
+/*
+ * 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.client.impl.consumer;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageAccessor;
+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.logging.org.slf4j.Logger;
+import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
+import org.apache.rocketmq.remoting.exception.RemotingException;
+
+final class OrderlyRetryMessageSender {
+    private static final Logger LOG = LoggerFactory.getLogger(OrderlyRetryMessageSender.class);
+    // A retry topic created by the broker always has queue 0, even before its route is published.
+    private static final int RETRY_QUEUE_ID = 0;
+
+    private OrderlyRetryMessageSender() {
+    }
+
+    static void send(DefaultMQPushConsumerImpl consumer, MessageExt originalMessage, Message retryMessage)
+        throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
+        DefaultMQProducer producer = consumer.getmQClientFactory().getDefaultMQProducer();
+        List<String> brokerNames = findTopicBrokerNames(consumer, originalMessage);
+        if (brokerNames.isEmpty()) {
+            producer.send(copyMessage(retryMessage));
+            return;
+        }
+
+        for (int i = 0; i < brokerNames.size(); i++) {
+            String brokerName = brokerNames.get(i);
+            MessageQueue retryQueue = new MessageQueue(retryMessage.getTopic(), brokerName, RETRY_QUEUE_ID);
+            Message messageToSend = copyMessage(retryMessage);
+            try {
+                producer.send(messageToSend, retryQueue);
+                return;
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw e;
+            } catch (MQClientException | RemotingException | MQBrokerException e) {
+                if (i == brokerNames.size() - 1) {
+                    throw e;
+                }
+                LOG.warn("Failed to send orderly retry message to broker {}, try next topic broker", brokerName, e);
+            }
+        }
+    }
+
+    private static List<String> findTopicBrokerNames(DefaultMQPushConsumerImpl consumer,
+        MessageExt originalMessage) {
+        Set<MessageQueue> messageQueues;
+        String originalTopic = originalMessage.getProperty(MessageConst.PROPERTY_RETRY_TOPIC);
+        if (originalTopic == null || originalTopic.isEmpty()) {
+            originalTopic = originalMessage.getTopic();
+        }
+        try {
+            messageQueues = consumer.fetchSubscribeMessageQueues(originalTopic);
+        } catch (MQClientException e) {
+            return Collections.emptyList();
+        }
+
+        Set<String> brokerNames = new HashSet<>();
+        for (MessageQueue messageQueue : messageQueues) {
+            if (messageQueue.getBrokerName() != null && !messageQueue.getBrokerName().isEmpty()) {
+                brokerNames.add(messageQueue.getBrokerName());
+            }
+        }
+
+        List<String> result = new ArrayList<>();
+        String originalBrokerName = originalMessage.getBrokerName();
+        if (brokerNames.remove(originalBrokerName)) {
+            result.add(originalBrokerName);
+        }
+        List<String> remainingBrokerNames = new ArrayList<>(brokerNames);
+        Collections.sort(remainingBrokerNames);
+        result.addAll(remainingBrokerNames);
+        return result;
+    }
+
+    private static Message copyMessage(Message message) {
+        Message result = MessageAccessor.cloneMessage(message);
+        MessageAccessor.setProperties(result, MessageAccessor.deepCopyProperties(message.getProperties()));
+        result.setTransactionId(message.getTransactionId());
+        return result;
+    }
+}

client/src/test/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessagePopOrderlyServiceTest.java

@@ -46,6 +46,7 @@
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
@@ -204,7 +205,11 @@ public void testGetConsumerStatsManager() {
 
     @Test
     public void testSendMessageBack() {
-        assertTrue(popService.sendMessageBack(createMessageExt()));
+        MessageExt message = createMessageExt();
+
+        assertTrue(popService.sendMessageBack(message));
+
+        assertNull(message.getProperty(MessageConst.PROPERTY_RETRY_TOPIC));
     }
 
     @Test

client/src/test/java/org/apache/rocketmq/client/impl/consumer/OrderlyRetryMessageSenderTest.java

@@ -0,0 +1,201 @@
+/*
+ * 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.client.impl.consumer;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.impl.factory.MQClientInstance;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageAccessor;
+import org.apache.rocketmq.common.message.MessageConst;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotSame;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
+public class OrderlyRetryMessageSenderTest {
+    private static final String ORIGINAL_TOPIC = "topic";
+    private static final String RETRY_TOPIC = "%RETRY%group";
+
+    @Mock
+    private DefaultMQPushConsumerImpl consumer;
+    @Mock
+    private MQClientInstance clientInstance;
+    @Mock
+    private DefaultMQProducer producer;
+
+    private MessageExt originalMessage;
+    private Message retryMessage;
+
+    @Before
+    public void setUp() {
+        when(consumer.getmQClientFactory()).thenReturn(clientInstance);
+        when(clientInstance.getDefaultMQProducer()).thenReturn(producer);
+
+        originalMessage = new MessageExt();
+        originalMessage.setTopic(ORIGINAL_TOPIC);
+        originalMessage.setBrokerName("broker-b");
+
+        retryMessage = new Message(RETRY_TOPIC, "body".getBytes(StandardCharsets.UTF_8));
+        retryMessage.putUserProperty("key", "value");
+        retryMessage.setTransactionId("transaction-id");
+    }
+
+    @Test
+    public void prefersOriginalBrokerAndCopiesMessage() throws Exception {
+        when(consumer.fetchSubscribeMessageQueues(ORIGINAL_TOPIC)).thenReturn(routes("broker-a", "broker-b"));
+
+        OrderlyRetryMessageSender.send(consumer, originalMessage, retryMessage);
+
+        ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class);
+        verify(producer).send(messageCaptor.capture(), eq(retryQueue("broker-b")));
+        verify(producer, never()).send(any(Message.class));
+        Message sentMessage = messageCaptor.getValue();
+        assertNotSame(retryMessage, sentMessage);
+        assertNotSame(retryMessage.getProperties(), sentMessage.getProperties());
+        assertEquals(retryMessage.getProperties(), sentMessage.getProperties());
+        assertEquals("transaction-id", sentMessage.getTransactionId());
+    }
+
+    @Test
+    public void ignoresOriginalBrokerWhenItIsNotInCurrentTopicRoute() throws Exception {
+        originalMessage.setBrokerName("broker-c");
+        when(consumer.fetchSubscribeMessageQueues(ORIGINAL_TOPIC)).thenReturn(routes("broker-b", "broker-a"));
+
+        OrderlyRetryMessageSender.send(consumer, originalMessage, retryMessage);
+
+        verify(producer).send(any(Message.class), eq(retryQueue("broker-a")));
+        verify(producer, never()).send(any(Message.class), eq(retryQueue("broker-c")));
+    }
+
+    @Test
+    public void usesBusinessTopicRouteWhenRetryMessageIsConsumedAgain() throws Exception {
+        originalMessage.setTopic(RETRY_TOPIC);
+        MessageAccessor.putProperty(originalMessage, MessageConst.PROPERTY_RETRY_TOPIC, ORIGINAL_TOPIC);
+        when(consumer.fetchSubscribeMessageQueues(ORIGINAL_TOPIC)).thenReturn(routes("broker-b"));
+
+        OrderlyRetryMessageSender.send(consumer, originalMessage, retryMessage);
+
+        verify(consumer).fetchSubscribeMessageQueues(ORIGINAL_TOPIC);
+        verify(producer).send(any(Message.class), eq(retryQueue("broker-b")));
+    }
+
+    @Test
+    public void retriesWithFreshMessageOnAnotherTopicBroker() throws Exception {
+        when(consumer.fetchSubscribeMessageQueues(ORIGINAL_TOPIC)).thenReturn(routes("broker-a", "broker-b"));
+        doAnswer(invocation -> {
+            Message failedMessage = invocation.getArgument(0);
+            failedMessage.setTopic("mutated-by-producer");
+            throw new MQClientException("first broker failed", null);
+        }).when(producer).send(any(Message.class), eq(retryQueue("broker-b")));
+
+        OrderlyRetryMessageSender.send(consumer, originalMessage, retryMessage);
+
+        ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class);
+        verify(producer).send(messageCaptor.capture(), eq(retryQueue("broker-a")));
+        assertEquals(RETRY_TOPIC, messageCaptor.getValue().getTopic());
+        assertEquals(RETRY_TOPIC, retryMessage.getTopic());
+    }
+
+    @Test
+    public void fallsBackToDefaultRoutingWhenOriginalTopicRouteIsUnavailable() throws Exception {
+        when(consumer.fetchSubscribeMessageQueues(ORIGINAL_TOPIC))
+            .thenThrow(new MQClientException("route unavailable", null));
+
+        OrderlyRetryMessageSender.send(consumer, originalMessage, retryMessage);
+
+        ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class);
+        verify(producer).send(messageCaptor.capture());
+        verify(producer, never()).send(any(Message.class), any(MessageQueue.class));
+        assertNotSame(retryMessage, messageCaptor.getValue());
+    }
+
+    @Test
+    public void fallsBackToDefaultRoutingWhenOriginalTopicRouteIsEmpty() throws Exception {
+        when(consumer.fetchSubscribeMessageQueues(ORIGINAL_TOPIC)).thenReturn(Collections.emptySet());
+
+        OrderlyRetryMessageSender.send(consumer, originalMessage, retryMessage);
+
+        verify(producer).send(any(Message.class));
+        verify(producer, never()).send(any(Message.class), any(MessageQueue.class));
+    }
+
+    @Test
+    public void doesNotFallBackOutsideTopicRouteWhenTopicBrokersFail() throws Exception {
+        when(consumer.fetchSubscribeMessageQueues(ORIGINAL_TOPIC)).thenReturn(routes("broker-a", "broker-b"));
+        doThrow(new MQClientException("broker-b failed", null))
+            .when(producer).send(any(Message.class), eq(retryQueue("broker-b")));
+        doThrow(new MQClientException("broker-a failed", null))
+            .when(producer).send(any(Message.class), eq(retryQueue("broker-a")));
+
+        MQClientException exception = assertThrows(MQClientException.class,
+            () -> OrderlyRetryMessageSender.send(consumer, originalMessage, retryMessage));
+
+        assertEquals("broker-a failed", exception.getErrorMessage());
+        verify(producer, never()).send(any(Message.class));
+    }
+
+    @Test
+    public void preservesInterruptionWithoutTryingAnotherBroker() throws Exception {
+        when(consumer.fetchSubscribeMessageQueues(ORIGINAL_TOPIC)).thenReturn(routes("broker-a", "broker-b"));
+        doThrow(new InterruptedException("interrupted"))
+            .when(producer).send(any(Message.class), eq(retryQueue("broker-b")));
+
+        try {
+            assertThrows(InterruptedException.class,
+                () -> OrderlyRetryMessageSender.send(consumer, originalMessage, retryMessage));
+
+            assertTrue(Thread.currentThread().isInterrupted());
+            verify(producer, never()).send(any(Message.class), eq(retryQueue("broker-a")));
+        } finally {
+            Thread.interrupted();
+        }
+    }
+
+    private Set<MessageQueue> routes(String... brokerNames) {
+        Set<MessageQueue> result = new HashSet<>();
+        Arrays.stream(brokerNames).forEach(brokerName ->
+            result.add(new MessageQueue(ORIGINAL_TOPIC, brokerName, 0)));
+        return result;
+    }
+
+    private MessageQueue retryQueue(String brokerName) {
+        return new MessageQueue(RETRY_TOPIC, brokerName, 0);
+    }
+}