#!/bin/bash
-IMAGE=juplo/spring-consumer:1.1-record-handler-SNAPSHOT
+IMAGE=juplo/spring-consumer:1.1-health-indicator-SNAPSHOT
if [ "$1" = "cleanup" ]
then
}
group = 'de.juplo.kafka'
-version = '1.1-record-handler-SNAPSHOT'
+version = '1.1-health-indicator-SNAPSHOT'
java {
toolchain {
command: kafka:9092 test producer
consumer:
- image: juplo/spring-consumer:1.1-record-handler-SNAPSHOT
+ image: juplo/spring-consumer:1.1-health-indicator-SNAPSHOT
environment:
spring.kafka.bootstrap-servers: kafka:9092
spring.kafka.client-id: consumer
juplo.consumer.topic: test
peter:
- image: juplo/spring-consumer:1.1-record-handler-SNAPSHOT
+ image: juplo/spring-consumer:1.1-health-indicator-SNAPSHOT
environment:
spring.kafka.bootstrap-servers: kafka:9092
spring.kafka.client-id: consumer
juplo.consumer.topic: test
ute:
- image: juplo/spring-consumer:1.1-record-handler-SNAPSHOT
+ image: juplo/spring-consumer:1.1-health-indicator-SNAPSHOT
environment:
spring.kafka.bootstrap-servers: kafka:9092
spring.kafka.client-id: consumer
<artifactId>spring-consumer</artifactId>
<name>Spring Consumer</name>
<description>Super Simple Consumer-Group, that is implemented as Spring-Boot application and configured by Spring Kafka</description>
- <version>1.1-record-handler-SNAPSHOT</version>
+ <version>1.1-health-indicator-SNAPSHOT</version>
<properties>
<java.version>21</java.version>
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.ConsumerFactory;
+import java.time.Clock;
+
@Configuration
@EnableConfigurationProperties(ApplicationProperties.class)
public ExampleConsumer<String, String> exampleConsumer(
Consumer<String, String> kafkaConsumer,
RecordHandler<String, String> recordHandler,
+ ConsumerHealthIndicatorAwareRebalanceListener consumerHealthIndicatorAwareRebalanceListener,
ApplicationProperties properties,
KafkaProperties kafkaProperties,
ConfigurableApplicationContext applicationContext)
properties.getConsumerProperties().getTopic(),
kafkaConsumer,
recordHandler,
+ consumerHealthIndicatorAwareRebalanceListener,
() -> applicationContext.close());
}
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 Consumer<?, ?> kafkaConsumer(ConsumerFactory<?, ?> consumerFactory)
{
--- /dev/null
+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;
+ };
+}
--- /dev/null
+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);
+ }
+}
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.WakeupException;
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;
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();
try
{
log.info("{} - Subscribing to topic {}", id, topic);
- consumer.subscribe(Arrays.asList(topic));
+ consumer.subscribe(Arrays.asList(topic), rebalanceListener);
while (true)
{
topic: test
management:
endpoint:
+ health:
+ show-details: ALWAYS
shutdown:
enabled: true
endpoints:
projects / demos / kafka / training / commitdiff