@Override
    public void collect(MetricsEmitter metricsEmitter) {
        for (Map.Entry<MetricKey, KafkaMetric> entry : ledger.getMetrics()) {
            MetricKey metricKey = entry.getKey();
            KafkaMetric metric = entry.getValue();

            try {
                collectMetric(metricsEmitter, metricKey, metric);
            } catch (Exception e) {
                // catch and log to continue processing remaining metrics
                log.error("Error processing Kafka metric {}", metricKey, e);
            }
        }
    }
@Test
    public void testMeasurableCounter() {
        Sensor sensor = metrics.sensor("test");
        sensor.add(metricName, new WindowedCount());

        sensor.record();
        sensor.record();

        time.sleep(60 * 1000L);

        // Collect metrics.
        collector.collect(testEmitter);
        List<SinglePointMetric> result = testEmitter.emittedMetrics();

        // Should get exactly 2 Kafka measurables since Metrics always includes a count measurable.
        assertEquals(2, result.size());

        Metric counter = result.stream()
            .flatMap(metrics -> Stream.of(metrics.builder().build()))
            .filter(metric -> metric.getName().equals("test.domain.group1.name1")).findFirst().get();

        assertTrue(counter.hasSum());
        assertEquals(tags, getTags(counter.getSum().getDataPoints(0).getAttributesList()));

        assertEquals(AggregationTemporality.AGGREGATION_TEMPORALITY_CUMULATIVE, counter.getSum().getAggregationTemporality());
        assertTrue(counter.getSum().getIsMonotonic());
        NumberDataPoint point = counter.getSum().getDataPoints(0);
        assertEquals(2d, point.getAsDouble(), 0.0);
        assertEquals(TimeUnit.SECONDS.toNanos(Instant.ofEpochSecond(61L).getEpochSecond()) +
            Instant.ofEpochSecond(61L).getNano(), point.getTimeUnixNano());
        assertEquals(TimeUnit.SECONDS.toNanos(Instant.ofEpochSecond(1L).getEpochSecond()) +
            Instant.ofEpochSecond(1L).getNano(), point.getStartTimeUnixNano());
    }