public CompletableFuture<Void> close() {
        return close(true, false);
    }
@Test
    public void testCompatibilityWithPartitionKeyword() throws PulsarAdminException, PulsarClientException {
        final String topicName = "persistent://prop/ns-abc/testCompatibilityWithPartitionKeyword";
        TopicName topicNameEntity = TopicName.get(topicName);
        String partition2 = topicNameEntity.getPartition(2).toString();
        // Create a non-partitioned topic with -partition- keyword
        Producer<byte[]> producer = pulsarClient.newProducer()
                .topic(partition2)
                .create();
        List<String> topics = admin.topics().getList("prop/ns-abc");
        // Close previous producer to simulate reconnect
        producer.close();
        // Disable auto topic creation
        conf.setAllowAutoTopicCreation(false);
        // Check the topic exist in the list.
        Assert.assertTrue(topics.contains(partition2));
        // Check this topic has no partition metadata.
        Assert.assertThrows(PulsarAdminException.NotFoundException.class,
                () -> admin.topics().getPartitionedTopicMetadata(topicName));
        // Reconnect to the broker and expect successful because the topic has existed in the broker.
        producer = pulsarClient.newProducer()
                .topic(partition2)
                .create();
        producer.close();
        // Check the topic exist in the list again.
        Assert.assertTrue(topics.contains(partition2));
        // Check this topic has no partition metadata again.
        Assert.assertThrows(PulsarAdminException.NotFoundException.class,
                () -> admin.topics().getPartitionedTopicMetadata(topicName));
    }