与 Kafka 生产者的 ACID 交易答案

【问题标题】：ACID transaction with Kafka producer与 Kafka 生产者的 ACID 交易
【发布时间】：2022-01-27 22:57:31
【问题描述】：

我有一个要发送的 Kafka 的 List<SourceRecord>，但我需要所有这些记录的发送都是 ATOMIC，所以如果在发送我的 Kafka 生产者 die 的过程中，我已经发送列表的一半，我需要能够恢复生产者已经发送的内容。

可以通过exactly once 策略和batch 进程来完成此任务吗？我找不到任何涵盖此场景的文档或示例。

【问题讨论】：

幂等生产者应该处理这个，是的。在最新的 kafka 客户端中，它默认启用。你能分享你的代码吗？您是在尝试编写自定义连接器，还是只是将 SourceRecord 与生产者 api 一起使用？
谢谢，Kafka Client 2.5.1版本默认包含？
最新，我的意思是 2.8+。否则，您需要手动创建交易。 2.0 以上的客户端应该可以工作。我想你已经找到了像这里展示的那样的 sn-ps？ confluent.io/blog/…

【解决方案1】：

您可以使用 Spring 使用本地事务轻松做到这一点：

boolean result = template.executeInTransaction(t -> {
    t.sendDefault("thing1", "thing2");
    t.sendDefault("cat", "hat");
    return true;
});

有关 Spring 事务的完整概述，请参阅

【讨论】：