@Override
    public synchronized ProducerStatsRecorderImpl getStats() {
        if (stats == null) {
            return null;
        }
        stats.reset();
        producers.forEach(
                (partition, producer) -> stats.updateCumulativeStats(producer.getTopic(), producer.getStats()));
        return stats;
    }
@Test
    public void testGetStats() throws Exception {
        String topicName = "test-stats";
        ClientConfigurationData conf = new ClientConfigurationData();
        conf.setServiceUrl("pulsar://localhost:6650");
        conf.setStatsIntervalSeconds(100);

        ThreadFactory threadFactory = new DefaultThreadFactory("client-test-stats", Thread.currentThread().isDaemon());
        @Cleanup("shutdownGracefully")
        EventLoopGroup eventLoopGroup = EventLoopUtil.newEventLoopGroup(conf.getNumIoThreads(), false, threadFactory);

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

        ProducerConfigurationData producerConfData = new ProducerConfigurationData();
        producerConfData.setMessageRoutingMode(MessageRoutingMode.CustomPartition);
        producerConfData.setCustomMessageRouter(new CustomMessageRouter());

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

        PartitionedProducerImpl impl = new PartitionedProducerImpl(
            clientImpl, topicName, producerConfData,
            1, null, null, null);

        impl.getStats();
    }