[ https://issues.apache.org/jira/browse/CAMEL-12732?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16582019#comment-16582019 ]
ASF GitHub Bot commented on CAMEL-12732: ---------------------------------------- dmvolod edited a comment on issue #2476: CAMEL-12732 Take into account autoCommitEnable for the state URL: https://github.com/apache/camel/pull/2476#issuecomment-413438782 If some parameters are incompatible you can detect this and raise an exception or waiting. Please also add a JUnit test for this option. You can add real test and mark it with @Ignored. I'm planning to add testcontainers option for Kafka component and able to run real test against container if possible. ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Kafka manual commit to file repository doesn't work properly (using Spring > boot) > -------------------------------------------------------------------------------- > > Key: CAMEL-12732 > URL: https://issues.apache.org/jira/browse/CAMEL-12732 > Project: Camel > Issue Type: Bug > Components: camel-kafka > Affects Versions: 2.22.0 > Environment: Spring boot > kafka_2.11-1.1.0 > Reporter: michael elbaz > Priority: Major > > I'im trying to save the Kafka offset into FileStateRepository, the offset is > correctly writing but it is not reading at route start so camel will read all > the topic every time > > {code:java} > @Component > public class Route extends RouteBuilder { > @Override > public void configure() throws Exception { > from(kafka()) > .to("log:TEST?level=INFO") > .process(Route::commitKafka); > } > private String kafka() { > String kafkaEndpoint = "kafka:"; > kafkaEndpoint += "topictest"; > kafkaEndpoint += "?brokers="; > kafkaEndpoint += "localhost:9092"; > kafkaEndpoint += "&groupId="; > kafkaEndpoint += "TEST"; > kafkaEndpoint += "&autoOffsetReset="; > kafkaEndpoint += "earliest"; > kafkaEndpoint += "&autoCommitEnable="; > kafkaEndpoint += false; > kafkaEndpoint += "&allowManualCommit="; > kafkaEndpoint += true; > kafkaEndpoint += "&offsetRepository="; > kafkaEndpoint += "#fileStore"; > return kafkaEndpoint; > } > @Bean(name = "fileStore") > private FileStateRepository fileStateRepository() { > FileStateRepository fileStateRepository = > FileStateRepository.fileStateRepository(new > File("/kafka/offset_repo/repo.dat")); > // This will be empty > // System.out.println(fileStateRepository.getCache()); > return fileStateRepository; > } > private static void commitKafka(Exchange exchange) { > KafkaManualCommit manual = > exchange.getIn().getHeader(KafkaConstants.MANUAL_COMMIT, > KafkaManualCommit.class); > manual.commitSync(); > } > } > {code} > -- This message was sent by Atlassian JIRA (v7.6.3#76005)