Kafka health checks and topics replication (#774)

* Factor in `min.insync.replicas` when calculating min node count * Update unit tests
micronaut-projects · Jul 27, 2023 · 7dd14ff · 7dd14ff
1 parent 57febe3
commit 7dd14ff
Show file tree

Hide file tree

Showing 2 changed files with 30 additions and 9 deletions.
diff --git a/kafka/src/main/java/io/micronaut/configuration/kafka/health/KafkaHealthIndicator.java b/kafka/src/main/java/io/micronaut/configuration/kafka/health/KafkaHealthIndicator.java
@@ -52,6 +52,7 @@
 public class KafkaHealthIndicator implements HealthIndicator {
 
     private static final String ID = "kafka";
+    private static final String MIN_INSYNC_REPLICAS_PROPERTY = "min.insync.replicas";
     private static final String REPLICATION_PROPERTY = "offsets.topic.replication.factor";
     private static final String DEFAULT_REPLICATION_PROPERTY = "default.replication.factor";
     private final AdminClient adminClient;
@@ -80,6 +81,19 @@ public static int getClusterReplicationFactor(Config config) {
         return ce != null ? Integer.parseInt(ce.value()) : Integer.MAX_VALUE;
     }
 
+    /**
+     * Retrieve the cluster "min.insync.replicas" for the given {@link Config}, falling back to
+     * "offsets.topic.replication.factor" or "default.replication.factor" if required, in order to
+     * support Confluent Cloud hosted Kafka.
+     *
+     * @param config the cluster {@link Config}
+     * @return the optional cluster minimum number of replicas that must acknowledge a write
+     */
+    public static int getMinNodeCount(Config config) {
+        return Optional.ofNullable(config.get(MIN_INSYNC_REPLICAS_PROPERTY)).map(ConfigEntry::value).map(Integer::parseInt)
+            .orElseGet(() -> getClusterReplicationFactor(config));
+    }
+
     @Override
     public Flux<HealthResult> getResult() {
         DescribeClusterResult result = adminClient.describeCluster(
@@ -99,11 +113,11 @@ public Flux<HealthResult> getResult() {
             Mono<Map<ConfigResource, Config>> configs = KafkaReactorUtil.fromKafkaFuture(configResult::all);
             return configs.flux().switchMap(resources -> {
                 Config config = resources.get(configResource);
-                int replicationFactor = getClusterReplicationFactor(config);
+                int minNodeCount = getMinNodeCount(config);
                 return nodes.flux().switchMap(nodeList -> clusterId.map(clusterIdString -> {
                     int nodeCount = nodeList.size();
                     HealthResult.Builder builder;
-                    if (nodeCount >= replicationFactor) {
+                    if (nodeCount >= minNodeCount) {
                         builder = HealthResult.builder(ID, HealthStatus.UP);
                     } else {
                         builder = HealthResult.builder(ID, HealthStatus.DOWN);

diff --git a/...a/src/test/groovy/io/micronaut/configuration/kafka/health/KafkaHealthIndicatorSpec.groovy b/...a/src/test/groovy/io/micronaut/configuration/kafka/health/KafkaHealthIndicatorSpec.groovy
@@ -12,6 +12,7 @@ import spock.lang.Specification
 import spock.lang.Unroll
 
 import static io.micronaut.configuration.kafka.health.KafkaHealthIndicator.DEFAULT_REPLICATION_PROPERTY
+import static io.micronaut.configuration.kafka.health.KafkaHealthIndicator.MIN_INSYNC_REPLICAS_PROPERTY
 import static io.micronaut.configuration.kafka.health.KafkaHealthIndicator.REPLICATION_PROPERTY
 import static io.micronaut.health.HealthStatus.DOWN
 import static io.micronaut.health.HealthStatus.UP
@@ -31,7 +32,6 @@ class KafkaHealthIndicatorSpec extends Specification {
         HealthResult result = healthIndicator.result.next().block()
 
         then:
-        // report down because the not enough nodes to meet replication factor
         result.status == UP
         result.details.nodes == 1
 
@@ -86,6 +86,9 @@ class KafkaHealthIndicatorSpec extends Specification {
     void "kafka health indicator handle missing replication factor config"() {
         given:
         Collection<ConfigEntry> configEntries = []
+        if (minReplicas) {
+            configEntries << new ConfigEntry(MIN_INSYNC_REPLICAS_PROPERTY, minReplicas)
+        }
         if (offsetFactor) {
             configEntries << new ConfigEntry(REPLICATION_PROPERTY, offsetFactor)
         }
@@ -95,16 +98,20 @@ class KafkaHealthIndicatorSpec extends Specification {
         Config config = new Config(configEntries)
 
         when:
-        int replicationFactor = KafkaHealthIndicator.getClusterReplicationFactor(config)
+        int replicationFactor = KafkaHealthIndicator.getMinNodeCount(config)
 
         then:
         replicationFactor == expected
 
         where:
-        offsetFactor | defaultFactor | expected
-        "10"         | null          | 10
-        "10"         | "8"           | 10
-        null         | "8"           | 8
-        null         | null          | Integer.MAX_VALUE
+        minReplicas | offsetFactor | defaultFactor | expected
+        "100"       | "10"         | null          | 100
+        "100"       | "10"         | "8"           | 100
+        "100"       | null         | "8"           | 100
+        "100"       | null         | null          | 100
+        null        | "10"         | null          | 10
+        null        | "10"         | "8"           | 10
+        null        | null         | "8"           | 8
+        null        | null         | null          | Integer.MAX_VALUE
     }
 }