@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 testCollectFilterWithDeltaTemporality() {
        MetricName name1 = metrics.metricName("nonMeasurable", "group1", tags);
        MetricName name2 = metrics.metricName("windowed", "group1", tags);
        MetricName name3 = metrics.metricName("cumulative", "group1", tags);

        metrics.addMetric(name1, (Gauge<Double>) (config, now) -> 99d);

        Sensor sensor = metrics.sensor("test");
        sensor.add(name2, new WindowedCount());
        sensor.add(name3, new CumulativeSum());

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

        // no-filter shall result in all 4 data metrics.
        assertEquals(4, result.size());

        testEmitter.reset();
        testEmitter.reconfigurePredicate(k -> !k.key().name().endsWith(".count"));
        collector.collect(testEmitter);
        result = testEmitter.emittedMetrics();

        // Drop metrics for Count type.
        assertEquals(3, result.size());

        testEmitter.reset();
        testEmitter.reconfigurePredicate(k -> !k.key().name().endsWith(".nonmeasurable"));
        collector.collect(testEmitter);
        result = testEmitter.emittedMetrics();

        // Drop non-measurable metric.
        assertEquals(3, result.size());

        testEmitter.reset();
        testEmitter.reconfigurePredicate(key -> true);
        collector.collect(testEmitter);
        result = testEmitter.emittedMetrics();

        // Again no filter.
        assertEquals(4, result.size());
    }