From edb5f294ef45174fbd9b7392e16ca17c9a15ebef Mon Sep 17 00:00:00 2001 From: jiacheo Date: Mon, 5 Nov 2018 22:00:56 +0800 Subject: [PATCH] =?UTF-8?q?fix=EF=BC=9Arollback=E4=B9=8B=E5=90=8E=E7=BB=A7?= =?UTF-8?q?=E7=BB=ADcommit=EF=BC=8C=E4=BC=9A=E4=BA=A7=E7=94=9F=20CanalServ?= =?UTF-8?q?erException:=20ack=20error=20,=20clientId:1001=20batchId:12=20i?= =?UTF-8?q?s=20not=20exist=20,=20please=20check=20=E7=9A=84=E9=94=99?= =?UTF-8?q?=E8=AF=AF=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../otter/canal/kafka/CanalKafkaProducer.java | 23 +++++++++++-------- .../canal/rocketmq/CanalRocketMQProducer.java | 18 ++++++++------- 2 files changed, 23 insertions(+), 18 deletions(-) diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java index 654a4b6b..8576835f 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java @@ -1,21 +1,20 @@ package com.alibaba.otter.canal.kafka; -import java.util.List; -import java.util.Properties; - -import org.apache.kafka.clients.producer.KafkaProducer; -import org.apache.kafka.clients.producer.Producer; -import org.apache.kafka.clients.producer.ProducerRecord; -import org.apache.kafka.common.serialization.StringSerializer; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.serializer.SerializerFeature; import com.alibaba.otter.canal.common.MQProperties; import com.alibaba.otter.canal.protocol.FlatMessage; import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.spi.CanalMQProducer; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.serialization.StringSerializer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.List; +import java.util.Properties; /** * kafka producer 主操作类 @@ -93,6 +92,7 @@ public class CanalKafkaProducer implements CanalMQProducer { logger.error(e.getMessage(), e); // producer.abortTransaction(); callback.rollback(); + return; } } else { // 发送扁平数据json @@ -110,6 +110,7 @@ public class CanalKafkaProducer implements CanalMQProducer { logger.error(e.getMessage(), e); // producer.abortTransaction(); callback.rollback(); + return; } } else { if (canalDestination.getPartitionHash() != null @@ -131,6 +132,7 @@ public class CanalKafkaProducer implements CanalMQProducer { logger.error(e.getMessage(), e); // producer.abortTransaction(); callback.rollback(); + return; } } } @@ -145,6 +147,7 @@ public class CanalKafkaProducer implements CanalMQProducer { logger.error(e.getMessage(), e); // producer.abortTransaction(); callback.rollback(); + return; } } } 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 6173cebd..d8bd45e4 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 @@ -1,7 +1,11 @@ package com.alibaba.otter.canal.rocketmq; -import java.util.List; - +import com.alibaba.fastjson.JSON; +import com.alibaba.otter.canal.common.CanalMessageSerializer; +import com.alibaba.otter.canal.common.MQProperties; +import com.alibaba.otter.canal.protocol.FlatMessage; +import com.alibaba.otter.canal.server.exception.CanalServerException; +import com.alibaba.otter.canal.spi.CanalMQProducer; import org.apache.rocketmq.client.exception.MQBrokerException; import org.apache.rocketmq.client.exception.MQClientException; import org.apache.rocketmq.client.producer.DefaultMQProducer; @@ -12,12 +16,7 @@ import org.apache.rocketmq.remoting.exception.RemotingException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.alibaba.fastjson.JSON; -import com.alibaba.otter.canal.common.CanalMessageSerializer; -import com.alibaba.otter.canal.common.MQProperties; -import com.alibaba.otter.canal.protocol.FlatMessage; -import com.alibaba.otter.canal.server.exception.CanalServerException; -import com.alibaba.otter.canal.spi.CanalMQProducer; +import java.util.List; public class CanalRocketMQProducer implements CanalMQProducer { @@ -67,6 +66,7 @@ public class CanalRocketMQProducer implements CanalMQProducer { } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) { logger.error("Send message error!", e); callback.rollback(); + return; } } else { List flatMessages = FlatMessage.messageConverter(data); @@ -90,6 +90,7 @@ public class CanalRocketMQProducer implements CanalMQProducer { } catch (Exception e) { logger.error("send flat message to fixed partition error", e); callback.rollback(); + return; } } else { if (destination.getPartitionHash() != null && !destination.getPartitionHash().isEmpty()) { @@ -124,6 +125,7 @@ public class CanalRocketMQProducer implements CanalMQProducer { } catch (Exception e) { logger.error("send flat message to hashed partition error", e); callback.rollback(); + return; } } }