@Override
    public CompletableFuture<Void> close(boolean closeWithoutWaitingClientDisconnect) {
        return close(true, closeWithoutWaitingClientDisconnect);
    }
@Test
    public void testRemoveProducerOnNonPersistentTopic() throws Exception {
        final String topicName = "non-persistent://prop/ns-abc/topic_" + UUID.randomUUID();

        Producer<byte[]> producer = pulsarClient.newProducer()
                .topic(topicName)
                .create();

        NonPersistentTopic topic = (NonPersistentTopic) pulsar.getBrokerService().getTopicReference(topicName).get();
        Field field = AbstractTopic.class.getDeclaredField("userCreatedProducerCount");
        field.setAccessible(true);
        int userCreatedProducerCount = (int) field.get(topic);
        assertEquals(userCreatedProducerCount, 1);

        producer.close();
        userCreatedProducerCount = (int) field.get(topic);
        assertEquals(userCreatedProducerCount, 0);
    }