@Test public void testIncrementNumAcksReceived() throws Exception { ClientConfigurationData conf = new ClientConfigurationData(); conf.setStatsIntervalSeconds(1); PulsarClientImpl client = mock(PulsarClientImpl.class); when(client.getConfiguration()).thenReturn(conf); 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); }
@Override public void incrementNumAcksReceived(long latencyNs) { numAcksReceived.increment(); synchronized (ds) { ds.update(TimeUnit.NANOSECONDS.toMillis(latencyNs)); } }
ProducerStatsRecorderImpl implements ProducerStatsRecorder { @Override public void incrementNumAcksReceived(long latencyNs) { numAcksReceived.increment(); synchronized (ds) { ds.update(TimeUnit.NANOSECONDS.toMillis(latencyNs)); } } }
ProducerStatsRecorderImpl implements ProducerStatsRecorder { @Override public void incrementNumAcksReceived(long latencyNs) { numAcksReceived.increment(); synchronized (ds) { ds.update(TimeUnit.NANOSECONDS.toMillis(latencyNs)); } } ProducerStatsRecorderImpl();  ProducerStatsRecorderImpl(PulsarClientImpl pulsarClient, ProducerConfigurationData conf,
            ProducerImpl<?> producer); }
ProducerStatsRecorderImpl implements ProducerStatsRecorder { @Override public void incrementNumAcksReceived(long latencyNs) { numAcksReceived.increment(); synchronized (ds) { ds.update(TimeUnit.NANOSECONDS.toMillis(latencyNs)); } } ProducerStatsRecorderImpl();  ProducerStatsRecorderImpl(PulsarClientImpl pulsarClient, ProducerConfigurationData conf,
            ProducerImpl<?> producer); @Override void updateNumMsgsSent(long numMsgs, long totalMsgsSize); @Override void incrementSendFailed(); @Override void incrementSendFailed(long numMsgs); @Override void incrementNumAcksReceived(long latencyNs); @Override long getNumMsgsSent(); @Override long getNumBytesSent(); @Override long getNumSendFailed(); @Override long getNumAcksReceived(); long getTotalMsgsSent(); long getTotalBytesSent(); long getTotalSendFailed(); long getTotalAcksReceived(); @Override double getSendMsgsRate(); @Override double getSendBytesRate(); @Override double getSendLatencyMillis50pct(); @Override double getSendLatencyMillis75pct(); @Override double getSendLatencyMillis95pct(); @Override double getSendLatencyMillis99pct(); @Override double getSendLatencyMillis999pct(); @Override double getSendLatencyMillisMax(); void cancelStatsTimeout(); }
ProducerStatsRecorderImpl implements ProducerStatsRecorder { @Override public void incrementNumAcksReceived(long latencyNs) { numAcksReceived.increment(); synchronized (ds) { ds.update(TimeUnit.NANOSECONDS.toMillis(latencyNs)); } } ProducerStatsRecorderImpl();  ProducerStatsRecorderImpl(PulsarClientImpl pulsarClient, ProducerConfigurationData conf,
            ProducerImpl<?> producer); @Override void updateNumMsgsSent(long numMsgs, long totalMsgsSize); @Override void incrementSendFailed(); @Override void incrementSendFailed(long numMsgs); @Override void incrementNumAcksReceived(long latencyNs); @Override long getNumMsgsSent(); @Override long getNumBytesSent(); @Override long getNumSendFailed(); @Override long getNumAcksReceived(); long getTotalMsgsSent(); long getTotalBytesSent(); long getTotalSendFailed(); long getTotalAcksReceived(); @Override double getSendMsgsRate(); @Override double getSendBytesRate(); @Override double getSendLatencyMillis50pct(); @Override double getSendLatencyMillis75pct(); @Override double getSendLatencyMillis95pct(); @Override double getSendLatencyMillis99pct(); @Override double getSendLatencyMillis999pct(); @Override double getSendLatencyMillisMax(); void cancelStatsTimeout();  }