chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

fixed Rocketmq can not batch send messages when producer set namespace (#5202)

Browse Source
This commit is contained in:
iamssx
2024-07-16 14:51:56 +08:00
committed by GitHub
parent 9373429015
commit 2b8a309ec7
1 changed files with 6 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/producer/CanalRocketMQProducer.java
View File

@@ -52,6 +52,7 @@ import java.util.stream.Collectors;
private DefaultMQProducer defaultMQProducer; private DefaultMQProducer defaultMQProducer;
private static final String CLOUD_ACCESS_CHANNEL = "cloud"; private static final String CLOUD_ACCESS_CHANNEL = "cloud";
private static final String NAMESPACE_SEPARATOR = "%";
protected ThreadPoolExecutor sendPartitionExecutor; protected ThreadPoolExecutor sendPartitionExecutor;
@Override public void init(Properties properties) { @Override public void init(Properties properties) {
@@ -311,7 +312,11 @@ import java.util.stream.Collectors;
// 获取一下messageQueue // 获取一下messageQueue
DefaultMQProducerImpl innerProducer = this.defaultMQProducer.getDefaultMQProducerImpl(); DefaultMQProducerImpl innerProducer = this.defaultMQProducer.getDefaultMQProducerImpl();
TopicPublishInfo topicInfo = innerProducer.getTopicPublishInfoTable().get(messages.get(0).getTopic()); String topic = messages.get(0).getTopic();
if (StringUtils.isNotBlank(this.defaultMQProducer.getNamespace())) {
topic = this.defaultMQProducer.getNamespace() + NAMESPACE_SEPARATOR + topic;
}
TopicPublishInfo topicInfo = innerProducer.getTopicPublishInfoTable().get(topic);
if (topicInfo == null) { if (topicInfo == null) {
for (Message message : messages) { for (Message message : messages) {
sendMessage(message, partition); sendMessage(message, partition);