@Override
    public synchronized ConsumerStats getStats() {
        if (stats == null) {
            return null;
        }
        stats.reset();

        consumers.forEach((partition, consumer) -> stats.updateCumulativeStats(partition, consumer.getStats()));
        return stats;
    }
@Test
    public void testGetStats() throws Exception {
        String topicName = "test-stats";
        ClientConfigurationData conf = new ClientConfigurationData();
        // ip and port is arbitrary since test will attempt to make a connection
        // and the connection is not needed for making the test to pass. This test should be improved.
        conf.setServiceUrl("pulsar://127.0.0.99:23456");
        conf.setStatsIntervalSeconds(100);

        ThreadFactory threadFactory = new DefaultThreadFactory("client-test-stats", Thread.currentThread().isDaemon());

        @Cleanup("shutdown")
        EventLoopGroup eventLoopGroup = EventLoopUtil.newEventLoopGroup(conf.getNumIoThreads(), false, threadFactory);

        @Cleanup("shutdownNow")
        ExecutorProvider executorProvider = new ExecutorProvider(1, "client-test-stats");

        @Cleanup
        PulsarClientImpl clientImpl = new PulsarClientImpl(conf, eventLoopGroup);

        ConsumerConfigurationData consumerConfData = new ConsumerConfigurationData();
        consumerConfData.setTopicNames(Sets.newHashSet(topicName));

        assertEquals(Long.parseLong("100"), clientImpl.getConfiguration().getStatsIntervalSeconds());

        MultiTopicsConsumerImpl impl = new MultiTopicsConsumerImpl(
            clientImpl, consumerConfData,
            executorProvider, null, Schema.BYTES, null, true);

        impl.getStats();

        clientImpl.close();
        executorProvider.shutdownNow();
        eventLoopGroup.shutdownGracefully().get();
    }