Apache Kafka

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

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

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

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

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

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