@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 testNonMeasurable() {
        metrics.addMetric(metrics.metricName("float", "group1", tags), (Gauge<Float>) (config, now) -> 99f);
        metrics.addMetric(metrics.metricName("double", "group1", tags), (Gauge<Double>) (config, now) -> 99d);
        metrics.addMetric(metrics.metricName("int", "group1", tags), (Gauge<Integer>) (config, now) -> 100);
        metrics.addMetric(metrics.metricName("long", "group1", tags), (Gauge<Long>) (config, now) -> 100L);

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

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

        result.stream()
            .flatMap(metrics -> Stream.of(metrics.builder().build()))
            .filter(metric -> metric.getName().equals("test.domain.group1.(float|double)")).forEach(
                doubleGauge -> {
                    assertTrue(doubleGauge.hasGauge());
                    assertEquals(tags, getTags(doubleGauge.getGauge().getDataPoints(0).getAttributesList()));
                    assertEquals(99d, doubleGauge.getGauge().getDataPoints(0).getAsDouble(), 0.0);
                });

        result.stream()
            .flatMap(metrics -> Stream.of(metrics.builder().build()))
            .filter(metric -> metric.getName().equals("test.domain.group1.(int|long)")).forEach(
                intGauge -> {
                    assertTrue(intGauge.hasGauge());
                    assertEquals(tags, getTags(intGauge.getGauge().getDataPoints(0).getAttributesList()));
                    assertEquals(100, intGauge.getGauge().getDataPoints(0).getAsDouble(), 0.0);
                });
    }