@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 testCollectFilter() {
        metrics.addMetric(metricName, (config, now) -> 100.0);

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

        // Should get exactly 1 Kafka measurables because we excluded the count measurable
        assertEquals(1, result.size());

        Metric counter = result.get(0).builder().build();

        assertTrue(counter.hasGauge());
        assertEquals(100L, counter.getGauge().getDataPoints(0).getAsDouble(), 0.0);
    }