BAEL-6566: Manage Kafka Consumer Groups (#15007)

* test removing consumer

* test removing consumer

* wrapping up

* fix main class

* addressing request changes

* introducing constants for numbers. optimizing imports

* introducing class level constants

* using constant naming conventions

spring-kafka-2/src/main/java/com/baeldung/spring/kafka/managingkafkaconsumergroups/KafkaConsumerConfiguration.java

@@ -0,0 +1,37 @@
+package com.baeldung.spring.kafka.managingkafkaconsumergroups;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.serialization.DoubleDeserializer;
+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.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@Configuration
+public class KafkaConsumerConfiguration {
+
+    @Value(value = "${spring.kafka.bootstrap-servers}")
+    private String bootstrapAddress;
+
+    @Bean
+    public ConsumerFactory<String, Double> kafkaConsumer() {
+        Map<String, Object> props = new HashMap<>();
+        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
+        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, DoubleDeserializer.class);
+        return new DefaultKafkaConsumerFactory<>(props);
+    }
+
+    @Bean
+    public ConcurrentKafkaListenerContainerFactory<String, Double> kafkaConsumerContainerFactory() {
+        ConcurrentKafkaListenerContainerFactory<String, Double> factory = new ConcurrentKafkaListenerContainerFactory<>();
+        factory.setConsumerFactory(kafkaConsumer());
+        return factory;
+    }
+}

spring-kafka-2/src/main/java/com/baeldung/spring/kafka/managingkafkaconsumergroups/KafkaProducerConfiguration.java

@@ -0,0 +1,35 @@
+package com.baeldung.spring.kafka.managingkafkaconsumergroups;
+
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.DoubleSerializer;
+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 java.util.HashMap;
+import java.util.Map;
+
+@Configuration
+public class KafkaProducerConfiguration {
+
+    @Value(value = "${spring.kafka.bootstrap-servers}")
+    private String bootstrapAddress;
+
+    @Bean
+    public ProducerFactory<String, Double> kafkaProducer() {
+        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, DoubleSerializer.class);
+        return new DefaultKafkaProducerFactory<>(configProps);
+    }
+
+    @Bean
+    public KafkaTemplate<String, Double> kafkaProducerTemplate() {
+        return new KafkaTemplate<>(kafkaProducer());
+    }
+}

spring-kafka-2/src/main/java/com/baeldung/spring/kafka/managingkafkaconsumergroups/KafkaTopicConfiguration.java

@@ -0,0 +1,30 @@
+package com.baeldung.spring.kafka.managingkafkaconsumergroups;
+
+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.Configuration;
+import org.springframework.kafka.config.TopicBuilder;
+import org.springframework.kafka.core.KafkaAdmin;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@Configuration
+public class KafkaTopicConfiguration {
+
+    @Value(value = "${spring.kafka.bootstrap-servers}")
+    private String bootstrapAddress;
+
+    public KafkaAdmin kafkaAdmin() {
+        Map<String, Object> configs = new HashMap<>();
+        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
+        return new KafkaAdmin(configs);
+    }
+
+    public NewTopic celciusTopic() {
+        return TopicBuilder.name("topic-1")
+                .partitions(2)
+                .build();
+    }
+}

spring-kafka-2/src/main/java/com/baeldung/spring/kafka/managingkafkaconsumergroups/ManagingConsumerGroupsApplicationKafkaApp.java

@@ -0,0 +1,13 @@
+package com.baeldung.spring.kafka.managingkafkaconsumergroups;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.Import;
+
+@SpringBootApplication
+@Import(value = {KafkaConsumerConfiguration.class, KafkaProducerConfiguration.class, KafkaTopicConfiguration.class})
+public class ManagingConsumerGroupsApplicationKafkaApp {
+    public static void main(String[] args) {
+        SpringApplication.run(ManagingConsumerGroupsApplicationKafkaApp.class, args);
+    }
+}

spring-kafka-2/src/main/java/com/baeldung/spring/kafka/managingkafkaconsumergroups/MessageConsumerService.java

@@ -0,0 +1,34 @@
+package com.baeldung.spring.kafka.managingkafkaconsumergroups;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Service;
+
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+@Service
+public class MessageConsumerService {
+
+    Map<String, Set<Integer>> consumedPartitions = new ConcurrentHashMap<>();
+
+    @KafkaListener(topics = "topic-1", groupId = "group-1")
+    public void consumer0(ConsumerRecord<?, ?> consumerRecord) {
+        trackConsumedPartitions("consumer-0", consumerRecord);
+    }
+
+    @KafkaListener(topics = "topic-1", groupId = "group-1")
+    public void consumer1(ConsumerRecord<?, ?> consumerRecord) {
+        trackConsumedPartitions("consumer-1", consumerRecord);
+    }
+
+    private void trackConsumedPartitions(String key, ConsumerRecord<?, ?> record) {
+        consumedPartitions.computeIfAbsent(key, k -> new HashSet<>());
+        consumedPartitions.computeIfPresent(key, (k, v) -> {
+            v.add(record.partition());
+            return v;
+        });
+    }
+}

spring-kafka-2/src/test/java/com/baeldung/spring/kafka/managingkafkaconsumergroups/ManagingConsumerGroupsIntegrationTest.java

@@ -0,0 +1,57 @@
+package com.baeldung.spring.kafka.managingkafkaconsumergroups;
+
+import org.apache.commons.lang3.RandomUtils;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.listener.MessageListenerContainer;
+import org.springframework.kafka.test.context.EmbeddedKafka;
+
+import java.util.Objects;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+@SpringBootTest(classes = ManagingConsumerGroupsApplicationKafkaApp.class)
+@EmbeddedKafka(partitions = 2, brokerProperties = {"listeners=PLAINTEXT://localhost:9092", "port=9092"})
+public class ManagingConsumerGroupsIntegrationTest {
+
+    private static final String CONSUMER_1_IDENTIFIER = "org.springframework.kafka.KafkaListenerEndpointContainer#1";
+    private static final int TOTAL_PRODUCED_MESSAGES = 50000;
+    private static final int MESSAGE_WHERE_CONSUMER_1_LEAVES_GROUP = 10000;
+
+    @Autowired
+    KafkaTemplate<String, Double> kafkaTemplate;
+
+    @Autowired
+    KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;
+
+    @Autowired
+    MessageConsumerService consumerService;
+
+    @Test
+    public void givenContinuousMessageFlow_whenAConsumerLeavesTheGroup_thenKafkaTriggersPartitionRebalance() throws InterruptedException {
+        int currentMessage = 0;
+
+        do {
+            kafkaTemplate.send("topic-1", RandomUtils.nextDouble(10.0, 20.0));
+            currentMessage++;
+
+            if (currentMessage == MESSAGE_WHERE_CONSUMER_1_LEAVES_GROUP) {
+                String containerId = kafkaListenerEndpointRegistry.getListenerContainerIds()
+                        .stream()
+                        .filter(a -> a.equals(CONSUMER_1_IDENTIFIER))
+                        .findFirst()
+                        .orElse("");
+                MessageListenerContainer container = kafkaListenerEndpointRegistry.getListenerContainer(containerId);
+                Thread.sleep(2000);
+                Objects.requireNonNull(container).stop();
+                kafkaListenerEndpointRegistry.unregisterListenerContainer(containerId);
+            }
+        } while (currentMessage != TOTAL_PRODUCED_MESSAGES);
+        Thread.sleep(2000);
+        assertEquals(1, consumerService.consumedPartitions.get("consumer-1").size());
+        assertEquals(2, consumerService.consumedPartitions.get("consumer-0").size());
+    }
+}