处理存储库和 Kafka 事务答案

【问题标题】：Handling Repository and Kafka Transactions处理存储库和 Kafka 事务
【发布时间】：2019-01-16 05:04:21
【问题描述】：

我有一个用例，我需要从一个 kafka 主题中消费，做一些工作，产生另一个只有一次语义的 kafka 主题并保存到一个 mongo 数据库。在通过文档后，我认为 kafka 事务和 mongo 事务可以同步，但它们仍然是两个不同的事务。在以下场景中，如果 mongo 提交失败，是否有办法回滚已提交到主题并从消费者重播的 kafka 记录。

producer.send()
producer.sendOffsetsToTransaction()
mongoDao.commit()

【问题讨论】：

标签： spring-kafka

【解决方案1】：

如果监听器抛出异常，kafka事务将回滚并重新传递。

如果mongo提交成功而kafka提交失败，则需要处理重复投递。

如果您将KafkaTransactionManager（或包含一个KafkaChainedTransactionManager）连接到侦听器容器中，则无需将偏移量发送到事务，容器会在提交之前为您完成。

【讨论】：

谢谢 Gary。我想如果 mongo 提交失败并且 kafka 成功，那么我将在 mongo 中丢失一个记录，这是我更关心的。即使我链接了 kafka tm 和 mongo tm 并且有一个带有@transactional 的发送方法，这是否适用。另一方面，我有另一种情况，我正在使用（kafkatemplate.send，kafkatemplate.sendoffsetsToTransaction，h@transactional）消费和生产另一个Kafka主题，因为我正在使用重试）如果kafkatemplate.send成功并且kafkatemplate.sendoffsetsToTransaction会发生什么失败？这种情况下事务会回滚吗？
不要把代码放在cmets中；很难阅读；改为编辑问题。不。如果 mongo 提交失败（并抛出异常），kafka 事务将回滚并重新传递消息。重试无关紧要 - 重试在 kafka 事务中执行。因此，如果您将@Transactional 与 mongo TM 一起使用，则每次交付尝试都会获得一个新的 mongo TX。当重试次数用尽时，kafka 事务将回滚。再次;最好让容器将偏移量发送给事务。
会记下 Gary。使用模板发送偏移量的想法是使重试具有事务性，再次从主题中读取并生成。此外，当使用带有 kafka 模板的 executeintransaction 和 @transactional 方法时记录是否真的被提交到 kafka。换句话说，使用 read_commited 的消费者何时能够看到这个事务。
再次；让容器管理事务。如文档/javadocs中所述，如果您使用template.executeIntransaction()，它将启动一个新事务，而不是参与容器启动的事务。