@SuppressWarnings("NPathComplexity")
    @Test
    public void testDescribeTopicsWithDescribeTopicPartitionsApiErrorHandling() throws InterruptedException {
        try (AdminClientUnitTestEnv env = mockClientEnv()) {
            env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
            String topicName0 = "test-0";
            String topicName1 = "test-1";
            Map<String, Uuid> topics = new HashMap<>();
            topics.put(topicName0, Uuid.randomUuid());
            topics.put(topicName1, Uuid.randomUuid());

            env.kafkaClient().prepareResponse(
                prepareDescribeClusterResponse(0,
                    env.cluster().nodes(),
                    env.cluster().clusterResource().clusterId(),
                    2,
                    MetadataResponse.AUTHORIZED_OPERATIONS_OMITTED)
            );

            DescribeTopicPartitionsResponseData dataFirstPart = new DescribeTopicPartitionsResponseData();
            dataFirstPart.topics().add(new DescribeTopicPartitionsResponseTopic()
                .setErrorCode((short) 0)
                .setTopicId(topics.get(topicName0))
                .setName(topicName0)
                .setIsInternal(false)
                .setPartitions(singletonList(new DescribeTopicPartitionsResponsePartition()
                    .setIsrNodes(singletonList(0))
                    .setErrorCode((short) 0)
                    .setLeaderEpoch(0)
                    .setLeaderId(0)
                    .setEligibleLeaderReplicas(singletonList(1))
                    .setLastKnownElr(singletonList(2))
                    .setPartitionIndex(0)
                    .setReplicaNodes(asList(0, 1, 2))))
            );
            dataFirstPart.topics().add(new DescribeTopicPartitionsResponseTopic()
                .setErrorCode((short) 29)
                .setTopicId(Uuid.ZERO_UUID)
                .setName(topicName1)
                .setIsInternal(false)
            );
            env.kafkaClient().prepareResponse(body -> {
                DescribeTopicPartitionsRequestData request = (DescribeTopicPartitionsRequestData) body.data();
                if (request.topics().size() != 2) return false;
                if (!request.topics().get(0).name().equals(topicName0)) return false;
                if (!request.topics().get(1).name().equals(topicName1)) return false;
                return request.cursor() == null;
            }, new DescribeTopicPartitionsResponse(dataFirstPart));
            DescribeTopicsResult result = env.adminClient().describeTopics(
                asList(topicName1, topicName0), new DescribeTopicsOptions()
            );

            TestUtils.assertFutureError(result.allTopicNames(), TopicAuthorizationException.class);

        }
    }
Test that we propagate exceptions encountered when fetching metadata.
clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java