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

问题描述

根据问题如何手动控制使用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.

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