@Test
    public void testListConsumerGroupsWithTypes() throws Exception {
        try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(mockCluster(1, 0))) {
            env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());

            // Test with a specific state filter but no type filter in list consumer group options.
            env.kafkaClient().prepareResponse(prepareMetadataResponse(env.cluster(), Errors.NONE));

            env.kafkaClient().prepareResponseFrom(
                expectListGroupsRequestWithFilters(singleton(ConsumerGroupState.STABLE.toString()), Collections.emptySet()),
                new ListGroupsResponse(new ListGroupsResponseData()
                    .setErrorCode(Errors.NONE.code())
                    .setGroups(singletonList(
                        new ListGroupsResponseData.ListedGroup()
                            .setGroupId("group-1")
                            .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
                            .setGroupState("Stable")
                            .setGroupType(GroupType.CLASSIC.toString())))),
                env.cluster().nodeById(0));

            final ListConsumerGroupsOptions options = new ListConsumerGroupsOptions().inStates(singleton(ConsumerGroupState.STABLE));
            final ListConsumerGroupsResult result = env.adminClient().listConsumerGroups(options);
            Collection<ConsumerGroupListing> listings = result.valid().get();

            assertEquals(1, listings.size());
            List<ConsumerGroupListing> expected = new ArrayList<>();
            expected.add(new ConsumerGroupListing("group-1", false, Optional.of(ConsumerGroupState.STABLE), Optional.of(GroupType.CLASSIC)));
            assertEquals(expected, listings);
            assertEquals(0, result.errors().get().size());

            // Test with list consumer group options.
            env.kafkaClient().prepareResponse(prepareMetadataResponse(env.cluster(), Errors.NONE));

            env.kafkaClient().prepareResponseFrom(
                expectListGroupsRequestWithFilters(Collections.emptySet(), singleton(GroupType.CONSUMER.toString())),
                new ListGroupsResponse(new ListGroupsResponseData()
                    .setErrorCode(Errors.NONE.code())
                    .setGroups(asList(
                        new ListGroupsResponseData.ListedGroup()
                            .setGroupId("group-1")
                            .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
                            .setGroupState("Stable")
                            .setGroupType(GroupType.CONSUMER.toString()),
                        new ListGroupsResponseData.ListedGroup()
                            .setGroupId("group-2")
                            .setGroupState("Empty")
                            .setGroupType(GroupType.CONSUMER.toString())))),
                env.cluster().nodeById(0));

            final ListConsumerGroupsOptions options2 = new ListConsumerGroupsOptions().withTypes(singleton(GroupType.CONSUMER));
            final ListConsumerGroupsResult result2 = env.adminClient().listConsumerGroups(options2);
            Collection<ConsumerGroupListing> listings2 = result2.valid().get();

            assertEquals(2, listings2.size());
            List<ConsumerGroupListing> expected2 = new ArrayList<>();
            expected2.add(new ConsumerGroupListing("group-2", true, Optional.of(ConsumerGroupState.EMPTY), Optional.of(GroupType.CONSUMER)));
            expected2.add(new ConsumerGroupListing("group-1", false, Optional.of(ConsumerGroupState.STABLE), Optional.of(GroupType.CONSUMER)));
            assertEquals(expected2, listings2);
            assertEquals(0, result.errors().get().size());
        }
    }
Test that we propagate exceptions encountered when fetching metadata.
clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java