[ISSUE #8584] fix missing brokerName in sendMessageBack request (#8585)

* fix missing brokerName in sendMessageBack request * fix
apache · Aug 30, 2024 · 720c87e · 720c87e
1 parent 71ec1ed
commit 720c87e
Show file tree

Hide file tree

Showing 4 changed files with 5 additions and 4 deletions.
diff --git a/client/src/main/java/org/apache/rocketmq/client/consumer/DefaultMQPullConsumer.java b/client/src/main/java/org/apache/rocketmq/client/consumer/DefaultMQPullConsumer.java
@@ -262,7 +262,7 @@ public void setRegisterTopics(Set<String> registerTopics) {
     public void sendMessageBack(MessageExt msg, int delayLevel)
         throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
         msg.setTopic(withNamespace(msg.getTopic()));
-        this.defaultMQPullConsumerImpl.sendMessageBack(msg, delayLevel, null);
+        this.defaultMQPullConsumerImpl.sendMessageBack(msg, delayLevel, msg.getBrokerName());
     }
 
     /**

diff --git a/client/src/main/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumer.java b/client/src/main/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumer.java
@@ -688,7 +688,7 @@ public void setSubscription(Map<String, String> subscription) {
     public void sendMessageBack(MessageExt msg, int delayLevel)
         throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
         msg.setTopic(withNamespace(msg.getTopic()));
-        this.defaultMQPushConsumerImpl.sendMessageBack(msg, delayLevel, (String) null);
+        this.defaultMQPushConsumerImpl.sendMessageBack(msg, delayLevel, msg.getBrokerName());
     }
 
     /**

diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java
@@ -752,7 +752,7 @@ public void sendMessageBack(MessageExt msg, int delayLevel, final String brokerN
 
     public void sendMessageBack(MessageExt msg, int delayLevel, final MessageQueue mq)
             throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
-        sendMessageBack(msg, delayLevel, null, mq);
+        sendMessageBack(msg, delayLevel, msg.getBrokerName(), mq);
     }
 
 

diff --git a/...src/test/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImplTest.java b/...src/test/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImplTest.java
@@ -651,10 +651,11 @@ public void testQueryMessageByUniqKey() throws InterruptedException, MQClientExc
 
     @Test
     public void testSendMessageBack() throws InterruptedException, MQClientException, MQBrokerException, RemotingException {
+        when(mQClientFactory.findBrokerAddressInPublish(anyString())).thenReturn(defaultBrokerAddr);
         defaultMQPushConsumerImpl.sendMessageBack(createMessageExt(), 1, createMessageQueue());
         verify(mqClientAPIImpl).consumerSendMessageBack(
                 eq(defaultBrokerAddr),
-                any(),
+                eq(defaultBroker),
                 any(MessageExt.class),
                 any(),
                 eq(1),