You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by du...@apache.org on 2022/03/30 06:18:50 UTC
[rocketmq] branch 5.0.0-beta updated: fix brokerName of msg is null in same case (#4066)
This is an automated email from the ASF dual-hosted git repository.
duhengforever pushed a commit to branch 5.0.0-beta
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/5.0.0-beta by this push:
new 52482d4 fix brokerName of msg is null in same case (#4066)
52482d4 is described below
commit 52482d4e0d610f45c139dc35b8d2d8784452331e
Author: cserwen <cs...@163.com>
AuthorDate: Wed Mar 30 01:18:42 2022 -0500
fix brokerName of msg is null in same case (#4066)
---
.../src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java | 1 +
.../rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyService.java | 1 +
2 files changed, 2 insertions(+)
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
index 468f824..614ce9f 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
@@ -1113,6 +1113,7 @@ public class MQClientAPIImpl implements NameServerUpdateCallback {
messageExt.getProperties().put(MessageConst.PROPERTY_FIRST_POP_TIME, String.valueOf(responseHeader.getPopTime()));
}
}
+ messageExt.setBrokerName(brokerName);
messageExt.setTopic(NamespaceUtil.withoutNamespace(topic, this.clientConfig.getNamespace()));
}
}
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyService.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyService.java
index 90d348d..4aca8f3 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyService.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyService.java
@@ -138,6 +138,7 @@ public class ConsumeMessageConcurrentlyService implements ConsumeMessageService
result.setOrder(false);
result.setAutoCommit(true);
+ msg.setBrokerName(brokerName);
List<MessageExt> msgs = new ArrayList<MessageExt>();
msgs.add(msg);
MessageQueue mq = new MessageQueue();