X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FApplication.java;h=0069257fe27025546c1f89a8124c9114ac88f830;hb=f9890a95d6672e1847e1d9f53a76c95ade877a9b;hp=5226d6bd062bee4003ef4ca00f8f1c37597af3d8;hpb=5a2c467b5b299f975f22d6c0e761686067634adc;p=demos%2Fkafka%2Ftraining diff --git a/src/main/java/de/juplo/kafka/Application.java b/src/main/java/de/juplo/kafka/Application.java index 5226d6b..0069257 100644 --- a/src/main/java/de/juplo/kafka/Application.java +++ b/src/main/java/de/juplo/kafka/Application.java @@ -1,55 +1,12 @@ package de.juplo.kafka; -import org.apache.kafka.clients.consumer.KafkaConsumer; -import org.apache.kafka.common.serialization.LongDeserializer; -import org.apache.kafka.common.serialization.StringDeserializer; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; -import org.springframework.boot.context.properties.EnableConfigurationProperties; -import org.springframework.context.annotation.Bean; - -import java.util.Properties; -import java.util.concurrent.Executors; @SpringBootApplication -@EnableConfigurationProperties(ApplicationProperties.class) public class Application { - @Bean - public EndlessConsumer endlessConsumer( - KafkaConsumer kafkaConsumer, - ApplicationProperties properties) - { - EndlessConsumer consumer = - new EndlessConsumer( - Executors.newFixedThreadPool(1), - properties.getClientId(), - properties.getTopic(), - kafkaConsumer); - - consumer.start(); - - return consumer; - } - - @Bean(destroyMethod = "close") - public KafkaConsumer kafkaConsumer(ApplicationProperties properties) - { - Properties props = new Properties(); - - props.put("bootstrap.servers", properties.getBootstrapServer()); - props.put("group.id", properties.getGroupId()); - props.put("client.id", properties.getClientId()); - props.put("auto.offset.reset", properties.getAutoOffsetReset()); - props.put("metadata.max.age.ms", "1000"); - props.put("key.deserializer", StringDeserializer.class.getName()); - props.put("value.deserializer", LongDeserializer.class.getName()); - - return new KafkaConsumer<>(props); - } - - public static void main(String[] args) { SpringApplication.run(Application.class, args);