如何使用camel-kafka提交偏移线程安全? [英] How to commit offsets thread safe using camel-kafka?

问题描述

根据问题如何手动控制我要使用camel-kafka手动提交偏移量吗?我的路线:

As asked in question How to manually control the offset commit with camel-kafka? I want to commit offsets manually using camel-kafka. My route:

.from(kafka:topic1)
 .aggregate(new GroupByExchangeStrategy())
.to(kafka:topic2)
 .process(new ManualCommitProcessor())

，其中 ManualCommitProcessor 在将消息发送到另一个主题后将执行承诺.

, where ManualCommitProcessor will do the commitment after sending the message to another topic.

问题是聚合器和kafka生产者在与kafka消费者分离的线程中工作，后者负责抵消承诺.因此，我以

Problem is that aggregator and kafka producer are working in separated threads to the kafka consumer which is responsible for offset commitment. Hence, I am ending in

java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access

是否有可能在聚合并分派提交偏移量后再次调用使用者线程?

Is there a possibility to call the consumer thread again after aggregation and dispatching to commit offsets?

推荐答案

这是不可能的，使用者线程独立于聚合器的输出运行.

No this is not possible, the consumer thread runs independently of the output of the aggregator.

这篇关于如何使用camel-kafka提交偏移线程安全?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！