@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.");
    }
@Test
    public void testDescribeTopicsByIds() throws ExecutionException, InterruptedException {
        try (AdminClientUnitTestEnv env = mockClientEnv()) {
            env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());

            // Valid ID
            Uuid topicId = Uuid.randomUuid();
            String topicName = "test-topic";
            Node leader = env.cluster().nodes().get(0);
            MetadataResponse.PartitionMetadata partitionMetadata = new MetadataResponse.PartitionMetadata(
                    Errors.NONE,
                    new TopicPartition(topicName, 0),
                    Optional.of(leader.id()),
                    Optional.of(10),
                    singletonList(leader.id()),
                    singletonList(leader.id()),
                    singletonList(leader.id()));
            env.kafkaClient().prepareResponse(RequestTestUtils
                    .metadataResponse(
                            env.cluster().nodes(),
                            env.cluster().clusterResource().clusterId(),
                            env.cluster().controller().id(),
                            singletonList(new MetadataResponse.TopicMetadata(Errors.NONE, topicName, topicId, false,
                                    singletonList(partitionMetadata), MetadataResponse.AUTHORIZED_OPERATIONS_OMITTED))));
            TopicCollection.TopicIdCollection topicIds = TopicCollection.ofTopicIds(
                    singletonList(topicId));

            DescribeTopicsResult describeTopicsresult = env.adminClient().describeTopics(topicIds);
            Map<Uuid, TopicDescription> allTopicIds = describeTopicsresult.allTopicIds().get();
            assertEquals(topicName, allTopicIds.get(topicId).name());

            // ID not exist in brokers
            Uuid nonExistID = Uuid.randomUuid();
            env.kafkaClient().prepareResponse(RequestTestUtils
                    .metadataResponse(
                            env.cluster().nodes(),
                            env.cluster().clusterResource().clusterId(),
                            env.cluster().controller().id(),
                            emptyList()));

            DescribeTopicsResult result1 = env.adminClient().describeTopics(
                    TopicCollection.ofTopicIds(singletonList(nonExistID)));
            TestUtils.assertFutureError(result1.allTopicIds(), UnknownTopicIdException.class);
            Exception e = assertThrows(Exception.class, () -> result1.allTopicIds().get(), "describe with non-exist topic ID should throw exception");
            assertEquals(String.format("org.apache.kafka.common.errors.UnknownTopicIdException: TopicId %s not found.", nonExistID), e.getMessage());

            DescribeTopicsResult result2 = env.adminClient().describeTopics(
                    TopicCollection.ofTopicIds(singletonList(Uuid.ZERO_UUID)));
            TestUtils.assertFutureError(result2.allTopicIds(), InvalidTopicException.class);
            e = assertThrows(Exception.class, () -> result2.allTopicIds().get(), "describe with non-exist topic ID should throw exception");
            assertEquals("The given topic id 'AAAAAAAAAAAAAAAAAAAAAA' cannot be represented in a request.", e.getCause().getMessage());

        }
    }