I have something like below which works well, but I would prefer checking health without sending any message, (not only checking socket connection). I know Kafka has something like KafkaHealthIndicator out of the box, does someone have experience or example using it ?
public class KafkaHealthIndicator implements HealthIndicator {
private final Logger log = LoggerFactory.getLogger(KafkaHealthIndicator.class);
private KafkaTemplate<String, String> kafka;
public KafkaHealthIndicator(KafkaTemplate<String, String> kafka) {
this.kafka = kafka;
}
@Override
public Health health() {
try {
kafka.send("kafka-health-indicator", "❥").get(100, TimeUnit.MILLISECONDS);
} catch (InterruptedException | ExecutionException | TimeoutException e) {
return Health.down(e).build();
}
return Health.up().build();
}
}
kafkaAdminClient.describeCluster(..) is point where Kafka availability is being tested.
@Configuration
public class KafkaConfig {
@Autowired
private KafkaAdmin kafkaAdmin;
@Bean
public AdminClient kafkaAdminClient() {
return AdminClient.create(kafkaAdmin.getConfigurationProperties());
}
@Bean
public HealthIndicator kafkaHealthIndicator(AdminClient kafkaAdminClient) {
final DescribeClusterOptions options = new DescribeClusterOptions()
.timeoutMs(1000);
return new AbstractHealthIndicator() {
@Override
protected void doHealthCheck(Health.Builder builder) throws Exception {
// When Kafka is not connected, describeCluster() method throws
// an exception which in turn sets this indicator as being DOWN.
kafkaAdminClient.describeCluster(options);
builder.up().build();
}
};
}
}
For more verbose probe add:
DescribeClusterResult clusterDesc = kafkaAdminClient.describeCluster(options);
builder.up()
.withDetail("clusterId", clusterDesc.clusterId().get())
.withDetail("nodeCount", clusterDesc.nodes().get().size())
.build();
Use the AdminClient API to check the health of the cluster via describing the cluster and/or the topic(s) you'll be interacting with, and verifying those topics have the required number of insync replicas, for example
Kafka has something like KafkaHealthIndicator out of the box
It doesn't. Spring's Kafka integration might