@Override
    public DescribeTopicsResult describeTopics(final TopicCollection topics, DescribeTopicsOptions options) {
        if (topics instanceof TopicIdCollection)
            return DescribeTopicsResult.ofTopicIds(handleDescribeTopicsByIds(((TopicIdCollection) topics).topicIds(), options));
        else if (topics instanceof TopicNameCollection)
            return DescribeTopicsResult.ofTopicNames(handleDescribeTopicsByNamesWithDescribeTopicPartitionsApi(((TopicNameCollection) topics).topicNames(), options));
        else
            throw new IllegalArgumentException("The TopicCollection: " + topics + " provided did not match any supported classes for describeTopics.");
    }
@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);

        }
    }