You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by ji...@apache.org on 2022/08/27 03:00:40 UTC
[rocketmq] branch develop updated: [ISSUE #4841] Fix the reply message is the original request msg. (#4896)
This is an automated email from the ASF dual-hosted git repository.
jinrongtong 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 f7c2072a2 [ISSUE #4841] Fix the reply message is the original request msg. (#4896)
f7c2072a2 is described below
commit f7c2072a27773d5feefeb5cb6cd5dead0450e3b7
Author: echooymxq <ec...@gmail.com>
AuthorDate: Sat Aug 27 11:00:20 2022 +0800
[ISSUE #4841] Fix the reply message is the original request msg. (#4896)
---
.../client/impl/producer/DefaultMQProducerImpl.java | 1 -
.../rocketmq/client/producer/DefaultMQProducerTest.java | 16 +++++++++++++---
2 files changed, 13 insertions(+), 4 deletions(-)
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java
index 7ec8e0033..739ab0a73 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java
@@ -1374,7 +1374,6 @@ public class DefaultMQProducerImpl implements MQProducerInner {
@Override
public void onSuccess(SendResult sendResult) {
requestResponseFuture.setSendRequestOk(true);
- requestResponseFuture.putResponseMessage(msg);
}
@Override
diff --git a/client/src/test/java/org/apache/rocketmq/client/producer/DefaultMQProducerTest.java b/client/src/test/java/org/apache/rocketmq/client/producer/DefaultMQProducerTest.java
index 14d3339fc..919fb009f 100644
--- a/client/src/test/java/org/apache/rocketmq/client/producer/DefaultMQProducerTest.java
+++ b/client/src/test/java/org/apache/rocketmq/client/producer/DefaultMQProducerTest.java
@@ -42,6 +42,7 @@ import org.apache.rocketmq.client.impl.factory.MQClientInstance;
import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl;
import org.apache.rocketmq.client.impl.producer.TopicPublishInfo;
import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeader;
import org.apache.rocketmq.common.protocol.route.BrokerData;
@@ -377,15 +378,19 @@ public class DefaultMQProducerTest {
Thread.sleep(10);
} catch (InterruptedException e) {
}
+ MessageExt responseMsg = new MessageExt();
+ responseMsg.setTopic(message.getTopic());
+ responseMsg.setBody(message.getBody());
for (Map.Entry<String, RequestResponseFuture> entry : responseMap.entrySet()) {
RequestResponseFuture future = entry.getValue();
- future.putResponseMessage(message);
+ future.putResponseMessage(responseMsg);
}
}
}
}).start();
Message result = producer.request(message, 3 * 1000L);
finish.getAndSet(true);
+ assertThat(result).isExactlyInstanceOf(MessageExt.class);
assertThat(result.getTopic()).isEqualTo("FooBar");
assertThat(result.getBody()).isEqualTo(new byte[] {'a'});
}
@@ -403,6 +408,7 @@ public class DefaultMQProducerTest {
RequestCallback requestCallback = new RequestCallback() {
@Override
public void onSuccess(Message message) {
+ assertThat(message).isExactlyInstanceOf(MessageExt.class);
assertThat(message.getTopic()).isEqualTo("FooBar");
assertThat(message.getBody()).isEqualTo(new byte[] {'a'});
assertThat(message.getFlag()).isEqualTo(1);
@@ -416,11 +422,15 @@ public class DefaultMQProducerTest {
producer.request(message, requestCallback, 3 * 1000L);
ConcurrentHashMap<String, RequestResponseFuture> responseMap = RequestFutureHolder.getInstance().getRequestFutureTable();
assertThat(responseMap).isNotNull();
+
+ MessageExt responseMsg = new MessageExt();
+ responseMsg.setTopic(message.getTopic());
+ responseMsg.setBody(message.getBody());
+ responseMsg.setFlag(1);
for (Map.Entry<String, RequestResponseFuture> entry : responseMap.entrySet()) {
RequestResponseFuture future = entry.getValue();
future.setSendRequestOk(true);
- message.setFlag(1);
- future.getRequestCallback().onSuccess(message);
+ future.getRequestCallback().onSuccess(responseMsg);
}
countDownLatch.await(3000L, TimeUnit.MILLISECONDS);
}