KAFKA-8341. Retry Consumer group operation for NOT_COORDINATOR error

[soondenana]

An api call for consumer groups is made up of two calls:

1. Find the consumer group coordinator
2. Send the request to the node found in step 1

But the coordinator can get moved between step 1 and 2. In that case we
currently fail. This change fixes that by detecting this error and then
retrying.

Following APIs are impacted by this behavior:

1. listConsumerGroupOffsets
2. deleteConsumerGroups
3. describeConsumerGroups

Each of these call result in AdminClient making multiple calls to the backend.
As AdminClient code invokes each backend api in a separate event loop, the code
that detects the error (step 2) need to restart whole operation including
step 1. This needed a change to capture the "Call" object for step 1 in
step 2.

This change thus refactors the code to make it easy to perform a retry of
whole operation. It creates a Context object to capture the api arguments
that can then be referred by each "Call" objects. This is just for convenience
and makes method signature simpler as we only need to pass one object instead
of multiple api arguments.

The creation of each "Call" object is done in a new method, so we can
easily resubmit step 1 in step 2.

This change also modifies corresponding unit test to test this scenario.

More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.

Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[soondenana]

retest this please

[soondenana]

retest this please

[cmccabe]

Since we're only using a batch size of one, we should be able to just verify that we got a list of size 1 and it contained what we wanted, right? Also, we should also try to be robust against malformed responses that don't include the group we asked for (this will currently throw an NPE if we get one of those).

[soondenana]

Done. After adding the check one unit test failed, I have fixed it (need to delete some code) and added another unit test that incorporates the deleted code and tests it in correct way. Please take a look at that.

[cmccabe]

Thanks, @soondenana. I still wish for an implementation of KAFKA-6788, but this is a nice step in the right direction, and a much smaller change than that would be :) I left a comment.

[soondenana]

Messed up when adding a new commit to take care of Colin's comments. Reopening it.

[abbccdda]

Thanks for the PR. Just some minor comments

[abbccdda]

nit: also add java doc for type T, O here

[soondenana]

Done.

[abbccdda]

Do we have NPE risk here?

[soondenana]

findCoordinator request should take care of this and we should not have a null/invalid node.

[soondenana]

Thanks for the PR. Just some minor comments

Thanks Boyang for taking a look. Added the javadoc and updated the second commit where you should be able to see the change (I have left original commit as is).

[soondenana]

retest this please