public CompletableFuture<Optional<Topic>> getTopic(final String topic, boolean createIfMissing) {
        return getTopic(topic, createIfMissing, null);
    }
@Test
    public void testGetTopic() throws Exception {
        final String ns = "prop/ns-test";
        admin.namespaces().createNamespace(ns, 2);
        final String topicName = ns + "/topic-1";
        admin.topics().createNonPartitionedTopic(String.format("persistent://%s", topicName));
        Producer<String> producer1 = pulsarClient.newProducer(Schema.STRING).topic(topicName).create();
        producer1.close();
        PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService().getTopic(topicName.toString(), false).get().get();
        persistentTopic.close().join();
        List<String> topics = new ArrayList<>(pulsar.getBrokerService().getTopics().keys());
        topics.removeIf(item -> item.contains(SystemTopicNames.NAMESPACE_EVENTS_LOCAL_NAME));
        Assert.assertEquals(topics.size(), 0);
        @Cleanup
        Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
                .topic(topicName)
                .subscriptionName("sub-1")
                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
                .subscriptionType(SubscriptionType.Shared)
                .subscribe();
    }