Migrate Kafka metrics to KafkaClientMetrics

Closes gh-20838

Author: snicoll

spring-boot-project/spring-boot-actuator-autoconfigure/build.gradle

@@ -101,6 +101,7 @@ dependencies {
 	optional("org.springframework.data:spring-data-redis")
 	optional("org.springframework.data:spring-data-solr")
 	optional("org.springframework.integration:spring-integration-core")
+	optional("org.springframework.kafka:spring-kafka")
 	optional("org.springframework.security:spring-security-config")
 	optional("org.springframework.security:spring-security-web")
 	optional("org.springframework.session:spring-session-core")

spring-boot-project/spring-boot-actuator-autoconfigure/src/main/java/org/springframework/boot/actuate/autoconfigure/metrics/KafkaMetricsAutoConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2019 the original author or authors.
+ * Copyright 2012-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,37 +18,37 @@
 
 import java.util.Collections;
 
-import javax.management.MBeanServer;
-
 import io.micrometer.core.instrument.MeterRegistry;
-import io.micrometer.core.instrument.binder.kafka.KafkaConsumerMetrics;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
+import io.micrometer.core.instrument.binder.kafka.KafkaClientMetrics;
 
 import org.springframework.boot.autoconfigure.AutoConfigureAfter;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
-import org.springframework.boot.autoconfigure.jmx.JmxAutoConfiguration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnSingleCandidate;
+import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.ProducerFactory;
 
 /**
  * Auto-configuration for Kafka metrics.
  *
  * @author Andy Wilkinson
+ * @author Stephane Nicoll
  * @since 2.1.0
  */
 @Configuration(proxyBeanMethods = false)
-@AutoConfigureAfter({ MetricsAutoConfiguration.class, JmxAutoConfiguration.class })
-@ConditionalOnClass({ KafkaConsumerMetrics.class, KafkaConsumer.class })
+@AutoConfigureAfter({ MetricsAutoConfiguration.class, KafkaAutoConfiguration.class })
+@ConditionalOnClass({ KafkaClientMetrics.class, ProducerFactory.class })
 @ConditionalOnBean(MeterRegistry.class)
 public class KafkaMetricsAutoConfiguration {
 
 	@Bean
 	@ConditionalOnMissingBean
-	@ConditionalOnBean(MBeanServer.class)
-	public KafkaConsumerMetrics kafkaConsumerMetrics(MBeanServer mbeanServer) {
-		return new KafkaConsumerMetrics(mbeanServer, Collections.emptyList());
+	@ConditionalOnSingleCandidate(ProducerFactory.class)
+	public KafkaClientMetrics kafkaClientMetrics(ProducerFactory<?, ?> producerFactory) {
+		return new KafkaClientMetrics(producerFactory.createProducer(), Collections.emptyList());
 	}
 
 }

spring-boot-project/spring-boot-actuator-autoconfigure/src/test/java/org/springframework/boot/actuate/autoconfigure/metrics/KafkaMetricsAutoConfigurationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2019 the original author or authors.
+ * Copyright 2012-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,54 +16,54 @@
 
 package org.springframework.boot.actuate.autoconfigure.metrics;
 
-import io.micrometer.core.instrument.binder.kafka.KafkaConsumerMetrics;
+import io.micrometer.core.instrument.binder.kafka.KafkaClientMetrics;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.boot.actuate.autoconfigure.metrics.test.MetricsRun;
 import org.springframework.boot.autoconfigure.AutoConfigurations;
-import org.springframework.boot.autoconfigure.jmx.JmxAutoConfiguration;
+import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
 import org.springframework.boot.test.context.runner.ApplicationContextRunner;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
 
 /**
  * Tests for {@link KafkaMetricsAutoConfiguration}.
  *
  * @author Andy Wilkinson
+ * @author Stephane Nicoll
  */
 class KafkaMetricsAutoConfigurationTests {
 
 	private final ApplicationContextRunner contextRunner = new ApplicationContextRunner().with(MetricsRun.simple())
-			.withPropertyValues("spring.jmx.enabled=true")
 			.withConfiguration(AutoConfigurations.of(KafkaMetricsAutoConfiguration.class));
 
 	@Test
-	void whenThereIsNoMBeanServerAutoConfigurationBacksOff() {
-		this.contextRunner.run((context) -> assertThat(context).doesNotHaveBean(KafkaConsumerMetrics.class));
+	void whenThereIsNoProducerFactoryAutoConfigurationBacksOff() {
+		this.contextRunner.run((context) -> assertThat(context).doesNotHaveBean(KafkaClientMetrics.class));
 	}
 
 	@Test
-	void whenThereIsAnMBeanServerKafkaConsumerMetricsIsConfigured() {
-		this.contextRunner.withConfiguration(AutoConfigurations.of(JmxAutoConfiguration.class))
-				.run((context) -> assertThat(context).hasSingleBean(KafkaConsumerMetrics.class));
+	void whenThereIsAnAProducerFactoryKafkaClientMetricsIsConfigured() {
+		this.contextRunner.withConfiguration(AutoConfigurations.of(KafkaAutoConfiguration.class))
+				.run((context) -> assertThat(context).hasSingleBean(KafkaClientMetrics.class));
 	}
 
 	@Test
-	void allowsCustomKafkaConsumerMetricsToBeUsed() {
-		this.contextRunner.withConfiguration(AutoConfigurations.of(JmxAutoConfiguration.class))
-				.withUserConfiguration(CustomKafkaConsumerMetricsConfiguration.class)
-				.run((context) -> assertThat(context).hasSingleBean(KafkaConsumerMetrics.class)
-						.hasBean("customKafkaConsumerMetrics"));
+	void allowsCustomKafkaClientMetricsToBeUsed() {
+		this.contextRunner.withConfiguration(AutoConfigurations.of(KafkaAutoConfiguration.class))
+				.withUserConfiguration(CustomKafkaClientMetricsConfiguration.class).run((context) -> assertThat(context)
+						.hasSingleBean(KafkaClientMetrics.class).hasBean("customKafkaClientMetrics"));
 	}
 
 	@Configuration(proxyBeanMethods = false)
-	static class CustomKafkaConsumerMetricsConfiguration {
+	static class CustomKafkaClientMetricsConfiguration {
 
 		@Bean
-		KafkaConsumerMetrics customKafkaConsumerMetrics() {
-			return new KafkaConsumerMetrics();
+		KafkaClientMetrics customKafkaClientMetrics() {
+			return mock(KafkaClientMetrics.class);
 		}
 
 	}

spring-boot-project/spring-boot-docs/src/docs/asciidoc/production-ready-features.adoc

@@ -1791,7 +1791,7 @@ Spring Boot registers the following core metrics when applicable:
 ** Number of classes loaded/unloaded
 * CPU metrics
 * File descriptor metrics
-* Kafka consumer metrics (<<production-ready-jmx,JMX support>> should be enabled)
+* Kafka consumer metrics
 * Log4j2 metrics: record the number of events logged to Log4j2 at each level
 * Logback metrics: record the number of events logged to Logback at each level
 * Uptime metrics: report a gauge for uptime and a fixed gauge representing the application's absolute start time