我正在使用camel的kafka组件在spring引导应用程序中为kafka生成事件。它是通过camel文档中描述的标准uri格式配置的,它是一个生产者。我已将其设置为“幂等”生产者,以确保具有以下属性的“精确一次交付”:
enable-idempotence: true
max-in-flight-request: 1
request-required-acks: all
retries: 100
我还想使用kafkaproducer的事务特性。当我在camel kafka uri上定义附加属性时:
additionalProperties.transactional.id=my-unique-transactional-id
我收到以下例外情况:
java.lang.IllegalStateException: Cannot perform a 'send' before completing a call to initTransactions when transactions are enabled.
使用标准的kafkaproducerjavaapi,应该调用inittransactions()并开始事务。
如何使用camel kafka组件实现这一点?
暂无答案!
目前还没有任何答案,快来回答吧!