From b64a3e301adbe3af60311ab6cc7882a7013cdab7 Mon Sep 17 00:00:00 2001 From: agapple Date: Thu, 10 Oct 2019 15:37:17 +0800 Subject: [PATCH] fixed issue #2286 , ArithmeticException zero --- .../canal/rocketmq/CanalRocketMQProducer.java | 33 +++++++++++-------- 1 file changed, 20 insertions(+), 13 deletions(-) diff --git a/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java b/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java index d6c8a3b2..50c6a6fd 100644 --- a/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java @@ -249,20 +249,27 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ // 批量发送 List queues = topicInfo.getMessageQueueList(); int size = queues.size(); - MessageQueue queue = null; - if (partition > queues.size()) { - queue = queues.get(partition % size); - } else { - queue = queues.get(partition); - } - - try { - SendResult sendResult = this.defaultMQProducer.send(messages, queue); - if (logger.isDebugEnabled()) { - logger.debug("Send Message Result: {}", sendResult); + if (size <= 0) { + // 可能是第一次创建 + for (Message message : messages) { + sendMessage(message, partition); + } + } else { + MessageQueue queue = null; + if (partition > size) { + queue = queues.get(partition % size); + } else { + queue = queues.get(partition); + } + + try { + SendResult sendResult = this.defaultMQProducer.send(messages, queue); + if (logger.isDebugEnabled()) { + logger.debug("Send Message Result: {}", sendResult); + } + } catch (Throwable e) { + throw new RuntimeException(e); } - } catch (Throwable e) { - throw new RuntimeException(e); } } }