我们正在为 Kafka 使用 Spring Cloud Stream，我们正在寻找具有消费者 API 的 Exactly Once Semantics答案

【问题标题】：We are using Spring Cloud Stream for Kafka and we are looking for Exactly Once Semantics with consumer API我们正在为 Kafka 使用 Spring Cloud Stream，我们正在寻找具有消费者 API 的 Exactly Once Semantics
【发布时间】：2019-08-18 05:52:39
【问题描述】：

我们正在使用 Spring Cloud Stream for Kafka 并寻找 Exactly Once Semantics。我们有一种解决方案可以按预期正常工作 1）从生产者启用幂等和事务 2) 使用 MetaDataStore 使用键 (offsetId + partitionId + topicName) 检查来自消费者端的重复消息使用上述解决方案，我们没有任何消息丢失和重复处理

但现在我们发现有一个属性 (producer.sendOffsetsToTransaction) Kafka API 可以帮助我们修复来自消费者端的重复处理，而无需任何元数据存储逻辑。现在不确定我们如何使用带有此属性.sendOffsetsToTransaction的spring cloud stream来做到这一点@

【问题讨论】：

标签： kafka-consumer-api kafka-producer-api spring-cloud-stream

【解决方案1】：

如果您将KafkaTransactionManager 添加到应用程序上下文，它会由框架自动处理。

您必须在配置中添加事务 ID 前缀。

spring.kafka.producer.transaction-id-prefix

Boot 会自动添加事务管理器。

见producer properties。

spring.cloud.stream.kafka.binder.transaction.transactionIdPrefix

在活页夹中启用事务。请参阅 Kafka 文档中的 transaction.id 和 spring-kafka 文档中的 Transactions。启用事务后，将忽略单个生产者属性，所有生产者都使用 spring.cloud.stream.kafka.binder.transaction.producer.* 属性。

当监听器正常退出时，监听器容器在提交事务之前将偏移量发送给事务。

【讨论】：

感谢加里的回复。因此，如果我在 application.yml 中启用了此属性（ kafka: binder: transaction: transaction-id-prefix:) 它将创建 transactionmanager，实习生将负责调用此方法并将值（偏移量和 groupId）传递给' sendOffsetsToTransaction' 对吗？
否；您要么需要手动配置事务管理器，要么还需要设置 Boot spring.kafka.producer.transaction-id-prefix 属性，以便他为您声明事务管理器。然后，容器会自动为您发送偏移量。
Gary，感谢您的重播，我能够做出改变，并且运行良好。但是我有不同的情况，并且看到了不同的行为。总共有 3 个不同的服务 1）第一个服务将从 Solace 队列中侦听并将其生成到 kafka topic-1（启用事务）2）第二个服务将从 kafka topic-1 上方侦听并将其写入另一个 kafka topic-2 （我们没有手动提交，启用生成其他主题的事务，自动提交偏移为 false &isolation.level 设置为 read_commited）
3) 第三个服务将从 kafka topic-2 侦听并将其写回 Solace 队列（我们没有手动提交，自动提交偏移为 false &isolation.level 设置为 read_commited）。现在我在第二个服务中启用事务和隔离级别后的问题无法读取任何消息，如果我在第二个服务中禁用事务能够读取所有消息。 1) 我们能否在一个服务中启用事务和隔离级别 2) 如果我的服务只是生产者或消费者，它是如何工作的（如何保证这些服务的 EoS）
>and it's working fine...。评论不是为提出其他问题而设计的；你应该关闭这个并开始一个新的。