Merge branch rebalance-listener into counting-consumer
[demos/kafka/training] / src / main / java / de / juplo / kafka / Application.java
1 package de.juplo.kafka;
2
3 import org.springframework.beans.factory.annotation.Autowired;
4 import org.springframework.boot.SpringApplication;
5 import org.springframework.boot.autoconfigure.SpringBootApplication;
6 import org.springframework.boot.context.properties.EnableConfigurationProperties;
7 import org.springframework.context.annotation.Bean;
8 import org.springframework.util.Assert;
9
10 import java.util.concurrent.Executors;
11
12
13 @SpringBootApplication
14 @EnableConfigurationProperties(ApplicationProperties.class)
15 public class Application
16 {
17   @Autowired
18   ApplicationProperties properties;
19
20
21   @Bean
22   public EndlessConsumer consumer()
23   {
24     Assert.hasText(properties.getBootstrapServer(), "consumer.bootstrap-server must be set");
25     Assert.hasText(properties.getGroupId(), "consumer.group-id must be set");
26     Assert.hasText(properties.getClientId(), "consumer.client-id must be set");
27     Assert.hasText(properties.getTopic(), "consumer.topic must be set");
28
29     EndlessConsumer consumer =
30         new EndlessConsumer(
31             Executors.newFixedThreadPool(1),
32             properties.getBootstrapServer(),
33             properties.getGroupId(),
34             properties.getClientId(),
35             properties.getTopic(),
36             properties.getAutoOffsetReset());
37
38     consumer.start();
39
40     return consumer;
41   }
42
43   public static void main(String[] args)
44   {
45     SpringApplication.run(Application.class, args);
46   }
47 }