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

            env.kafkaClient().prepareResponse(new FindCoordinatorResponse(
                new FindCoordinatorResponseData()
                    .setCoordinators(asList(
                        FindCoordinatorResponse.prepareCoordinatorResponse(Errors.NONE, "grp1", env.cluster().controller()),
                        FindCoordinatorResponse.prepareCoordinatorResponse(Errors.NONE, "grp2", env.cluster().controller())
                    ))
            ));

            env.kafkaClient().prepareResponse(new ConsumerGroupDescribeResponse(
                new ConsumerGroupDescribeResponseData()
                    .setGroups(asList(
                        new ConsumerGroupDescribeResponseData.DescribedGroup()
                            .setGroupId("grp1")
                            .setGroupState("Stable")
                            .setGroupEpoch(10)
                            .setAssignmentEpoch(10)
                            .setAssignorName("range")
                            .setAuthorizedOperations(Utils.to32BitField(emptySet()))
                            .setMembers(singletonList(
                                new ConsumerGroupDescribeResponseData.Member()
                                    .setMemberId("memberId")
                                    .setInstanceId("instanceId")
                                    .setClientHost("host")
                                    .setClientId("clientId")
                                    .setMemberEpoch(10)
                                    .setRackId("rackid")
                                    .setSubscribedTopicNames(singletonList("foo"))
                                    .setSubscribedTopicRegex("regex")
                                    .setAssignment(new ConsumerGroupDescribeResponseData.Assignment()
                                        .setTopicPartitions(singletonList(
                                            new ConsumerGroupDescribeResponseData.TopicPartitions()
                                                .setTopicId(Uuid.randomUuid())
                                                .setTopicName("foo")
                                                .setPartitions(singletonList(0))
                                        )))
                                    .setTargetAssignment(new ConsumerGroupDescribeResponseData.Assignment()
                                        .setTopicPartitions(singletonList(
                                            new ConsumerGroupDescribeResponseData.TopicPartitions()
                                                .setTopicId(Uuid.randomUuid())
                                                .setTopicName("foo")
                                                .setPartitions(singletonList(1))
                                        )))
                            )),
                        new ConsumerGroupDescribeResponseData.DescribedGroup()
                            .setGroupId("grp2")
                            .setErrorCode(Errors.GROUP_ID_NOT_FOUND.code())
                    ))
            ));

            env.kafkaClient().prepareResponse(new DescribeGroupsResponse(
                new DescribeGroupsResponseData()
                    .setGroups(Collections.singletonList(
                        DescribeGroupsResponse.groupMetadata(
                            "grp2",
                            Errors.NONE,
                            "Stable",
                            ConsumerProtocol.PROTOCOL_TYPE,
                            "range",
                            singletonList(
                                DescribeGroupsResponse.groupMember(
                                    "0",
                                    null,
                                    "clientId0",
                                    "clientHost",
                                    ConsumerProtocol.serializeAssignment(
                                        new ConsumerPartitionAssignor.Assignment(
                                            Collections.singletonList(new TopicPartition("bar", 0))
                                        )
                                    ).array(),
                                    null
                                )
                            ),
                            Collections.emptySet()
                        )
                    ))
            ));

            DescribeConsumerGroupsResult result = env.adminClient()
                .describeConsumerGroups(asList("grp1", "grp2"));

            Map<String, ConsumerGroupDescription> expectedResult = new HashMap<>();
            expectedResult.put("grp1", new ConsumerGroupDescription(
                "grp1",
                false,
                Collections.singletonList(
                    new MemberDescription(
                        "memberId",
                        Optional.of("instanceId"),
                        "clientId",
                        "host",
                        new MemberAssignment(
                            Collections.singleton(new TopicPartition("foo", 0))
                        ),
                        Optional.of(new MemberAssignment(
                            Collections.singleton(new TopicPartition("foo", 1))
                        ))
                    )
                ),
                "range",
                GroupType.CONSUMER,
                ConsumerGroupState.STABLE,
                env.cluster().controller(),
                Collections.emptySet()
            ));
            expectedResult.put("grp2", new ConsumerGroupDescription(
                "grp2",
                false,
                Collections.singletonList(
                    new MemberDescription(
                        "0",
                        Optional.empty(),
                        "clientId0",
                        "clientHost",
                        new MemberAssignment(
                            Collections.singleton(new TopicPartition("bar", 0))
                        )
                    )
                ),
                "range",
                GroupType.CLASSIC,
                ConsumerGroupState.STABLE,
                env.cluster().controller(),
                Collections.emptySet()
            ));

            assertEquals(expectedResult, result.all().get());
        }
    }