Hi, After reviewing the code, I think that it is, indeed, a bug. As such, I created a PR which should solve the problem: https://github.com/apache/camel/pull/8746.
This is a patch for the upcoming 3.20.0. I would kindly appreciate it if you could give a try with this version of Camel, so I can safely backport for Camel 3.18 (LTS). Kind regards On Sat, Nov 19, 2022 at 12:21 PM Otavio Rodolfo Piske <angusyo...@gmail.com> wrote: > Hi, > > I am glad that you managed to adjust it to work the way you need. > > "In my opinion, the offset repository should be used in case we define it > in the Kafka endpoint no matter if it is AsyncCommitManager, > SyncCommitManager or NoopCommitManager. > > I’m not so sure if my thought is correct. Could you please let me know the > right behaviour of the CommitManagers?" > > I think it's a bug on our code - but I am not 100% sure yet. If the user > provides a state repository, then it should be updated when the commit > happens. I'll take a look at this during this week, so, please keep an eye > on the ticket as I'll provide updates there if/when I fix it. > > Thanks for reporting this. > > Kind regards > > > On Wed, Nov 16, 2022 at 11:10 AM Yuttana Sangchoei <tanay...@gmail.com> > wrote: > >> Hi Otavio, >> >> Thank you for creating the ticket. I found where the problem might be in >> 3.18.3. At CommitManager in createCommitManager static method, there are >> some check starting by checking if isAllowManualCommit, if so it continues >> checking the type of manualCommitFactory. However, if not, it checks the >> configuration if there is offset repository defined, then return >> CommitToOffsetManager object. >> >> I want CommitToOffsetManager so I removed the flag, >> allowManualCommit=true, out of the Kafka endpoint. It works as expect. The >> offset in the offset repository gets increased. >> >> In my opinion, the offset repository should be used in case we define it >> in the Kafka endpoint no matter if it is AsyncCommitManager, >> SyncCommitManager or NoopCommitManager. >> >> I’m not so sure if my thought is correct. Could you please let me know >> the right behaviour of the CommitManagers? >> >> Thank you very much and best regards, >> Yuttana >> >> On 2022/11/12 08:05:14 Otavio Rodolfo Piske wrote: >> > Thanks. I can't confirm whether it's an user error, bug or anything >> else, >> > but I created a ticket so I can investigate this: >> > https://issues.apache.org/jira/browse/CAMEL-18717 >> > >> > I also can't provide an ETA yet, as I have a busy week ahead, but I'll >> keep >> > it on my radar. In the meantime, please do feel free to contribute with >> a >> > reproducer or even propose a bug fix if you want as any of those will >> > certainly make the resolution faster. We love receiving contributions. >> > >> > Kind regards >> > >> > On Fri, Nov 11, 2022 at 10:28 AM Sangchoei, Yuttana >> > <yu...@anz.com.invalid> wrote: >> > >> > > Hi Otavio, Clays, >> > > >> > > >> > > Thank you very much for your hands. I tried 3.18.3. It didn't >> increase the >> > > offset. >> > > >> > > Best regards, >> > > Yuttana >> > > >> > > >> > > On 2022/11/10 06:36:08 Otavio Rodolfo Piske wrote: >> > > > Hi, before I go ahead and investigate this further ... Could you >> possibly >> > > > try with Camel 3.18.3? We have made a bunch of fixes for Kafka in >> 3.18.3. >> > > > >> > > > Kind regards >> > > > >> > > > On Wed, Nov 9, 2022 at 3:56 PM Claus Ibsen <cl...@gmail.com> wrote: >> > > > >> > > > > Hi >> > > > > >> > > > > I am not sure, there are some improvements and refactorings going >> on in >> > > > > camel-kafka and offset management. >> > > > > I think Otavio would be better to help look into this. >> > > > > >> > > > > On Wed, Nov 9, 2022 at 11:36 AM Sangchoei, Yuttana >> > > > > <yu...@anz.com.invalid> wrote: >> > > > > >> > > > > > Hi All, >> > > > > > >> > > > > > I'm upgrading Camel from 3.16 to 3.17 and found a problem that >> > > > > > OffsetRepository didn't get update after manual commit. >> > > > > > It is a Spring boot project and MemoryStateRepository as >> follows. >> > > > > > >> > > > > > @Bean >> > > > > > public MemoryStateRepository offsetRepo() { >> > > > > > MemoryStateRepository stateRepository = new >> > > > > > MemoryStateRepository(); >> > > > > > stateRepository.setState(topic + "/0", ""); >> > > > > > return stateRepository; >> > > > > > } >> > > > > > >> > > > > > Here is the method I use when commit: >> > > > > > >> > > > > > private void commitOffsetToKafka(Exchange exchange) { >> > > > > > KafkaManualCommit manual = >> > > > > > exchange.getIn().getHeader(KafkaConstants.MANUAL_COMMIT, >> > > > > > KafkaManualCommit.class); >> > > > > > manual.commit(); >> > > > > > } >> > > > > > >> > > > > > Here is my Kafka endpoint >> > > > > > >> > > > > > kafka:integration_test _topic?brokers=127.0.0.1:41478 >> > > > > > >> > > > > >> > > >> &autoCommitEnable=false&allowManualCommit=true&seekTo=beginning&maxPollRecords=2&groupId=ANE&autoOffsetReset=earliest&offsetRepository=#offsetRepo&kafkaManualCommitFactory=#class:org.apache.camel.component.kafka.consumer.DefaultKafkaManualCommitFactory >> > > > > > >> > > > > > It is working fine in 3.16. When >> > > DefaultkafkaManualSyncCommit.commit() is >> > > > > > invoked. The offsetRepository.setState() get invoked so the >> offset >> > > > > > increased. >> > > > > > However, in 3.17, DefaultkafkaManualSyncCommit.commit() doesn't >> have >> > > a >> > > > > > reference to offsetRepository so the offset in offsetRepository >> > > won't get >> > > > > > increased. >> > > > > > >> > > > > > Is this a bug or there is a different way to get the offset >> after >> > > > > > committing. Could you please suggest what I should do to test >> if the >> > > > > offset >> > > > > > get increased after manual commit? >> > > > > > >> > > > > > Best regards, >> > > > > > Yuttana Sangchoei >> > > > > > >> > > > > > This e-mail and any attachments to it (the "Communication") is, >> > > unless >> > > > > > otherwise stated, confidential, may contain copyright material >> and >> > > is for >> > > > > > the use only of the intended recipient. If you receive the >> > > Communication >> > > > > in >> > > > > > error, please notify the sender immediately by return e-mail, >> delete >> > > the >> > > > > > Communication and the return e-mail, and do not read, copy, >> > > retransmit or >> > > > > > otherwise deal with it. Any views expressed in the >> Communication are >> > > > > those >> > > > > > of the individual sender only, unless expressly stated to be >> those of >> > > > > > Australia and New Zealand Banking Group Limited ABN 11 005 357 >> 522, >> > > or >> > > > > any >> > > > > > of its related entities including ANZ Bank New Zealand Limited >> > > (together >> > > > > > "ANZ"). ANZ does not accept liability in connection with the >> > > integrity of >> > > > > > or errors in the Communication, computer virus, data corruption, >> > > > > > interference or delay arising from or in respect of the >> > > Communication. >> > > > > > >> > > > > >> > > > > >> > > > > -- >> > > > > Claus Ibsen >> > > > > ----------------- >> > > > > @davsclaus >> > > > > Camel in Action 2: https://www.manning.com/ibsen2 >> > > > > >> > > > >> > > > >> > > > -- >> > > > Otavio R. Piske >> > > > http://orpiske.net >> > > > >> > > >> > > Get Outlook for Android<https://aka.ms/AAb9ysg> >> > > This e-mail and any attachments to it (the "Communication") is, unless >> > > otherwise stated, confidential, may contain copyright material and is >> for >> > > the use only of the intended recipient. If you receive the >> Communication in >> > > error, please notify the sender immediately by return e-mail, delete >> the >> > > Communication and the return e-mail, and do not read, copy, >> retransmit or >> > > otherwise deal with it. Any views expressed in the Communication are >> those >> > > of the individual sender only, unless expressly stated to be those of >> > > Australia and New Zealand Banking Group Limited ABN 11 005 357 522, >> or any >> > > of its related entities including ANZ Bank New Zealand Limited >> (together >> > > "ANZ"). ANZ does not accept liability in connection with the >> integrity of >> > > or errors in the Communication, computer virus, data corruption, >> > > interference or delay arising from or in respect of the Communication. >> > > >> > >> > >> > -- >> > Otavio R. Piske >> > http://orpiske.net >> > > > > > -- > Otavio R. Piske > http://orpiske.net > -- Otavio R. Piske http://orpiske.net
