kafka-go

kafka-go copied to clipboard

Suggestion to fix issue https://github.com/segmentio/kafka-go/issues/1314

Steps to reproduce the issue:

1. Create topic
2. Simultaneously connect to this topic with consumer group
3. At some time, if Kafka didn't assigned partitions to topic (so topic is exists, but has zero partitions), you won't receive topic to consumer group because of empty assignment. So Kafka after that assigning partitions to topic
4. At the start of partitionWatcher it scrapes assigned number of partitions and starts loop where monitors number of partitions assigned to topic

After all we have empty consumer group, and partitionWatcher monitoring for changing partition number with non-zero start number of partitions. So we have to increase partitions to receive topic into consumer group.

Solution: We need to assign partitions number to topic in assignTopicPartitions function. So we run partitionWatcher with assigned in assignTopicPartition function number of partitions and don't scrape it inside partitionWatcher

P.S. In this PR I also changed partitionWatcher. Now it runs only if consumer is leader of consumer group. Kafka says this is correct