Add Kafka health indicator

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

@@ -193,6 +193,11 @@
 			<artifactId>elasticsearch</artifactId>
 			<optional>true</optional>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+			<optional>true</optional>
+		</dependency>
 		<dependency>
 			<groupId>org.flywaydb</groupId>
 			<artifactId>flyway-core</artifactId>

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

@@ -0,0 +1,81 @@
+/*
+ * Copyright 2012-2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.actuate.autoconfigure.kafka;
+
+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;
+import org.springframework.boot.actuate.health.HealthIndicator;
+import org.springframework.boot.actuate.kafka.KafkaHealthIndicator;
+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+import org.springframework.boot.autoconfigure.AutoConfigureBefore;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * {@link EnableAutoConfiguration Auto-configuration} for {@link KafkaHealthIndicator}.
+ *
+ * @author Juan Rada
+ */
+@Configuration
+@ConditionalOnEnabledHealthIndicator("kafka")
+@AutoConfigureBefore(HealthIndicatorAutoConfiguration.class)
+@AutoConfigureAfter(KafkaAutoConfiguration.class)
+public class KafkaHealthIndicatorAutoConfiguration {
+
+	@Configuration
+	@ConditionalOnBean(AdminClient.class)
+	@EnableConfigurationProperties(KafkaHealthIndicatorProperties.class)
+	static class KafkaClientHealthIndicatorConfiguration extends
+			CompositeHealthIndicatorConfiguration<KafkaHealthIndicator, AdminClient> {
+
+		private final Map<String, AdminClient> admins;
+
+		private final KafkaHealthIndicatorProperties properties;
+
+		KafkaClientHealthIndicatorConfiguration(Map<String, AdminClient> admins,
+				KafkaHealthIndicatorProperties properties) {
+			this.admins = admins;
+			this.properties = properties;
+		}
+
+		@Bean
+		@ConditionalOnMissingBean(name = "kafkaHealthIndicator")
+		public HealthIndicator kafkaHealthIndicator() {
+			return createHealthIndicator(this.admins);
+		}
+
+		@Override
+		protected KafkaHealthIndicator createHealthIndicator(AdminClient source) {
+			Duration responseTimeout = this.properties.getResponseTimeout();
+
+			return new KafkaHealthIndicator(source,
+					responseTimeout == null ? 100L : responseTimeout.toMillis());
+		}
+	}
+
+}

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

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2012-2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.actuate.autoconfigure.kafka;
+
+import java.time.Duration;
+
+import org.springframework.boot.actuate.kafka.KafkaHealthIndicator;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * External configuration properties for {@link KafkaHealthIndicator}.
+ *
+ * @author Juan Rada
+ */
+@ConfigurationProperties(prefix = "management.health.kafka", ignoreUnknownFields = false)
+public class KafkaHealthIndicatorProperties {
+
+	/**
+	 * Time to wait for a response from the cluster description operation.
+	 */
+	private Duration responseTimeout = Duration.ofMillis(100);
+
+	public Duration getResponseTimeout() {
+		return this.responseTimeout;
+	}
+
+	public void setResponseTimeout(Duration responseTimeout) {
+		this.responseTimeout = responseTimeout;
+	}
+
+}

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

@@ -0,0 +1,20 @@
+/*
+ * Copyright 2012-2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Auto-configuration for actuator JDBC concerns.
+ */
+package org.springframework.boot.actuate.autoconfigure.kafka;

spring-boot-project/spring-boot-actuator-autoconfigure/src/main/resources/META-INF/additional-spring-configuration-metadata.json

@@ -157,6 +157,12 @@
       "description": "Whether to enable Neo4j health check.",
       "defaultValue": true
     },
+    {
+      "name": "management.health.kafka.enabled",
+      "type": "java.lang.Boolean",
+      "description": "Whether to enable kafka health check.",
+      "defaultValue": true
+    },
     {
       "name": "management.info.build.enabled",
       "type": "java.lang.Boolean",

spring-boot-project/spring-boot-actuator-autoconfigure/src/main/resources/META-INF/spring.factories

@@ -12,6 +12,7 @@ org.springframework.boot.actuate.autoconfigure.context.properties.ConfigurationP
 org.springframework.boot.actuate.autoconfigure.context.ShutdownEndpointAutoConfiguration,\
 org.springframework.boot.actuate.autoconfigure.couchbase.CouchbaseHealthIndicatorAutoConfiguration,\
 org.springframework.boot.actuate.autoconfigure.elasticsearch.ElasticsearchHealthIndicatorAutoConfiguration,\
+org.springframework.boot.actuate.autoconfigure.kafka.KafkaHealthIndicatorAutoConfiguration,\
 org.springframework.boot.actuate.autoconfigure.endpoint.EndpointAutoConfiguration,\
 org.springframework.boot.actuate.autoconfigure.endpoint.jmx.JmxEndpointAutoConfiguration,\
 org.springframework.boot.actuate.autoconfigure.endpoint.web.WebEndpointAutoConfiguration,\

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

@@ -0,0 +1,71 @@
+/*
+ * Copyright 2012-2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+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.AutoConfigureBefore;
+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}.
+ *
+ * @author Juan Rada
+ */
+public class KafkaHealthIndicatorAutoConfigurationTests {
+
+	private ApplicationContextRunner contextRunner = new ApplicationContextRunner()
+			.withConfiguration(AutoConfigurations.of(KafkaConfiguration.class,
+					KafkaHealthIndicatorAutoConfiguration.class,
+					HealthIndicatorAutoConfiguration.class));
+
+	@Test
+	public void runShouldCreateIndicator() {
+		this.contextRunner.run((context) -> assertThat(context)
+				.hasSingleBean(KafkaHealthIndicator.class)
+				.doesNotHaveBean(ApplicationHealthIndicator.class));
+	}
+
+	@Test
+	public void runWhenDisabledShouldNotCreateIndicator() {
+		this.contextRunner.withPropertyValues("management.health.kafka.enabled:false")
+				.run((context) -> assertThat(context)
+						.doesNotHaveBean(KafkaHealthIndicator.class)
+						.hasSingleBean(ApplicationHealthIndicator.class));
+	}
+
+	@Configuration
+	@AutoConfigureBefore(KafkaHealthIndicatorAutoConfiguration.class)
+	protected static class KafkaConfiguration {
+
+		@Bean
+		public AdminClient kafkaAdminClient() {
+			return mock(AdminClient.class);
+		}
+
+	}
+}

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

@@ -161,6 +161,11 @@
 			<artifactId>spring-rabbit</artifactId>
 			<optional>true</optional>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+			<optional>true</optional>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.data</groupId>
 			<artifactId>spring-data-cassandra</artifactId>

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

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2012-2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+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.springframework.boot.actuate.health.AbstractHealthIndicator;
+import org.springframework.boot.actuate.health.Health.Builder;
+import org.springframework.boot.actuate.health.HealthIndicator;
+import org.springframework.util.Assert;
+
+/**
+ * {@link HealthIndicator} for Kafka cluster.
+ *
+ * @author Juan Rada
+ */
+public class KafkaHealthIndicator extends AbstractHealthIndicator {
+
+	private final AdminClient adminClient;
+	private final DescribeClusterOptions describeOptions;
+
+	/**
+	 * Create a new {@link KafkaHealthIndicator} instance.
+	 *
+	 * @param adminClient the kafka admin client
+	 * @param responseTimeout the describe cluster request timeout in milliseconds
+	 */
+	public KafkaHealthIndicator(AdminClient adminClient, long responseTimeout) {
+		Assert.notNull(adminClient, "KafkaAdmin must not be null");
+
+		this.adminClient = adminClient;
+		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());
+	}
+}
+

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

@@ -0,0 +1,20 @@
+/*
+ * Copyright 2012-2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Actuator support for Kafka.
+ */
+package org.springframework.boot.actuate.kafka;