@Test
    public void testDescribeClusterHandleError() {
        try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(mockCluster(4, 0),
            AdminClientConfig.RETRIES_CONFIG, "2")) {
            env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());

            // Prepare the describe cluster response used for the first describe cluster
            String errorMessage = "my error";
            env.kafkaClient().prepareResponse(
                new DescribeClusterResponse(new DescribeClusterResponseData()
                    .setErrorCode(Errors.INVALID_REQUEST.code())
                    .setErrorMessage(errorMessage)));

            final DescribeClusterResult result = env.adminClient().describeCluster();
            TestUtils.assertFutureThrows(result.clusterId(),
                InvalidRequestException.class, errorMessage);
            TestUtils.assertFutureThrows(result.controller(),
                InvalidRequestException.class, errorMessage);
            TestUtils.assertFutureThrows(result.nodes(),
                InvalidRequestException.class, errorMessage);
            TestUtils.assertFutureThrows(result.authorizedOperations(),
                InvalidRequestException.class, errorMessage);
        }
    }
Test that we propagate exceptions encountered when fetching metadata.
clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java