public Optional<Topic> getTopicReference(String topic) {
          CompletableFuture<Optional<Topic>> future = topics.get(topic);
        if (future != null && future.isDone() && !future.isCompletedExceptionally()) {
            return future.join();
        } else {
            return Optional.empty();
        }
    }
@Test
    public void testStatsOfStorageSizeWithSubscription() throws Exception {
        final String topicName = "persistent://prop/ns-abc/no-subscription";
        Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
        PersistentTopic topicRef = (PersistentTopic) pulsar.getBrokerService().getTopicReference(topicName).get();

        assertNotNull(topicRef);
        assertEquals(topicRef.getStats(false, false, false).storageSize, 0);

        for (int i = 0; i < 10; i++) {
            producer.send(new byte[10]);
        }

        assertTrue(topicRef.getStats(false, false, false).storageSize > 0);
    }