From: Kai Moritz <kai@juplo.de>
Date: Fri, 22 Apr 2022 15:15:36 +0000 (+0200)
Subject: Springify: Merge der Umstellung auf die Auto-Konfiguration von Spring-Boot
X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=02bf5ef5c451eab56f01b2bd82941587383c0ede;p=demos%2Fkafka%2Ftraining

Springify: Merge der Umstellung auf die Auto-Konfiguration von Spring-Boot
---

02bf5ef5c451eab56f01b2bd82941587383c0ede
diff --cc src/main/java/de/juplo/kafka/EndlessConsumer.java
index 15e1b4e,b173b12..09d08da
--- a/src/main/java/de/juplo/kafka/EndlessConsumer.java
+++ b/src/main/java/de/juplo/kafka/EndlessConsumer.java
@@@ -2,81 -2,285 +2,81 @@@ package de.juplo.kafka
  
  import lombok.RequiredArgsConstructor;
  import lombok.extern.slf4j.Slf4j;
 -import org.apache.kafka.clients.consumer.*;
 -import org.apache.kafka.common.TopicPartition;
 -import org.apache.kafka.common.errors.RecordDeserializationException;
 -import org.apache.kafka.common.errors.WakeupException;
 +import org.apache.kafka.clients.consumer.ConsumerRecord;
 +import org.springframework.beans.factory.annotation.Autowired;
 +import org.springframework.beans.factory.annotation.Value;
 +import org.springframework.kafka.annotation.KafkaListener;
 +import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
 +import org.springframework.stereotype.Component;
  
 -import javax.annotation.PreDestroy;
 -import java.time.Duration;
 -import java.util.*;
 -import java.util.concurrent.ExecutionException;
 -import java.util.concurrent.ExecutorService;
 -import java.util.concurrent.locks.Condition;
 -import java.util.concurrent.locks.Lock;
 -import java.util.concurrent.locks.ReentrantLock;
 +import java.util.Optional;
 +import java.util.function.Consumer;
  
  
 +@Component
  @Slf4j
  @RequiredArgsConstructor
 -public class EndlessConsumer<K, V> implements Runnable
 +public class EndlessConsumer<K, V>
  {
 -  private final ExecutorService executor;
 -  private final String id;
 -  private final String topic;
 -  private final Consumer<K, V> consumer;
 -  private final java.util.function.Consumer<ConsumerRecord<K, V>> handler;
 +  @Autowired
 +  private KafkaListenerEndpointRegistry registry;
-   @Value("${consumer.client-id}")
++  @Value("${spring.kafka.consumer.client-id}")
 +  String id;
 +  @Autowired
 +  Consumer<ConsumerRecord<K, V>> handler;
 +  @Autowired
 +  ApplicationErrorHandler errorHandler;
  
 -  private final Lock lock = new ReentrantLock();
 -  private final Condition condition = lock.newCondition();
 -  private boolean running = false;
 -  private Exception exception;
    private long consumed = 0;
  
 -  private final Map<Integer, Map<String, Long>> seen = new HashMap<>();
 -  private final Map<Integer, Long> offsets = new HashMap<>();
 -
 -
 -  @Override
 -  public void run()
 +  @KafkaListener(
-       id = "${consumer.client-id}",
++      id = "${spring.kafka.consumer.client-id}",
 +      idIsGroup = false,
 +      topics = "${consumer.topic}",
 +      autoStartup = "false")
 +  public void receive(ConsumerRecord<K, V> record)
    {
 -    try
 -    {
 -      log.info("{} - Subscribing to topic {}", id, topic);
 -      consumer.subscribe(Arrays.asList(topic), new ConsumerRebalanceListener()
 -      {
 -        @Override
 -        public void onPartitionsRevoked(Collection<TopicPartition> partitions)
 -        {
 -          partitions.forEach(tp ->
 -          {
 -            Integer partition = tp.partition();
 -            Long newOffset = consumer.position(tp);
 -            Long oldOffset = offsets.remove(partition);
 -            log.info(
 -                "{} - removing partition: {}, consumed {} records (offset {} -> {})",
 -                id,
 -                partition,
 -                newOffset - oldOffset,
 -                oldOffset,
 -                newOffset);
 -            Map<String, Long> removed = seen.remove(partition);
 -            for (String key : removed.keySet())
 -            {
 -              log.info(
 -                  "{} - Seen {} messages for partition={}|key={}",
 -                  id,
 -                  removed.get(key),
 -                  partition,
 -                  key);
 -            }
 -          });
 -        }
 -
 -        @Override
 -        public void onPartitionsAssigned(Collection<TopicPartition> partitions)
 -        {
 -          partitions.forEach(tp ->
 -          {
 -            Integer partition = tp.partition();
 -            Long offset = consumer.position(tp);
 -            log.info("{} - adding partition: {}, offset={}", id, partition, offset);
 -            offsets.put(partition, offset);
 -            seen.put(partition, new HashMap<>());
 -          });
 -        }
 -      });
 -
 -      while (true)
 -      {
 -        ConsumerRecords<K, V> records =
 -            consumer.poll(Duration.ofSeconds(1));
 -
 -        // Do something with the data...
 -        log.info("{} - Received {} messages", id, records.count());
 -        for (ConsumerRecord<K, V> record : records)
 -        {
 -          log.info(
 -              "{} - {}: {}/{} - {}={}",
 -              id,
 -              record.offset(),
 -              record.topic(),
 -              record.partition(),
 -              record.key(),
 -              record.value()
 -          );
 -
 -          handler.accept(record);
 -
 -          consumed++;
 -
 -          Integer partition = record.partition();
 -          String key = record.key() == null ? "NULL" : record.key().toString();
 -          Map<String, Long> byKey = seen.get(partition);
 -
 -          if (!byKey.containsKey(key))
 -            byKey.put(key, 0l);
 -
 -          long seenByKey = byKey.get(key);
 -          seenByKey++;
 -          byKey.put(key, seenByKey);
 -        }
 -      }
 -    }
 -    catch(WakeupException e)
 -    {
 -      log.info("{} - RIIING! Request to stop consumption - commiting current offsets!", id);
 -      consumer.commitSync();
 -      shutdown();
 -    }
 -    catch(RecordDeserializationException e)
 -    {
 -      TopicPartition tp = e.topicPartition();
 -      long offset = e.offset();
 -      log.error(
 -          "{} - Could not deserialize  message on topic {} with offset={}: {}",
 -          id,
 -          tp,
 -          offset,
 -          e.getCause().toString());
 -
 -      consumer.commitSync();
 -      shutdown(e);
 -    }
 -    catch(Exception e)
 -    {
 -      log.error("{} - Unexpected error: {}", id, e.toString(), e);
 -      shutdown(e);
 -    }
 -    finally
 -    {
 -      log.info("{} - Consumer-Thread exiting", id);
 -    }
 +    log.info(
 +        "{} - {}: {}/{} - {}={}",
 +        id,
 +        record.offset(),
 +        record.topic(),
 +        record.partition(),
 +        record.key(),
 +        record.value()
 +    );
 +
 +    handler.accept(record);
 +
 +    consumed++;
    }
  
 -  private void shutdown()
 -  {
 -    shutdown(null);
 -  }
  
 -  private void shutdown(Exception e)
 +  public synchronized void start()
    {
 -    lock.lock();
 -    try
 -    {
 -      try
 -      {
 -        log.info("{} - Unsubscribing from topic {}", id, topic);
 -        consumer.unsubscribe();
 -      }
 -      catch (Exception ue)
 -      {
 -        log.error(
 -            "{} - Error while unsubscribing from topic {}: {}",
 -            id,
 -            topic,
 -            ue.toString());
 -      }
 -      finally
 -      {
 -        running = false;
 -        exception = e;
 -        condition.signal();
 -      }
 -    }
 -    finally
 -    {
 -      lock.unlock();
 -    }
 -  }
 -
 -  public Map<Integer, Map<String, Long>> getSeen()
 -  {
 -    return seen;
 -  }
 -
 -  public void start()
 -  {
 -    lock.lock();
 -    try
 -    {
 -      if (running)
 -        throw new IllegalStateException("Consumer instance " + id + " is already running!");
 +    if (registry.getListenerContainer(id).isChildRunning())
 +      throw new IllegalStateException("Consumer instance " + id + " is already running!");
  
 -      log.info("{} - Starting - consumed {} messages before", id, consumed);
 -      running = true;
 -      exception = null;
 -      executor.submit(this);
 -    }
 -    finally
 -    {
 -      lock.unlock();
 -    }
 +    log.info("{} - Starting - consumed {} messages before", id, consumed);
 +    errorHandler.clearException();
 +    registry.getListenerContainer(id).start();
    }
  
 -  public synchronized void stop() throws ExecutionException, InterruptedException
 +  public synchronized void stop()
    {
 -    lock.lock();
 -    try
 -    {
 -      if (!running)
 -        throw new IllegalStateException("Consumer instance " + id + " is not running!");
 +    if (!registry.getListenerContainer(id).isChildRunning())
 +      throw new IllegalStateException("Consumer instance " + id + " is not running!");
  
 -      log.info("{} - Stopping", id);
 -      consumer.wakeup();
 -      condition.await();
 -      log.info("{} - Stopped - consumed {} messages so far", id, consumed);
 -    }
 -    finally
 -    {
 -      lock.unlock();
 -    }
 -  }
 -
 -  @PreDestroy
 -  public void destroy() throws ExecutionException, InterruptedException
 -  {
 -    log.info("{} - Destroy!", id);
 -    try
 -    {
 -      stop();
 -    }
 -    catch (IllegalStateException e)
 -    {
 -      log.info("{} - Was already stopped", id);
 -    }
 -    catch (Exception e)
 -    {
 -      log.error("{} - Unexpected exception while trying to stop the consumer", id, e);
 -    }
 -    finally
 -    {
 -      log.info("{}: Consumed {} messages in total, exiting!", id, consumed);
 -    }
 -  }
 -
 -  public boolean running()
 -  {
 -    lock.lock();
 -    try
 -    {
 -      return running;
 -    }
 -    finally
 -    {
 -      lock.unlock();
 -    }
 +    log.info("{} - Stopping", id);
 +    registry.getListenerContainer(id).stop();
 +    log.info("{} - Stopped - consumed {} messages so far", id, consumed);
    }
  
 -  public Optional<Exception> exitStatus()
 +  public synchronized Optional<Exception> exitStatus()
    {
 -    lock.lock();
 -    try
 -    {
 -      if (running)
 -        throw new IllegalStateException("No exit-status available: Consumer instance " + id + " is running!");
 +    if (registry.getListenerContainer(id).isChildRunning())
 +      throw new IllegalStateException("No exit-status available: Consumer instance " + id + " is running!");
  
 -      return Optional.ofNullable(exception);
 -    }
 -    finally
 -    {
 -      lock.unlock();
 -    }
 +    return errorHandler.getException();
    }
  }
diff --cc src/main/resources/application.yml
index 081fe1a,07fb162..43cde87
--- a/src/main/resources/application.yml
+++ b/src/main/resources/application.yml
@@@ -27,14 -23,11 +23,14 @@@ info
  spring:
    kafka:
      consumer:
-       bootstrap-servers: ${consumer.bootstrap-server}
-       client-id: ${consumer.client-id}
-       auto-offset-reset: ${consumer.auto-offset-reset}
-       group-id: ${consumer.group-id}
+       bootstrap-servers: :9092
+       client-id: DEV
+       auto-offset-reset: earliest
+       group-id: my-group
 -      value-deserializer: org.apache.kafka.common.serialization.LongDeserializer
 +      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
 +      properties:
 +        spring.json.type.mapping: "message:de.juplo.kafka.ClientMessage"
 +        spring.json.trusted.packages: "de.juplo.kafka"
  logging:
    level:
      root: INFO
diff --cc src/test/java/de/juplo/kafka/ApplicationTests.java
index ee8e8c4,a185b72..3f6a6a8
--- a/src/test/java/de/juplo/kafka/ApplicationTests.java
+++ b/src/test/java/de/juplo/kafka/ApplicationTests.java
@@@ -39,8 -39,7 +40,8 @@@ import static org.awaitility.Awaitility
  
  @SpringJUnitConfig(
  		initializers = ConfigDataApplicationContextInitializer.class,
-     classes = {
+ 		classes = {
 +				EndlessConsumer.class,
  				KafkaAutoConfiguration.class,
  				ApplicationTests.Configuration.class })
  @TestMethodOrder(MethodOrderer.OrderAnnotation.class)
@@@ -63,19 -62,21 +64,21 @@@ class ApplicationTest
  	@Autowired
  	KafkaProducer<String, Bytes> kafkaProducer;
  	@Autowired
- 	KafkaConsumer<String, ClientMessage> kafkaConsumer;
 -	org.apache.kafka.clients.consumer.Consumer<String, Long> kafkaConsumer;
++	org.apache.kafka.clients.consumer.Consumer<String, ClientMessage> kafkaConsumer;
  	@Autowired
  	KafkaConsumer<Bytes, Bytes> offsetConsumer;
  	@Autowired
- 	ApplicationProperties properties;
+ 	ApplicationProperties applicationProperties;
+ 	@Autowired
+ 	KafkaProperties kafkaProperties;
  	@Autowired
 -	ExecutorService executor;
 +	EndlessConsumer endlessConsumer;
 +	@Autowired
 +	RecordHandler recordHandler;
  
 -	Consumer<ConsumerRecord<String, Long>> testHandler;
 -	EndlessConsumer<String, Long> endlessConsumer;
  	Map<TopicPartition, Long> oldOffsets;
  	Map<TopicPartition, Long> newOffsets;
 -	Set<ConsumerRecord<String, Long>> receivedRecords;
 +	Set<ConsumerRecord<String, ClientMessage>> receivedRecords;
  
  
  	/** Tests methods */