如何使用camel-kafka提交偏移线程安全? [英] 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.

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