Integrating Apache Kafka with Spring Boot

This guide walks through the process of setting up Apache Kafka in Docker and integrating it with a Spring Boot application using spring-kafka.

---

🐳 Step 1: Set Up Kafka with Docker

1.1 Pull the Kafka Docker Image

docker pull apache/kafka

1.2 Start a Kafka Broker

docker run -d --name broker -p 9092:9092 apache/kafka:latest

1.3 Open a Shell in the Broker Container

docker exec --workdir /opt/kafka/bin/ -it broker sh

1.4 Create a Kafka Topic (test-topic)

./kafka-topics.sh --bootstrap-server localhost:9092 --create --topic test-topic

---

⚙️ Step 2: Set Up Spring Boot Project

Use Spring Initializr or Maven to create a Spring Boot project with the following dependencies:

2.1 Required Dependencies

<dependencies>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
</dependencies>

---

🔧 Step 3: Configure Kafka in Spring Boot

application.properties

spring.kafka.bootstrap-servers=localhost:9092 spring.kafka.consumer.group-id=my-group-id

---

📨 Step 4: Kafka Producer Setup

4.1 KafkaProducerConfig

package com.example.kafka.producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
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 KafkaProducerConfig {
    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        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());
    }
}

4.2 MessageProducer

package com.example.kafka.producer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
@Component
public class MessageProducer {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    public void sendMessage(String topic, String message) {
        kafkaTemplate.send(topic, message);
    }
}

---

📥 Step 5: Kafka Consumer Setup

5.1 KafkaConsumerConfig

package com.example.kafka.consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
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 KafkaConsumerConfig {
    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configProps.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group-id");
        configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return new DefaultKafkaConsumerFactory<>(configProps);
    }
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}

5.2 MessageConsumer

package com.example.kafka.consumer;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
@Component
public class MessageConsumer {
    @KafkaListener(topics = "test-topic", groupId = "my-group-id")
    public void listen(String message) {
        System.out.println("Received message: " + message);
    }
}

---

🌐 Step 6: REST Controller for Testing

package com.example.kafka.controller;
import com.example.kafka.producer.MessageProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class KafkaController {
    @Autowired
    private MessageProducer messageProducer;
    @PostMapping("/send")
    public String sendMessage(@RequestParam("message") String message) {
        messageProducer.sendMessage("test-topic", message);
        return "Message sent: " + message;
    }
}

---

✅ Testing the Setup

1. Run your Spring Boot application.

2. Use Postman or curl to test:

   curl -X POST "http://localhost:8080/send?message=Hello_Kafka"

3. Check the application logs — the consumer should print:

   Received message: Hello_Kafka

---

📌 Notes

• Ensure the Kafka container is running and accessible at localhost:9092.

• Use the same topic name (test-topic) in both producer and consumer.

• If you want to avoid hardcoding localhost:9092 multiple times, consider moving the config to application.properties.