@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 testSecondDeltaCollectDouble() {
        Sensor sensor = metrics.sensor("test");
        sensor.add(metricName, new CumulativeSum());

        sensor.record();
        sensor.record();
        time.sleep(60 * 1000L);

        testEmitter.onlyDeltaMetrics(true);
        collector.collect(testEmitter);

        // Update it again by 5 and advance time by another 60 seconds.
        sensor.record();
        sensor.record();
        sensor.record();
        sensor.record();
        sensor.record();
        time.sleep(60 * 1000L);

        testEmitter.reset();
        collector.collect(testEmitter);
        List<SinglePointMetric> result = testEmitter.emittedMetrics();

        assertEquals(2, result.size());

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

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