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:未捕获kafka事务异常，造成canal ack异常 (#1814)

Browse Source
This commit is contained in:
nieyuan1991
2019-05-28 00:02:25 +08:00
committed by agapple
parent 4075a0623f
commit f9989e0856
1 changed files with 8 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -102,10 +102,10 @@ public class CanalKafkaProducer implements CanalMQProducer {
producerTmp = producer2; producerTmp = producer2;
} }
try {
if (kafkaProperties.getTransaction()) { if (kafkaProperties.getTransaction()) {
producerTmp.beginTransaction(); producerTmp.beginTransaction();
} }
try {
if (!StringUtils.isEmpty(canalDestination.getDynamicTopic())) { if (!StringUtils.isEmpty(canalDestination.getDynamicTopic())) {
// 动态topic // 动态topic
Map<String, Message> messageMap = MQMessageUtils.messageTopics(message, Map<String, Message> messageMap = MQMessageUtils.messageTopics(message,
@@ -130,7 +130,11 @@ public class CanalKafkaProducer implements CanalMQProducer {
} catch (Throwable e) { } catch (Throwable e) {
logger.error(e.getMessage(), e); logger.error(e.getMessage(), e);
if (kafkaProperties.getTransaction()) { if (kafkaProperties.getTransaction()) {
try {
producerTmp.abortTransaction(); producerTmp.abortTransaction();
} catch (Exception e1) {
logger.error(e1.getMessage(), e1);
}
} }
callback.rollback(); callback.rollback();
} }