From b0fb7740b780460c2ec409186f63b0233281823b Mon Sep 17 00:00:00 2001
From: Claudio Silva Junior <42524939+Claudio-code@users.noreply.github.com>
Date: Mon, 31 Jul 2023 11:39:44 -0300
Subject: [PATCH] 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
---
 .../main/asciidoc/kafka/kafka_overview.adoc   | 56 +++++++++++++++++++
 1 file changed, 56 insertions(+)

diff --git a/docs/src/main/asciidoc/kafka/kafka_overview.adoc b/docs/src/main/asciidoc/kafka/kafka_overview.adoc
index 2eb4acfb5..fba71214c 100644
--- a/docs/src/main/asciidoc/kafka/kafka_overview.adoc
+++ b/docs/src/main/asciidoc/kafka/kafka_overview.adoc
@@ -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);
+	}
+}
+```