juplo.de Git - demos/kafka/training/blob - src/main/java/de/juplo/kafka/ApplicationConfiguration.java

   1 package de.juplo.kafka;
   2
   3 import org.apache.kafka.clients.consumer.ConsumerRecord;
   4 import org.apache.kafka.common.TopicPartition;
   5 import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
   6 import org.springframework.boot.context.properties.EnableConfigurationProperties;
   7 import org.springframework.context.annotation.Bean;
   8 import org.springframework.context.annotation.Configuration;
   9 import org.springframework.kafka.core.ConsumerFactory;
  10 import org.springframework.kafka.core.KafkaOperations;
  11 import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
  12 import org.springframework.kafka.listener.DefaultErrorHandler;
  13 import org.springframework.util.backoff.FixedBackOff;
  14
  15 import java.util.function.Consumer;
  16
  17
  18 @Configuration
  19 @EnableConfigurationProperties({ KafkaProperties.class, ApplicationProperties.class })
  20 public class ApplicationConfiguration
  21 {
  22   @Bean
  23   public Consumer<ConsumerRecord<String, ClientMessage>> consumer()
  24   {
  25     return (record) ->
  26     {
  27       // Handle record
  28     };
  29   }
  30
  31   @Bean
  32   public DeadLetterPublishingRecoverer recoverer(
  33       ApplicationProperties properties,
  34       KafkaOperations<?, ?> template)
  35   {
  36     return new DeadLetterPublishingRecoverer(
  37         template,
  38         (record, exception) -> new TopicPartition(properties.getDlqTopic(), record.partition()));
  39   }
  40
  41   @Bean
  42   public DefaultErrorHandler errorHandler(DeadLetterPublishingRecoverer recoverer)
  43   {
  44     return new DefaultErrorHandler(recoverer, new FixedBackOff(0l, 0l));
  45   }
  46
  47   @Bean(destroyMethod = "close")
  48   public org.apache.kafka.clients.consumer.Consumer<String, ClientMessage> kafkaConsumer(ConsumerFactory<String, ClientMessage> factory)
  49   {
  50     return factory.createConsumer();
  51   }
  52 }