BTW, yes that indicates the record in the changelog was already truncated (logically). But since we only physically truncate logs by segments, which is 1GB by default, it should still be physically on the log. Are you enabling EOS on Streams, and when you shutdown the streams app, is that a clean shutdown?
On Thu, Mar 25, 2021 at 4:22 PM Guozhang Wang <wangg...@gmail.com> wrote: > That's indeed weird. > > Have you tried to run Kafka brokers with 2.6 while Kafka Streams client > with 2.7? > > On Thu, Mar 25, 2021 at 2:34 PM Upesh Desai <ude...@itrsgroup.com> wrote: > >> Hello Guozhang, >> >> >> >> I have tried your suggestions with an inMemoryStore FYI and seen the >> following: >> >> >> >> 1. I have the record added to the state store, stopped the >> application, and check the earliest and latest offsets via the command >> line >> tools. This shows that the earliest offset is 1, and the latest offset is >> also 1. Does this mean that the record has been marked for deletion >> already? My retention.ms config is set to 3 days (259200000 ms), so >> it should not be marked for deletion if added a couple minutes prior? >> 2. Following the above, this makes sense as well. When logging the >> starting offset, it is not 0, but rather 1: >> >> *topic: streamapp-teststore-changelog, partition: 4, start offset: 1, >> end offset: 1* >> >> >> >> I also confirmed different behavior when we change the changelog topic >> cleanup policy from “*compact,delete”* to just “*compact”*. We DO NOT >> see this issue when the changelog is just set to compact. We also confirmed >> that this does not happen when we run everything on Kafka version 2.6. >> >> >> >> Thanks, >> >> Upesh >> >> >> Upesh Desai | Senior Software Developer | *ude...@itrsgroup.com* >> <ude...@itrsgroup.com> >> *www.itrsgroup.com* <https://www.itrsgroup.com/> >> <https://www.itrsgroup.com/> >> >> *From: *Guozhang Wang <wangg...@gmail.com> >> *Date: *Thursday, March 25, 2021 at 4:01 PM >> *To: *Users <users@kafka.apache.org> >> *Cc: *Bart Lilje <bli...@itrsgroup.com> >> *Subject: *Re: Kafka Streams Processor API state stores not restored via >> changelog topics >> >> Hello Upesh, >> >> Could you confirm a few more things for me: >> >> 1. After you stopped the application, and wiped out the state dir; check >> if >> the corresponding changelog topic has one record indeed at offset 0 --- >> this can be done via the admin#listOffsets (get the earliest and latest >> offset, which should be 0 and 1 correspondingly). >> 2. After you resumed the application, check from which starting position >> we >> are restoring the changelog --- this can be done via implementing the >> `stateRestoreListener.onRestoreStart(partition, storeName, startOffset, >> restoreEndOffset);`, should be 0 >> >> If both of them check out fine as expected, then from the code I think >> bufferedLimitIndex should be updated to 1. >> >> >> Guozhang >> >> On Wed, Mar 24, 2021 at 5:14 PM Upesh Desai <ude...@itrsgroup.com> wrote: >> >> > Hi Guozhang, >> > >> > >> > >> > Here are some of the answers to your questions I see during my testing: >> > >> > >> > >> > 1. ChangelogMetadata#restoreEndOffset == 1 ; This is expected as in >> my >> > test 1 record had been added to the store. However the numRecords >> variable >> > is still set to 0 >> > 2. For that particular test, `hasRestoredToEnd()` indeed returns true >> > as well. But it is confusing since the store is actually empty / that >> > record I added does not exist in the store when trying to check for >> it. >> > 3. N/A >> > >> > >> > >> > A little more information, the records we add to this store/changelog >> are >> > of type <CustomKey,byte[]> where the value is always set to an empty >> byte >> > array `new byte[0]`. A couple other variations I have tried are setting >> to >> > a non-empty static byte array such as `new byte[1]` or `new byte[]{1}`. >> > >> > >> > >> > Hope this gives a little more clarity and hope to hear from you soon. >> > >> > >> > >> > Thanks, >> > >> > Upesh >> > >> > >> > Upesh Desai | Senior Software Developer | *ude...@itrsgroup.com* >> > <ude...@itrsgroup.com> >> > *www.itrsgroup.com* <https://www.itrsgroup.com/> >> > <https://www.itrsgroup.com/> >> > >> > *From: *Guozhang Wang <wangg...@gmail.com> >> > *Date: *Wednesday, March 24, 2021 at 1:37 PM >> > *To: *Users <users@kafka.apache.org> >> > *Cc: *Bart Lilje <bli...@itrsgroup.com> >> > *Subject: *Re: Kafka Streams Processor API state stores not restored via >> > changelog topics >> > >> > Hello Upesh, >> > >> > Thanks for the detailed report. I looked through the code and tried to >> > reproduce the issue, but so far have not succeeded. I think I may need >> some >> > further information from you to help my further investigation. >> > >> > 1) The `bufferedLimitIndex == 0` itself does not necessarily mean >> there's >> > an issue, as long as it could still be bumped later (i.e. it is possible >> > that the restore consumer has not fetched data yet). What's key though, >> is >> > to check `ChangelogMetadata#restoreEndOffset`: for active tasks, it >> would >> > be created with null value, and then been initialized once. >> ChangelogReader >> > would stop restoring once the current offset has reached beyond this >> value >> > or if this value itself is 0. >> > >> > 2) If `restoreEndOffset` is initialized to a non-zero value, then check >> if >> > the restoration indeed completed without applying any records, this is >> > determined as `hasRestoredToEnd()` returning true. >> > >> > 3) If `restoreEndOffset` is initialized to 0, then we need to check >> why: on >> > top of my head I can only think of that the consumer's end offset >> request >> > gets the response with 0, indicating the changelog is now empty. >> > >> > >> > Guozhang >> > >> > >> > On Tue, Mar 23, 2021 at 8:44 AM Upesh Desai <ude...@itrsgroup.com> >> wrote: >> > >> > > Hi all, >> > > >> > > >> > > >> > > Our team think we discovered a bug over the weekend withing the Kafka >> > > Streams / Processor API. We are running 2.7.0. >> > > >> > > >> > > >> > > When configuring a state store backed by a changelog topic with the >> > > cleanup policy configuration set to “compact,delete”: >> > > >> > > >> > > >> > > final StoreBuilder<KeyValueStore<k,v>> store = Stores >> > > .*keyValueStoreBuilder*( >> > > Stores.*persistentKeyValueStore*(*STORE_ID*), >> > > kSerde, >> > > vSerde) >> > > .withLoggingEnabled(Map.*of*( >> > > *RETENTION_MS_CONFIG*, "90000000"), >> > > *CLEANUP_POLICY_CONFIG*, "compact,delete")) >> > > .withCachingEnabled(); >> > > >> > > >> > > >> > > Here is how we reproduced the problem: >> > > >> > > 1. Records are written to the state store, and subsequently >> produced >> > > to the changelog topic. >> > > 2. Store streams application >> > > 3. Delete state.dir directory >> > > 4. Restart streams application >> > > 5. Confirm state store is initialized empty with no records >> restored >> > > from changelog >> > > >> > > >> > > >> > > We see this problem with both in-memory and RocksDB backed state >> stores. >> > > For persistent state store, if the streams application is restarted >> > without >> > > the state dir being deleted, the application still does not “restore” >> > from >> > > the changelog, but records are still seen in the state store. >> > > >> > > >> > > >> > > When rolling back to 2.6, we do not see this issue. >> > > >> > > >> > > >> > > Doing some debugging in the source code, in the StoreChangelogReader >> > class >> > > I found that the number of records to restore is always 0 based on the >> > > below snippet: >> > > >> > > >> > > >> > > private void restoreChangelog(final ChangelogMetadata >> changelogMetadata) >> > { >> > > final ProcessorStateManager stateManager = >> > changelogMetadata.stateManager; >> > > final StateStoreMetadata storeMetadata = >> > changelogMetadata.storeMetadata; >> > > final TopicPartition partition = >> storeMetadata.changelogPartition(); >> > > final String storeName = storeMetadata.store().name(); >> > > final int numRecords = changelogMetadata.bufferedLimitIndex; >> > > >> > > >> > > >> > > Where ‘changelogMetadata.bufferedLimitIndex’ always == 0. >> > > >> > > >> > > >> > > My question to you all is, 1) Is this expected behavior? 2) If not, >> is it >> > > a bug? >> > > >> > > >> > > >> > > Hope to get some clarity, and thanks in advance! >> > > >> > > >> > > >> > > Best, >> > > Upesh >> > > <https://www.itrsgroup.com/> >> > > Upesh Desai >> > > Senior Software Developer >> > > *ude...@itrsgroup.com* <ude...@itrsgroup.com> >> > > *www.itrsgroup.com* <https://www.itrsgroup.com/> >> > > Internet communications are not secure and therefore the ITRS Group >> does >> > > not accept legal responsibility for the contents of this message. Any >> > view >> > > or opinions presented are solely those of the author and do not >> > necessarily >> > > represent those of the ITRS Group unless otherwise specifically >> stated. >> > > [itrs.email.signature] >> > > >> > > >> > > *Disclaimer* >> > > >> > > The information contained in this communication from the sender is >> > > confidential. It is intended solely for use by the recipient and >> others >> > > authorized to receive it. If you are not the recipient, you are hereby >> > > notified that any disclosure, copying, distribution or taking action >> in >> > > relation of the contents of this information is strictly prohibited >> and >> > may >> > > be unlawful. >> > > >> > > This email has been scanned for viruses and malware, and may have been >> > > automatically archived by *Mimecast Ltd*, an innovator in Software as >> a >> > > Service (SaaS) for business. Providing a *safer* and *more useful* >> place >> > > for your human generated data. Specializing in; Security, archiving >> and >> > > compliance. >> > > >> > >> > >> > -- >> > -- Guozhang >> > >> > >> > *Disclaimer* >> > >> > The information contained in this communication from the sender is >> > confidential. It is intended solely for use by the recipient and others >> > authorized to receive it. If you are not the recipient, you are hereby >> > notified that any disclosure, copying, distribution or taking action in >> > relation of the contents of this information is strictly prohibited and >> may >> > be unlawful. >> > >> > This email has been scanned for viruses and malware, and may have been >> > automatically archived by *Mimecast Ltd*, an innovator in Software as a >> > Service (SaaS) for business. Providing a *safer* and *more useful* place >> > for your human generated data. Specializing in; Security, archiving and >> > compliance. >> > >> >> >> -- >> -- Guozhang >> >> >> *Disclaimer* >> >> The information contained in this communication from the sender is >> confidential. It is intended solely for use by the recipient and others >> authorized to receive it. If you are not the recipient, you are hereby >> notified that any disclosure, copying, distribution or taking action in >> relation of the contents of this information is strictly prohibited and may >> be unlawful. >> >> This email has been scanned for viruses and malware, and may have been >> automatically archived by *Mimecast Ltd*, an innovator in Software as a >> Service (SaaS) for business. Providing a *safer* and *more useful* place >> for your human generated data. Specializing in; Security, archiving and >> compliance. >> > > > -- > -- Guozhang > -- -- Guozhang
