### Question:
AggregatedMetricsCollectionService extends AbstractExecutionThreadService implements MetricsCollectionService { protected abstract void publish(Iterator<MetricValues> metrics) throws Exception; AggregatedMetricsCollectionService(long publishIntervalInMillis); @Override final MetricsContext getContext(final Map<String, String> tags);  }### Answer:
@Category(SlowTests.class) @Test public void testPublish() throws InterruptedException { final BlockingQueue<MetricValues> published = new LinkedBlockingQueue<>(); AggregatedMetricsCollectionService service = new AggregatedMetricsCollectionService(1000L) { @Override protected void publish(Iterator<MetricValues> metrics) { Iterators.addAll(published, metrics); } }; service.startAndWait(); final Map<String, String> baseTags = ImmutableMap.of(Constants.Metrics.Tag.NAMESPACE, NAMESPACE, Constants.Metrics.Tag.APP, APP, Constants.Metrics.Tag.SERVICE, SERVICE, Constants.Metrics.Tag.RUN_ID, RUNID); try { service.getContext(EMPTY_TAGS).increment(METRIC, Integer.MAX_VALUE); service.getContext(EMPTY_TAGS).increment(METRIC, 2); service.getContext(EMPTY_TAGS).increment(METRIC, 3); service.getContext(EMPTY_TAGS).increment(METRIC, 4); verifyCounterMetricsValue(published, ImmutableMap.of(0, ImmutableMap.of(METRIC, 9L + Integer.MAX_VALUE))); Assert.assertNull(published.poll(3, TimeUnit.SECONDS)); service.getContext(EMPTY_TAGS).gauge(GAUGE_METRIC, 1); service.getContext(EMPTY_TAGS).gauge(GAUGE_METRIC, 2); service.getContext(EMPTY_TAGS).gauge(GAUGE_METRIC, 3); verifyGaugeMetricsValue(published, ImmutableMap.of(0, 3L)); MetricsContext baseCollector = service.getContext(baseTags); MetricsContext metricsContext = baseCollector.childContext(Constants.Metrics.Tag.HANDLER, HANDLER) .childContext(Constants.Metrics.Tag.INSTANCE_ID, INSTANCE); baseCollector.increment(METRIC, Integer.MAX_VALUE); metricsContext.increment(METRIC, 5); baseCollector.increment(METRIC, 10); baseCollector.increment(METRIC, 3); metricsContext.increment(METRIC, 2); metricsContext.increment(METRIC, 4); metricsContext.increment(METRIC, 3); metricsContext.increment(METRIC, 1); verifyCounterMetricsValue(published, ImmutableMap.of(4, ImmutableMap.of(METRIC, 13L + Integer.MAX_VALUE), 6, ImmutableMap.of(METRIC, 15L))); Assert.assertNull(published.poll(3, TimeUnit.SECONDS)); baseCollector.gauge(GAUGE_METRIC, Integer.MAX_VALUE); baseCollector.gauge(GAUGE_METRIC, 3); metricsContext.gauge(GAUGE_METRIC, 6); metricsContext.gauge(GAUGE_METRIC, 2); baseCollector.gauge(GAUGE_METRIC, 1); metricsContext.gauge(GAUGE_METRIC, Integer.MAX_VALUE); verifyGaugeMetricsValue(published, ImmutableMap.of(4, 1L, 6, (long) Integer.MAX_VALUE)); metricsContext.gauge(GAUGE_METRIC, 0); verifyCounterMetricsValue(published, ImmutableMap.of(6, ImmutableMap.of(GAUGE_METRIC, 0L))); } finally { service.stopAndWait(); } }