Hi Bruno,
"As far as I understand, the issue is that bulk loading as done in Kafka
Streams does work as expected if FIFO compaction is used."
You meant "doesn't" right ?
Ok, I will open a ticket, but I don't think my "fix" is the correct one.
Just ignoring the issue doesn't seem to be a correct solution :)
Le jeu. 16 avr. 2020 à 11:49, Bruno Cadonna <br...@confluent.io> a écrit :
> Hi Nicolas,
>
> Thank you for reporting this issue.
>
> As far as I understand, the issue is that bulk loading as done in Kafka
> Streams does work as expected if FIFO compaction is used.
>
> I would propose that you open a bug ticket. Please make sure to include
> steps to reproduce the issue in the ticket. Additionally, since you already
> have some code that solves the issue, you could assign the ticket to
> yourself and open a PR for it. Would you be interested in contributing?
>
> If you cannot assign the ticket to yourself, ask to be added to the list of
> contributors on the dev mailing list.
>
> Best,
> Bruno
>
> On Thu, Apr 16, 2020 at 11:06 AM Nicolas Carlot <
> nicolas.car...@chronopost.fr> wrote:
>
> > # Bump #
> > So is this a bug ? Should I file a ticket ?
> > Any idea ? I don't like the idea of having to patch Kafka libraries...
> >
> > Le mer. 1 avr. 2020 à 16:33, Nicolas Carlot <
> nicolas.car...@chronopost.fr>
> > a écrit :
> >
> >> Added some nasty code in kafka 2.4.1. Seems to work fine for now... From
> >> my understanding, the compaction process when restoring a store is only
> >> done to speed up things.
> >> So I guess this kind of "hack" isn't such a big deal ?
> >>
> >> [image: image.png]
> >>
> >> Le mer. 1 avr. 2020 à 10:44, Nicolas Carlot <
> nicolas.car...@chronopost.fr>
> >> a écrit :
> >>
> >>> Here is the full configuration of Rocks.
> >>>
> >>> [image: image.png]
> >>>
> >>> Le mer. 1 avr. 2020 à 10:41, Nicolas Carlot <
> >>> nicolas.car...@chronopost.fr> a écrit :
> >>>
> >>>> It's not that I cannot turn on compaction.
> >>>> Compaction works fine.
> >>>> The issue is with the restoration process of the state store, which
> >>>> tries to compact the store to a single level:
> db.compactRange(columnFamily,
> >>>> true, 1, 0) before bulk loading the data.
> >>>> It automatically fails when I'm using FIFO compaction.
> >>>>
> >>>>
> >>>> Le mer. 1 avr. 2020 à 10:26, Nicolas Carlot <
> >>>> nicolas.car...@chronopost.fr> a écrit :
> >>>>
> >>>>> My current workaround is to completely delete the state store and
> >>>>> rebuild it from scratch.
> >>>>>
> >>>>> Le mar. 31 mars 2020 à 21:39, Boyang Chen <
> reluctanthero...@gmail.com>
> >>>>> a écrit :
> >>>>>
> >>>>>> Thanks Nicolas for the report, so are you suggesting that you
> >>>>>> couldn't turn
> >>>>>> on compactions for the state store? Is there a workaround?
> >>>>>>
> >>>>>> On Tue, Mar 31, 2020 at 9:54 AM Nicolas Carlot <
> >>>>>> nicolas.car...@chronopost.fr>
> >>>>>> wrote:
> >>>>>>
> >>>>>> > After some more testing and debugging, it seems that it is caused
> >>>>>> by the
> >>>>>> > compaction option I've configured for RocksDB. When removed
> >>>>>> everything is
> >>>>>> > fine...
> >>>>>> > The option is as follow:
> >>>>>> >
> >>>>>> > CompactionOptionsFIFO fifoOptions = new CompactionOptionsFIFO();
> >>>>>> > fifoOptions.setMaxTableFilesSize(maxSize);
> >>>>>> > fifoOptions.setAllowCompaction(true);
> >>>>>> > options.setCompactionOptionsFIFO(fifoOptions);
> >>>>>> > options.setCompactionStyle(CompactionStyle.FIFO);
> >>>>>> >
> >>>>>> > Le mar. 31 mars 2020 à 16:27, Nicolas Carlot <
> >>>>>> nicolas.car...@chronopost.fr
> >>>>>> > >
> >>>>>> > a écrit :
> >>>>>> >
> >>>>>> > > Hello everyone,
> >>>>>> > >
> >>>>>> > > I'm currently facing an issue with RocksDb internal compaction
> >>>>>> process,
> >>>>>> > > which occurs when the local state store of several of my
> >>>>>> KafkaStream
> >>>>>> > > applications are being restored. This is sadly a huge concern as
> >>>>>> it
> >>>>>> > > completely discard resiliency over node failure as those often
> >>>>>> lead to a
> >>>>>> > > state store restoration. The only workaround I currently have is
> >>>>>> to
> >>>>>> > delete
> >>>>>> > > the local store to restore it from scratch. I'm using version
> >>>>>> 2.4.1 of
> >>>>>> > the
> >>>>>> > > Java libraries.
> >>>>>> > >
> >>>>>> > > The exception thrown by the KStream process is:
> >>>>>> > > org.apache.kafka.streams.errors.ProcessorStateException: Error
> >>>>>> while
> >>>>>> > range
> >>>>>> > > compacting during restoring store merge_store
> >>>>>> > > at
> >>>>>> > >
> >>>>>> >
> >>>>>>
> org.apache.kafka.streams.state.internals.RocksDBStore$SingleColumnFamilyAccessor.toggleDbForBulkLoading(RocksDBStore.java:615)
> >>>>>> > > ~[kafka-stream-router.jar:?]
> >>>>>> > > at
> >>>>>> > >
> >>>>>> >
> >>>>>>
> org.apache.kafka.streams.state.internals.RocksDBStore.toggleDbForBulkLoading(RocksDBStore.java:398)
> >>>>>> > > ~[kafka-stream-router.jar:?]
> >>>>>> > > at
> >>>>>> > >
> >>>>>> >
> >>>>>>
> org.apache.kafka.streams.state.internals.RocksDBStore$RocksDBBatchingRestoreCallback.onRestoreStart(RocksDBStore.java:644)
> >>>>>> > > ~[kafka-stream-router.jar:?]
> >>>>>> > > at
> >>>>>> > >
> >>>>>> >
> >>>>>>
> org.apache.kafka.streams.processor.internals.CompositeRestoreListener.onRestoreStart(CompositeRestoreListener.java:59)
> >>>>>> > > ~[kafka-stream-router.jar:?]
> >>>>>> > > at
> >>>>>> > >
> >>>>>> >
> >>>>>>
> org.apache.kafka.streams.processor.internals.StateRestorer.restoreStarted(StateRestorer.java:76)
> >>>>>> > > ~[kafka-stream-router.jar:?]
> >>>>>> > > at
> >>>>>> > >
> >>>>>> >
> >>>>>>
> org.apache.kafka.streams.processor.internals.StoreChangelogReader.startRestoration(StoreChangelogReader.java:211)
> >>>>>> > > ~[kafka-stream-router.jar:?]
> >>>>>> > > at
> >>>>>> > >
> >>>>>> >
> >>>>>>
> org.apache.kafka.streams.processor.internals.StoreChangelogReader.initialize(StoreChangelogReader.java:185)
> >>>>>> > > ~[kafka-stream-router.jar:?]
> >>>>>> > > at
> >>>>>> > >
> >>>>>> >
> >>>>>>
> org.apache.kafka.streams.processor.internals.StoreChangelogReader.restore(StoreChangelogReader.java:81)
> >>>>>> > > ~[kafka-stream-router.jar:?]
> >>>>>> > > at
> >>>>>> > >
> >>>>>> >
> >>>>>>
> org.apache.kafka.streams.processor.internals.TaskManager.updateNewAndRestoringTasks(TaskManager.java:389)
> >>>>>> > > ~[kafka-stream-router.jar:?]
> >>>>>> > > at
> >>>>>> > >
> >>>>>> >
> >>>>>>
> org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:769)
> >>>>>> > > ~[kafka-stream-router.jar:?]
> >>>>>> > > at
> >>>>>> > >
> >>>>>> >
> >>>>>>
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:698)
> >>>>>> > > ~[kafka-stream-router.jar:?]
> >>>>>> > > at
> >>>>>> > >
> >>>>>> >
> >>>>>>
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:671)
> >>>>>> > > [kafka-stream-router.jar:?]
> >>>>>> > > Caused by: org.rocksdb.RocksDBException: Target level exceeds
> >>>>>> number of
> >>>>>> > > levels
> >>>>>> > > at org.rocksdb.RocksDB.compactRange(Native Method)
> >>>>>> > > ~[kafka-stream-router.jar:?]
> >>>>>> > > at org.rocksdb.RocksDB.compactRange(RocksDB.java:2636)
> >>>>>> > > ~[kafka-stream-router.jar:?]
> >>>>>> > > at
> >>>>>> > >
> >>>>>> >
> >>>>>>
> org.apache.kafka.streams.state.internals.RocksDBStore$SingleColumnFamilyAccessor.toggleDbForBulkLoading(RocksDBStore.java:613)
> >>>>>> > > ~[kafka-stream-router.jar:?]
> >>>>>> > > ... 11 more
> >>>>>> > >
> >>>>>> > > Here is the state of partition 0_0 of one of those stores:
> >>>>>> > > -rw-r--r-- 1 java j2ee 148210568 31 mars 10:49 000139.sst
> >>>>>> > > -rw-r--r-- 1 java j2ee 21620385 31 mars 11:06 000184.sst
> >>>>>> > > -rw-r--r-- 1 java j2ee 0 31 mars 11:11 000198.log
> >>>>>> > > -rw-r--r-- 1 java j2ee 31602468 31 mars 11:31 000251.sst
> >>>>>> > > -rw-r--r-- 1 java j2ee 37856549 31 mars 12:00 000324.sst
> >>>>>> > > -rw-r--r-- 1 java j2ee 33498822 31 mars 12:26 000393.sst
> >>>>>> > > -rw-r--r-- 1 java j2ee 34368461 31 mars 12:49 000450.sst
> >>>>>> > > -rw-r--r-- 1 java j2ee 11371247 31 mars 12:55 000467.sst
> >>>>>> > > -rw-r--r-- 1 java j2ee 14356435 31 mars 13:04 000489.sst
> >>>>>> > > -rw-r--r-- 1 java j2ee 5858737 31 mars 13:05 000494.sst
> >>>>>> > > -rw-r--r-- 1 java j2ee 2545952 31 mars 14:08 000659.sst
> >>>>>> > > -rw-r--r-- 1 java j2ee 3187275 31 mars 15:27 000868.sst
> >>>>>> > > -rw-r--r-- 1 java j2ee 407017 31 mars 15:34 000885.sst
> >>>>>> > > -rw-r--r-- 1 java j2ee 590190 31 mars 15:45 000914.sst
> >>>>>> > > -rw-r--r-- 1 java j2ee 154471 31 mars 15:47 000919.sst
> >>>>>> > > -rw-r--r-- 1 java j2ee 139838 31 mars 15:49 000924.sst
> >>>>>> > > -rw-r--r-- 1 java j2ee 35058 31 mars 15:49 000925.sst
> >>>>>> > > -rw-r--r-- 1 java j2ee 33987 31 mars 15:50 000926.sst
> >>>>>> > > -rw-r--r-- 1 java j2ee 16 31 mars 11:11 CURRENT
> >>>>>> > > -rw-r--r-- 1 java j2ee 37 31 mars 10:33 IDENTITY
> >>>>>> > > -rw-r--r-- 1 java j2ee 0 31 mars 10:33 LOCK
> >>>>>> > > -rw-r--r-- 1 java j2ee 15340 31 mars 11:11 LOG
> >>>>>> > > -rw-r--r-- 1 java j2ee 15046 31 mars 10:33
> >>>>>> LOG.old.1585643630145007
> >>>>>> > > -rw-r--r-- 1 java j2ee 15290 31 mars 10:33
> >>>>>> LOG.old.1585643826265995
> >>>>>> > > -rw-r--r-- 1 java j2ee 15384 31 mars 10:37
> >>>>>> LOG.old.1585645861692248
> >>>>>> > > -rw-r--r-- 1 java j2ee 60767 31 mars 15:55 MANIFEST-000197
> >>>>>> > > -rw-r--r-- 1 java j2ee 4857 31 mars 10:37 OPTIONS-000107
> >>>>>> > > -rw-r--r-- 1 java j2ee 4857 31 mars 11:11 OPTIONS-000200
> >>>>>> > >
> >>>>>> > > I can see that Kafka is running RocksDB's compaction with:
> >>>>>> > > public void toggleDbForBulkLoading() {
> >>>>>> > > try {
> >>>>>> > > db.compactRange(columnFamily, true, 1, 0);
> >>>>>> > > } catch (final RocksDBException e) {
> >>>>>> > > throw new ProcessorStateException("Error while
> >>>>>> range
> >>>>>> > > compacting during restoring store " + name, e);
> >>>>>> > > }
> >>>>>> > > }
> >>>>>> > >
> >>>>>> > > Seems related to an issue with RocksDB itself:
> >>>>>> > > https://github.com/facebook/rocksdb/issues/2734
> >>>>>> > > But i'm not sure of this.
> >>>>>> > > Any help would be greatly appreciated here :)
> >>>>>> > >
> >>>>>> >
> >>>>>> >
> >>>>>> > --
> >>>>>> > *Nicolas Carlot*
> >>>>>> >
> >>>>>> > Lead dev
> >>>>>> > | | nicolas.car...@chronopost.fr
> >>>>>> >
> >>>>>> >
> >>>>>> > *Veuillez noter qu'à partir du 20 mai, le siège Chronopost
> >>>>>> déménage. La
> >>>>>> > nouvelle adresse est : 3 boulevard Romain Rolland 75014 Paris*
> >>>>>> >
> >>>>>> > [image: Logo Chronopost]
> >>>>>> > | chronopost.fr <http://www.chronopost.fr/>
> >>>>>> > Suivez nous sur Facebook <https://fr-fr.facebook.com/chronopost>
> et
> >>>>>> > Twitter
> >>>>>> > <https://twitter.com/chronopost>.
> >>>>>> >
> >>>>>> > [image: DPD Group]
> >>>>>> >
> >>>>>>
> >>>>>
> >>>>>
> >>>>> --
> >>>>> *Nicolas Carlot*
> >>>>>
> >>>>> Lead dev
> >>>>> | | nicolas.car...@chronopost.fr
> >>>>>
> >>>>>
> >>>>> *Veuillez noter qu'à partir du 20 mai, le siège Chronopost déménage.
> >>>>> La nouvelle adresse est : 3 boulevard Romain Rolland 75014 Paris*
> >>>>>
> >>>>> [image: Logo Chronopost]
> >>>>> | chronopost.fr <http://www.chronopost.fr/>
> >>>>> Suivez nous sur Facebook <https://fr-fr.facebook.com/chronopost> et
> >>>>> Twitter <https://twitter.com/chronopost>.
> >>>>>
> >>>>> [image: DPD Group]
> >>>>>
> >>>>
> >>>>
> >>>> --
> >>>> *Nicolas Carlot*
> >>>>
> >>>> Lead dev
> >>>> | | nicolas.car...@chronopost.fr
> >>>>
> >>>>
> >>>> *Veuillez noter qu'à partir du 20 mai, le siège Chronopost déménage.
> La
> >>>> nouvelle adresse est : 3 boulevard Romain Rolland 75014 Paris*
> >>>>
> >>>> [image: Logo Chronopost]
> >>>> | chronopost.fr <http://www.chronopost.fr/>
> >>>> Suivez nous sur Facebook <https://fr-fr.facebook.com/chronopost> et
> >>>> Twitter <https://twitter.com/chronopost>.
> >>>>
> >>>> [image: DPD Group]
> >>>>
> >>>
> >>>
> >>> --
> >>> *Nicolas Carlot*
> >>>
> >>> Lead dev
> >>> | | nicolas.car...@chronopost.fr
> >>>
> >>>
> >>> *Veuillez noter qu'à partir du 20 mai, le siège Chronopost déménage. La
> >>> nouvelle adresse est : 3 boulevard Romain Rolland 75014 Paris*
> >>>
> >>> [image: Logo Chronopost]
> >>> | chronopost.fr <http://www.chronopost.fr/>
> >>> Suivez nous sur Facebook <https://fr-fr.facebook.com/chronopost> et
> >>> Twitter <https://twitter.com/chronopost>.
> >>>
> >>> [image: DPD Group]
> >>>
> >>
> >>
> >> --
> >> *Nicolas Carlot*
> >>
> >> Lead dev
> >> | | nicolas.car...@chronopost.fr
> >>
> >>
> >> *Veuillez noter qu'à partir du 20 mai, le siège Chronopost déménage. La
> >> nouvelle adresse est : 3 boulevard Romain Rolland 75014 Paris*
> >>
> >> [image: Logo Chronopost]
> >> | chronopost.fr <http://www.chronopost.fr/>
> >> Suivez nous sur Facebook <https://fr-fr.facebook.com/chronopost> et
> >> Twitter <https://twitter.com/chronopost>.
> >>
> >> [image: DPD Group]
> >>
> >
> >
> > --
> > *Nicolas Carlot*
> >
> > Lead dev
> > | | nicolas.car...@chronopost.fr
> >
> >
> > *Veuillez noter qu'à partir du 20 mai, le siège Chronopost déménage. La
> > nouvelle adresse est : 3 boulevard Romain Rolland 75014 Paris*
> >
> > [image: Logo Chronopost]
> > | chronopost.fr <http://www.chronopost.fr/>
> > Suivez nous sur Facebook <https://fr-fr.facebook.com/chronopost> et
> > Twitter <https://twitter.com/chronopost>.
> >
> > [image: DPD Group]
> >
>
--
*Nicolas Carlot*
Lead dev
| | nicolas.car...@chronopost.fr
*Veuillez noter qu'à partir du 20 mai, le siège Chronopost déménage. La
nouvelle adresse est : 3 boulevard Romain Rolland 75014 Paris*
[image: Logo Chronopost]
| chronopost.fr <http://www.chronopost.fr/>
Suivez nous sur Facebook <https://fr-fr.facebook.com/chronopost> et Twitter
<https://twitter.com/chronopost>.
[image: DPD Group]
