From 2b8a309ec70370fc3edec113e0fd36b60ba60f67 Mon Sep 17 00:00:00 2001 From: iamssx <1120460561@qq.com> Date: Tue, 16 Jul 2024 14:51:56 +0800 Subject: [PATCH] fixed Rocketmq can not batch send messages when producer set namespace (#5202) --- .../connector/rocketmq/producer/CanalRocketMQProducer.java | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/producer/CanalRocketMQProducer.java b/connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/producer/CanalRocketMQProducer.java index b2024e03..3c1e609e 100644 --- a/connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/producer/CanalRocketMQProducer.java +++ b/connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/producer/CanalRocketMQProducer.java @@ -52,6 +52,7 @@ import java.util.stream.Collectors; private DefaultMQProducer defaultMQProducer; private static final String CLOUD_ACCESS_CHANNEL = "cloud"; + private static final String NAMESPACE_SEPARATOR = "%"; protected ThreadPoolExecutor sendPartitionExecutor; @Override public void init(Properties properties) { @@ -311,7 +312,11 @@ import java.util.stream.Collectors; // 获取一下messageQueue 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) { for (Message message : messages) { sendMessage(message, partition);