Merge pull request #8125 from eugenp/revert-8119-BAEL-3275-2

Revert "BAEL-3275: Using blocking queue for pub-sub"

spring-kafka/src/main/java/com/baeldung/spring/kafka/Greeting.java

@@ -0,0 +1,37 @@
+package com.baeldung.spring.kafka;
+
+public class Greeting {
+
+    private String msg;
+    private String name;
+
+    public Greeting() {
+
+    }
+
+    public Greeting(String msg, String name) {
+        this.msg = msg;
+        this.name = name;
+    }
+
+    public String getMsg() {
+        return msg;
+    }
+
+    public void setMsg(String msg) {
+        this.msg = msg;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    @Override
+    public String toString() {
+        return msg + ", " + name + "!";
+    }
+}

spring-kafka/src/main/java/com/baeldung/spring/kafka/KafkaApplication.java

@@ -0,0 +1,181 @@
+package com.baeldung.spring.kafka;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.annotation.TopicPartition;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.kafka.support.SendResult;
+import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.messaging.handler.annotation.Payload;
+import org.springframework.util.concurrent.ListenableFuture;
+import org.springframework.util.concurrent.ListenableFutureCallback;
+
+@SpringBootApplication
+public class KafkaApplication {
+
+    public static void main(String[] args) throws Exception {
+
+        ConfigurableApplicationContext context = SpringApplication.run(KafkaApplication.class, args);
+        
+        MessageProducer producer = context.getBean(MessageProducer.class);
+        MessageListener listener = context.getBean(MessageListener.class);
+        /*
+         * Sending a Hello World message to topic 'baeldung'. 
+         * Must be recieved by both listeners with group foo
+         * and bar with containerFactory fooKafkaListenerContainerFactory
+         * and barKafkaListenerContainerFactory respectively.
+         * It will also be recieved by the listener with
+         * headersKafkaListenerContainerFactory as container factory
+         */
+        producer.sendMessage("Hello, World!");
+        listener.latch.await(10, TimeUnit.SECONDS);
+
+        /*
+         * Sending message to a topic with 5 partition,
+         * each message to a different partition. But as per
+         * listener configuration, only the messages from
+         * partition 0 and 3 will be consumed.
+         */
+        for (int i = 0; i < 5; i++) {
+            producer.sendMessageToPartion("Hello To Partioned Topic!", i);
+        }
+        listener.partitionLatch.await(10, TimeUnit.SECONDS);
+
+        /*
+         * Sending message to 'filtered' topic. As per listener
+         * configuration,  all messages with char sequence
+         * 'World' will be discarded.
+         */
+        producer.sendMessageToFiltered("Hello Baeldung!");
+        producer.sendMessageToFiltered("Hello World!");
+        listener.filterLatch.await(10, TimeUnit.SECONDS);
+
+        /*
+         * Sending message to 'greeting' topic. This will send
+         * and recieved a java object with the help of 
+         * greetingKafkaListenerContainerFactory.
+         */
+        producer.sendGreetingMessage(new Greeting("Greetings", "World!"));
+        listener.greetingLatch.await(10, TimeUnit.SECONDS);
+
+        context.close();
+    }
+
+    @Bean
+    public MessageProducer messageProducer() {
+        return new MessageProducer();
+    }
+
+    @Bean
+    public MessageListener messageListener() {
+        return new MessageListener();
+    }
+
+    public static class MessageProducer {
+
+        @Autowired
+        private KafkaTemplate<String, String> kafkaTemplate;
+
+        @Autowired
+        private KafkaTemplate<String, Greeting> greetingKafkaTemplate;
+
+        @Value(value = "${message.topic.name}")
+        private String topicName;
+
+        @Value(value = "${partitioned.topic.name}")
+        private String partionedTopicName;
+
+        @Value(value = "${filtered.topic.name}")
+        private String filteredTopicName;
+
+        @Value(value = "${greeting.topic.name}")
+        private String greetingTopicName;
+
+        public void sendMessage(String message) {
+            
+            ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topicName, message);
+            
+            future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
+
+                @Override
+                public void onSuccess(SendResult<String, String> result) {
+                    System.out.println("Sent message=[" + message + "] with offset=[" + result.getRecordMetadata().offset() + "]");
+                }
+                @Override
+                public void onFailure(Throwable ex) {
+                    System.out.println("Unable to send message=[" + message + "] due to : " + ex.getMessage());
+                }
+            });
+        }
+
+        public void sendMessageToPartion(String message, int partition) {
+            kafkaTemplate.send(partionedTopicName, partition, null, message);
+        }
+
+        public void sendMessageToFiltered(String message) {
+            kafkaTemplate.send(filteredTopicName, message);
+        }
+
+        public void sendGreetingMessage(Greeting greeting) {
+            greetingKafkaTemplate.send(greetingTopicName, greeting);
+        }
+    }
+
+    public static class MessageListener {
+
+        private CountDownLatch latch = new CountDownLatch(3);
+
+        private CountDownLatch partitionLatch = new CountDownLatch(2);
+
+        private CountDownLatch filterLatch = new CountDownLatch(2);
+
+        private CountDownLatch greetingLatch = new CountDownLatch(1);
+
+        @KafkaListener(topics = "${message.topic.name}", groupId = "foo", containerFactory = "fooKafkaListenerContainerFactory")
+        public void listenGroupFoo(String message) {
+            System.out.println("Received Messasge in group 'foo': " + message);
+            latch.countDown();
+        }
+
+        @KafkaListener(topics = "${message.topic.name}", groupId = "bar", containerFactory = "barKafkaListenerContainerFactory")
+        public void listenGroupBar(String message) {
+            System.out.println("Received Messasge in group 'bar': " + message);
+            latch.countDown();
+        }
+
+        @KafkaListener(topics = "${message.topic.name}", containerFactory = "headersKafkaListenerContainerFactory")
+        public void listenWithHeaders(@Payload String message, @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {
+            System.out.println("Received Messasge: " + message + " from partition: " + partition);
+            latch.countDown();
+        }
+
+        @KafkaListener(topicPartitions = @TopicPartition(topic = "${partitioned.topic.name}", partitions = { "0", "3" }))
+        public void listenToParition(@Payload String message, @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {
+            System.out.println("Received Message: " + message + " from partition: " + partition);
+            this.partitionLatch.countDown();
+        }
+
+        @KafkaListener(topics = "${filtered.topic.name}", containerFactory = "filterKafkaListenerContainerFactory")
+        public void listenWithFilter(String message) {
+            System.out.println("Recieved Message in filtered listener: " + message);
+            this.filterLatch.countDown();
+        }
+
+        @KafkaListener(topics = "${greeting.topic.name}", containerFactory = "greetingKafkaListenerContainerFactory")
+        public void greetingListener(Greeting greeting) {
+            System.out.println("Recieved greeting message: " + greeting);
+            this.greetingLatch.countDown();
+        }
+
+    }
+
+}

spring-kafka/src/main/java/com/baeldung/spring/kafka/KafkaConsumerConfig.java

@@ -0,0 +1,81 @@
+package com.baeldung.spring.kafka;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.annotation.EnableKafka;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.support.serializer.JsonDeserializer;
+
+@EnableKafka
+@Configuration
+public class KafkaConsumerConfig {
+
+    @Value(value = "${kafka.bootstrapAddress}")
+    private String bootstrapAddress;
+
+    public ConsumerFactory<String, String> consumerFactory(String groupId) {
+        Map<String, Object> props = new HashMap<>();
+        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
+        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
+        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+        return new DefaultKafkaConsumerFactory<>(props);
+    }
+    
+    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(String groupId) {
+        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
+        factory.setConsumerFactory(consumerFactory(groupId));
+        return factory;
+    }
+
+    @Bean
+    public ConcurrentKafkaListenerContainerFactory<String, String> fooKafkaListenerContainerFactory() {
+        return kafkaListenerContainerFactory("foo");
+    }
+
+    @Bean
+    public ConcurrentKafkaListenerContainerFactory<String, String> barKafkaListenerContainerFactory() {
+        return kafkaListenerContainerFactory("bar");
+    }
+
+    @Bean
+    public ConcurrentKafkaListenerContainerFactory<String, String> headersKafkaListenerContainerFactory() {
+        return kafkaListenerContainerFactory("headers");
+    }
+    
+    @Bean
+    public ConcurrentKafkaListenerContainerFactory<String, String> partitionsKafkaListenerContainerFactory() {
+        return kafkaListenerContainerFactory("partitions");
+    }
+    
+    @Bean
+    public ConcurrentKafkaListenerContainerFactory<String, String> filterKafkaListenerContainerFactory() {
+        ConcurrentKafkaListenerContainerFactory<String, String> factory = kafkaListenerContainerFactory("filter");
+        factory.setRecordFilterStrategy(record -> record.value()
+            .contains("World"));
+        return factory;
+    }
+
+    public ConsumerFactory<String, Greeting> greetingConsumerFactory() {
+        Map<String, Object> props = new HashMap<>();
+        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
+        props.put(ConsumerConfig.GROUP_ID_CONFIG, "greeting");
+        return new DefaultKafkaConsumerFactory<>(props, new StringDeserializer(), new JsonDeserializer<>(Greeting.class));
+    }
+
+    @Bean
+    public ConcurrentKafkaListenerContainerFactory<String, Greeting> greetingKafkaListenerContainerFactory() {
+        ConcurrentKafkaListenerContainerFactory<String, Greeting> factory = new ConcurrentKafkaListenerContainerFactory<>();
+        factory.setConsumerFactory(greetingConsumerFactory());
+        return factory;
+    }
+
+}

spring-kafka/src/main/java/com/baeldung/spring/kafka/KafkaProducerConfig.java

@@ -0,0 +1,50 @@
+package com.baeldung.spring.kafka;
+
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.support.serializer.JsonSerializer;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@Configuration
+public class KafkaProducerConfig {
+
+    @Value(value = "${kafka.bootstrapAddress}")
+    private String bootstrapAddress;
+
+    @Bean
+    public ProducerFactory<String, String> producerFactory() {
+        Map<String, Object> configProps = new HashMap<>();
+        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
+        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        return new DefaultKafkaProducerFactory<>(configProps);
+    }
+
+    @Bean
+    public KafkaTemplate<String, String> kafkaTemplate() {
+        return new KafkaTemplate<>(producerFactory());
+    }
+    
+    @Bean
+    public ProducerFactory<String, Greeting> greetingProducerFactory() {
+        Map<String, Object> configProps = new HashMap<>();
+        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
+        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
+        return new DefaultKafkaProducerFactory<>(configProps);
+    }
+    
+    @Bean
+    public KafkaTemplate<String, Greeting> greetingKafkaTemplate() {
+        return new KafkaTemplate<>(greetingProducerFactory());
+    }
+    
+}

spring-kafka/src/main/java/com/baeldung/spring/kafka/KafkaTopicConfig.java

@@ -0,0 +1,57 @@
+package com.baeldung.spring.kafka;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.KafkaAdmin;
+
+@Configuration
+public class KafkaTopicConfig {
+    
+    @Value(value = "${kafka.bootstrapAddress}")
+    private String bootstrapAddress;
+    
+    @Value(value = "${message.topic.name}")
+    private String topicName;
+
+    @Value(value = "${partitioned.topic.name}")
+    private String partionedTopicName;
+
+    @Value(value = "${filtered.topic.name}")
+    private String filteredTopicName;
+
+    @Value(value = "${greeting.topic.name}")
+    private String greetingTopicName;
+    
+    @Bean
+    public KafkaAdmin kafkaAdmin() {
+        Map<String, Object> configs = new HashMap<>();
+        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
+        return new KafkaAdmin(configs);
+    }
+    
+    @Bean
+    public NewTopic topic1() {
+         return new NewTopic(topicName, 1, (short) 1);
+    }
+    
+    @Bean
+    public NewTopic topic2() {
+         return new NewTopic(partionedTopicName, 6, (short) 1);
+    }
+    
+    @Bean
+    public NewTopic topic3() {
+         return new NewTopic(filteredTopicName, 1, (short) 1);
+    }
+    
+    @Bean
+    public NewTopic topic4() {
+         return new NewTopic(greetingTopicName, 1, (short) 1);
+    }
+}