public CompletableFuture<TopicExistsInfo> checkTopicExists(TopicName topic) {
        return pulsar.getBrokerService()
            .fetchPartitionedTopicMetadataAsync(TopicName.get(topic.toString()))
            .thenCompose(metadata -> {
                if (metadata.partitions > 0) {
                    return CompletableFuture.completedFuture(
                            TopicExistsInfo.newPartitionedTopicExists(metadata.partitions));
                }
                return checkNonPartitionedTopicExists(topic)
                    .thenApply(b -> b ? TopicExistsInfo.newNonPartitionedTopicExists()
                            : TopicExistsInfo.newTopicNotExists());
            });
    }
@Test(dataProvider = "topicDomain")
    public void testCheckTopicExists(String topicDomain) throws Exception {
        String topic = topicDomain + "://prop/ns-abc/" + UUID.randomUUID();
        admin.topics().createNonPartitionedTopic(topic);
        Awaitility.await().untilAsserted(() -> {
            assertTrue(pulsar.getNamespaceService().checkTopicExists(TopicName.get(topic)).get().isExists());
        });

        String partitionedTopic = topicDomain + "://prop/ns-abc/" + UUID.randomUUID();
        admin.topics().createPartitionedTopic(partitionedTopic, 5);
        Awaitility.await().untilAsserted(() -> {
            assertTrue(pulsar.getNamespaceService().checkTopicExists(TopicName.get(partitionedTopic)).get().isExists());
            assertTrue(pulsar.getNamespaceService()
                    .checkTopicExists(TopicName.get(partitionedTopic + "-partition-2")).get().isExists());
        });
    }