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.KafkaConsumer;
   4 import org.apache.kafka.clients.producer.KafkaProducer;
   5 import org.apache.kafka.common.serialization.IntegerDeserializer;
   6 import org.apache.kafka.common.serialization.StringDeserializer;
   7 import org.apache.kafka.common.serialization.StringSerializer;
   8 import org.springframework.boot.context.properties.EnableConfigurationProperties;
   9 import org.springframework.context.annotation.Bean;
  10 import org.springframework.context.annotation.Configuration;
  11 import org.springframework.kafka.support.serializer.JsonSerializer;
  12 import java.util.Properties;
  13 import java.util.concurrent.ExecutorService;
  14 import java.util.concurrent.Executors;
  15
  16
  17 @Configuration
  18 @EnableConfigurationProperties(ApplicationProperties.class)
  19 public class ApplicationConfiguration
  20 {
  21   @Bean
  22   public ApplicationRecordHandler recordHandler(
  23       KafkaProducer<String, Object> kafkaProducer,
  24       ApplicationProperties properties)
  25   {
  26     return new ApplicationRecordHandler(
  27         kafkaProducer,
  28         properties.getClientId(),
  29         properties.getTopicOut());
  30   }
  31
  32   @Bean
  33   public EndlessConsumer<String, Integer> endlessConsumer(
  34       KafkaConsumer<String, Integer> kafkaConsumer,
  35       ExecutorService executor,
  36       ApplicationRecordHandler recordHandler,
  37       ApplicationProperties properties)
  38   {
  39     return
  40         new EndlessConsumer<>(
  41             executor,
  42             properties.getClientId(),
  43             properties.getTopicIn(),
  44             kafkaConsumer,
  45             recordHandler);
  46   }
  47
  48   @Bean
  49   public ExecutorService executor()
  50   {
  51     return Executors.newSingleThreadExecutor();
  52   }
  53
  54   @Bean(destroyMethod = "close")
  55   public KafkaConsumer<String, Integer> kafkaConsumer(ApplicationProperties properties)
  56   {
  57     Properties props = new Properties();
  58
  59     props.put("bootstrap.servers", properties.getBootstrapServer());
  60     props.put("partition.assignment.strategy", "org.apache.kafka.clients.consumer.CooperativeStickyAssignor");
  61     props.put("group.id", properties.getGroupId());
  62     props.put("client.id", properties.getClientId());
  63     props.put("auto.offset.reset", properties.getAutoOffsetReset());
  64     props.put("auto.commit.interval.ms", (int)properties.getCommitInterval().toMillis());
  65     props.put("metadata.max.age.ms", "1000");
  66     props.put("key.deserializer", StringDeserializer.class.getName());
  67     props.put("value.deserializer", IntegerDeserializer.class.getName());
  68
  69     return new KafkaConsumer<>(props);
  70   }
  71
  72   @Bean(destroyMethod = "close")
  73   public KafkaProducer<String, Object> kafkaProducer(ApplicationProperties properties)
  74   {
  75     Properties props = new Properties();
  76     props.put("bootstrap.servers", properties.getBootstrapServer());
  77     props.put("client.id", properties.getClientId());
  78     props.put("acks", properties.getAcks());
  79     props.put("batch.size", properties.getBatchSize());
  80     props.put("delivery.timeout.ms", 20000); // 20 Sekunden
  81     props.put("request.timeout.ms",  10000); // 10 Sekunden
  82     props.put("linger.ms", properties.getLingerMs());
  83     props.put("compression.type", properties.getCompressionType());
  84     props.put("key.serializer", StringSerializer.class.getName());
  85     props.put("value.serializer", "TODO: JsonSerializer konfigurieren");
  86
  87     return new KafkaProducer<>(props);
  88   }
  89 }