confluentinc / confluentinc/parallel-consumer
Apply backpressure per partition instead of the entire assignment
- Dominant language
- Java
- Stars
- 299
- Forks
- 172
- PR merge metrics
- No merged PRs in 30d
Description
Currently when the BrokerPollSystem considers there is [too much work pending](https://github.com/confluentinc/parallel-consumer/blob/754e29f0aece070ef86814b53e1613d888d45905/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/WorkManager.java#L215) already [it pauses](https://github.com/confluentinc/parallel-consumer/blob/754e29f0aece070ef86814b53e1613d888d45905/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/BrokerPollSystem.java#L273) all the partitions assigned.
I think we could improve parallelization by applying backpressure per partition instead, and would be specially helpful for the case described in the [KIP-41](https://cwiki.apache.org/confluence/display/KAFKA/KIP-41%3A+KafkaConsumer+Max+Records#KIP41:KafkaConsumerMaxRecords-EnsuringFairConsumption)
> Prefetching is skipped when there are enough records already available from any partition to satisfy the next call to poll(). When this number dips below max.poll.records, we fetch all partitions as in the current implementation. The only downside to this approach is that it could lead to some partitions going unconsumed for an extended amount of time **when there is a large imbalance between the partition's respective message rates**. For example, suppose that a consumer with max messages set to 1 fetches data from partitions A and B. If the returned fetch includes 1000 records from A and no records from B, the consumer will have to process all 1000 available records from A before fetching on partition B again.
I'm considering the following hypothetical scenario
- Topic with 2 partitions
- Imbalance in message rates: each poll will fetch 1000 records from p0 and 10 from p1.
- Average processing time: 500 millis
- max.poll.records = 10
- maxConcurrency = 2
- Partition ordering in parallel consumer
In this scenario we won't be achieving good parallelization since the partition ordering will force us to process all the records from partition 0 in serial order, which will lead us to have one thread idle most of the time until the last poll (corresponding to a single fetch of 1000 and 10 records from p0 and p1 respectively) gets records from both partitions.
Of course increasing `max.poll.records` will reduce the probability of this happening, for example if we were using `max.poll.records=501` the third poll will be already be able to parallelize work. And being the poll loop being detached from the processing we don't incur in the risk of being removed from the group by increasing the amount of work until the next poll. However this requires a better understanding of actual message rates and more fine tuning, and a good configuration of max.poll.records today could not work well in the future if the rates change.
I think applying backpressure per partition is a more robust way to handle this scenario, which may not be a common one but I don't see this affecting other scenarios, the only disadvantage being a more complex implementation since we need to track load per partition instead of global load as it is today.
I've focused mainly in the partition ordering for my analysis and I'm not sure if the proposal may have a negative impact on the other two ordering modes, my initial though is that it should be ok (the backpressure won't have anything to do with the ordering since it can only be applied per partition, but it will benefit mostly the partition ordering).
Contributor guide
No contributing guide indexed for this repository
Assessment
This issue has not been assessed yet.