Added example to identify if kafka error is in brokers or in topic (#2778)

* Added example to identify Kafka error

* removed implementation and added doc links and added bean example

* changed doc link to the official doc

* Added custom message in the down status of the actuator
This commit is contained in:
Claudio Silva Junior
2023-07-31 11:39:44 -03:00
committed by GitHub
parent 0a02ffcb0a
commit b0fb7740b7

View File

@@ -1016,3 +1016,59 @@ public KafkaBinderHealth kafkaBinderHealthIndicator() {
};
}
```
[[custom-kafka-binder-health-indicator-example]]
=== Custom kafka Binder Health Indicator Example
Here is the pseudo-code for writing a custom Kafka binder HealthIndicator.
In this example, we try to override the binder provided Kafka HealthIndicator by specifically checking first for cluster connectivity and then followed by topic-related issues.
1. First we need create a custom implementation of the `KafkaBinderHealth` interface.
```
public class KafkaBinderHealthImplementation implements KafkaBinderHealth {
@Value("${spring.cloud.bus.destination}")
private String topic;
private final AdminClient client;
public KafkaBinderHealthImplementation(final KafkaAdmin admin) {
// More about configuring Kafka
// https://docs.spring.io/spring-kafka/reference/html/#configuring-topics
this.client = AdminClient.create(admin.getConfigurationProperties());
}
@Override
public Health health() {
if (!checkBrokersConnection()) {
logger.error("Error when connect brokers");
return Health.down().withDetail("BrokersConnectionError", "Error message").build();
}
if (!checkTopicConnection()) {
logger.error("Error when trying to connect with specific topic");
return Health.down().withDetail("TopicError", "Error message with topic name").build();
}
return Health.up().build();
}
public boolean checkBrokersConnection() {
// Your implementation
}
public boolean checkTopicConnection() {
// Your implementation
}
}
```
2. Then we need to create a bean for the custom implementation.
```
@Configuration
public class KafkaBinderHealthIndicatorConfiguration {
@Bean
public KafkaBinderHealth kafkaBinderHealthIndicator(final KafkaAdmin admin) {
return new KafkaBinderHealthImplementation(admin);
}
}
```