Micronaut kafka：无法完全使用一次 kafka 消息语义

Question

我正在使用 micronaut kafka 来设置我的制作人。我正在使用 @KafkaClient 注释来设置所有生产者配置 Micronaut kafka 使我能够设置所有参数来设置事务生产者。 当我推送消息时，我得到一个异常提示

io.micronaut.messaging.exceptions.MessagingClientException: Exception sending producer record: Cannot perform 'send' before completing a call to initTransactions when transactions are enabled.

参考 mirconaut 文档 section 看起来它要求您使用 KafkaProducer API 来实现此功能。

据我所知，需要在开始事务之前调用 KafkaProducer.initTransactions() 方法，并且看起来并没有发生。

有没有人在实现这个时遇到过类似的问题？

Answer 1

我猜，您正在使用单节点集群进行开发，对吧？如果是这样，您应该在本地集群上配置 transaction.state.log.min.isr=1 和 transaction.state.log.replication.factor=1。默认情况下，它们都预配置为 3。

还有一段来自confluenthttps://docs.confluent.io/current/streams/developer-guide/config-streams.html

processing.guarantee
应使用的加工保证。可能的值为“at_least_once”（默认）和“exactly_once”。请注意，如果启用了一次性处理，则参数 commit.interval.ms 的默认值将更改为 100 毫秒。此外，消费者配置了isolation.level="read_committed"，生产者配置了retries=Integer.MAX_VALUE 和enable.idempotence=true。请注意，“exactly_once”处理默认情况下需要至少三个代理的集群，这是生产的推荐设置。对于开发，您可以通过将 transaction.state.log.replication.factor 和 transaction.state.log.min.isr 中的代理设置调整为您要使用的代理数量来更改此设置。