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.springframework.boot.autoconfigure.kafka.KafkaProperties;
   5 import org.springframework.boot.context.properties.EnableConfigurationProperties;
   6 import org.springframework.context.annotation.Bean;
   7 import org.springframework.context.annotation.Configuration;
   8 import org.springframework.kafka.core.ConsumerFactory;
   9
  10 import java.util.concurrent.ExecutorService;
  11 import java.util.concurrent.Executors;
  12 import java.util.function.Consumer;
  13
  14
  15 @Configuration
  16 @EnableConfigurationProperties({ KafkaProperties.class, ApplicationProperties.class })
  17 public class ApplicationConfiguration
  18 {
  19   @Bean
  20   public Consumer<ConsumerRecord<String, Long>> consumer()
  21   {
  22     return (record) ->
  23     {
  24       // Handle record
  25     };
  26   }
  27
  28   @Bean
  29   public EndlessConsumer<String, Long> endlessConsumer(
  30       org.apache.kafka.clients.consumer.Consumer<String, Long> kafkaConsumer,
  31       ExecutorService executor,
  32       Consumer<ConsumerRecord<String, Long>> handler,
  33       KafkaProperties kafkaProperties,
  34       ApplicationProperties applicationProperties)
  35   {
  36     return
  37         new EndlessConsumer<>(
  38             executor,
  39             kafkaProperties.getConsumer().getClientId(),
  40             applicationProperties.getTopic(),
  41             kafkaConsumer,
  42             handler);
  43   }
  44
  45   @Bean
  46   public ExecutorService executor()
  47   {
  48     return Executors.newSingleThreadExecutor();
  49   }
  50
  51   @Bean(destroyMethod = "close")
  52   public org.apache.kafka.clients.consumer.Consumer<String, Long> kafkaConsumer(ConsumerFactory<String, Long> factory)
  53   {
  54     return factory.createConsumer();
  55   }
  56 }