`HealthIndicator` mit Details zu Rebalances des Consumers implementiert

diff --git a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
index 107c342..8308b00 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 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..47d7593
--- /dev/null
+++ b/src/main/java/de/juplo/kafka/ConsumerHealthIndicator.java
@@ -0,0 +1,122 @@
+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 State state;
+  private volatile List<StateAtTime> history;
+  private volatile List<Partition> assignedPartitions;
+
+
+  public ConsumerHealthIndicator(Clock clock)
+  {
+    this.clock = clock;
+
+    state = State.STARTING;
+    StateAtTime sat = new StateAtTime(ZonedDateTime.now(clock), state, List.of());
+    history = List.of(sat);
+  }
+
+
+  @Override
+  public Health getHealth(boolean includeDetails)
+  {
+    Health.Builder healthBuilder = getHealthBuilder();
+
+    if (includeDetails)
+    {
+      healthBuilder.withDetail("status", state);
+      healthBuilder.withDetail("history", history);
+    }
+
+    return healthBuilder.build();
+  }
+
+  @Override
+  public Health health()
+  {
+    return getHealthBuilder().build();
+  }
+
+
+  public void partitionsAssigned(Collection<TopicPartition> partitions)
+  {
+    this.state = State.RUNNING;
+    List<Partition> assignedPartitions = new LinkedList<>();
+    assignedPartitions.addAll(assignedPartitions);
+    partitions.forEach(tp -> assignedPartitions.add(new Partition(tp.topic(), tp.partition())));
+    Collections.sort(assignedPartitions, partitionComparator);
+    this.assignedPartitions = assignedPartitions;
+    List<StateAtTime> history = new LinkedList<>();
+    history.add(new StateAtTime(ZonedDateTime.now(clock), state, assignedPartitions));
+    history.addAll(this.history);
+    if(history.size() > 10)
+    {
+      history.removeLast();
+    }
+    this.history = history;
+  }
+
+  public void partitionsRevoked(Collection<TopicPartition> partitions)
+  {
+    this.state = State.REBALANCING;
+    List<Partition> assignedPartitions = new LinkedList<>();
+    assignedPartitions.addAll(assignedPartitions);
+    partitions.forEach(tp -> assignedPartitions.remove(new Partition(tp.topic(), tp.partition())));
+    this.assignedPartitions = assignedPartitions;
+    List<StateAtTime> history = new LinkedList<>();
+    history.add(new StateAtTime(ZonedDateTime.now(clock), state, assignedPartitions));
+    history.addAll(this.history);
+    if(history.size() > 10)
+    {
+      history.removeLast();
+    }
+    this.history = history;
+  }
+
+  public void partitionsLost(Collection<TopicPartition> partitions)
+  {
+    this.state = State.FENCED;
+    this.assignedPartitions = List.of();
+    List<StateAtTime> history = new LinkedList<>();
+    history.add(new StateAtTime(ZonedDateTime.now(clock), state, assignedPartitions));
+    history.addAll(this.history);
+    if(history.size() > 10)
+    {
+      history.removeLast();
+    }
+    this.history = history;
+  }
+
+
+  private Health.Builder getHealthBuilder()
+  {
+    return state == State.RUNNING
+      ? new Health.Builder().up()
+      : new Health.Builder().status(state.name());
+  }
+
+  enum State { STARTING, FENCED, REBALANCING, RUNNING }
+
+  public record Partition(String topic, Integer partition) {}
+  public record StateAtTime(ZonedDateTime time, State state, 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 149395a..8b2611c 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 implements Runnable
   private final String topic;
   private final Consumer<String, Long> consumer;
   private final RecordHandler<String, Long> recordHandler;
+  private final ConsumerRebalanceListener rebalanceListener;
   private final Thread workerThread;
   private final Runnable closeCallback;
 
@@ -30,12 +32,14 @@ public class ExampleConsumer implements Runnable
     String topic,
     Consumer<String, Long> consumer,
     RecordHandler<String, Long> 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 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: