Kafka 生产者抛出错误尝试从状态 IN_TRANSACTION 到状态 IN_TRANSACTION 的无效转换答案

【问题标题】：Kafka producer throws an error Invalid transition attempted from state IN_TRANSACTION to state IN_TRANSACTIONKafka 生产者抛出错误尝试从状态 IN_TRANSACTION 到状态 IN_TRANSACTION 的无效转换
【发布时间】：2020-05-08 21:33:43
【问题描述】：

我的 kafka 生产者抛出错误“尝试从状态 IN_TRANSACTION 到状态 IN_TRANSACTION 的无效转换”。这是我想要实现的目标 -

KafkaProducer producer = new KafkaProducer<>(props);
producer.initTransactions();
//transaction 1
producer.beginTransaction();
//send some messages
producer.commitTransaction();

//transaction 2
producer.beginTransaction(); //here it throws an exception "Invalid transition attempted from state IN_TRANSACTION to state IN_TRANSACTION".
//send some messages
producer.commitTransaction();

producer.close();

如果我在开始事务 2 之前再次调用 producer.initTransactions();，它会引发异常“尝试从状态 READY 到状态 INITIALIZING 的无效转换”。

我做错了什么？

【问题讨论】：

标签： apache-kafka transactions kafka-producer-api

【解决方案1】：

每次只为每笔交易创建新的生产者。

KafkaProducer producer = new KafkaProducer<>(props);
producer.initTransactions();
//transaction 1
producer.beginTransaction();
//send some messages
producer.commitTransaction();
producer.close();

KafkaProducer producer = new KafkaProducer<>(props);
producer.initTransactions();

//transaction 2
producer.beginTransaction(); //here it throws an exception "Invalid transition attempted from state IN_TRANSACTION to state IN_TRANSACTION".
//send some messages
producer.commitTransaction();

producer.close();

【讨论】：

嗨，为每笔交易创建新的生产者看起来不切实际，知道为什么会抛出这个错误。当它提交事务时，它不应该准备好开始另一个事务吗？

【解决方案2】：

producer.initTransactions();

这会将生产者向代理注册为可以使用事务的生产者，通过其 transactional.id 和序列号或纪元来标识它。反过来，代理将使用这些将任何操作预先写入事务日志。

因此，代理将从该日志中删除属于具有相同事务 ID 和更早纪元的生产者的所有操作，假定它们来自已失效的事务。

https://www.baeldung.com/kafka-exactly-once

在初始化生产者之后，我在构造函数本身中调用 initTransactions ->

Properties properties = getProperties();
kafkaProducer = new org.apache.kafka.clients.producer.KafkaProducer<>(properties);
kafkaProducer.initTransactions();
return kafkaProducer;

【讨论】：