KafkaMetricsReporter implements Consumer<MetricsDump> { @Override public void accept(MetricsDump metricsDump) { try { kafkaAdapter.sendMessage(topicName, SerializableMetricsDump.createFor(metricsDump)).subscribe( metaData -> logger.trace("Successfully sent message {}", metaData), error -> logger.error("Unable to send MetricsDump to topic {}", topicName, error) ); } catch (Exception e) { logger.error("Unable to send metrics dump", e); } } KafkaMetricsReporter(KafkaAdapter kafkaAdapter, String topicName); @Override void accept(MetricsDump metricsDump);  }
@Test void dumpIsTransmittedAsExpected() throws Exception { int numDumps = 3; CountDownLatch cdl = new CountDownLatch(numDumps); kafkaAdapter.messages("test.metrics").subscribe(dump -> cdl.countDown()); Nova nova = Nova.builder().build(); for (int i=0; i< numDumps; i++) { sut.accept(nova.metrics.dump()); Thread.sleep(700); } cdl.await(20, TimeUnit.SECONDS); assertThat(cdl.getCount(), is(0L)); }