KAFKA-17465: Make getMembersFromGroup be non-blocking

[TaiJuWu]

Jira: https://issues.apache.org/jira/browse/KAFKA-17465

Committer Checklist (excluded from commit message)

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

[frankvicky]

Hi @TaiJuWu
I have few comments, PTAL

[frankvicky]

@TaiJuWu
Please fix the PR title, correct JIRA number is 17465

[TaiJuWu]

@TaiJuWu Please fix the PR title, correct JIRA number is 17465

Thanks for your remainder, I will address your comments later.

[TaiJuWu]

Hi @TaiJuWu I have few comments, PTAL

Hi @frankvicky ,thanks for review, all comments are addressed.

[frankvicky]

LGTM

[chia7712]

@TaiJuWu thanks for your updates. a couple of comments are left.

[chia7712]

Could you please initialize the size by res.members()?

[chia7712]

If you prefer to use lambda function, please try to eliminate the local variable. for example:`

                    membersToRemove.add(member.groupInstanceId()
                            .map(id -> new MemberIdentity().setGroupInstanceId(id))
                            .orElseGet(() -> new MemberIdentity().setMemberId(member.consumerId()))
                            .setReason(reason));

[TaiJuWu]

Hi @chia7712 , thanks for review.
I rewrite this part, please take a look.

[frankvicky]

Thanks for update.
I have some comments.

[frankvicky]

nit: I would recommend not omitting the curly braces, or at the very least, it should break the line.

[frankvicky]

Please rename this variable.

[TaiJuWu]

All comments are addressed, thanks!