Kafka Connect: Don't check that consumer group is stable for coordinator leader election

[fenil25]

This PR - changed the logic of how we close the coordinator in ICR mode.

Initially leaderPartition was only checked while assigning new partitions to the task during rebalancing.
While closing the partitions for a task, the coordinator was killed. This did not work during incremental co-operative rebalancing. This PR - #12372 fixed this issue.

However, we are being too conservative right now in checking whether there is leaderPartition. When, Kafka connect consumers are rebalancing, there can be a possibility that partition-0 is assigned to task A, and we start the coordinator. During a rebalance operation, partition 0 can move from task A to task B. When it calls close() on A, the group is rebalancing, and we never close the coordinator.
This removes the criteria that consumer needs to be stable while opening or closing partitions. Even during continuous rebalance, Kafka will always provide new partitions in opening and closing call. So, we don't have to check whether consumer group is stable while closing the coordinator.

Currently, without this, during rebalancing, there is a race condition and there is a possibility of having multiple coordinators around. This fixes the issue.

This PR #13756 addresses some of the scenarios where there can be multiple coordinators.
This PR handles the case while rebalancing.

[fenil25]

@kumarpritam863 @bryanck will appreciate a review! 🙇

[bryanck]

Maybe change the parameter name to onlyIfStable to be a little bit more clear? Alternatively the method could return an enum indicating the state.

[fenil25]

Good idea. Updated. ✅

[bryanck]

@kumarpritam863 Do you have a minute to take a look at this PR?

[fenil25]

I realized if we always close the coordinator when a partiion is revoked; during race conditions, we can also end up with no coordinators if we check for stability during open call for new partitions.
This ensures we never check for stability and just let Kafka Connect provide the list of partitions during open and close call while electing leader partition for coordinators.

@kumarpritam863 let me know your thoughts! Thank you. 🙇

[kumarpritam863]

@fenil25 Thank you so much for this thoughtful change — really appreciate your effort!
Just wanted to share that this was already discussed with @bryanck a while back, and we’d agreed that the check in CommitterImpl is indeed redundant. I was about to make this change but was a little busy and a bunch of resiliency tests were pending to cover some scenarios. The client can’t guarantee server-side invariants, so removing it makes perfect sense.
Also if you see the consumer coordinator class of kafka, it does check stability before proceeding with the open and close calls.
Your change aligns perfectly with that direction.
Just before going ahead with this, it would be nice if we can add some unit and integration tests on this one.

[bryanck]

I don't think these need to be member variables, I feel it would be better to initialize these in the test.

[fenil25]

I use these variables in both the tests so to avoid code repetition, I added them in Before call. I can add them individually in each test.

[fenil25]

@fenil25 Thank you so much for this thoughtful change — really appreciate your effort! Just wanted to share that this was already discussed with @bryanck a while back, and we’d agreed that the check in CommitterImpl is indeed redundant. I was about to make this change but was a little busy and a bunch of resiliency tests were pending to cover some scenarios. The client can’t guarantee server-side invariants, so removing it makes perfect sense. Also if you see the consumer coordinator class of kafka, it does check stability before proceeding with the open and close calls. Your change aligns perfectly with that direction. Just before going ahead with this, it would be nice if we can add some unit and integration tests on this one.

Perfect. Thanks a lot for the insights! Really appreciate your eyes. I added some tests 🙇‍♂️

[bryanck]

LGTM. Thanks @fenil25 for the contribution, and @kumarpritam863 for the review!