Partition allocation and even distribution and resource usage across consumers in a group

[rajatgoyal247341]

Let's say I have topic with 15 partitions, which is currently consumed by a group of 3 consumers and each running on a difference instance.

Now with parallel consumers :

a) For achieving parallel partition order, is it always a good practice to start with good high max-concurrency ( even if some threads are idle). If I execute 20 max-concurrency on each instance, 5 threads will run on each consumer and rest will be idle. I am asking this because if we downscale our system ( remove an application node) from 3 to 2, will it be running 7 & 8 (total 15) threads per instance ? Does it auto-adjust the thread-pool size on every instance, based on no-of consumers in group ?

b) In same fashion (for achieving parallel partition order) , if we add more consumer ( add an instance) to the group, does load sharing among instances change automatically ? For example if I increase consumers from 3 to 4, will each instance be running 4,4,4,3 threads respectively,

c) For achieving unordered / key ordering : In case of upscale / downscale, does load distribution adjustment happens automatically among instances.

[astubbs]

Hi @rajatgoyal247341 , do you not want any help with this anymore?

The max concurrency setting is per instance only, not global. However, this could potentially be changed to an optionally global setting. Is that something that you'd find useful?

[Jayesh-pict]

what if I dynamically add more partitions to my Kafka topic then how should I assign consumers to this topic? And if I want to stop consuming from a particular topic how should I achieve it as the confluent parallel consumer does not have consumer.assign() method as of the KafkaConsumer<>.

[rkolesnev]

Hi @Jayesh-pict - i do not fully follow the question - consumers spread partitions across consumer instances in the same consumer group.
Assigning consumers to topics is done using subscribe call in Parallel Consumer.
I do not follow how consumer.assign() is relevant to stopping consuming from a particular topic - to stop consuming from topic - you would stop/close parallel consumer instance that is subscribed to that topic.

[jayeshpandav]

I am allocating memory to my jvm using -xms attribute how can I distribute this allocate memory to the internal concurrent threads of the parallel consumer? I have some minimum amount of memory requirement for each thread as I am using key based ordering and each key corresponds to unique sender_id. There is max_cap on the messsage size for every sender stored in a db and I want to allocate that minimum of that much amount of memory to that particular thread to ensure the successful message loading into ram. how can i achieve this?

[rkolesnev]

Hi @jayeshpandav - i would suggest that you ask a new question in the future.
I am a bit confused on what is the question or the issue you are trying to solve - do you get OutOfMemory errors? exceeding heap?
In Java applications memory is not allocated per Thread - it is allocated per application overall.
In Parallel Consumer - most of the memory use will come from buffering messages in the internal queue - when they are polled from Kafka and buffered for execution on User Function threads.
Size of the buffer is as number of messages buffered - calculated as: maxConcurrency * batchSize * loadingFactor - where maxConcurrency, batchSize, minLoadFactor and maxLoadFactor can be specified through ParallelConsumer options with defaults of (maxConcurrency=16, batchSize=1, minLoadFactor=2, maxLoadFactor=100). Loading factor will change from min towards max at steps as data is processed if it is processed faster than polled from Kafka to ensure that processing threads are not idle.
You can roughly calculate memory requirements based on your message size and those parameters allowing some extra for overhead / other parts of application.
You should be setting max heap size to allow for estimated memory use using -Xmx flag.