consumer: consumer group cannot automatically fetch newly-added partitions

[napallday]

Versions

Please specify real version numbers or git SHAs, not just "Latest" since that changes fairly regularly.

Sarama	Kafka	Go
v1.40.1

Configuration

What configuration values are you using for Sarama and Kafka?

Logs

When filing an issue please provide logs from Sarama and Kafka if at all
possible. You can set sarama.Logger to a log.Logger to capture Sarama debug
output.

logs: CLICK ME

Problem Description

The fix #1525 hasn't completely fixed the problem.

And the issue can be reproduced easily(A functional test is provided in the fix #2563):

1. Set config.Metadata.RefreshFrequency = 10 * time.Second(to trigger the checking logic in loopCheckPartitionNumbers more frequently)
2. Two consumer c1 and c2 subscribe to t1p0 and t2p0 respectively in the same consumer group.
3. Let's say if c1 is the leader, after adding a new partition in t2p1, the new partition cannot be fetched until restart.

The root cause is that:

1. Although loopCheckPartitionNumbers in c2 will detect the new partition t2p1 and try to send JoinGroupRequest to the server, the server won't trigger a rebalance because c2 is not the leader, and will just return the previous assignment to c2.
2. Although c1 is the leader, goroutine loopCheckPartitionNumbers still won't exit to trigger a rebalance, because it will only check the topics it subscribes to instead of all the subscribed topics in the consumer group.