canal集成kafka，配置多个destinations，造成kafka事务异常

环境信息

canal version 1.1.3 mysql version 5.7.22 kafka version 1.1.1

问题描述

canal集成kafka，配置多个destinations，造成kafka事务异常。 每一个destination起一个线程订阅canal，当线程1执行完beginTransaction()时，线程2执行beginTransaction()就会造成kafka事务异常，如下：

exception trace:

ERROR com.alibaba.otter.canal.server.CanalMQStarter - TransactionalId canal-transactional-id: Invalid transition attempted from state IN_TRANSACT ION to state IN_TRANSACTION org.apache.kafka.common.KafkaException: TransactionalId canal-transactional-id: Invalid transition attempted from state IN_TRANSACTION to state IN_TRANSACTION at org.apache.kafka.clients.producer.internals.TransactionManager.transitionTo(TransactionManager.java:758) ~[kafka-clients-1.1.1.jar:na] at org.apache.kafka.clients.producer.internals.TransactionManager.transitionTo(TransactionManager.java:751) ~[kafka-clients-1.1.1.jar:na] at org.apache.kafka.clients.producer.internals.TransactionManager.beginTransaction(TransactionManager.java:216) ~[kafka-clients-1.1.1.jar:na] at org.apache.kafka.clients.producer.KafkaProducer.beginTransaction(KafkaProducer.java:587) ~[kafka-clients-1.1.1.jar:na] at com.alibaba.otter.canal.kafka.CanalKafkaProducer.send(CanalKafkaProducer.java:106) ~[canal.server-1.1.3.jar:na] at com.alibaba.otter.canal.server.CanalMQStarter.worker(CanalMQStarter.java:182) [canal.server-1.1.3.jar:na] at com.alibaba.otter.canal.server.CanalMQStarter.access$500(CanalMQStarter.java:22) [canal.server-1.1.3.jar:na] at com.alibaba.otter.canal.server.CanalMQStarter$CanalMQRunnable.run(CanalMQStarter.java:224) [canal.server-1.1.3.jar:na] at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_171] at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_171] at java.lang.Thread.run(Thread.java:748) [na:1.8.0_171]

原提问者GitHub用户nieyuan1991