[
https://issues.apache.org/jira/browse/CAMEL-12732?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16581612#comment-16581612
]
ASF GitHub Bot commented on CAMEL-12732:
----------------------------------------
GitHub user mikadev opened a pull request:
https://github.com/apache/camel/pull/2476
CAMEL-12732 Take into account autoCommitEnable for the state
Check if autocommitstate is enabled before setting the state
(Fix the manual commit for the state)
Warning! if the AutoCommitStateEnable is true and manual.commitSync is
called when AllowManualCommit is true the state will be committed twice
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/mikadev/camel master
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/camel/pull/2476.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #2476
----
commit c3d89a59eb6bd9d78042e7e38972bc9d0ebd3dfc
Author: elbaz michael <elbazmichael92@...>
Date: 2018-08-15T21:02:39Z
CAMEL-12732 Take into account autoCommitEnable for the state
Check if autocommitstate is enabled before setting the state
(Fix the manual commit for the state)
Warning! if the AutoCommitStateEnable is true and manual.commitSync is
called when AllowManualCommit is true the state will be committed twice
----
> 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)
