users:1.0.0 - users are directly send to kafka

[demos/kafka/wordcount] / src / main / java / de / juplo / kafka / wordcount / users / UsersApplication.java

diff --git a/src/main/java/de/juplo/kafka/wordcount/users/UsersApplication.java b/src/main/java/de/juplo/kafka/wordcount/users/UsersApplication.java
new file mode 100644 (file)
index 0000000..9ae44e1
--- /dev/null
+++ b/src/main/java/de/juplo/kafka/wordcount/users/UsersApplication.java
@@ -0,0 +1,36 @@
+package de.juplo.kafka.wordcount.users;
+
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringSerializer;
+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 org.springframework.util.Assert;
+
+import java.util.Properties;
+
+
+@SpringBootApplication
+@EnableConfigurationProperties(UsersApplicationProperties.class)
+public class UsersApplication
+{
+       @Bean(destroyMethod = "close")
+       KafkaProducer<String, String> producer(UsersApplicationProperties properties)
+       {
+               Assert.hasText(properties.getBootstrapServer(), "juplo.wordcount.recorder.bootstrap-server must be set");
+
+               Properties props = new Properties();
+               props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServer());
+               props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+               props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+
+               return new KafkaProducer<>(props);
+       }
+
+       public static void main(String[] args)
+       {
+               SpringApplication.run(UsersApplication.class, args);
+       }
+}