@Override
    public void incrementNumAcksReceived(long latencyNs) {
        numAcksReceived.increment();
        synchronized (ds) {
            ds.update(TimeUnit.NANOSECONDS.toMillis(latencyNs));
        }
    }
@Test
    public void testIncrementNumAcksReceived() throws Exception {
        ClientConfigurationData conf = new ClientConfigurationData();
        conf.setStatsIntervalSeconds(1);
        PulsarClientImpl client = mock(PulsarClientImpl.class);
        ConnectionPool connectionPool = mock(ConnectionPool.class);
        when(client.getCnxPool()).thenReturn(connectionPool);
        when(client.getConfiguration()).thenReturn(conf);
        @Cleanup("stop")
        Timer timer = new HashedWheelTimer();
        when(client.timer()).thenReturn(timer);
        ProducerImpl<?> producer = mock(ProducerImpl.class);
        when(producer.getTopic()).thenReturn("topic-test");
        when(producer.getProducerName()).thenReturn("producer-test");
        when(producer.getPendingQueueSize()).thenReturn(1);
        ProducerConfigurationData producerConfigurationData = new ProducerConfigurationData();
        ProducerStatsRecorderImpl recorder = new ProducerStatsRecorderImpl(client, producerConfigurationData, producer);
        long latencyNs = TimeUnit.SECONDS.toNanos(1);
        recorder.incrementNumAcksReceived(latencyNs);
        Thread.sleep(1200);
        assertEquals(1000.0, recorder.getSendLatencyMillisMax(), 0.5);
        recorder.cancelStatsTimeout();
    }