[rocketmq] branch develop updated: [ISSUE #2724] Resend message to DLQ directly when max reconsume times reach (#3318)

[vongosling]

This is an automated email from the ASF dual-hosted git repository.

vongosling pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git


The following commit(s) were added to refs/heads/develop by this push:
     new 8434658  [ISSUE #2724] Resend message to DLQ directly when max 
reconsume times reach (#3318)
8434658 is described below

commit 8434658798ad5a77f9f2137c72a82b157dab5398
Author: 张旭 <maixiaoha...@gmail.com>
AuthorDate: Sat Sep 11 17:21:50 2021 +0800

    [ISSUE #2724] Resend message to DLQ directly when max reconsume times reach 
(#3318)
    
    * [Store] Resend message directly to DLQ when max reconsume times reach
    
    * set msg delaylevel=0 when topic is DLQ
    
    Co-authored-by: zhangxu16 <zhangx...@xiaomi.com>
---
 .../java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java | 2 ++
 1 file changed, 2 insertions(+)

diff --git 
a/broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java
 
b/broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java
index a76c144..1b1cf4b 100644
--- 
a/broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java
+++ 
b/broker/src/main/java/org/apache/rocketmq/broker/processor/SendMessageProcessor.java
@@ -197,6 +197,7 @@ public class SendMessageProcessor extends 
AbstractSendMessageProcessor implement
                 response.setRemark("topic[" + newTopic + "] not exist");
                 return CompletableFuture.completedFuture(response);
             }
+            msgExt.setDelayTimeLevel(0);
         } else {
             if (0 == delayLevel) {
                 delayLevel = 3 + msgExt.getReconsumeTimes();
@@ -358,6 +359,7 @@ public class SendMessageProcessor extends 
AbstractSendMessageProcessor implement
                 );
                 msg.setTopic(newTopic);
                 msg.setQueueId(queueIdInt);
+                msg.setDelayTimeLevel(0);
                 if (null == topicConfig) {
                     response.setCode(ResponseCode.SYSTEM_ERROR);
                     response.setRemark("topic[" + newTopic + "] not exist");