@Test
    public void testDescribeMetadataQuorumFailure() {
        try (final AdminClientUnitTestEnv env = mockClientEnv()) {
            env.kafkaClient().setNodeApiVersions(NodeApiVersions.create(ApiKeys.DESCRIBE_QUORUM.id,
                        ApiKeys.DESCRIBE_QUORUM.oldestVersion(),
                        ApiKeys.DESCRIBE_QUORUM.latestVersion()));

            // Test top level error
            env.kafkaClient().prepareResponse(
                    body -> body instanceof DescribeQuorumRequest,
                    prepareDescribeQuorumResponse(Errors.INVALID_REQUEST, Errors.NONE, false, false, false, false, false));
            KafkaFuture<QuorumInfo> future = env.adminClient().describeMetadataQuorum().quorumInfo();
            TestUtils.assertFutureThrows(future, InvalidRequestException.class);

            // Test incorrect topic count
            env.kafkaClient().prepareResponse(
                    body -> body instanceof DescribeQuorumRequest,
                    prepareDescribeQuorumResponse(Errors.NONE, Errors.NONE, true, false, false, false, false));
            future = env.adminClient().describeMetadataQuorum().quorumInfo();
            TestUtils.assertFutureThrows(future, UnknownServerException.class);

            // Test incorrect topic name
            env.kafkaClient().prepareResponse(
                    body -> body instanceof DescribeQuorumRequest,
                    prepareDescribeQuorumResponse(Errors.NONE, Errors.NONE, false, true, false, false, false));
            future = env.adminClient().describeMetadataQuorum().quorumInfo();
            TestUtils.assertFutureThrows(future, UnknownServerException.class);

            // Test incorrect partition count
            env.kafkaClient().prepareResponse(
                    body -> body instanceof DescribeQuorumRequest,
                    prepareDescribeQuorumResponse(Errors.NONE, Errors.NONE, false, false, true, false, false));
            future = env.adminClient().describeMetadataQuorum().quorumInfo();
            TestUtils.assertFutureThrows(future, UnknownServerException.class);

            // Test incorrect partition index
            env.kafkaClient().prepareResponse(
                    body -> body instanceof DescribeQuorumRequest,
                    prepareDescribeQuorumResponse(Errors.NONE, Errors.NONE, false, false, false, true, false));
            future = env.adminClient().describeMetadataQuorum().quorumInfo();
            TestUtils.assertFutureThrows(future, UnknownServerException.class);

            // Test partition level error
            env.kafkaClient().prepareResponse(
                    body -> body instanceof DescribeQuorumRequest,
                    prepareDescribeQuorumResponse(Errors.NONE, Errors.INVALID_REQUEST, false, false, false, false, false));
            future = env.adminClient().describeMetadataQuorum().quorumInfo();
            TestUtils.assertFutureThrows(future, InvalidRequestException.class);

            // Test all incorrect and no errors
            env.kafkaClient().prepareResponse(
                    body -> body instanceof DescribeQuorumRequest,
                    prepareDescribeQuorumResponse(Errors.NONE, Errors.NONE, true, true, true, true, false));
            future = env.adminClient().describeMetadataQuorum().quorumInfo();
            TestUtils.assertFutureThrows(future, UnknownServerException.class);

            // Test all incorrect and both errors
            env.kafkaClient().prepareResponse(
                    body -> body instanceof DescribeQuorumRequest,
                    prepareDescribeQuorumResponse(Errors.INVALID_REQUEST, Errors.INVALID_REQUEST, true, true, true, true, false));
            future = env.adminClient().describeMetadataQuorum().quorumInfo();
            TestUtils.assertFutureThrows(future, Errors.INVALID_REQUEST.exception().getClass());
        }
    }
Test that we propagate exceptions encountered when fetching metadata.
clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java