@Test
    public void testListConsumerGroupsWithTypesOlderBrokerVersion() throws Exception {
        ApiVersion listGroupV4 = new ApiVersion()
            .setApiKey(ApiKeys.LIST_GROUPS.id)
            .setMinVersion((short) 0)
            .setMaxVersion((short) 4);
        try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(mockCluster(1, 0))) {
            env.kafkaClient().setNodeApiVersions(NodeApiVersions.create(Collections.singletonList(listGroupV4)));

            env.kafkaClient().prepareResponse(prepareMetadataResponse(env.cluster(), Errors.NONE));

            // Check if we can list groups with older broker if we specify states and don't specify types.
            env.kafkaClient().prepareResponseFrom(
                expectListGroupsRequestWithFilters(singleton(ConsumerGroupState.STABLE.toString()), Collections.emptySet()),
                new ListGroupsResponse(new ListGroupsResponseData()
                    .setErrorCode(Errors.NONE.code())
                    .setGroups(Collections.singletonList(
                        new ListGroupsResponseData.ListedGroup()
                            .setGroupId("group-1")
                            .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
                            .setGroupState(ConsumerGroupState.STABLE.toString())))),
                env.cluster().nodeById(0));

            ListConsumerGroupsOptions options = new ListConsumerGroupsOptions().inStates(singleton(ConsumerGroupState.STABLE));
            ListConsumerGroupsResult result = env.adminClient().listConsumerGroups(options);

            Collection<ConsumerGroupListing> listing = result.all().get();
            assertEquals(1, listing.size());
            List<ConsumerGroupListing> expected = Collections.singletonList(
                new ConsumerGroupListing("group-1", false, Optional.of(ConsumerGroupState.STABLE))
            );
            assertEquals(expected, listing);

            // Check that we cannot set a type filter with an older broker.
            env.kafkaClient().prepareResponse(prepareMetadataResponse(env.cluster(), Errors.NONE));
            env.kafkaClient().prepareUnsupportedVersionResponse(request ->
                request instanceof ListGroupsRequest && !((ListGroupsRequest) request).data().typesFilter().isEmpty()
            );

            options = new ListConsumerGroupsOptions().withTypes(singleton(GroupType.CLASSIC));
            result = env.adminClient().listConsumerGroups(options);
            TestUtils.assertFutureThrows(result.all(), UnsupportedVersionException.class);
        }
    }
Test that we propagate exceptions encountered when fetching metadata.
clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java