Integrating Apache Kafka with Spring Boot: A Step-by-Step Guide

In today's data-driven world, real-time data processing is crucial for building responsive and scalable applications. Apache Kafka, a distributed streaming platform, has become the go-to solution for handling real-time data feeds. When combined with Spring Boot, a powerful framework for building Java applications, you can create robust and efficient applications that leverage the power of Kafka.

In this article, we'll explore how to integrate Apache Kafka with a Spring Boot application. We'll cover the basics of Kafka, set up a Kafka server, and build a simple producer and consumer application using Spring Boot.

What is Apache Kafka?

Apache Kafka is an open-source distributed event streaming platform used by thousands of companies for high-performance data pipelines, streaming analytics, data integration, and mission-critical applications.

Key Features:

• Scalability: Kafka can handle large volumes of data with minimal latency.
• Fault-Tolerance: Data is replicated within the cluster to prevent data loss.
• High Throughput: Capable of handling millions of messages per second.

Common Use Cases:

• Real-time analytics
• Log aggregation
• Stream processing
• Event sourcing

Setting Up Apache Kafka:

Before integrating Kafka with Spring Boot, you need a running Kafka instance.

Step 1: Download Kafka

Download the latest stable version of Kafka from the official website.

Step 2: Start Zookeeper and Kafka Server

Kafka uses Zookeeper for coordinating and managing the cluster.

• Start Zookeeper:

bin/zookeeper-server-start.sh config/zookeeper.properties        

• Start Kafka Server:

bin/kafka-server-start.sh config/server.properties        

Configuring Kafka in Spring Boot

Configure the necessary Kafka properties in application.properties.

# Kafka Properties
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=optima
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer        

Creating Kafka Producer and Consumer

1. Kafka Producer

Create a service that will act as a Kafka producer.

package com.prajtech.kafka;

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class KafkaProducer {
    private static final String TOPIC = "my_topic";
    private final KafkaTemplate<String, String> kafkaTemplate;
    
    public KafkaProducer(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }
    
    public void sendMessage(String message) {
        kafkaTemplate.send(TOPIC, message);
        System.out.println("Sent message: " + message);
    }
}        

2. Kafka Consumer

Create a listener that will act as a Kafka consumer.

package com.prajtech.kafka;

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

@Service
public class KafkaConsumer {
    @KafkaListener(topics = "my_topic", groupId = "my-group")
    public void listen(String message) {
        System.out.println("Received message: " + message);
    }
}        

Testing the Application

Create a REST Controller

Create a controller to publish messages to Kafka.

package com.prajtech.kafka;

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
class KafkaController {
    private final KafkaProducer kafkaProducer;
    
    public KafkaController(KafkaProducer kafkaProducer) {
        this.kafkaProducer = kafkaProducer;
    }
    
    @GetMapping("/publish")
    public String publish(@RequestParam("message") String message) {
        kafkaProducer.sendMessage(message);
        return "Message published successfully";
    }
}        

Run the Application

• Start your Spring Boot application.
• Open a web browser or use Postman to send a GET request:

https://localhost:8080/publish?message=Hello, Kafka!        

• Check your application logs. You should see the sent and received messages.

Advanced Configurations

Partitioning and Replication

• Partitions: Divide a topic log into multiple logs for scalability.
• Replication: Duplicate data across brokers for fault tolerance.

Configure partitions and replication in Kafka when creating a topic:

bin/kafka-topics.sh --create --topic my_topic --bootstrap-server localhost:9092 --partitions 3 --replication-factor 2        

Error Handling and Retries

Implement error handlers in your consumer configuration to handle exceptions and retry logic.

Message Filtering

Use RecordFilterStrategy to filter out messages before they reach the consumer.

Conclusion

In this article, we've walked through integrating Apache Kafka with a Spring Boot application. We covered setting up Kafka, creating a producer and consumer, and touched on advanced configurations. With this foundation, you're well on your way to building scalable and real-time applications using Kafka and Spring Boot.

References

• Apache Kafka Documentation
• Spring for Apache Kafka Documentation