Apache Kafka

   1 // src/main/java/com/example/demo/DemoApplication.java
   2 package com.example.demo;
   3 
   4 import org.springframework.boot.SpringApplication;
   5 import org.springframework.boot.autoconfigure.SpringBootApplication;
   6 
   7 @SpringBootApplication
   8 public class DemoApplication {
   9 
  10         public static void main(String[] args) {
  11                 SpringApplication.run(DemoApplication.class, args);
  12         }
  13 }

   1 // src/main/java/com/example/demo/DemoController.java
   2 package com.example.demo;
   3 
   4 import org.springframework.kafka.core.KafkaTemplate;
   5 import org.springframework.stereotype.Controller;
   6 import org.springframework.web.bind.annotation.GetMapping;
   7 import org.springframework.web.bind.annotation.PathVariable;
   8 import org.springframework.web.bind.annotation.ResponseBody;
   9 
  10 @Controller
  11 public class DemoController {
  12     private KafkaTemplate<String, String> kafkaTemplate;
  13 
  14     public DemoController(KafkaTemplate<String, String> kafkaTemplate) {
  15         this.kafkaTemplate = kafkaTemplate;
  16     }
  17 
  18     @GetMapping("/uppercase/{text}")
  19     @ResponseBody
  20     public String uppercase(@PathVariable String text) {
  21         String message = String.format("text to be sent in uppercase %s", text);
  22         kafkaTemplate.send(KafkaTopicConfig.TOPIC_TASK, message);
  23         return message;
  24     }
  25 
  26 }

   1 // src/main/java/com/example/demo/KafkaConsumerConfig.java
   2 package com.example.demo;
   3 
   4 import java.util.HashMap;
   5 import java.util.Map;
   6 import org.apache.kafka.clients.consumer.ConsumerConfig;
   7 import org.apache.kafka.common.serialization.StringDeserializer;
   8 import org.springframework.beans.factory.annotation.Value;
   9 import org.springframework.context.annotation.Bean;
  10 import org.springframework.context.annotation.Configuration;
  11 import org.springframework.kafka.annotation.EnableKafka;
  12 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
  13 import org.springframework.kafka.core.ConsumerFactory;
  14 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
  15 
  16 @EnableKafka
  17 @Configuration
  18 public class KafkaConsumerConfig {
  19     @Value(value = "${spring.kafka.bootstrap-servers}")
  20     private String bootstrapAddress;
  21     @Value(value = "${spring.kafka.consumer.group-id}")
  22     private String groupId;
  23 
  24     @Bean
  25     public ConsumerFactory<String, String> consumerFactory() {
  26         Map<String, Object> props = new HashMap<>();
  27         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
  28         props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
  29         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  30         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  31         return new DefaultKafkaConsumerFactory<>(props);
  32     }
  33 
  34     @Bean
  35     public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
  36         ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
  37         factory.setConsumerFactory(consumerFactory());
  38         return factory;
  39     }    
  40 }

   1 // src/main/java/com/example/demo/KafkaMessageListener.java
   2 package com.example.demo;
   3 
   4 import org.slf4j.Logger;
   5 import org.slf4j.LoggerFactory;
   6 import org.springframework.kafka.annotation.KafkaListener;
   7 import org.springframework.kafka.listener.MessageListener;
   8 import org.springframework.stereotype.Component;
   9 
  10 @Component
  11 public class KafkaMessageListener {
  12     private Logger logger;
  13 
  14     public KafkaMessageListener() {
  15         this.logger = LoggerFactory.getLogger(MessageListener.class);
  16         this.logger.info("Created rest MessageListener");
  17     }
  18 
  19     @KafkaListener(topics = KafkaTopicConfig.TOPIC_TASK)
  20     public void listen(String message) {
  21         System.out.println("Received Message in topicTask: " + message + " in uppercase " + message.toUpperCase());
  22     }
  23 }

   1 // src/main/java/com/example/demo/KafkaProducerConfig.java
   2 package com.example.demo;
   3 
   4 import java.util.HashMap;
   5 import java.util.Map;
   6 import org.apache.kafka.clients.producer.ProducerConfig;
   7 import org.apache.kafka.common.serialization.StringSerializer;
   8 import org.springframework.beans.factory.annotation.Value;
   9 import org.springframework.context.annotation.Bean;
  10 import org.springframework.context.annotation.Configuration;
  11 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
  12 import org.springframework.kafka.core.KafkaTemplate;
  13 import org.springframework.kafka.core.ProducerFactory;
  14 
  15 @Configuration
  16 public class KafkaProducerConfig {
  17     @Value(value = "${spring.kafka.bootstrap-servers}")
  18     private String bootstrapAddress;
  19 
  20     @Bean
  21     public ProducerFactory<String, String> producerFactory() {
  22         Map<String, Object> configProps = new HashMap<>();
  23         configProps.put(
  24                 ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
  25                 bootstrapAddress);
  26         configProps.put(
  27                 ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
  28                 StringSerializer.class);
  29         configProps.put(
  30                 ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
  31                 StringSerializer.class);
  32         return new DefaultKafkaProducerFactory<>(configProps);
  33     }
  34 
  35     @Bean
  36     public KafkaTemplate<String, String> kafkaTemplate() {
  37         return new KafkaTemplate<>(producerFactory());
  38     }
  39 }

   1 // src/main/java/com/example/demo/KafkaTopicConfig.java
   2 package com.example.demo;
   3 
   4 import java.util.HashMap;
   5 import java.util.Map;
   6 
   7 import org.apache.kafka.clients.admin.AdminClientConfig;
   8 import org.apache.kafka.clients.admin.NewTopic;
   9 import org.springframework.beans.factory.annotation.Value;
  10 import org.springframework.context.annotation.Bean;
  11 import org.springframework.context.annotation.Configuration;
  12 import org.springframework.kafka.core.KafkaAdmin;
  13 
  14 @Configuration
  15 public class KafkaTopicConfig {
  16     public static final String TOPIC_TASK = "topicTask";
  17     @Value(value = "${spring.kafka.bootstrap-servers}")
  18     private String bootstrapAddress;
  19 
  20     @Bean
  21     public KafkaAdmin kafkaAdmin() {
  22         Map<String, Object> configs = new HashMap<>();
  23         configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
  24         return new KafkaAdmin(configs);
  25     }
  26 
  27     @Bean
  28     public NewTopic topicTask() {
  29         return new NewTopic(TOPIC_TASK, 1, (short) 1);
  30     }
  31 }