How to add Kafka listener in Spring Framework

To use Kafka with Spring, we need to add the spring-kafka dependency to our project’s build file (e.g., pom.xml for Maven or build.gradle for Gradle). This dependency includes the necessary classes and interfaces for us to interact with Kafka using Spring’s abstractions.

Here’s an example of how we can define a Kafka listener in Spring Framework:

package com.example;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class MyKafkaListener {

    @KafkaListener(topics = "myTopic")
    public void listen(ConsumerRecord<String, String> record) {
        System.out.printf("Received message: %s%n", record.value());
    }
}

In this code, we define a MyKafkaListener class and annotate it with @Component to indicate that it should be registered as a Spring bean. We then define a listen method and annotate it with @KafkaListener to indicate that it should be invoked when a message is received from the myTopic Kafka topic.

The listen method takes a ConsumerRecord object as a parameter, which contains the details of the received message (e.g., the key and value of the message, the topic and partition it came from, etc.). In this example, we simply print the value of the message to the console, but we could have performed any other operation with the message data.

Note that we could have used other configuration options to customize the behavior of the Kafka listener. For example, we could have specified a group ID, partition assignment strategy, error handling behavior, or message filtering criteria. These options are all available through the @KafkaListener annotation and its related annotations.

To run this Kafka listener in a Spring Boot application, we need to make sure that we have the necessary dependencies in our project’s build file (e.g., pom.xml for Maven or build.gradle for Gradle). We also need to configure our Kafka connection properties, such as the bootstrap servers, group ID, and topic names, in our application.properties or application.yml file. Here’s an example of how we can configure our Spring Boot application to use this Kafka listener:

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: myGroupId
      auto-offset-reset: earliest

In this code, we specify the bootstrap servers for our Kafka cluster and configure our Kafka consumer properties, such as the group ID and auto-offset-reset behavior. These properties tell Spring how to connect to Kafka and receive messages from the myTopic topic.

With these code snippets, we can create a simple Spring Boot application that listens to Kafka messages using a Kafka listener. When we run our application and send messages to the myTopic topic, we’ll see the message data printed to the console by our Kafka listener.

Congratulations, you’ve just added a Kafka listener to a Spring Framework application! You can now use this knowledge to build more complex Kafka-based systems that process real-time data, integrate with external services, or react to user events.

Here is a post How to work with Kafka in Java.