Apache Kafka is a popular distributed messaging system that has become a cornerstone for many modern data architectures due to its high throughput, built-in partitioning, replication, and inherent fault tolerance. Spring Boot, on the other hand, greatly simplifies the development of complex applications with its convention-over-configuration approach, including the integration with Apache Kafka through the Spring Kafka project. This article discusses how multiple consumers can listen to multiple topics in a Spring Boot Kafka application, enhancing scalability and flexibility.

Understanding Kafka Consumers and Topics

In Kafka, a topic is a category or feed name to which records are published. Topics in Kafka are always multi-subscriber; that is, a topic can have zero, one, or many consumers that subscribe to the data written to it. Each topic can be subdivided into a number of partitions, allowing the data to be scaled. Each partition is an ordered, immutable sequence of records that is continually appended to.

A Kafka consumer pulls records off a Kafka topic. Consumers can work alone or can be part of a consumer group. When consumers are part of a group, each consumer within the group reads from exclusive partitions of the topic, ensuring that every message is effectively processed by exactly one consumer. If all consumers are in different groups, then each consumer will effectively receive all the messages on the topic.

Configuring Kafka Consumers in Spring Boot

To allow multiple consumers to listen to multiple topics in Spring Boot, you must configure each consumer and assign them to the desired topics. The @KafkaListener annotation provides an easy way to do this in Spring Boot applications.

Basic Configuration

1. Add Dependencies: Include spring-kafka and kafka-clients in your pom.xml or build.gradle file.

1   <!-- Maven dependency -->
2   <dependency>
3       <groupId>org.springframework.kafka</groupId>
4       <artifactId>spring-kafka</artifactId>
5       <version>2.8.0</version>
6   </dependency>

1. Configure Application Properties: In your application.yml or application.properties file, define the necessary Kafka properties.

1   spring:
2     kafka:
3       bootstrap-servers: localhost:9092
4       consumer:
5         group-id: group-id-1
6         auto-offset-reset: earliest
7         key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
8         value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

1. Create Kafka Listeners: Use the @KafkaListener annotation to define methods that will be triggered when a message is received.

1   @Service
2   public class KafkaConsumers {
3      
4       @KafkaListener(topics = "topic1", groupId = "group1")
5       public void listenTopic1(String message) {
6           System.out.println("Received Message in group 'group1': " + message);
7       }
8
9       @KafkaListener(topics = "topic2", groupId = "group2")
10       public void listenTopic2(String message) {
11           System.out.println("Received Message in group 'group2': " + message);
12       }
13   }

Advanced Configurations

For more complex scenarios, such as multiple methods listening to multiple topics within the same service class or dynamically allocating topics based on some conditions, you need to employ some additional configuration.

Using @KafkaListener with Multiple Topics

You can configure a single method to listen to multiple topics by specifying them in the topics attribute of the @KafkaListener.

1@KafkaListener(topics = {"topic1", "topic3"}, groupId = "group1")
2public void listenMultipleTopics(String message) {
3    System.out.println("Received Message: " + message);
4}

Programmatically Configuring Listeners

You may want to configure listeners at runtime based on some business logic or external configuration. You can do this programmatically by using ConcurrentKafkaListenerContainerFactory to create ConcurrentMessageListenerContainer.

1@Autowired
2private ConcurrentKafkaListenerContainerFactory<String, String> factory;
3
4public void addListener(String topic) {
5    ConcurrentMessageListenerContainer<String, String> container = factory.createContainer(topic);
6    container.setupMessageListener((MessageListener<String, String>) record -> {
7        System.out.println("Received: " + record.value());
8    });
9    container.start();
10}

Summary of Key Points

Conclusion

Listening to multiple topics with multiple consumers in a Spring Boot application using Kafka involves understanding the fundamental concepts of consumers and topics in Kafka, and then leveraging Spring Boot's @KafkaListener to hook into this functionality easily. Whether using annotation-driven approach or programmatically configuring consumers, Spring Boot with Kafka provides a powerful framework to handle vast amounts of data efficiently, distributed across various topics and consumer groups.