Question: Parallel polling and multiple consumers with confluent-kafka-go v2

[nurlybekovnt]

Hi there, @emasab @rayokota! 👋

In the official examples for confluent-kafka-go v2 a single Consumer is created with kafka.NewConsumer, and consumer.Poll() is called in the main goroutine. My kafka topic has four partitions, so I'm trying to understand the recommended approach for higher throughput and less RAM concumption:

1. Parallel polling:
   • Is it OK (or even expected) to call consumer.Poll() from four separate goroutines, one per partition?
2. Multiple consumers:
   • Alternatively, should I create four separate Consumer instances (in the same consumer group)?
   • If I do this, will each instance be assigned a dedicated partition, or will messages still be interleaved across them?
3. Best practice:
   • What is the canonical way to scale reads when a single consumer can't keep up and Kafka lag starts to grow?

Any guidance or links to relevant docs or examples would be much appreciated. Thanks in advance!

[nurlybekovnt]

Hi @emasab @rayokota,
Can I call consumer.Poll() and consumer.CommitOffsets() from different goroutines safely? Is CommitOffsets() thread-safe, and can it block or interfere with Poll()?

Thanks!

[milindl]

I think from the point of view of thread-safety, it would be okay to call Poll() from different goroutines (in terms of there being no race conditions or inconsistent internal states).

However, I would not recommend it. The first reason is that we don't test it for the go client specifically for this. The second being that, in case of crashes of your application, this has the possibility to lead to dropped messages.

For example, if two different goroutines call Poll(), and obtain messages at offset 2 and 3 respectively. Message at offset 3 is processed successfully, and committed. However, message at offset 2 is not processed yet when the application crashes for some reason. Since the commit was made with offset 3, the next message you fetch when you restart the application will be 4. The message at 2 will be skipped.

In general, if resources permit, you should increase the number of consumers like (2). Partitions are owned wholly by consumers, ie messages from a single partition will not be sent to multiple consumers (as long as all consumers are connected).
If not, I would say to have a single goroutine for polling, and push messages from every partition into a different channel, where they can be processed by concurrently running goroutines. This will allow you to commit safely as the order of messages on a partition level is maintained.