ZookeeperCounterReporter extends ScheduledReporter { @Override public void report(SortedMap<String, Gauge> gauges, SortedMap<String, Counter> counters, SortedMap<String, Histogram> histograms, SortedMap<String, Meter> meters, SortedMap<String, Timer> timers) { counters.forEach((name, counter) -> reportCounter(name, counter)); meters .entrySet() .stream() .filter(meterEntry -> meterEntry.getKey().startsWith(Meters.THROUGHPUT_BYTES)) .forEach(meterEntry -> reportVolumeCounter(meterEntry.getKey(), meterEntry.getValue().getCount())); } ZookeeperCounterReporter(MetricRegistry registry,
                                    CounterStorage counterStorage,
                                    ConfigFactory config
    ); @Override void report(SortedMap<String, Gauge> gauges,
                       SortedMap<String, Counter> counters,
                       SortedMap<String, Histogram> histograms,
                       SortedMap<String, Meter> meters,
                       SortedMap<String, Timer> timers);  }
@Test public void shouldReportPublishedMessages() { SortedMap<String, Counter> counters = prepareCounters(METRIC_NAME_FOR_PUBLISHED); when(counter.getCount()).thenReturn(COUNT); zookeeperCounterReporter.report(EMPTY_GAUGES, counters, EMPTY_HISTOGRAMS, EMPTY_METERS, EMPTY_TIMERS); verify(counterStorage).setTopicPublishedCounter(QUALIFIED_TOPIC_NAME, COUNT); }
@Test public void shouldReportDeliveredMessages() { SortedMap<String, Counter> counters = prepareCounters(METRIC_NAME_FOR_DELIVERED); when(counter.getCount()).thenReturn(COUNT); zookeeperCounterReporter.report(EMPTY_GAUGES, counters, EMPTY_HISTOGRAMS, EMPTY_METERS, EMPTY_TIMERS); verify(counterStorage).setSubscriptionDeliveredCounter(QUALIFIED_TOPIC_NAME, SUBSCRIPTION_NAME, COUNT); }
@Test public void shouldReportDiscardedMessages() { SortedMap<String, Counter> counters = prepareCounters(METRIC_NAME_FOR_DISCARDED); when(counter.getCount()).thenReturn(COUNT); zookeeperCounterReporter.report(EMPTY_GAUGES, counters, EMPTY_HISTOGRAMS, EMPTY_METERS, EMPTY_TIMERS); verify(counterStorage).setSubscriptionDiscardedCounter( QUALIFIED_TOPIC_NAME, SUBSCRIPTION_NAME, COUNT ); }
@Test public void shouldReportSubscriptionVolumeCounter() { SortedMap<String, Meter> meters = new TreeMap<>(); meters.put(METRIC_NAME_FOR_SUBSCRIPTION_THROUGHPUT, meter); when(meter.getCount()).thenReturn(COUNT); zookeeperCounterReporter.report(EMPTY_GAUGES, EMPTY_COUNTERS, EMPTY_HISTOGRAMS, meters, EMPTY_TIMERS); verify(counterStorage).incrementVolumeCounter( QUALIFIED_TOPIC_NAME, SUBSCRIPTION_NAME, COUNT ); }
@Test public void shouldReportTopicVolumeCounter() { SortedMap<String, Meter> meters = new TreeMap<>(); meters.put(METRIC_NAME_FOR_TOPIC_THRESHOLD, meter); when(meter.getCount()).thenReturn(COUNT); zookeeperCounterReporter.report(EMPTY_GAUGES, EMPTY_COUNTERS, EMPTY_HISTOGRAMS, meters, EMPTY_TIMERS); verify(counterStorage).incrementVolumeCounter( QUALIFIED_TOPIC_NAME, COUNT ); }