`HealthIndicator` mit Details zu Rebalances des Consumers implementiert

* `ConsumerHealthIndicator` refaktorisiert (DRY)
* `ConsumerHealthIndicator` refaktorisiert (Aufgaben klarer getrennt)
* `ConsumerHealthIndicator` refaktorisiert (Klareres Benennungs-Schema)
* `ConsumerHealthIndicator` refaktorisiert (Methoden aufgeräumt)

diff --git a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
index c417484..bdcd94e 100644 (file)
--- a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
+++ b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
@@ -11,6 +11,7 @@ import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
+import java.time.Clock;
 import java.util.Properties;
 
 
@@ -23,6 +24,7 @@ public class ApplicationConfiguration
   public ExampleConsumer<String, Long> exampleConsumer(
     Consumer<String, Long> kafkaConsumer,
     RecordHandler<String, Long> recordHandler,
+    ConsumerHealthIndicatorAwareRebalanceListener consumerHealthIndicatorAwareRebalanceListener,
     ApplicationProperties properties,
     ConfigurableApplicationContext applicationContext)
   {
@@ -32,6 +34,7 @@ public class ApplicationConfiguration
         properties.getConsumerProperties().getTopic(),
         kafkaConsumer,
         recordHandler,
+        consumerHealthIndicatorAwareRebalanceListener,
         () -> applicationContext.close());
   }
 
@@ -41,6 +44,24 @@ public class ApplicationConfiguration
     return (topic, partition, offset, key, value) -> log.info("No-Ops Handler called for {}={}", key, value);
   }
 
+  @Bean
+  public ConsumerHealthIndicatorAwareRebalanceListener rebalanceListener(ConsumerHealthIndicator consumerHealthIndicator)
+  {
+    return new ConsumerHealthIndicatorAwareRebalanceListener(consumerHealthIndicator);
+  }
+
+  @Bean
+  public ConsumerHealthIndicator consumerHealthIndicator(Clock clock)
+  {
+    return new ConsumerHealthIndicator(clock);
+  }
+
+  @Bean
+  public Clock clock()
+  {
+    return Clock.systemDefaultZone();
+  }
+
   @Bean(destroyMethod = "")
   public KafkaConsumer<String, Long> kafkaConsumer(ApplicationProperties properties)
   {

diff --git a/src/main/java/de/juplo/kafka/ConsumerHealthIndicator.java b/src/main/java/de/juplo/kafka/ConsumerHealthIndicator.java
new file mode 100644 (file)
index 0000000..30fb732
--- /dev/null
+++ b/src/main/java/de/juplo/kafka/ConsumerHealthIndicator.java
@@ -0,0 +1,116 @@
+package de.juplo.kafka;
+
+import org.apache.kafka.common.TopicPartition;
+import org.springframework.boot.actuate.health.Health;
+import org.springframework.boot.actuate.health.HealthIndicator;
+
+import java.time.Clock;
+import java.time.ZonedDateTime;
+import java.util.*;
+
+
+public class ConsumerHealthIndicator implements HealthIndicator
+{
+  private final Clock clock;
+
+  private volatile RebalancingState rebalancingState;
+  private volatile List<RecordedState> history;
+  private volatile List<Partition> assignedPartitions = List.of();
+
+
+  public ConsumerHealthIndicator(Clock clock)
+  {
+    this.clock = clock;
+
+    rebalancingState = RebalancingState.STARTING;
+    RecordedState sat = new RecordedState(ZonedDateTime.now(clock), rebalancingState, List.of());
+    history = List.of(sat);
+  }
+
+
+  @Override
+  public Health getHealth(boolean includeDetails)
+  {
+    Health.Builder healthBuilder = getHealthBuilder();
+
+    if (includeDetails)
+    {
+      healthBuilder.withDetail("rebalancing_state", rebalancingState);
+      healthBuilder.withDetail("history", history);
+    }
+
+    return healthBuilder.build();
+  }
+
+  @Override
+  public Health health()
+  {
+    return getHealthBuilder().build();
+  }
+
+  private Health.Builder getHealthBuilder()
+  {
+    return rebalancingState == RebalancingState.RUNNING
+      ? new Health.Builder().up()
+      : new Health.Builder().status(rebalancingState.name());
+  }
+
+
+  public void partitionsAssigned(Collection<TopicPartition> partitions)
+  {
+    List<Partition> newAssignedPartitions = new LinkedList<>(this.assignedPartitions);
+    partitions.forEach(tp -> newAssignedPartitions.add(new Partition(tp.topic(), tp.partition())));
+    Collections.sort(newAssignedPartitions, partitionComparator);
+
+    updateAndRecordState(RebalancingState.RUNNING, newAssignedPartitions);
+  }
+
+  public void partitionsRevoked(Collection<TopicPartition> partitions)
+  {
+    List<Partition> newAssignedPartitions = new LinkedList<>(this.assignedPartitions);
+    partitions.forEach(tp -> newAssignedPartitions.remove(new Partition(tp.topic(), tp.partition())));
+
+    updateAndRecordState(RebalancingState.REBALANCING, newAssignedPartitions);
+  }
+
+  public void partitionsLost(Collection<TopicPartition> partitions)
+  {
+    updateAndRecordState(RebalancingState.FENCED, List.of());
+  }
+
+  private void updateAndRecordState(
+    RebalancingState newRebalancingState,
+    List<Partition> newAssignedPartitions)
+  {
+    List<RecordedState> newHistory = new LinkedList<>();
+    newHistory.add(new RecordedState(
+      ZonedDateTime.now(clock),
+      newRebalancingState,
+      newAssignedPartitions));
+    newHistory.addAll(this.history);
+    if(newHistory.size() > 10)
+    {
+      newHistory.removeLast();
+    }
+    this.rebalancingState = newRebalancingState;
+    this.assignedPartitions = newAssignedPartitions;
+    this.history = newHistory;
+  }
+
+
+  enum RebalancingState { STARTING, FENCED, REBALANCING, RUNNING }
+
+  public record Partition(String topic, Integer partition) {}
+  public record RecordedState(
+    ZonedDateTime time,
+    RebalancingState rebalancingState,
+    List<Partition> assignedPartitions) {}
+
+  private final static Comparator<Partition> partitionComparator = (tp1, tp2) ->
+  {
+    int result = tp1.topic().compareTo(tp2.topic());
+    return result == 0
+      ? tp1.partition() - tp2.partition()
+      : result;
+  };
+}

diff --git a/src/main/java/de/juplo/kafka/ConsumerHealthIndicatorAwareRebalanceListener.java b/src/main/java/de/juplo/kafka/ConsumerHealthIndicatorAwareRebalanceListener.java
new file mode 100644 (file)
index 0000000..3339ac9
--- /dev/null
+++ b/src/main/java/de/juplo/kafka/ConsumerHealthIndicatorAwareRebalanceListener.java
@@ -0,0 +1,33 @@
+package de.juplo.kafka;
+
+import lombok.RequiredArgsConstructor;
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+import org.apache.kafka.common.TopicPartition;
+
+import java.util.Collection;
+
+
+@RequiredArgsConstructor
+public class ConsumerHealthIndicatorAwareRebalanceListener implements ConsumerRebalanceListener
+{
+  private final ConsumerHealthIndicator consumerHealthIndicator;
+
+
+  @Override
+  public void onPartitionsAssigned(Collection<TopicPartition> partitions)
+  {
+    consumerHealthIndicator.partitionsAssigned(partitions);
+  }
+
+  @Override
+  public void onPartitionsRevoked(Collection<TopicPartition> partitions)
+  {
+    consumerHealthIndicator.partitionsRevoked(partitions);
+  }
+
+  @Override
+  public void onPartitionsLost(Collection<TopicPartition> partitions)
+  {
+    consumerHealthIndicator.partitionsLost(partitions);
+  }
+}

diff --git a/src/main/java/de/juplo/kafka/ExampleConsumer.java b/src/main/java/de/juplo/kafka/ExampleConsumer.java
index b2c59f8..08a17d7 100644 (file)
--- a/src/main/java/de/juplo/kafka/ExampleConsumer.java
+++ b/src/main/java/de/juplo/kafka/ExampleConsumer.java
@@ -2,6 +2,7 @@ package de.juplo.kafka;
 
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.common.errors.RecordDeserializationException;
@@ -18,6 +19,7 @@ public class ExampleConsumer<K, V> implements Runnable
   private final String topic;
   private final Consumer<K, V> consumer;
   private final RecordHandler<K, V> recordHandler;
+  private final ConsumerRebalanceListener rebalanceListener;
   private final Thread workerThread;
   private final Runnable closeCallback;
 
@@ -30,12 +32,14 @@ public class ExampleConsumer<K, V> implements Runnable
     String topic,
     Consumer<K, V> consumer,
     RecordHandler<K, V> recordHandler,
+    ConsumerRebalanceListener rebalanceListener,
     Runnable closeCallback)
   {
     this.id = clientId;
     this.topic = topic;
     this.consumer = consumer;
     this.recordHandler = recordHandler;
+    this.rebalanceListener = rebalanceListener;
 
     workerThread = new Thread(this, "ExampleConsumer Worker-Thread");
     workerThread.start();
@@ -50,7 +54,7 @@ public class ExampleConsumer<K, V> implements Runnable
     try
     {
       log.info("{} - Subscribing to topic {}", id, topic);
-      consumer.subscribe(Arrays.asList(topic));
+      consumer.subscribe(Arrays.asList(topic), rebalanceListener);
       running = true;
 
       while (running)

diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml
index 7a06731..09791a7 100644 (file)
--- a/src/main/resources/application.yml
+++ b/src/main/resources/application.yml
@@ -8,6 +8,8 @@ juplo:
     auto-commit-interval: 5s
 management:
   endpoint:
+    health:
+      show-details: ALWAYS
     shutdown:
       enabled: true
   endpoints:

diff --git a/src/test/java/de/juplo/kafka/ExampleConsumerTest.java b/src/test/java/de/juplo/kafka/ExampleConsumerTest.java
index 590c9cd..1307d61 100644 (file)
--- a/src/test/java/de/juplo/kafka/ExampleConsumerTest.java
+++ b/src/test/java/de/juplo/kafka/ExampleConsumerTest.java
@@ -4,6 +4,7 @@ import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.AdminClientConfig;
 import org.apache.kafka.clients.admin.RecordsToDelete;
 import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.serialization.LongSerializer;
@@ -22,6 +23,7 @@ import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.test.context.EmbeddedKafka;
 
 import java.time.Duration;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
@@ -171,6 +173,14 @@ public class ExampleConsumerTest
       TOPIC,
       consumer,
       mockRecordHandler,
+      new ConsumerRebalanceListener()
+      {
+        @Override
+        public void onPartitionsRevoked(Collection<TopicPartition> collection) {}
+
+        @Override
+        public void onPartitionsAssigned(Collection<TopicPartition> collection) {}
+      },
       () -> isTerminatedExceptionally.set(true));
   }