[ https://issues.apache.org/jira/browse/KAFKA-14081?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17568387#comment-17568387 ]
Gian Luca commented on KAFKA-14081: ----------------------------------- This is my report implementation: {code:java} package kpm; import org.apache.kafka.common.metrics.KafkaMetric; import org.apache.kafka.common.metrics.MetricsReporter; import java.util.List; import java.util.Map; public class TestMetricsReporter implements MetricsReporter { public TestMetricsReporter() { System.out.println("creation"); } @Override public void init(List<KafkaMetric> list) { System.out.println("init"); dump(list); } @Override public void metricChange(KafkaMetric kafkaMetric) { if (kafkaMetric.metricName().name().equals("request-total")) { System.out.println("metric change"); dump(kafkaMetric); } } @Override public void metricRemoval(KafkaMetric kafkaMetric) { } @Override public void close() { } @Override public void configure(Map<String, ?> map) { } private void dump(List<KafkaMetric> list) { for (KafkaMetric m: list) { dump(m); } } private void dump(KafkaMetric m) { System.out.printf("%s/%s -> %s\n", m.metricName().name(), m.metricName().tags(), m.metricValue()); } } {code} And this is the main class: {code:java} package kpm; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.serialization.Serializer; import org.apache.kafka.common.serialization.StringSerializer; import java.util.Properties; import java.util.concurrent.Future; public class Main { public static void main(String[] args) throws InterruptedException { Serializer<Void> keySerializer = (s, o) -> new byte[0]; Properties properties = new Properties(); properties.setProperty("bootstrap.servers", "localhost:9092"); properties.setProperty("topic", "source"); properties.setProperty("metric.reporters", TestMetricsReporter.class.getCanonicalName()); KafkaProducer<Void, String> producer = new KafkaProducer<>( properties, keySerializer, new StringSerializer()); for (int i = 0; i < 1000; i++) { Future f = producer.send( new ProducerRecord("source", "{}"), (recordMetadata, e) -> { if (e == null) System.out.println("SUCCESS"); else System.out.println("FAIL"); } ); Thread.sleep(1000); } Thread.sleep(5000); System.out.println("Closing..."); producer.close(); } } {code} On execution, the values of the 'request-total' metric are notified once (through the metricChange() method) with value 0.0, then no more updates happen. > Cannot get my MetricsReporter implementation to receive meaningful metrics > -------------------------------------------------------------------------- > > Key: KAFKA-14081 > URL: https://issues.apache.org/jira/browse/KAFKA-14081 > Project: Kafka > Issue Type: Bug > Components: clients > Affects Versions: 3.3.0 > Reporter: Gian Luca > Priority: Minor > > I want to extract metrics from KafkaProducer to export them to our company > monitoring solution. At first I went for implementing {{MetricsReporter}} and > registering my implementation through the "metric.reporters" config property. > The class is correctly registered as it receives metric updates through > {{metricChange()}} while KafkaProducer is being used. The problem is that all > the metric values are stuck at zero (NaN in older versions of Kafka), even > the most trivial (e.g. 'record-send-total'). > If instead of using a report I simply poll the {{metrics()}} method of the > KafkaProducer, then I see meaningful values: counters increasing over time, > etc. -- This message was sent by Atlassian Jira (v8.20.10#820010)