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

fix：rollback之后继续commit，会产生 CanalServerException: ack error , clientId:1001 batchId:12 is not exist , please check 的错误。

Browse Source
This commit is contained in:
jiacheo
2018-11-05 22:00:56 +08:00
parent 5a76e2aa58
commit edb5f294ef
2 changed files with 23 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java
View File

@@ -1,21 +1,20 @@
package com.alibaba.otter.canal.kafka; 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.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature; import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.otter.canal.common.MQProperties; import com.alibaba.otter.canal.common.MQProperties;
import com.alibaba.otter.canal.protocol.FlatMessage; import com.alibaba.otter.canal.protocol.FlatMessage;
import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.spi.CanalMQProducer; 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 主操作类 * kafka producer 主操作类
@@ -93,6 +92,7 @@ public class CanalKafkaProducer implements CanalMQProducer {
logger.error(e.getMessage(), e); logger.error(e.getMessage(), e);
// producer.abortTransaction(); // producer.abortTransaction();
callback.rollback(); callback.rollback();
return;
} }
} else { } else {
// 发送扁平数据json // 发送扁平数据json
@@ -110,6 +110,7 @@ public class CanalKafkaProducer implements CanalMQProducer {
logger.error(e.getMessage(), e); logger.error(e.getMessage(), e);
// producer.abortTransaction(); // producer.abortTransaction();
callback.rollback(); callback.rollback();
return;
} }
} else { } else {
if (canalDestination.getPartitionHash() != null if (canalDestination.getPartitionHash() != null
@@ -131,6 +132,7 @@ public class CanalKafkaProducer implements CanalMQProducer {
logger.error(e.getMessage(), e); logger.error(e.getMessage(), e);
// producer.abortTransaction(); // producer.abortTransaction();
callback.rollback(); callback.rollback();
return;
} }
} }
} }
@@ -145,6 +147,7 @@ public class CanalKafkaProducer implements CanalMQProducer {
logger.error(e.getMessage(), e); logger.error(e.getMessage(), e);
// producer.abortTransaction(); // producer.abortTransaction();
callback.rollback(); callback.rollback();
return;
} }
} }
} }

18
server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java
View File

@@ -1,7 +1,11 @@
package com.alibaba.otter.canal.rocketmq; 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.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException; import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer; 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.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSON; import java.util.List;
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;
public class CanalRocketMQProducer implements CanalMQProducer { public class CanalRocketMQProducer implements CanalMQProducer {
@@ -67,6 +66,7 @@ public class CanalRocketMQProducer implements CanalMQProducer {
} catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) { } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
logger.error("Send message error!", e); logger.error("Send message error!", e);
callback.rollback(); callback.rollback();
return;
} }
} else { } else {
List<FlatMessage> flatMessages = FlatMessage.messageConverter(data); List<FlatMessage> flatMessages = FlatMessage.messageConverter(data);
@@ -90,6 +90,7 @@ public class CanalRocketMQProducer implements CanalMQProducer {
} catch (Exception e) { } catch (Exception e) {
logger.error("send flat message to fixed partition error", e); logger.error("send flat message to fixed partition error", e);
callback.rollback(); callback.rollback();
return;
} }
} else { } else {
if (destination.getPartitionHash() != null && !destination.getPartitionHash().isEmpty()) { if (destination.getPartitionHash() != null && !destination.getPartitionHash().isEmpty()) {
@@ -124,6 +125,7 @@ public class CanalRocketMQProducer implements CanalMQProducer {
} catch (Exception e) { } catch (Exception e) {
logger.error("send flat message to hashed partition error", e); logger.error("send flat message to hashed partition error", e);
callback.rollback(); callback.rollback();
return;
} }
} }
} }