@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 testDescribeMultipleShareGroups() {
        try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(mockCluster(1, 0))) {
            env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());

            env.kafkaClient().prepareResponse(prepareFindCoordinatorResponse(Errors.NONE, env.cluster().controller()));

            ShareGroupDescribeResponseData.TopicPartitions topicPartitions = new ShareGroupDescribeResponseData.TopicPartitions()
                .setTopicName("my_topic")
                .setPartitions(asList(0, 1, 2));
            final ShareGroupDescribeResponseData.Assignment memberAssignment = new ShareGroupDescribeResponseData.Assignment()
                .setTopicPartitions(asList(topicPartitions));
            ShareGroupDescribeResponseData group0Data = new ShareGroupDescribeResponseData();
            group0Data.groups().add(new ShareGroupDescribeResponseData.DescribedGroup()
                .setGroupId(GROUP_ID)
                .setGroupState(ShareGroupState.STABLE.toString())
                .setMembers(asList(
                    new ShareGroupDescribeResponseData.Member()
                        .setMemberId("0")
                        .setClientId("clientId0")
                        .setClientHost("clientHost")
                        .setAssignment(memberAssignment),
                    new ShareGroupDescribeResponseData.Member()
                        .setMemberId("1")
                        .setClientId("clientId1")
                        .setClientHost("clientHost")
                        .setAssignment(memberAssignment))));

            ShareGroupDescribeResponseData group1Data = new ShareGroupDescribeResponseData();
            group1Data.groups().add(new ShareGroupDescribeResponseData.DescribedGroup()
                .setGroupId("group-1")
                .setGroupState(ShareGroupState.STABLE.toString())
                .setMembers(asList(
                    new ShareGroupDescribeResponseData.Member()
                        .setMemberId("0")
                        .setClientId("clientId0")
                        .setClientHost("clientHost")
                        .setAssignment(memberAssignment),
                    new ShareGroupDescribeResponseData.Member()
                        .setMemberId("1")
                        .setClientId("clientId1")
                        .setClientHost("clientHost")
                        .setAssignment(memberAssignment))));

            env.kafkaClient().prepareResponse(new ShareGroupDescribeResponse(group0Data));
            env.kafkaClient().prepareResponse(new ShareGroupDescribeResponse(group1Data));

            Collection<String> groups = new HashSet<>();
            groups.add(GROUP_ID);
            groups.add("group-1");
            final DescribeShareGroupsResult result = env.adminClient().describeShareGroups(groups);
            assertEquals(2, result.describedGroups().size());
            assertEquals(groups, result.describedGroups().keySet());
        }
    }