I am using apache camel to integerate with my kafka messaging. Also i am using JAVA DSL to consume messages from kafka endpoint.
Using apache kafka API's its known how to commit consumer offsets with given properties toggling.
if i disbale auto commit in camel-kafka component how can i commit offsets in apcahe camel then?
I am using below endpoint to disable auto commit in Apache Camel
kafka.consumer.uri = kafka://{{kafka.host}}?zookeeperHost={{zookeeper.host}}&zookeeperPort={{zookeeper.port}}&groupId={{kafka.consumerGroupId}}&topic={{kafka.topic.transformer_t}}&{{kafka.uri.options}}
kafka.uri.options = autoCommitEnable=false
Support for manual offset commit was added in CAMEL-11933 and is supported since version 2.21.0. You can enable it with option allowManualCommit=true. This option will expose header CamelKafkaManualCommit
, holding instance of KafkaManualCommit
, which you can then use e.g. in Processor
.
from("kafka:topic?groupId=group&autoCommitEnable=false&allowManualCommit=true")
//...
.process(exchange -> {
exchange.getIn().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class).commitSync();
});
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With