@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 testMeasurableWithException() {
        metrics.addMetric(metricName, null, (config, now) -> {
            throw new RuntimeException();
        });

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

        //Verify only the global count of metrics exist
        assertEquals(1, result.size());
        // Group is registered as kafka-metrics-count
        assertEquals("test.domain.kafka.count.count", result.get(0).builder().build().getName());
        //Verify metrics with measure() method throw exception is not returned
        assertFalse(result.stream()
            .flatMap(metrics -> Stream.of(metrics.builder().build()))
            .anyMatch(metric -> metric.getName().equals("test.domain.group1.name1")));
    }