Merge der überarbeiteten Compose-Konfiguration ('rebalance-listener')
[demos/kafka/training] / 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.clients.consumer.KafkaConsumer;
5 import org.apache.kafka.common.serialization.LongDeserializer;
6 import org.apache.kafka.common.serialization.StringDeserializer;
7 import org.springframework.boot.context.properties.EnableConfigurationProperties;
8 import org.springframework.context.annotation.Bean;
9 import org.springframework.context.annotation.Configuration;
10
11 import java.util.Properties;
12 import java.util.concurrent.ExecutorService;
13 import java.util.concurrent.Executors;
14 import java.util.function.Consumer;
15
16
17 @Configuration
18 @EnableConfigurationProperties(ApplicationProperties.class)
19 public class ApplicationConfiguration
20 {
21   @Bean
22   public Consumer<ConsumerRecord<String, Long>> consumer()
23   {
24     return (record) ->
25     {
26       // Handle record
27     };
28   }
29
30   @Bean
31   public EndlessConsumer<String, Long> endlessConsumer(
32       KafkaConsumer<String, Long> kafkaConsumer,
33       ExecutorService executor,
34       Consumer<ConsumerRecord<String, Long>> handler,
35       ApplicationProperties properties)
36   {
37     return
38         new EndlessConsumer<>(
39             executor,
40             properties.getClientId(),
41             properties.getTopic(),
42             kafkaConsumer,
43             handler);
44   }
45
46   @Bean
47   public ExecutorService executor()
48   {
49     return Executors.newSingleThreadExecutor();
50   }
51
52   @Bean(destroyMethod = "close")
53   public KafkaConsumer<String, Long> kafkaConsumer(ApplicationProperties properties)
54   {
55     Properties props = new Properties();
56
57     props.put("bootstrap.servers", properties.getBootstrapServer());
58     props.put("group.id", properties.getGroupId());
59     props.put("client.id", properties.getClientId());
60     props.put("auto.offset.reset", properties.getAutoOffsetReset());
61     props.put("metadata.max.age.ms", "1000");
62     props.put("key.deserializer", StringDeserializer.class.getName());
63     props.put("value.deserializer", LongDeserializer.class.getName());
64
65     return new KafkaConsumer<>(props);
66   }
67 }