Fix: #1314 partition watcher doesn't reacts on partition number changing

[arxon31]

Suggestion to fix issue #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

[arxon31]

@erikdw @petedannemann
Sorry for pinging, but I noticed that you reviewed the last few PRs that were merged.
Could you please consider my proposal to improve the work of the consumer group?

[petedannemann]

Why is this being set to zero instead of the actual value?

[arxon31]

@petedannemann
This code can be removed. I forgot that I fill this map in the assignTopicPartitions function and only after that I run partitionWatcher

Should I remove this?

[petedannemann]

Yeah if it's not necessary let's remove this

[petedannemann]

Can you provide where you found this in the Kafka docs or code base? I guess this makes sense as the leader will trigger a consumer balance, which will let follows pick up partition changes

[arxon31]

@petedannemann
So it's my mistake, when I said that kafka says that this is correct. I misunderstood Jun Rao from this video. He did not say who exactly should run the partition watcher mechanism. But it still seems to me that partitionWatcher should only be launched by the leader. This way we create less load on the cluster and if leader disconnects broker also triggers rebalance for group and leader election

[petedannemann]

I am a bit worried about this causing some unintended regression. Could we break that change out into a separate PR and let us think it through for a bit?