Spring Boot Integration with Kafka

Last updated: November 1, 2026Nov 1, 2026

Spring Boot integration with Kafka lets one service publish events, such as "an order was placed", and lets other services react to them in their own time. Apache Kafka stores these events durably and delivers them to every interested service, so your applications stay loosely coupled and keep working even when one of them is slow or down.

This tutorial builds a small shop service that publishes order events and reserves stock when they arrive. You will set up the Kafka starter, send JSON messages with KafkaTemplate, consume them with @KafkaListener, add retries and a dead letter topic, and test the whole flow against a real broker with Testcontainers.

1. What Is Spring Boot Kafka Integration?

Apache Kafka is a distributed event log. Producers append messages, called records, to named streams called topics. Consumers read those records at their own pace. Unlike a classic queue, Kafka keeps records after they are read, for a configured retention time, so several independent services can read the same events.

Spring for Apache Kafka (the spring-kafka library) wraps the plain Kafka Java client in familiar Spring ideas: a template for sending and an annotation for listening. Spring Boot then auto-configures it from spring.kafka.* properties, so most apps need very little Java configuration.

If Kafka itself is new to you, the Apache Kafka tutorial covers brokers, topics and the KRaft architecture in depth.

2. Why Use Kafka in a Spring Boot Application

Think of a shop. When a customer places an order, the inventory, payment, email and analytics services all need to know. Calling each of them over HTTP from the order service has problems:

Tight coupling. The order service must know every downstream service, and change whenever a new one is added.
Fragility. If the email service is down, the order request fails or needs complex retry code.
Slow responses. The customer waits for every call to finish.

With Kafka, the order service publishes one OrderPlacedEvent and returns. Each other service consumes it independently. A service that is down simply catches up when it comes back, because Kafka kept the events. If you are choosing between brokers, see Kafka vs RabbitMQ.

3. How Kafka Works: Topics, Partitions and Consumer Groups

Figure 1: the key picks the partition, and each partition is read by exactly one listener thread in the consumer group.

Four ideas explain almost everything Kafka does:

Partition. A topic is split into partitions, ordered logs that can live on different brokers. Partitions are how Kafka scales. Read more in Kafka partitions.
Key. Each record can have a key. Kafka hashes the key to choose the partition, so all records with the same key (here, the same orderId) go to the same partition and keep their order.
Offset. Each record in a partition gets a sequence number called an offset. A consumer remembers how far it has read by committing its offset.
Consumer group. Consumers that share a group.id split the partitions between them, and each partition goes to exactly one member. A different group, such as an email service, gets its own copy of every record. The Kafka consumer guide explains group rebalancing.

A consequence worth remembering: a group can't run more useful listener threads than the topic has partitions. Extra threads sit idle.

4. Setup: Dependencies and a Local Kafka Broker

4.1 Dependencies

pom.xml

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

    <!-- Test -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-kafka-test</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-testcontainers</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.testcontainers</groupId>
        <artifactId>testcontainers-junit-jupiter</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.testcontainers</groupId>
        <artifactId>testcontainers-kafka</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.awaitility</groupId>
        <artifactId>awaitility</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

spring-boot-starter-kafka brings spring-kafka and the Kafka client. The web starter is only needed for the REST endpoint, and it also brings Jackson, which the JSON serializer uses. The test dependencies are used in Section 10.

Version note:

Starters: Spring Boot 4 adds a dedicated spring-boot-starter-kafka and spring-boot-starter-kafka-test. In Spring Boot 3.x you added org.springframework.kafka:spring-kafka directly, and spring-kafka-test for tests.
JSON classes: Spring Kafka 4 is built for Jackson 3. Use JacksonJsonSerializer and JacksonJsonDeserializer. The older JsonSerializer and JsonDeserializer, which you will see in many 3.x tutorials, are deprecated.
Kafka 4: Kafka 4 runs only in KRaft mode, so there is no ZooKeeper in the setup or in embedded test brokers.
Testcontainers 2: the Kafka module is testcontainers-kafka, and the container class is org.testcontainers.kafka.KafkaContainer.

4.2 A Local Kafka Broker

The official apache/kafka image starts a single-node broker in KRaft mode, reachable on localhost:9092:

compose.yaml

services:
  kafka:
    image: apache/kafka:4.2.0
    ports:
      - "9092:9092"

Terminal

docker compose up -d

4.3 Application Properties

application.properties

spring.application.name=shop-service
spring.kafka.bootstrap-servers=localhost:9092

# Producer: String keys (the default) and JSON values
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JacksonJsonSerializer

# Consumer: read from the beginning when the group is new
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
spring.kafka.consumer.properties[spring.deserializer.value.delegate.class]=org.springframework.kafka.support.serializer.JacksonJsonDeserializer
spring.kafka.consumer.properties[spring.json.trusted.packages]=com.javabytechie.shop.order

# One listener thread per partition
spring.kafka.listener.concurrency=3

Keys use Spring Boot's default StringSerializer and StringDeserializer. Values are JSON. On the consumer side, ErrorHandlingDeserializer wraps the JSON deserializer so that a broken message causes a handled error instead of an endless loop (see Section 9). trusted.packages limits which classes may be created from incoming JSON.

5. Core Building Blocks of Spring Kafka

5.1 KafkaTemplate: Sending Messages

Spring Boot auto-configures a KafkaTemplate, so you just inject it. send(topic, key, value) doesn't block. It returns a CompletableFuture<SendResult> that completes when the broker confirms the write, or fails if it can't.

OrderEventPublisher.java (excerpt)

kafkaTemplate.send("orders", event.orderId(), event)
        .whenComplete((result, ex) -> {
            if (ex == null) {
                log.info("Stored at offset {}", result.getRecordMetadata().offset());
            }
        });

5.2 @KafkaListener: Receiving Messages

@KafkaListener turns a method into a consumer. Spring creates a listener container, a background loop that polls Kafka, converts each record and calls your method. By default, the offset is committed after the records from a poll have been processed successfully, so a crash mid-processing means the records are delivered again.

InventoryListener.java (excerpt)

@KafkaListener(topics = "orders", groupId = "inventory-service")
public void onOrderPlaced(OrderPlacedEvent event) {
    stockService.reserve(event.orderId(), event.sku(), event.quantity());
}

Attribute	What it does
`topics`	Topic names to subscribe to
`groupId`	Consumer group; overrides `spring.kafka.consumer.group-id`
`id`	Listener ID; also used as the group ID unless `groupId` is set
`concurrency`	Number of consumer threads for this listener; overrides `spring.kafka.listener.concurrency`
`containerFactory`	Bean name of a custom listener container factory
`autoStartup`	Set to `"false"` to start the listener manually later

5.3 NewTopic and TopicBuilder: Creating Topics

Any NewTopic bean is created on the broker at startup by the auto-configured KafkaAdmin. Topics that already exist are left alone. TopicBuilder gives a fluent way to set partitions and replicas. In production, topics are often created by a platform team, but declaring them keeps local development simple.

5.4 JSON Serialization with JacksonJsonSerializer

Kafka stores bytes. A serializer turns your object into bytes on the way in, and a deserializer turns them back on the way out. JacksonJsonSerializer writes Java records as JSON and, by default, adds a type header naming the class. JacksonJsonDeserializer reads that header, but only creates classes from the trusted packages you list.

5.5 Message Keys and Ordering

Kafka guarantees order only within a partition. If an order can produce several events (placed, paid, shipped), use the orderId as the key. They then share a partition and are consumed in the order they were sent. Records with no key are spread across partitions, with no ordering between them.

6. Spring Boot Kafka Producer and Consumer Example

The example keeps the producer and consumer in one application so you can run it with one command. In a real system, InventoryListener would live in a separate inventory service. The REST layer follows the same style as the Spring Boot REST API tutorial.

6.1 Application and Topic Configuration

ShopApplication.java

package com.javabytechie.shop;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class ShopApplication {

    public static void main(String[] args) {
        SpringApplication.run(ShopApplication.class, args);
    }
}

KafkaTopicConfig.java

package com.javabytechie.shop.messaging;

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;

@Configuration
public class KafkaTopicConfig {

    public static final String ORDERS_TOPIC = "orders";
    public static final String ORDERS_DLT_TOPIC = "orders-dlt";

    @Bean
    public NewTopic ordersTopic() {
        return TopicBuilder.name(ORDERS_TOPIC).partitions(3).replicas(1).build();
    }

    @Bean
    public NewTopic ordersDeadLetterTopic() {
        return TopicBuilder.name(ORDERS_DLT_TOPIC).partitions(3).replicas(1).build();
    }
}

6.2 Request, Event and Response Records

PlaceOrderRequest.java

package com.javabytechie.shop.order;

import java.math.BigDecimal;

public record PlaceOrderRequest(String customerId, String sku, int quantity, BigDecimal totalAmount) {
}

OrderPlacedEvent.java

package com.javabytechie.shop.order;

import java.math.BigDecimal;
import java.time.Instant;

public record OrderPlacedEvent(
        String orderId,
        String customerId,
        String sku,
        int quantity,
        BigDecimal totalAmount,
        Instant placedAt) {
}

OrderAcceptedResponse.java

package com.javabytechie.shop.order;

public record OrderAcceptedResponse(String orderId, String status) {
}

6.3 The Producer

OrderEventPublisher.java

package com.javabytechie.shop.order;

import java.util.concurrent.CompletableFuture;

import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import com.javabytechie.shop.messaging.KafkaTopicConfig;

@Service
public class OrderEventPublisher {

    private static final Logger log = LoggerFactory.getLogger(OrderEventPublisher.class);

    private final KafkaTemplate<String, OrderPlacedEvent> kafkaTemplate;

    public OrderEventPublisher(KafkaTemplate<String, OrderPlacedEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public CompletableFuture<SendResult<String, OrderPlacedEvent>> publish(OrderPlacedEvent event) {
        return kafkaTemplate.send(KafkaTopicConfig.ORDERS_TOPIC, event.orderId(), event)
                .whenComplete((result, ex) -> {
                    if (ex != null) {
                        log.error("Could not publish order {}", event.orderId(), ex);
                    } else {
                        RecordMetadata metadata = result.getRecordMetadata();
                        log.info("Published order {} to {}-{} at offset {}",
                                event.orderId(), metadata.topic(), metadata.partition(), metadata.offset());
                    }
                });
    }
}

OrderController.java

package com.javabytechie.shop.order;

import java.time.Instant;
import java.util.UUID;

import org.springframework.http.HttpStatus;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("/api/orders")
public class OrderController {

    private final OrderEventPublisher publisher;

    public OrderController(OrderEventPublisher publisher) {
        this.publisher = publisher;
    }

    @PostMapping
    @ResponseStatus(HttpStatus.ACCEPTED)
    public OrderAcceptedResponse placeOrder(@RequestBody PlaceOrderRequest request) {
        String orderId = UUID.randomUUID().toString();
        OrderPlacedEvent event = new OrderPlacedEvent(orderId, request.customerId(), request.sku(),
                request.quantity(), request.totalAmount(), Instant.now());

        // Wait for the broker to confirm, so 202 means the event is safely stored
        publisher.publish(event).join();
        return new OrderAcceptedResponse(orderId, "PENDING");
    }
}

The controller waits for the broker's confirmation with join(). If Kafka is unreachable, the request fails with 500 instead of returning 202 Accepted for an event that was never stored.

6.4 The Consumer

StockService.java

package com.javabytechie.shop.inventory;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

@Service
public class StockService {

    private static final Logger log = LoggerFactory.getLogger(StockService.class);

    // In-memory for the example; a real service would use a database table
    private final Map<String, Integer> stock = new HashMap<>(Map.of("KB-100", 50, "MS-200", 20));
    private final Set<String> processedOrders = new HashSet<>();

    public synchronized void reserve(String orderId, String sku, int quantity) {
        if (processedOrders.contains(orderId)) {
            log.info("Order {} was already processed, skipping duplicate", orderId);
            return;
        }
        int available = stock.getOrDefault(sku, 0);
        if (available < quantity) {
            throw new InsufficientStockException(sku, quantity, available);
        }
        stock.put(sku, available - quantity);
        processedOrders.add(orderId);
        log.info("Reserved {} x {} for order {} ({} left)", quantity, sku, orderId, available - quantity);
    }

    public synchronized int availableStock(String sku) {
        return stock.getOrDefault(sku, 0);
    }
}

InsufficientStockException.java

package com.javabytechie.shop.inventory;

public class InsufficientStockException extends RuntimeException {

    public InsufficientStockException(String sku, int requested, int available) {
        super("Not enough stock for " + sku + ": requested " + requested + ", available " + available);
    }
}

InventoryListener.java

package com.javabytechie.shop.inventory;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import com.javabytechie.shop.messaging.KafkaTopicConfig;
import com.javabytechie.shop.order.OrderPlacedEvent;

@Component
public class InventoryListener {

    private static final Logger log = LoggerFactory.getLogger(InventoryListener.class);

    private final StockService stockService;

    public InventoryListener(StockService stockService) {
        this.stockService = stockService;
    }

    @KafkaListener(topics = KafkaTopicConfig.ORDERS_TOPIC, groupId = "inventory-service")
    public void onOrderPlaced(OrderPlacedEvent event,
                              @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
                              @Header(KafkaHeaders.OFFSET) long offset) {
        log.info("Received order {} from partition {} at offset {}", event.orderId(), partition, offset);
        stockService.reserve(event.orderId(), event.sku(), event.quantity());
    }
}

Kafka delivers at least once: after a crash or rebalance, a record can arrive twice. StockService remembers processed order IDs, so a duplicate is skipped instead of reserving stock twice. This property is called idempotency. synchronized protects the maps, because three listener threads call the service at the same time.

6.5 Error Handling and the Dead Letter Listener

KafkaErrorHandlingConfig.java

package com.javabytechie.shop.messaging;

import org.apache.kafka.common.TopicPartition;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.util.backoff.FixedBackOff;

import com.javabytechie.shop.inventory.InsufficientStockException;
import com.javabytechie.shop.order.OrderPlacedEvent;

@Configuration
public class KafkaErrorHandlingConfig {

    @Bean
    public DefaultErrorHandler kafkaErrorHandler(KafkaTemplate<String, OrderPlacedEvent> kafkaTemplate) {
        DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate,
                (record, exception) -> new TopicPartition(KafkaTopicConfig.ORDERS_DLT_TOPIC, record.partition()));

        // 1 second between attempts, 2 retries = 3 attempts in total
        DefaultErrorHandler errorHandler = new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 2L));

        // Retrying will not create stock, so send these straight to the dead letter topic
        errorHandler.addNotRetryableExceptions(InsufficientStockException.class);
        return errorHandler;
    }
}

OrderDeadLetterListener.java

package com.javabytechie.shop.messaging;

import java.nio.charset.StandardCharsets;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Header;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.stereotype.Component;

import com.javabytechie.shop.order.OrderPlacedEvent;

@Component
public class OrderDeadLetterListener {

    private static final Logger log = LoggerFactory.getLogger(OrderDeadLetterListener.class);

    @KafkaListener(topics = KafkaTopicConfig.ORDERS_DLT_TOPIC, groupId = "orders-dlt-monitor")
    public void onDeadLetter(ConsumerRecord<String, OrderPlacedEvent> record) {
        Header cause = record.headers().lastHeader(KafkaHeaders.DLT_EXCEPTION_CAUSE_FQCN);
        String causeName = cause == null ? "unknown" : new String(cause.value(), StandardCharsets.UTF_8);
        log.warn("Dead letter for order {} (cause: {})", record.key(), causeName);
    }
}

Spring Boot attaches any CommonErrorHandler bean, such as this DefaultErrorHandler, to the auto-configured listener container factory. Section 7 explains how it works.

6.6 Run It

Terminal

docker compose up -d
./mvnw spring-boot:run

curl -i -X POST http://localhost:8080/api/orders \
  -H "Content-Type: application/json" \
  -d '{"customerId":"C-1001","sku":"KB-100","quantity":2,"totalAmount":178.00}'

Console

HTTP/1.1 202
{"orderId":"4f6c2a9e-8d1b-4c7a-9f2e-1b3d5a7c9e01","status":"PENDING"}

Console

INFO c.j.shop.order.OrderEventPublisher     : Published order 4f6c2a9e-8d1b-4c7a-9f2e-1b3d5a7c9e01 to orders-1 at offset 0
INFO c.j.shop.inventory.InventoryListener   : Received order 4f6c2a9e-8d1b-4c7a-9f2e-1b3d5a7c9e01 from partition 1 at offset 0
INFO c.j.shop.inventory.StockService        : Reserved 2 x KB-100 for order 4f6c2a9e-8d1b-4c7a-9f2e-1b3d5a7c9e01 (48 left)

Now order more MS-200 mice than are in stock:

Terminal

curl -i -X POST http://localhost:8080/api/orders \
  -H "Content-Type: application/json" \
  -d '{"customerId":"C-1002","sku":"MS-200","quantity":500,"totalAmount":12500.00}'

Console

INFO c.j.shop.inventory.InventoryListener   : Received order 9b2d7e41-3a5c-4f8e-b6d0-7c1e2f3a4b5c from partition 0 at offset 0
WARN c.j.shop.messaging.OrderDeadLetterListener : Dead letter for order 9b2d7e41-3a5c-4f8e-b6d0-7c1e2f3a4b5c (cause: com.javabytechie.shop.inventory.InsufficientStockException)

The API still returns 202, because publishing worked. The failure happens later, in the consumer. InsufficientStockException is not retryable, so the record goes straight to orders-dlt. Order IDs, partitions and offsets will differ on your machine.

7. Error Handling: Retries and Dead Letter Topics

Figure 2: retryable failures get two more tries one second apart; everything else goes straight to the dead letter topic.

When a listener throws, the listener container hands the record to the error handler. DefaultErrorHandler then decides what happens next:

Retryable exceptions are retried by seeking back to the failed record and polling it again. With new FixedBackOff(1000L, 2L), that means two retries one second apart, so three attempts in total. Without your own bean, Spring Kafka's default is ten attempts with no delay, after which the record is only logged.
Non-retryable exceptions skip the retries. Some are non-retryable by default, such as DeserializationException, MessageConversionException and ClassCastException. You can add your own with addNotRetryableExceptions(...).
The recoverer runs when retries are over. DeadLetterPublishingRecoverer copies the record to a dead letter topic (DLT), a parking topic for records that could not be processed. It adds headers describing the exception, then the offset is committed and the listener moves on.

The example routes failures to orders-dlt on the same partition number as the original record. That is why the DLT was created with the same three partitions.

Retries block the partition while they wait, because Kafka keeps records in order. Keep the back-off short. For long waits, look at Spring Kafka's @RetryableTopic, which moves failing records to separate retry topics so the main partition keeps flowing.

8. Spring Boot Kafka Configuration Properties

Property	Default	What it controls
`spring.kafka.bootstrap-servers`	`localhost:9092`	Broker addresses the client connects to first
`spring.kafka.consumer.group-id`	none	Default consumer group for listeners without `groupId`
`spring.kafka.consumer.auto-offset-reset`	`latest` (Kafka default)	Where a new group starts: `earliest` or `latest`
`spring.kafka.consumer.max-poll-records`	500 (Kafka default)	Maximum records returned by one poll
`spring.kafka.producer.acks`	`all` (Kafka 3.0+ default)	How many replicas must confirm a write
`spring.kafka.producer.value-serializer`	`StringSerializer`	How values are turned into bytes
`spring.kafka.consumer.value-deserializer`	`StringDeserializer`	How bytes are turned back into values
`spring.kafka.listener.concurrency`	1	Consumer threads per listener
`spring.kafka.listener.ack-mode`	`batch`	When offsets are committed
`spring.kafka.producer.transaction-id-prefix`	none	Enables Kafka transactions and auto-configures a `KafkaTransactionManager`
`spring.kafka.consumer.properties[...]`	none	Any Kafka or Spring Kafka property without a dedicated key

Kafka producers have used idempotent writes with acks=all by default since client 3.0, so a retried send doesn't create a duplicate in the topic. Consumer-side duplicates are still possible, which is why the listener itself must be idempotent.

9. Common Kafka Errors and How to Fix Them

Error or symptom	Cause	Fix
`The class '...' is not in the trusted packages`	The JSON deserializer refuses to create a class from an unlisted package	Add the event's package to `spring.json.trusted.packages`; avoid `*` in production
The same broken record is logged over and over	A record that can't be deserialized fails on every poll	Wrap the deserializer in `ErrorHandlingDeserializer`, as in Section 4.3
Listener never receives old messages	A new group starts at `latest` by default	Set `auto-offset-reset=earliest`, or check the group's committed offsets
Some listener threads are idle	More concurrency than partitions	Match `concurrency` to the partition count, or add partitions
Events for one order are processed out of order	No key, or different keys, so the events land in different partitions	Use the business ID (`orderId`) as the record key
Stock reserved twice after a restart	At-least-once delivery redelivered a processed record	Make listeners idempotent, for example by tracking processed IDs
`TimeoutException` when sending	Broker unreachable, or wrong `bootstrap-servers`	Check the address and that the broker is running; handle the failed future

10. Testing Kafka Listeners with Testcontainers

Integration tests should run against a real broker. @ServiceConnection on a Testcontainers KafkaContainer starts Kafka in Docker and points spring.kafka.bootstrap-servers at it automatically. The Testcontainers tutorial covers the setup in more detail.

OrderFlowIntegrationTest.java

package com.javabytechie.shop;

import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;

import java.math.BigDecimal;
import java.time.Duration;
import java.time.Instant;

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.kafka.KafkaContainer;

import com.javabytechie.shop.inventory.StockService;
import com.javabytechie.shop.order.OrderEventPublisher;
import com.javabytechie.shop.order.OrderPlacedEvent;

@SpringBootTest
@Testcontainers
class OrderFlowIntegrationTest {

    @Container
    @ServiceConnection
    static KafkaContainer kafka = new KafkaContainer("apache/kafka:4.2.0");

    @Autowired
    private OrderEventPublisher publisher;

    @Autowired
    private StockService stockService;

    @Test
    void reservesStockWhenOrderIsPublished() {
        int before = stockService.availableStock("KB-100");

        publisher.publish(order("order-test-1", "KB-100", 2)).join();

        await().atMost(Duration.ofSeconds(15))
                .untilAsserted(() -> assertThat(stockService.availableStock("KB-100")).isEqualTo(before - 2));
    }

    @Test
    void reservesStockOnlyOnceForDuplicateEvents() {
        int before = stockService.availableStock("MS-200");
        OrderPlacedEvent event = order("order-test-2", "MS-200", 3);

        publisher.publish(event).join();
        publisher.publish(event).join();

        // The stock must drop by 3 and then stay there, even after the duplicate is consumed
        await().atMost(Duration.ofSeconds(15))
                .during(Duration.ofSeconds(2))
                .until(() -> stockService.availableStock("MS-200") == before - 3);
    }

    private static OrderPlacedEvent order(String orderId, String sku, int quantity) {
        return new OrderPlacedEvent(orderId, "C-TEST", sku, quantity, new BigDecimal("10.00"), Instant.now());
    }
}

Both tests read the stock before publishing and assert the change after it. Because the listener runs on another thread, Awaitility polls until the expected value appears. The second test publishes the same event twice, and during(...) checks that the stock stays at the expected value for two seconds, proving the duplicate was ignored.

Console

[INFO] Tests run: 2, Failures: 0, Errors: 0, Skipped: 0

spring-boot-starter-kafka-test also provides @EmbeddedKafka, an in-memory KRaft broker that needs no Docker. It is faster to start, but Testcontainers runs the same Kafka image you use in production.

11. When to Use Kafka and When to Avoid It

Good use cases	Poor use cases
Several services reacting to the same business events	A simple request that needs an immediate answer
High-volume streams: clicks, logs, sensor readings	A few messages a day between two services
Replaying past events to rebuild a read model	Task queues that need per-message priority or delays
Decoupling services that release independently	A small team with no one to run or pay for a Kafka cluster

If a plain HTTP call or Spring's in-process application events solve the problem, start there. Add Kafka when several consumers, durability or replay are real requirements.

12. Best Practices

Key records by the business ID. Using orderId as the key keeps all events for one order in order.
Make every listener idempotent. Expect each record at least once, and skip IDs you have already processed.
Wrap deserializers in ErrorHandlingDeserializer. One bad message should end up in the DLT, not block a partition.
Send failures to a dead letter topic. Mark errors that retrying can't fix as non-retryable, and monitor the DLT.
Keep in-place retries short. Retries block the partition; use @RetryableTopic for long back-offs.
Trust only your own event packages. List exact packages in spring.json.trusted.packages instead of *.
Handle the send result. Log or act on a failed CompletableFuture, and wait for it when the caller needs a guarantee.
Match concurrency to partitions. Threads beyond the partition count do nothing.
Test against a real broker. Use Testcontainers with @ServiceConnection for listener tests.

13. Summary

Spring Boot integration with Kafka comes down to a few steps: add spring-boot-starter-kafka, point spring.kafka.bootstrap-servers at your broker, publish events with KafkaTemplate using a business ID as the key, and consume them with @KafkaListener in a consumer group sized to your partitions. Use JacksonJsonSerializer for JSON payloads, guard consumers with ErrorHandlingDeserializer, and give the listener container a DefaultErrorHandler with a short back-off and a dead letter topic. The main recommendation: assume every message can arrive twice and some will fail, so make listeners idempotent and route failures to a DLT from day one, then prove the flow with a Testcontainers integration test.

Was this tutorial helpful?

Help Us Improve

Spring Boot Integration with Kafka

1. What Is Spring Boot Kafka Integration?

2. Why Use Kafka in a Spring Boot Application

3. How Kafka Works: Topics, Partitions and Consumer Groups

4. Setup: Dependencies and a Local Kafka Broker

4.1 Dependencies

4.2 A Local Kafka Broker

4.3 Application Properties

5. Core Building Blocks of Spring Kafka

5.1 KafkaTemplate: Sending Messages

5.2 @KafkaListener: Receiving Messages

5.3 NewTopic and TopicBuilder: Creating Topics

5.4 JSON Serialization with JacksonJsonSerializer

5.5 Message Keys and Ordering

6. Spring Boot Kafka Producer and Consumer Example

6.1 Application and Topic Configuration

6.2 Request, Event and Response Records

6.3 The Producer

6.4 The Consumer

6.5 Error Handling and the Dead Letter Listener

6.6 Run It

7. Error Handling: Retries and Dead Letter Topics

8. Spring Boot Kafka Configuration Properties

9. Common Kafka Errors and How to Fix Them

10. Testing Kafka Listeners with Testcontainers

11. When to Use Kafka and When to Avoid It

12. Best Practices

13. Summary