### Question:
MicrometerMetricsReporter implements MetricsReporter { @Override public final void init(List<KafkaMetric> metrics) { metrics.forEach(this::metricChange); }  @Override void configure(Map<String, ?> configs); @Override final void init(List<KafkaMetric> metrics); @Override final void metricChange(KafkaMetric metric); @Override void metricRemoval(KafkaMetric metric); @Override void close(); static final String FILTER_NAMES_INCLUSION_CONFIG; static final String FILTER_NAMES_CONFIG; }### Answer:
@Test public void metricsAreReportedViaMicrometer() { Tag tag = Tag.of("key", "value"); double value = Math.random(); KafkaMetric metric = createMetric("group", "name", Collections.singletonMap(tag.getKey(), tag.getValue()), () -> value); reporter.init(Collections.singletonList(metric)); Gauge gauge = meterRegistry.find("kafka.group.name").tags(Collections.singletonList(tag)).gauge(); assertNotNull(gauge); assertEquals(value, gauge.value(), 0D); }