@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", "CyclomaticComplexity"})
    @Test
    public void testDescribeTopicsWithDescribeTopicPartitionsApiEdgeCase() throws ExecutionException, InterruptedException {
        try (AdminClientUnitTestEnv env = mockClientEnv()) {
            env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
            String topicName0 = "test-0";
            String topicName1 = "test-1";
            String topicName2 = "test-2";
            Map<String, Uuid> topics = new HashMap<>();
            topics.put(topicName0, Uuid.randomUuid());
            topics.put(topicName1, Uuid.randomUuid());
            topics.put(topicName2, Uuid.randomUuid());

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

            DescribeTopicPartitionsResponseData dataFirstPart = new DescribeTopicPartitionsResponseData();
            addPartitionToDescribeTopicPartitionsResponse(dataFirstPart, topicName0, topics.get(topicName0), singletonList(0));
            addPartitionToDescribeTopicPartitionsResponse(dataFirstPart, topicName1, topics.get(topicName1), singletonList(0));
            dataFirstPart.setNextCursor(new DescribeTopicPartitionsResponseData.Cursor()
                .setTopicName(topicName1)
                .setPartitionIndex(1));
            env.kafkaClient().prepareResponse(body -> {
                DescribeTopicPartitionsRequestData request = (DescribeTopicPartitionsRequestData) body.data();
                if (request.topics().size() != 3) return false;
                if (!request.topics().get(0).name().equals(topicName0)) return false;
                if (!request.topics().get(1).name().equals(topicName1)) return false;
                if (!request.topics().get(2).name().equals(topicName2)) return false;
                return request.cursor() == null;
            }, new DescribeTopicPartitionsResponse(dataFirstPart));

            DescribeTopicPartitionsResponseData dataSecondPart = new DescribeTopicPartitionsResponseData();
            addPartitionToDescribeTopicPartitionsResponse(dataSecondPart, topicName1, topics.get(topicName1), singletonList(1));
            addPartitionToDescribeTopicPartitionsResponse(dataSecondPart, topicName2, topics.get(topicName2), singletonList(0));
            dataSecondPart.setNextCursor(new DescribeTopicPartitionsResponseData.Cursor()
                .setTopicName(topicName2)
                .setPartitionIndex(1));
            env.kafkaClient().prepareResponse(body -> {
                DescribeTopicPartitionsRequestData request = (DescribeTopicPartitionsRequestData) body.data();
                if (request.topics().size() != 2) return false;
                if (!request.topics().get(0).name().equals(topicName1)) return false;
                if (!request.topics().get(1).name().equals(topicName2)) return false;
                DescribeTopicPartitionsRequestData.Cursor cursor = request.cursor();
                return cursor != null && cursor.topicName().equals(topicName1) && cursor.partitionIndex() == 1;
            }, new DescribeTopicPartitionsResponse(dataSecondPart));

            DescribeTopicPartitionsResponseData dataThirdPart = new DescribeTopicPartitionsResponseData();
            addPartitionToDescribeTopicPartitionsResponse(dataThirdPart, topicName2, topics.get(topicName2), singletonList(1));
            env.kafkaClient().prepareResponse(body -> {
                DescribeTopicPartitionsRequestData request = (DescribeTopicPartitionsRequestData) body.data();
                if (request.topics().size() != 1) return false;
                if (!request.topics().get(0).name().equals(topicName2)) return false;
                DescribeTopicPartitionsRequestData.Cursor cursor = request.cursor();
                return cursor != null && cursor.topicName().equals(topicName2) && cursor.partitionIndex() == 1;
            }, new DescribeTopicPartitionsResponse(dataThirdPart));

            DescribeTopicsResult result = env.adminClient().describeTopics(
                asList(topicName1, topicName0, topicName2), new DescribeTopicsOptions()
            );
            Map<String, TopicDescription> topicDescriptions = result.allTopicNames().get();
            assertEquals(3, topicDescriptions.size());
            TopicDescription topicDescription = topicDescriptions.get(topicName0);
            assertEquals(1, topicDescription.partitions().size());
            assertEquals(0, topicDescription.partitions().get(0).partition());
            topicDescription = topicDescriptions.get(topicName1);
            assertEquals(2, topicDescription.partitions().size());
            topicDescription = topicDescriptions.get(topicName2);
            assertEquals(2, topicDescription.partitions().size());
            assertNull(topicDescription.authorizedOperations());
        }
    }