Kafka consumer: proper parallelism for partition processing #814
Conversation
| partitioner := kgo.BasicConsistentPartitioner(func(topic string) func(*kgo.Record, int) int { | ||
| return func(r *kgo.Record, n int) int { | ||
| return int(r.Partition) | ||
| } | ||
| }) |
There was a problem hiding this comment.
I think this is just the kgo.ManualPartitioner()?
There was a problem hiding this comment.
Many thanks @twmb - fixing in #823
What do you think about the approach here BTW? We originally implemented consumer based on the example in franz-go repo, but without channel buffer for the partition consumer (i.e. recs channel). This quickly resulted into poll blocked when partition processing was long, and consumer excluded from group by Kafka due to poll timeout. franz-go uses buffered channel for a partition consumer, but it seems it has the same issue - just with less probability, as soon as channel buffer (5 in the example) is full - poll will block. I believe the approach here fully solves these concerns. WDYT?
There was a problem hiding this comment.
I think it works. It may cause more bandwidth to be used, but there isn't a way to really avoid that.
There may be data that is already buffered, or a fetch already in progress, for partitions you are pausing. The data is just dropped when being polled. Once you resume the fetch partition, the partition is re-fetched at the prior offset.
It may be worth having a buffer of 1 or 5 on the channel, to allow for the chance that processing finishes before you need to pause.
Another thing to do may be some select/default case before the existing select.
select {
<-done:
<-quit:
processor <- records:
default:
}
cl.Pause...
select {
<-done:
<-quit:
processor <- records:
}
The first select default would allow you to avoid pausing if things are keeping up, then you only pause if needed when things are backed up.
There was a problem hiding this comment.
Thanks a lot, yeah - looks like using a buffer makes sense, addressing in #829
Proposed changes
Pause processing partition to not block poll for a long time and not poll partition while it's being processed. This improves parallelism of partition processing and avoids theoretical stack of Kafka polling when parititon is processing message for a long time.