MINOR: Add toString() to KafkaMetric for readable logging

clients/src/main/java/org/apache/kafka/common/metrics/KafkaMetric.java

@@ -124,4 +124,20 @@ public void config(MetricConfig config) {
             this.config = config;
         }
     }
+
+    /**
+     * Returns a human-readable representation of this metric, primarily useful for logging
+     * in contexts like {@link MetricsReporter#metricChange(KafkaMetric)}.
+     *
+     * <p>The metricValueProvider is represented by its simple class name rather than its full
+     * toString() to avoid dumping internal stat state (e.g. SampledStat's samples list) into
+     * logs, which could be verbose and is rarely useful for identifying which metric changed.
+     */
+    @Override
+    public String toString() {
+        return "KafkaMetric [" +
+                "metricName=" + metricName +
+                ", metricValueProvider=" + metricValueProvider.getClass().getSimpleName() +
+                ']';
+    }
 }

clients/src/test/java/org/apache/kafka/common/metrics/KafkaMetricTest.java

@@ -17,6 +17,7 @@
 package org.apache.kafka.common.metrics;
 
 import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.stats.Avg;
 import org.apache.kafka.common.utils.MockTime;
 
 import org.junit.jupiter.api.Test;
@@ -30,20 +31,26 @@
 
 public class KafkaMetricTest {
 
-    private static final MetricName METRIC_NAME = new MetricName("name", "group", "description", Collections.emptyMap());
+    private static final MetricName METRIC_NAME_1 = new MetricName("name", "group", "description", Collections.emptyMap());
+    private static final MetricName METRIC_NAME_2 = new MetricName(
+            "request-latency-avg",
+            "consumer-fetch-manager-metrics",
+            "The average request latency in ms",
+            Collections.singletonMap("client-id", "consumer-1")
+    );
 
     @Test
     public void testIsMeasurable() {
         Measurable metricValueProvider = (config, now) -> 0;
-        KafkaMetric metric = new KafkaMetric(new Object(), METRIC_NAME, metricValueProvider, new MetricConfig(), new MockTime());
+        KafkaMetric metric = new KafkaMetric(new Object(), METRIC_NAME_1, metricValueProvider, new MetricConfig(), new MockTime());
         assertTrue(metric.isMeasurable());
         assertEquals(metricValueProvider, metric.measurable());
     }
 
     @Test
     public void testIsMeasurableWithGaugeProvider() {
         Gauge<Double> metricValueProvider = (config, now) -> 0.0;
-        KafkaMetric metric = new KafkaMetric(new Object(), METRIC_NAME, metricValueProvider, new MetricConfig(), new MockTime());
+        KafkaMetric metric = new KafkaMetric(new Object(), METRIC_NAME_1, metricValueProvider, new MetricConfig(), new MockTime());
         assertFalse(metric.isMeasurable());
         assertThrows(IllegalStateException.class, metric::measurable);
     }
@@ -54,14 +61,14 @@ public void testMeasurableValueReturnsZeroWhenNotMeasurable() {
         MetricConfig config = new MetricConfig();
         Gauge<Integer> gauge = (c, now) -> 7;
 
-        KafkaMetric metric = new KafkaMetric(new Object(), METRIC_NAME, gauge, config, time);
+        KafkaMetric metric = new KafkaMetric(new Object(), METRIC_NAME_1, gauge, config, time);
         assertEquals(0.0d, metric.measurableValue(time.milliseconds()), 0.0d);
     }
 
     @Test
     public void testKafkaMetricAcceptsNonMeasurableNonGaugeProvider() {
         MetricValueProvider<String> provider = (config, now) -> "metric value provider";
-        KafkaMetric metric = new KafkaMetric(new Object(), METRIC_NAME, provider, new MetricConfig(), new MockTime());
+        KafkaMetric metric = new KafkaMetric(new Object(), METRIC_NAME_1, provider, new MetricConfig(), new MockTime());
 
         Object value = metric.metricValue();
         assertEquals("metric value provider", value);
@@ -70,7 +77,43 @@ public void testKafkaMetricAcceptsNonMeasurableNonGaugeProvider() {
     @Test
     public void testConstructorWithNullProvider() {
         assertThrows(NullPointerException.class, () ->
-                new KafkaMetric(new Object(), METRIC_NAME, null, new MetricConfig(), new MockTime())
+                new KafkaMetric(new Object(), METRIC_NAME_1, null, new MetricConfig(), new MockTime())
         );
     }
+
+    /**
+     * Verifies that toString produces a human-readable representation suitable for logging,
+     * e.g. in {@link org.apache.kafka.common.metrics.MetricsReporter#metricChange(KafkaMetric)}.
+     * Note that metricValueProvider may render as a lambda reference rather than a meaningful
+     * value, but this is still a significant improvement over the default Object.toString()
+     * output (e.g. "KafkaMetric@62a7d6c6").
+     */
+    @Test
+    public void testToStringWithLambdaProvider() {
+        Measurable metricValueProvider = (config, now) -> 0;
+        KafkaMetric metric = new KafkaMetric(
+                new Object(), METRIC_NAME_2, metricValueProvider, new MetricConfig(), new MockTime());
+
+        String result = metric.toString();
+        assertTrue(result.startsWith(
+                "KafkaMetric [metricName=MetricName [name=request-latency-avg, "
+                + "group=consumer-fetch-manager-metrics, "
+                + "description=The average request latency in ms, "
+                + "tags={client-id=consumer-1}], "
+                + "metricValueProvider="));
+        assertTrue(result.endsWith("]"));
+    }
+
+    @Test
+    public void testToStringWithRealStat() {
+        Avg avg = new Avg();
+        KafkaMetric metric = new KafkaMetric(new Object(), METRIC_NAME_2, avg, new MetricConfig(), new MockTime());
+
+        assertEquals("KafkaMetric [metricName=MetricName [name=request-latency-avg, "
+                + "group=consumer-fetch-manager-metrics, "
+                + "description=The average request latency in ms, "
+                + "tags={client-id=consumer-1}], "
+                + "metricValueProvider=Avg]",
+                metric.toString());
+    }
 }