Hi Guozhang I have created https://issues.apache.org/jira/browse/KAFKA-8023 and by accident found https://issues.apache.org/jira/browse/KAFKA-6721 which was what I was looking for at the beginning. Does this need a KIP? I can maybe help with the writeup but I am not sure I should help with the code ;-)
6721 might indirectly cover point 1) from above as currently (if I understand Taylor correctly) it seems a bit inconsistent that the normal Tables have split config for processing and restore while in the global case both are shared, although it is understandably just a result of using only one consumer for the global state stores. best regards Patrik On Thu, 28 Feb 2019 at 23:46, Guozhang Wang <wangg...@gmail.com> wrote: > Hi Taylor, > > 1) Yes we do allow users to have separate config values for global > consumers / restore consumers via StreamsConfig#restoreConsumerPrefix and > StreamsConfig#globalConsumerPrefix, as Patrik pointed out. > > 2) I think I agree with you that for global consumer, it is worth while to > allow one than one update thread (for restore consumer though we have the > same stream thread for it by design, so that is much harder to > re-architecture). Would you mind creating a JIRA ticket for it so we do not > forget about this potential improvement? > > Guozhang > > On Wed, Feb 27, 2019 at 2:02 PM Taylor P <tdp002...@gmail.com> wrote: > > > Hi Guozhang, Patrik, > > > > Yes, the global consumer setting is what needs to be changed for these > > settings. The restore consumer configs aren't used at all since a > separate > > restore consumer is not initialized for global state store restoration - > > the global consumer is used. I think it would be an improvement to allow > > for using different configs for the global consumer between restoration > and > > regular processing. > > > > I previously tried tweaking fetch.max.bytes and receive.buffer.bytes, but > > if I recall correctly, I was still capped around 100K records/sec. I will > > try tweaking them again when I get time. > > > > Is there anything major that would prevent parallelizing the restoration > of > > each partition of the global state store? It looks like that would be a > > decent chunk of work to refactor, but I think that would have the biggest > > impact in reducing global state restoration times for the scenario where > > the keyspace of the global state store is very large. > > > > Taylor > > > > > > On Thu, Feb 21, 2019 at 6:31 AM Patrik Kleindl <pklei...@gmail.com> > wrote: > > > > > Hello Guozhang, > > > > > > thanks, that might help us too. > > > Just to confirm, this depends on KTable/GlobalKTable usage, right? > > > I did a test with > > > > > > > > > > > > streamsConfiguration.put(StreamsConfig.restoreConsumerPrefix(StreamsConfig.RECEIVE_BUFFER_CONFIG), > > > 65536); > > > > > > > > > streamsConfiguration.put(StreamsConfig.restoreConsumerPrefix(ConsumerConfig.FETCH_MAX_BYTES_CONFIG), > > > 52428800); > > > > > > > > > streamsConfiguration.put(StreamsConfig.globalConsumerPrefix(StreamsConfig.RECEIVE_BUFFER_CONFIG), > > > 65536); > > > > > > > > > streamsConfiguration.put(StreamsConfig.globalConsumerPrefix(ConsumerConfig.FETCH_MAX_BYTES_CONFIG), > > > 52428800); > > > > > > and only if I change the globalConsumerPrefix values I see a change for > > the > > > GlobalKTable restoration. > > > > > > br, Patrik > > > > > > PS: Logs from the test, seems to work fine and get faster/slower > > depending > > > on the change: > > > > > > Default Values: > > > > > > 743 > > > > > > > > > [globalTableRocksDB-7c733295-b253-4c0a-a5bd-81e02b2bd4b7-GlobalStreamThread] > > > INFO org.apache.kafka.clients.Metadata - Cluster ID: > > s2467KdmTlKV5b2YGe831g > > > 936 > > > > > > > > > [globalTableRocksDB-7c733295-b253-4c0a-a5bd-81e02b2bd4b7-GlobalStreamThread] > > > INFO org.apache.kafka.clients.consumer.internals.Fetcher - [Consumer > > > > > > > > > clientId=globalTableRocksDB-7c733295-b253-4c0a-a5bd-81e02b2bd4b7-global-consumer, > > > groupId=] Resetting offset for partition topic-0 to offset 2480157. > > > 13378 > > > > > > > > > [globalTableRocksDB-7c733295-b253-4c0a-a5bd-81e02b2bd4b7-GlobalStreamThread] > > > INFO org.apache.kafka.clients.consumer.internals.Fetcher - [Consumer > > > > > > > > > clientId=globalTableRocksDB-7c733295-b253-4c0a-a5bd-81e02b2bd4b7-global-consumer, > > > groupId=] Resetting offset for partition topic-5 to offset 2311478. > > > 26459 > > > > > > > > > [globalTableRocksDB-7c733295-b253-4c0a-a5bd-81e02b2bd4b7-GlobalStreamThread] > > > INFO org.apache.kafka.clients.consumer.internals.Fetcher - [Consumer > > > > > > > > > clientId=globalTableRocksDB-7c733295-b253-4c0a-a5bd-81e02b2bd4b7-global-consumer, > > > groupId=] Resetting offset for partition topic-10 to offset 2430617. > > > 38178 > > > > > > > > > [globalTableRocksDB-7c733295-b253-4c0a-a5bd-81e02b2bd4b7-GlobalStreamThread] > > > INFO org.apache.kafka.clients.consumer.internals.Fetcher - [Consumer > > > > > > > > > clientId=globalTableRocksDB-7c733295-b253-4c0a-a5bd-81e02b2bd4b7-global-consumer, > > > groupId=] Resetting offset for partition topic-8 to offset 2295704. > > > > > > Default * 4: > > > 714 > > > > > > > > > [globalTableRocksDB-65bf45ed-d044-41fa-9346-90bf64cb0e0d-GlobalStreamThread] > > > INFO org.apache.kafka.clients.Metadata - Cluster ID: > > s2467KdmTlKV5b2YGe831g > > > 862 > > > > > > > > > [globalTableRocksDB-65bf45ed-d044-41fa-9346-90bf64cb0e0d-GlobalStreamThread] > > > INFO org.apache.kafka.clients.consumer.internals.Fetcher - [Consumer > > > > > > > > > clientId=globalTableRocksDB-65bf45ed-d044-41fa-9346-90bf64cb0e0d-global-consumer, > > > groupId=] Resetting offset for partition topic-0 to offset 2480157. > > > 10465 > > > > > > > > > [globalTableRocksDB-65bf45ed-d044-41fa-9346-90bf64cb0e0d-GlobalStreamThread] > > > INFO org.apache.kafka.clients.consumer.internals.Fetcher - [Consumer > > > > > > > > > clientId=globalTableRocksDB-65bf45ed-d044-41fa-9346-90bf64cb0e0d-global-consumer, > > > groupId=] Resetting offset for partition topic-5 to offset 2311478. > > > 20014 > > > > > > > > > [globalTableRocksDB-65bf45ed-d044-41fa-9346-90bf64cb0e0d-GlobalStreamThread] > > > INFO org.apache.kafka.clients.consumer.internals.Fetcher - [Consumer > > > > > > > > > clientId=globalTableRocksDB-65bf45ed-d044-41fa-9346-90bf64cb0e0d-global-consumer, > > > groupId=] Resetting offset for partition topic-10 to offset 2430617. > > > 29570 > > > > > > > > > [globalTableRocksDB-65bf45ed-d044-41fa-9346-90bf64cb0e0d-GlobalStreamThread] > > > INFO org.apache.kafka.clients.consumer.internals.Fetcher - [Consumer > > > > > > > > > clientId=globalTableRocksDB-65bf45ed-d044-41fa-9346-90bf64cb0e0d-global-consumer, > > > groupId=] Resetting offset for partition topic-8 to offset 2295704. > > > 40066 > > > > > > > > > [globalTableRocksDB-65bf45ed-d044-41fa-9346-90bf64cb0e0d-GlobalStreamThread] > > > INFO org.apache.kafka.clients.consumer.internals.Fetcher - [Consumer > > > > > > > > > clientId=globalTableRocksDB-65bf45ed-d044-41fa-9346-90bf64cb0e0d-global-consumer, > > > > > > Default / 4: > > > 679 > > > > > > > > > [globalTableRocksDB-ffdfb1ea-fba8-4034-af8e-159d35231ec6-GlobalStreamThread] > > > INFO > org.apache.kafka.streams.processor.internals.GlobalStateManagerImpl > > - > > > global-stream-thread > > > > > > > > > [globalTableRocksDB-ffdfb1ea-fba8-4034-af8e-159d35231ec6-GlobalStreamThread] > > > Restoring state for global store topic-STATE-STORE-0000000000 > > > 725 > > > > > > > > > [globalTableRocksDB-ffdfb1ea-fba8-4034-af8e-159d35231ec6-GlobalStreamThread] > > > INFO org.apache.kafka.clients.Metadata - Cluster ID: > > s2467KdmTlKV5b2YGe831g > > > 848 > > > > > > > > > [globalTableRocksDB-ffdfb1ea-fba8-4034-af8e-159d35231ec6-GlobalStreamThread] > > > INFO org.apache.kafka.clients.consumer.internals.Fetcher - [Consumer > > > > > > > > > clientId=globalTableRocksDB-ffdfb1ea-fba8-4034-af8e-159d35231ec6-global-consumer, > > > groupId=] Resetting offset for partition topic-0 to offset 2480157. > > > 29283 > > > > > > > > > [globalTableRocksDB-ffdfb1ea-fba8-4034-af8e-159d35231ec6-GlobalStreamThread] > > > INFO org.apache.kafka.clients.consumer.internals.Fetcher - [Consumer > > > > > > > > > clientId=globalTableRocksDB-ffdfb1ea-fba8-4034-af8e-159d35231ec6-global-consumer, > > > groupId=] Resetting offset for partition topic-5 to offset 2311478. > > > 56349 > > > > > > > > > [globalTableRocksDB-ffdfb1ea-fba8-4034-af8e-159d35231ec6-GlobalStreamThread] > > > INFO org.apache.kafka.clients.consumer.internals.Fetcher - [Consumer > > > > > > > > > clientId=globalTableRocksDB-ffdfb1ea-fba8-4034-af8e-159d35231ec6-global-consumer, > > > groupId=] Resetting offset for partition topic-10 to offset 2430617. > > > > > > On Wed, 20 Feb 2019 at 19:16, Guozhang Wang <wangg...@gmail.com> > wrote: > > > > > > > Hello Taylor, > > > > > > > > Sorry for the late reply! And thanks for the updated information. > > > > > > > > I'd recommend overriding some consumer configs via `StreamsConfig` > (you > > > can > > > > use the StreamsConfig#restoreConsumerPrefix for that) for the > following > > > > props: > > > > > > > > 1) increase RECEIVE_BUFFER_CONFIG (64K may cause poll to return early > > > than > > > > necessary) > > > > 2) increase FETCH_MAX_BYTES_CONFIG if you'd expect the total size of > > your > > > > maximum 2000 records to possibly exceed it (default is 50Mb). > > > > > > > > > > > > Guozhang > > > > > > > > > > > > > > > > On Fri, Feb 8, 2019 at 12:43 AM Patrik Kleindl <pklei...@gmail.com> > > > wrote: > > > > > > > > > Hi Taylor > > > > > You are right, the parallel processing is not mentioned in this > > issue, > > > if > > > > > I remember correctly it was in the thread that lead to it as a > > > > possibility > > > > > when changing to the restoration listeners. > > > > > Best regards > > > > > Patrik > > > > > > > > > > > Am 07.02.2019 um 00:47 schrieb Taylor P <tdp002...@gmail.com>: > > > > > > > > > > > > Hi Patrik, > > > > > > > > > > > > I am not sure that > > https://issues.apache.org/jira/browse/KAFKA-7380 > > > > will > > > > > > resolve this issue since our application is dependent on the > global > > > > store > > > > > > being fully restored before the application can be considered > > > healthy. > > > > It > > > > > > does not seem like KAFKA-7380 is aiming to address the nature of > > > global > > > > > > stores restoring each partition sequentially - it is aiming to > > change > > > > the > > > > > > blocking nature of #start(). Restoring the global store > partitions > > in > > > > > > parallel would definitely speed things up, though, and admittedly > > my > > > > > first > > > > > > thought when debugging this was "why isn't this restoring each > > > > partition > > > > > in > > > > > > parallel?". > > > > > > > > > > > > Changing our streams topology to avoid using a global store for > > such > > > a > > > > > > large amount of data would be doable but it does seem like a > > > > significant > > > > > > amount of work. I am curious to know if anyone else is storing > > large > > > > > > amounts of data in global stores and whether there are any > inherent > > > > > > limitations to the size of global stores. > > > > > > > > > > > > Our topic is already using compaction. > > > > > > > > > > > > Taylor > > > > > > > > > > > >> On Wed, Feb 6, 2019 at 2:41 AM Patrik Kleindl < > pklei...@gmail.com > > > > > > > > wrote: > > > > > >> > > > > > >> Hi Taylor > > > > > >> > > > > > >> We are facing the same issue, although on a smaller scale. > > > > > >> The main problem as you found is that the restoration is running > > > > > >> sequentially, this should be addressed in > > > > > >> https://issues.apache.org/jira/browse/KAFKA-7380, although > there > > > has > > > > > been > > > > > >> no progress lately. > > > > > >> > > > > > >> On the other hand you could try re-evaluate if your problem can > > only > > > > be > > > > > >> solved with global state stores, in our case (both in streams as > > > well > > > > as > > > > > >> for interactive queries) we could solve it with local state > stores > > > > too, > > > > > >> although only with more changes and more complexity in the > > topology. > > > > > >> > > > > > >> Not sure if it is applicable for your case, but have you looked > > into > > > > > >> compression for the topics? > > > > > >> > > > > > >> best regards > > > > > >> > > > > > >> Patrik > > > > > >> > > > > > >>> On Tue, 5 Feb 2019 at 22:37, Taylor P <tdp002...@gmail.com> > > wrote: > > > > > >>> > > > > > >>> Hi, > > > > > >>> > > > > > >>> I am having issues with the global store taking a very long > time > > to > > > > > >> restore > > > > > >>> during startup of a Kafka Streams 2.0.1 application. The global > > > store > > > > > is > > > > > >>> backed by a RocksDB persistent store and is added to the > Streams > > > > > topology > > > > > >>> in the following manner: https://pastebin.com/raw/VJutDyYe The > > > > global > > > > > >>> store > > > > > >>> topic has approximately 15 million records per partition and 18 > > > > > >> partitions. > > > > > >>> The following global consumer settings are specified: > > > > > >>> > > > > > >>> poll.timeout.ms = 10 > > > > > >>> max.poll.records = 2000 > > > > > >>> max.partition.fetch.bytes = 1048576 > > > > > >>> fetch.max.bytes = 52428800 > > > > > >>> receive.buffer.bytes = 65536 > > > > > >>> > > > > > >>> I have tried tweaking the settings above on the consumer side, > > such > > > > as > > > > > >>> increasing poll.timeout.ms to 2000, max.poll.records to 10000, > > and > > > > > >>> max.partition.fetch.bytes to 52428800, but it seems that I keep > > > > > hitting a > > > > > >>> ceiling of restoring approximately 100,000 records per second. > > With > > > > 15 > > > > > >>> million records per partition, it takes approximately 150 > seconds > > > to > > > > > >>> restore a single partition. With 18 partitions, it takes > roughly > > 45 > > > > > >> minutes > > > > > >>> to fully restore the global store. > > > > > >>> > > > > > >>> Switching from HDDs to SSDs on the brokers' log directories > made > > > > > >>> restoration roughly 25% faster overall, but this still feels > > slow. > > > It > > > > > >> seems > > > > > >>> that I am hitting IOPS limits on the disks and am not even > close > > to > > > > > >> hitting > > > > > >>> the throughput limits of the disks on either the broker or > > streams > > > > > >>> application side. > > > > > >>> > > > > > >>> How can I minimize restoration time of a global store? Are > there > > > > > settings > > > > > >>> that can increase throughput with the same number of IOPS? > > Ideally > > > > > >>> restoration of each partition could be done in parallel but I > > > > recognize > > > > > >>> there is only a single global store thread. Bringing up a new > > > > instance > > > > > of > > > > > >>> the Kafka Streams application occurs on a potentially daily > > basis, > > > so > > > > > the > > > > > >>> restoration time is becoming more and more of a hassle. > > > > > >>> > > > > > >>> Thanks. > > > > > >>> > > > > > >>> Taylor > > > > > >>> > > > > > >> > > > > > > > > > > > > > > > > > -- > > > > -- Guozhang > > > > > > > > > > > > -- > -- Guozhang >
