[
https://issues.apache.org/jira/browse/CAMEL-12732?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16582071#comment-16582071
]
ASF GitHub Bot commented on CAMEL-12732:
----------------------------------------
davsclaus commented on a change in pull request #2476: CAMEL-12732 Take into
account autoCommitEnable for the state
URL: https://github.com/apache/camel/pull/2476#discussion_r210497745
##########
File path:
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java
##########
@@ -403,6 +405,7 @@ public Properties createConsumerProperties() {
addPropertyIfNotNull(props, ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
getAutoOffsetReset());
addPropertyIfNotNull(props,
ConsumerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG, getConnectionMaxIdleMs());
addPropertyIfNotNull(props, ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,
isAutoCommitEnable());
+ addPropertyIfNotNull(props, ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,
isAutoCommitStateEnable());
Review comment:
This is wrong where you override the same option
----------------------------------------------------------------
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
> Assignee: Claus Ibsen
> Priority: Major
> Fix For: 2.23.0
>
>
> 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)
