spring-projectsGH-11435: Add Kafka health indicator

- refactor code to create and destroy admin client when health monitoring

Author: Jcamilorada

spring-boot-project/spring-boot-actuator-autoconfigure/src/main/java/org/springframework/boot/actuate/autoconfigure/kafka/KafkaHealthIndicatorAutoConfiguration.java

@@ -19,8 +19,6 @@
 import java.time.Duration;
 import java.util.Map;
 
-import org.apache.kafka.clients.admin.AdminClient;
-
 import org.springframework.boot.actuate.autoconfigure.health.CompositeHealthIndicatorConfiguration;
 import org.springframework.boot.actuate.autoconfigure.health.ConditionalOnEnabledHealthIndicator;
 import org.springframework.boot.actuate.autoconfigure.health.HealthIndicatorAutoConfiguration;
@@ -35,6 +33,7 @@
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.KafkaAdmin;
 
 /**
  * {@link EnableAutoConfiguration Auto-configuration} for {@link KafkaHealthIndicator}.
@@ -48,16 +47,16 @@
 public class KafkaHealthIndicatorAutoConfiguration {
 
 	@Configuration
-	@ConditionalOnBean(AdminClient.class)
+	@ConditionalOnBean(KafkaAdmin.class)
 	@EnableConfigurationProperties(KafkaHealthIndicatorProperties.class)
 	static class KafkaClientHealthIndicatorConfiguration extends
-			CompositeHealthIndicatorConfiguration<KafkaHealthIndicator, AdminClient> {
+			CompositeHealthIndicatorConfiguration<KafkaHealthIndicator, KafkaAdmin> {
 
-		private final Map<String, AdminClient> admins;
+		private final Map<String, KafkaAdmin> admins;
 
 		private final KafkaHealthIndicatorProperties properties;
 
-		KafkaClientHealthIndicatorConfiguration(Map<String, AdminClient> admins,
+		KafkaClientHealthIndicatorConfiguration(Map<String, KafkaAdmin> admins,
 				KafkaHealthIndicatorProperties properties) {
 			this.admins = admins;
 			this.properties = properties;
@@ -70,7 +69,7 @@ public HealthIndicator kafkaHealthIndicator() {
 		}
 
 		@Override
-		protected KafkaHealthIndicator createHealthIndicator(AdminClient source) {
+		protected KafkaHealthIndicator createHealthIndicator(KafkaAdmin source) {
 			Duration responseTimeout = this.properties.getResponseTimeout();
 
 			return new KafkaHealthIndicator(source,

spring-boot-project/spring-boot-actuator-autoconfigure/src/main/java/org/springframework/boot/actuate/autoconfigure/kafka/package-info.java

@@ -15,6 +15,6 @@
  */
 
 /**
- * Auto-configuration for actuator JDBC concerns.
+ * Auto-configuration for actuator kafka support.
  */
 package org.springframework.boot.actuate.autoconfigure.kafka;

spring-boot-project/spring-boot-actuator-autoconfigure/src/test/java/org/springframework/boot/actuate/autoconfigure/kafka/KafkaHealthIndicatorAutoConfigurationTests.java

@@ -16,19 +16,16 @@
 
 package org.springframework.boot.actuate.autoconfigure.kafka;
 
-import org.apache.kafka.clients.admin.AdminClient;
 import org.junit.Test;
 
 import org.springframework.boot.actuate.autoconfigure.health.HealthIndicatorAutoConfiguration;
 import org.springframework.boot.actuate.health.ApplicationHealthIndicator;
 import org.springframework.boot.actuate.kafka.KafkaHealthIndicator;
 import org.springframework.boot.autoconfigure.AutoConfigurations;
+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 KafkaHealthIndicatorAutoConfiguration}.
@@ -38,7 +35,7 @@
 public class KafkaHealthIndicatorAutoConfigurationTests {
 
 	private ApplicationContextRunner contextRunner = new ApplicationContextRunner()
-			.withConfiguration(AutoConfigurations.of(KafkaConfiguration.class,
+			.withConfiguration(AutoConfigurations.of(KafkaAutoConfiguration.class,
 					KafkaHealthIndicatorAutoConfiguration.class,
 					HealthIndicatorAutoConfiguration.class));
 
@@ -56,14 +53,4 @@ public void runWhenDisabledShouldNotCreateIndicator() {
 						.doesNotHaveBean(KafkaHealthIndicator.class)
 						.hasSingleBean(ApplicationHealthIndicator.class));
 	}
-
-	@Configuration
-	protected static class KafkaConfiguration {
-
-		@Bean
-		public AdminClient kafkaAdminClient() {
-			return mock(AdminClient.class);
-		}
-
-	}
 }

spring-boot-project/spring-boot-actuator/pom.xml

@@ -246,6 +246,11 @@
 			<artifactId>spring-boot-test-support</artifactId>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka-test</artifactId>
+			<scope>test</scope>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-autoconfigure</artifactId>
@@ -256,6 +261,11 @@
 			<artifactId>log4j-slf4j-impl</artifactId>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.slf4j</groupId>
+			<artifactId>log4j-over-slf4j</artifactId>
+			<scope>test</scope>
+		</dependency>
 		<dependency>
 			<groupId>org.apache.logging.log4j</groupId>
 			<artifactId>log4j-api</artifactId>

spring-boot-project/spring-boot-actuator/src/main/java/org/springframework/boot/actuate/kafka/KafkaHealthIndicator.java

@@ -23,6 +23,7 @@
 import org.springframework.boot.actuate.health.AbstractHealthIndicator;
 import org.springframework.boot.actuate.health.Health.Builder;
 import org.springframework.boot.actuate.health.HealthIndicator;
+import org.springframework.kafka.core.KafkaAdmin;
 import org.springframework.util.Assert;
 
 /**
@@ -32,26 +33,30 @@
  */
 public class KafkaHealthIndicator extends AbstractHealthIndicator {
 
-	private final AdminClient adminClient;
+	private final KafkaAdmin kafkaAdmin;
 	private final DescribeClusterOptions describeOptions;
 
 	/**
 	 * Create a new {@link KafkaHealthIndicator} instance.
 	 *
-	 * @param adminClient the kafka admin client
+	 * @param kafkaAdmin the kafka admin
 	 * @param responseTimeout the describe cluster request timeout in milliseconds
 	 */
-	public KafkaHealthIndicator(AdminClient adminClient, long responseTimeout) {
-		Assert.notNull(adminClient, "KafkaAdmin must not be null");
+	public KafkaHealthIndicator(KafkaAdmin kafkaAdmin, long responseTimeout) {
+		Assert.notNull(kafkaAdmin, "KafkaAdmin must not be null");
 
-		this.adminClient = adminClient;
-		this.describeOptions = new DescribeClusterOptions().timeoutMs((int) responseTimeout);
+		this.kafkaAdmin = kafkaAdmin;
+		this.describeOptions = new DescribeClusterOptions()
+				.timeoutMs((int) responseTimeout);
 	}
 
 	@Override
 	protected void doHealthCheck(Builder builder) throws Exception {
-		DescribeClusterResult result = this.adminClient.describeCluster(this.describeOptions);
-		builder.up().withDetail("clusterId", result.clusterId().get());
+		try (AdminClient adminClient = AdminClient.create(this.kafkaAdmin.getConfig())) {
+			DescribeClusterResult result = adminClient
+					.describeCluster(this.describeOptions);
+			builder.up().withDetail("clusterId", result.clusterId().get());
+		}
 	}
 }
 

spring-boot-project/spring-boot-actuator/src/test/java/org/springframework/boot/actuate/kafka/KafkaHealthIndicatorTests.java

@@ -16,25 +16,20 @@
 
 package org.springframework.boot.actuate.kafka;
 
-import org.apache.kafka.clients.admin.AdminClient;
-import org.apache.kafka.clients.admin.DescribeClusterOptions;
-import org.apache.kafka.clients.admin.DescribeClusterResult;
-import org.apache.kafka.common.KafkaFuture;
+import java.util.Collections;
+
+import org.apache.kafka.clients.producer.ProducerConfig;
 import org.assertj.core.data.MapEntry;
 import org.junit.Before;
+import org.junit.Rule;
 import org.junit.Test;
-import org.mockito.ArgumentCaptor;
-import org.mockito.Captor;
-import org.mockito.Mock;
-import org.mockito.MockitoAnnotations;
 
 import org.springframework.boot.actuate.health.Health;
 import org.springframework.boot.actuate.health.Status;
+import org.springframework.kafka.core.KafkaAdmin;
+import org.springframework.kafka.test.rule.KafkaEmbedded;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.BDDMockito.given;
-import static org.mockito.Mockito.verify;
 
 /**
  * Test for {@link KafkaHealthIndicator}
@@ -43,48 +38,34 @@
  */
 public class KafkaHealthIndicatorTests {
 
-	private static final Long RESPONSE_TIME = 10L;
-	private static final String CLUSTER_ID = "abc_123";
-
-	@Mock
-	private AdminClient adminClient;
-
-	@Mock
-	private DescribeClusterResult describeClusterResult;
-
-	@Mock
-	private KafkaFuture<String> clusterIdFuture;
+	private static final Long RESPONSE_TIME = 1000L;
 
-	@Captor
-	private ArgumentCaptor<DescribeClusterOptions> describeOptionsCaptor;
+	@Rule
+	public KafkaEmbedded kafkaEmbedded = new KafkaEmbedded(1, true);
 
-	private KafkaHealthIndicator healthIndicator;
+	private KafkaAdmin kafkaAdmin;
 
 	@Before
 	public void setup() {
-		MockitoAnnotations.initMocks(this);
-		this.healthIndicator = new KafkaHealthIndicator(this.adminClient, RESPONSE_TIME);
-		given(this.describeClusterResult.clusterId()).willReturn(this.clusterIdFuture);
+		this.kafkaAdmin = new KafkaAdmin(Collections.singletonMap(
+				ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.kafkaEmbedded.getBrokersAsString()));
 	}
 
 	@Test
-	public void kafkaIsUp() throws Exception {
-		given(this.adminClient.describeCluster(any(DescribeClusterOptions.class)))
-				.willReturn(this.describeClusterResult);
-		given(this.clusterIdFuture.get()).willReturn(CLUSTER_ID);
-		Health health = this.healthIndicator.health();
+	public void kafkaIsUp() {
+		KafkaHealthIndicator healthIndicator = new KafkaHealthIndicator(this.kafkaAdmin, RESPONSE_TIME);
+		Health health = healthIndicator.health();
 		assertThat(health.getStatus()).isEqualTo(Status.UP);
-		assertThat(health.getDetails()).containsOnly(MapEntry.entry("clusterId", CLUSTER_ID));
-		verify(this.adminClient).describeCluster(this.describeOptionsCaptor.capture());
-		assertThat(this.describeOptionsCaptor.getValue().timeoutMs()).isEqualTo(RESPONSE_TIME.intValue());
+		assertThat(health.getDetails()).containsOnly(MapEntry.entry(
+				"clusterId", this.kafkaEmbedded.getKafkaServer(0).clusterId()));
 	}
 
 	@Test
-	public void kafkaIsDown() {
-		given(this.adminClient.describeCluster(any(DescribeClusterOptions.class)))
-				.willThrow(new IllegalStateException("test, expected"));
-		Health health = this.healthIndicator.health();
+	public void kafkaIsDown() throws Exception {
+		this.kafkaEmbedded.destroy();
+		KafkaHealthIndicator healthIndicator = new KafkaHealthIndicator(this.kafkaAdmin, RESPONSE_TIME);
+		Health health = healthIndicator.health();
 		assertThat(health.getStatus()).isEqualTo(Status.DOWN);
-		assertThat((String) health.getDetails().get("error")).contains("test, expected");
+		assertThat((String) health.getDetails().get("error")).isNotEmpty();
 	}
 }