完全事务性的 Spring Kafka 消费者/侦听器

Question

目前，我有一个配置了 ConcurrentKafkaListenerContainerFactory 和 SeekToCurrentErrorHandler 的 Kafka 侦听器（DeadLetterPublishingRecoverer 配置了 1 次重试）。 我的 Listener 方法用 @Transactional 注释（以及我的服务中与数据库交互的所有方法）。

我的 Listener 方法执行以下操作：

1. 从 Kafka 接收消息
2. 与多个服务交互，将接收到的数据的不同部分保存到数据库中
3. Kafka 中的确认消息（即提交偏移量）

如果它在中间的某个地方失败，它应该回滚并重试，直到最大重试次数。 然后向 DLT 发送消息。

我正在尝试使此方法完全具有事务性，即，如果出现故障，所有以前的更改都会回滚。 但是，Listener 方法中的@Transactional 注解是不够的。 我怎样才能做到这一点？ 我应该采用哪些配置来使 Listener 方法完全事务化？

Answer 1

如果您还没有从侦听器发布到 Kafka，则没有必要（或好处）使用 Kafka 事务；只是开销。 STCEH + DLPR 就足够了。

如果您还发布到 Kafka（并且希望这些也回滚），请参阅 the documentation - 在侦听器容器中配置 KafkaTransactionManager。