Add Kafka Health Indicator

In previous versions of Spring-Boot there was an inbuild health indicator for Kafka, however somewhere along the way it was lost.

Refs:

• https://github.com/spring-projects/spring-boot/pull/11515
• KafkaHealthIndicator

Please add the HealthIndicator for Kafka again and add metrics as well. This can be achieved using the following code:

(includes both metrics and health)

@Configuration
public class KafkaConfig {

	@Autowired
	private KafkaAdmin admin;

	@Autowired
	private MeterRegistry meterRegistry;

	@Autowired
	private Map<String, KafkaTemplate<?, ?>> kafkaTemplates;

	@Bean
	public AdminClient kafkaAdminClient() {
		return AdminClient.create(admin.getConfig());
	}

	@SuppressWarnings("deprecation") // Can be avoided by relying on Double.NaN for non doubles.
	@PostConstruct
	private void initMetrics() {
		final String kafkaPrefix = "kafka.";
		for (Entry<String, KafkaTemplate<?, ?>> templateEntry : kafkaTemplates.entrySet()) {
			final String name = templateEntry.getKey();
			final KafkaTemplate<?, ?> kafkaTemplate = templateEntry.getValue();
			for (Metric metric : kafkaTemplate.metrics().values()) {
				final MetricName metricName = metric.metricName();
				final Builder<Metric> gaugeBuilder = Gauge
						.builder(kafkaPrefix + metricName.name(), metric, Metric::value) // <-- Here
						.description(metricName.description());
				for (Entry<String, String> tagEntry : metricName.tags().entrySet()) {
					gaugeBuilder.tag(kafkaPrefix + tagEntry.getKey(), tagEntry.getValue());
				}
				gaugeBuilder.tag("bean", name);
				gaugeBuilder.register(meterRegistry);
			}
		}
	}

	@Bean
	public HealthIndicator kafkaHealthIndicator() {
		final DescribeClusterOptions describeClusterOptions = new DescribeClusterOptions().timeoutMs(1000);
		final AdminClient adminClient = kafkaAdminClient();
		return () -> {
			final DescribeClusterResult describeCluster = adminClient.describeCluster(describeClusterOptions);
			try {
				final String clusterId = describeCluster.clusterId().get();
				final int nodeCount = describeCluster.nodes().get().size();
				return Health.up()
						.withDetail("clusterId", clusterId)
						.withDetail("nodeCount", nodeCount)
						.build();
			} catch (InterruptedException | ExecutionException e) {
				return Health.down()
						.withException(e)
						.build();
			}
		};

	}

}

Feel free to use or modify the code as you see fit.