@Override
    public DescribeShareGroupsResult describeShareGroups(final Collection<String> groupIds,
                                                         final DescribeShareGroupsOptions options) {
        SimpleAdminApiFuture<CoordinatorKey, ShareGroupDescription> future =
                DescribeShareGroupsHandler.newFuture(groupIds);
        DescribeShareGroupsHandler handler = new DescribeShareGroupsHandler(options.includeAuthorizedOperations(), logContext);
        invokeDriver(handler, future, options.timeoutMs);
        return new DescribeShareGroupsResult(future.all().entrySet().stream()
                .collect(Collectors.toMap(entry -> entry.getKey().idValue, Map.Entry::getValue)));
    }
@Test
    public void testDescribeShareGroups() throws Exception {
        try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(mockCluster(1, 0))) {
            env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());

            // Retriable FindCoordinatorResponse errors should be retried
            env.kafkaClient().prepareResponse(prepareFindCoordinatorResponse(Errors.COORDINATOR_NOT_AVAILABLE,  Node.noNode()));
            env.kafkaClient().prepareResponse(prepareFindCoordinatorResponse(Errors.COORDINATOR_LOAD_IN_PROGRESS,  Node.noNode()));
            env.kafkaClient().prepareResponse(prepareFindCoordinatorResponse(Errors.NONE, env.cluster().controller()));

            ShareGroupDescribeResponseData data = new ShareGroupDescribeResponseData();

            // Retriable errors should be retried
            data.groups().add(new ShareGroupDescribeResponseData.DescribedGroup()
                .setGroupId(GROUP_ID)
                .setErrorCode(Errors.COORDINATOR_LOAD_IN_PROGRESS.code()));
            env.kafkaClient().prepareResponse(new ShareGroupDescribeResponse(data));

            /*
             * We need to return two responses here, one with NOT_COORDINATOR error when calling describe share group
             * api using coordinator that has moved. This will retry whole operation. So we need to again respond with a
             * FindCoordinatorResponse.
             *
             * And the same reason for COORDINATOR_NOT_AVAILABLE error response
             */
            data = new ShareGroupDescribeResponseData();
            data.groups().add(new ShareGroupDescribeResponseData.DescribedGroup()
                .setGroupId(GROUP_ID)
                .setErrorCode(Errors.NOT_COORDINATOR.code()));
            env.kafkaClient().prepareResponse(new ShareGroupDescribeResponse(data));
            env.kafkaClient().prepareResponse(prepareFindCoordinatorResponse(Errors.NONE, env.cluster().controller()));

            data = new ShareGroupDescribeResponseData();
            data.groups().add(new ShareGroupDescribeResponseData.DescribedGroup()
                .setGroupId(GROUP_ID)
                .setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code()));
            env.kafkaClient().prepareResponse(new ShareGroupDescribeResponse(data));
            env.kafkaClient().prepareResponse(prepareFindCoordinatorResponse(Errors.NONE, env.cluster().controller()));

            data = new ShareGroupDescribeResponseData();
            ShareGroupDescribeResponseData.TopicPartitions topicPartitions = new ShareGroupDescribeResponseData.TopicPartitions()
                .setTopicName("my_topic")
                .setPartitions(asList(0, 1, 2));
            ShareGroupDescribeResponseData.Assignment memberAssignment = new ShareGroupDescribeResponseData.Assignment()
                .setTopicPartitions(asList(topicPartitions));
            ShareGroupDescribeResponseData.Member memberOne = new ShareGroupDescribeResponseData.Member()
                .setMemberId("0")
                .setClientId("clientId0")
                .setClientHost("clientHost")
                .setAssignment(memberAssignment);
            ShareGroupDescribeResponseData.Member memberTwo = new ShareGroupDescribeResponseData.Member()
                .setMemberId("1")
                .setClientId("clientId1")
                .setClientHost("clientHost")
                .setAssignment(memberAssignment);

            ShareGroupDescribeResponseData group0Data = new ShareGroupDescribeResponseData();
            group0Data.groups().add(new ShareGroupDescribeResponseData.DescribedGroup()
                .setGroupId(GROUP_ID)
                .setGroupState(ShareGroupState.STABLE.toString())
                .setMembers(asList(memberOne, memberTwo)));

            final List<TopicPartition> expectedTopicPartitions = new ArrayList<>();
            expectedTopicPartitions.add(0, new TopicPartition("my_topic", 0));
            expectedTopicPartitions.add(1, new TopicPartition("my_topic", 1));
            expectedTopicPartitions.add(2, new TopicPartition("my_topic", 2));

            List<MemberDescription> expectedMemberDescriptions = new ArrayList<>();
            expectedMemberDescriptions.add(convertToMemberDescriptions(memberOne,
                new MemberAssignment(new HashSet<>(expectedTopicPartitions))));
            expectedMemberDescriptions.add(convertToMemberDescriptions(memberTwo,
                new MemberAssignment(new HashSet<>(expectedTopicPartitions))));
            data.groups().add(new ShareGroupDescribeResponseData.DescribedGroup()
                .setGroupId(GROUP_ID)
                .setGroupState(ShareGroupState.STABLE.toString())
                .setMembers(asList(memberOne, memberTwo)));

            env.kafkaClient().prepareResponse(new ShareGroupDescribeResponse(data));

            final DescribeShareGroupsResult result = env.adminClient().describeShareGroups(singletonList(GROUP_ID));
            final ShareGroupDescription groupDescription = result.describedGroups().get(GROUP_ID).get();

            assertEquals(1, result.describedGroups().size());
            assertEquals(GROUP_ID, groupDescription.groupId());
            assertEquals(2, groupDescription.members().size());
            assertEquals(expectedMemberDescriptions, groupDescription.members());
        }
    }