Backend Engineering

Integrating Spring Boot with Apache Kafka Using Java 21 and Maven

A complete event-driven example covering Kafka concepts, local KRaft setup, typed JSON events, producer acknowledgements, consumer groups, retries, dead-letter handling, testing, and production controls.

Author
Published
Published
Updated
Updated
Reading time
15 minute read
Apache KafkaSpring BootSpring KafkaJava 21MavenMessaging
A Spring Boot producer writing to a three-partition Kafka topic, a three-member consumer group reading it, and one failed record routed to a dead-letter topic

Apache Kafka is a distributed event-streaming platform. A producer writes records to a topic, Kafka stores those records in ordered partitions, and consumers read them while tracking progress through offsets. Spring Boot and Spring for Apache Kafka remove much of the connection and listener boilerplate while leaving delivery semantics, event design, failure handling, and operations under application control.

This is a complete Spring Boot 4 Kafka example: a single Maven project with a local broker, a typed producer, a consumer group, retries, a dead-letter topic, and an embedded-broker integration test. It is written against Boot 4.1 and Spring Kafka 4.1 rather than the 3.x line, which matters here because the JSON serializers, the backoff annotation, and the test helper all changed names.

Kafka concepts and architecture

ConceptMeaningDesign implication
TopicA named stream of recordsUse names that describe business events, not implementation classes
PartitionAn ordered append-only log within a topicOrdering is guaranteed only inside one partition
Record keyA value used by the partitionerUse a stable aggregate key, such as orderId, when related events need ordering
Consumer groupConsumers cooperating to process a topicWithin a group, one partition is assigned to at most one consumer at a time
OffsetA record position inside a partitionCommitting an offset records processing progress, not business-side-effect success by itself
ReplicationCopies of partitions across brokersUse a production replication factor that tolerates broker failure; the local one-broker example uses one replica

The request flow in this example is: HTTP request -> validate payload -> create an immutable event -> publish with the order ID as key -> broker acknowledgement -> consumer group receives the record -> handler performs an idempotent action. If processing fails, a blocking retry policy replays the record and then publishes exhausted failures to a dead-letter topic.

Prerequisites

  • JDK 21 and Maven 3.9 or newer.
  • Docker for the local broker command below, or an existing Kafka cluster.
  • A basic understanding of asynchronous delivery and eventual consistency.
  • A decision about the event contract, key, retention, partition count, and ownership before production rollout.

Local Kafka setup

Kafka 4 runs without ZooKeeper. The official Apache image starts a single local broker suitable for development. Do not copy the one-broker topology or replication factor into production.

Start and inspect a local Kafka 4.3.1 broker
docker run --detach   --name kafka   --publish 9092:9092   apache/kafka:4.3.1

docker exec kafka   /opt/kafka/bin/kafka-topics.sh   --bootstrap-server localhost:9092   --list

Spring Boot project setup and Maven dependencies

Spring Boot 4.1 provides a Kafka starter and manages compatible library versions. spring-boot-starter-kafka-test supplies the embedded broker and Spring Kafka test utilities used later.

pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>4.1.1</version>
        <relativePath/>
    </parent>

    <groupId>com.example</groupId>
    <artifactId>kafka-orders</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>kafka-orders</name>

    <properties>
        <java.version>21</java.version>
    </properties>

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

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

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

A small package layout makes the event contract and operational configuration easy to find:

Suggested source layout
src/main/java/com/example/kafkaorders/
├── KafkaOrdersApplication.java
├── config/
│   ├── KafkaErrorConfiguration.java
│   ├── KafkaTopicConfiguration.java
│   └── KafkaTopicsProperties.java
├── events/
│   └── OrderCreatedEvent.java
├── messaging/
│   ├── DeadLetterMonitor.java
│   ├── OrderCreatedListener.java
│   ├── OrderEventPublisher.java
│   └── OrderNotificationHandler.java
└── web/
    ├── ApiError.java
    ├── CreateOrderEventRequest.java
    ├── GlobalExceptionHandler.java
    ├── OrderEventController.java
    └── PublishResponse.java

Application configuration

The producer requests acknowledgements from all in-sync replicas and enables idempotence. The consumer disables automatic commits and uses record-level acknowledgement. Type headers are disabled because this service owns one explicit event type; the consumer defines that type and limits trusted packages.

src/main/resources/application.yml
spring:
  application:
    name: kafka-orders
  kafka:
    bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}
    producer:
      acks: all
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JacksonJsonSerializer
      properties:
        "[enable.idempotence]": true
        "[spring.json.add.type.headers]": false
    consumer:
      group-id: order-notifications
      auto-offset-reset: earliest
      enable-auto-commit: false
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JacksonJsonDeserializer
      properties:
        "[spring.json.use.type.headers]": false
        "[spring.json.value.default.type]": com.example.kafkaorders.events.OrderCreatedEvent
        "[spring.json.trusted.packages]": com.example.kafkaorders.events
    listener:
      ack-mode: record
      observation-enabled: true

app:
  kafka:
    orders-topic: orders.created
    partitions: 3
    replicas: 1

management:
  endpoints:
    web:
      exposure:
        include: health,info,metrics
  endpoint:
    health:
      probes:
        enabled: true

server:
  shutdown: graceful

Application entry point and topic properties

src/main/java/com/example/kafkaorders/KafkaOrdersApplication.java
package com.example.kafkaorders;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.context.properties.ConfigurationPropertiesScan;

@SpringBootApplication
@ConfigurationPropertiesScan
public class KafkaOrdersApplication {

    public static void main(String[] args) {
        SpringApplication.run(KafkaOrdersApplication.class, args);
    }
}
src/main/java/com/example/kafkaorders/config/KafkaTopicsProperties.java
package com.example.kafkaorders.config;

import org.springframework.boot.context.properties.ConfigurationProperties;

@ConfigurationProperties("app.kafka")
public record KafkaTopicsProperties(
        String ordersTopic,
        int partitions,
        int replicas
) {
}

Topics, partitions, and consumer groups

The main and dead-letter topics use the same partition count because the dead-letter recoverer routes a failed record to the corresponding partition by default. NewTopic beans are convenient in development; many production teams provision topics through infrastructure-as-code and restrict application-level topic creation.

src/main/java/com/example/kafkaorders/config/KafkaTopicConfiguration.java
package com.example.kafkaorders.config;

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 KafkaTopicConfiguration {

    @Bean
    NewTopic ordersTopic(KafkaTopicsProperties properties) {
        return TopicBuilder.name(properties.ordersTopic())
                .partitions(properties.partitions())
                .replicas(properties.replicas())
                .build();
    }

    @Bean
    NewTopic ordersDeadLetterTopic(KafkaTopicsProperties properties) {
        return TopicBuilder.name(properties.ordersTopic() + "-dlt")
                .partitions(properties.partitions())
                .replicas(properties.replicas())
                .build();
    }
}

All instances using order-notifications belong to the same group and share partitions. A different group receives its own logical copy of the stream. Increasing consumer instances beyond the partition count does not increase active parallelism for that group.

Message serialisation and the event contract

Use an immutable event with an event ID for deduplication, an aggregate ID for partitioning, a timestamp, and business fields that describe what happened. Avoid serialising JPA entities or internal request classes directly.

src/main/java/com/example/kafkaorders/events/OrderCreatedEvent.java
package com.example.kafkaorders.events;

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

public record OrderCreatedEvent(
        UUID eventId,
        UUID orderId,
        String customerId,
        BigDecimal total,
        Instant occurredAt
) {
}

Producer implementation

KafkaTemplate.send returns a CompletableFuture. Waiting for it confirms that the broker acknowledged the record according to the producer configuration. It does not confirm that a downstream consumer completed its business action.

src/main/java/com/example/kafkaorders/messaging/OrderEventPublisher.java
package com.example.kafkaorders.messaging;

import com.example.kafkaorders.config.KafkaTopicsProperties;
import com.example.kafkaorders.events.OrderCreatedEvent;
import java.util.concurrent.CompletableFuture;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;

@Component
public class OrderEventPublisher {

    private final KafkaTemplate<String, OrderCreatedEvent> kafkaTemplate;
    private final KafkaTopicsProperties topics;

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

    public CompletableFuture<SendResult<String, OrderCreatedEvent>> publish(
            OrderCreatedEvent event
    ) {
        String key = event.orderId().toString();
        return kafkaTemplate.send(topics.ordersTopic(), key, event);
    }
}

REST API that publishes an event

src/main/java/com/example/kafkaorders/web/CreateOrderEventRequest.java
package com.example.kafkaorders.web;

import jakarta.validation.constraints.DecimalMin;
import jakarta.validation.constraints.NotBlank;
import jakarta.validation.constraints.NotNull;
import java.math.BigDecimal;
import java.util.UUID;

public record CreateOrderEventRequest(
        @NotNull UUID orderId,
        @NotBlank String customerId,
        @NotNull @DecimalMin("0.01") BigDecimal total
) {
}
src/main/java/com/example/kafkaorders/web/PublishResponse.java
package com.example.kafkaorders.web;

import java.util.UUID;

public record PublishResponse(
        UUID eventId,
        String topic,
        int partition,
        long offset
) {
}
src/main/java/com/example/kafkaorders/web/OrderEventController.java
package com.example.kafkaorders.web;

import com.example.kafkaorders.events.OrderCreatedEvent;
import com.example.kafkaorders.messaging.OrderEventPublisher;
import jakarta.validation.Valid;
import java.time.Instant;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import org.springframework.http.ResponseEntity;
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.RestController;

@RestController
@RequestMapping("/api/order-events")
public class OrderEventController {

    private final OrderEventPublisher publisher;

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

    @PostMapping
    public CompletableFuture<ResponseEntity<PublishResponse>> publish(
            @Valid @RequestBody CreateOrderEventRequest request
    ) {
        var event = new OrderCreatedEvent(
                UUID.randomUUID(),
                request.orderId(),
                request.customerId(),
                request.total(),
                Instant.now()
        );

        return publisher.publish(event).thenApply(result -> {
            var metadata = result.getRecordMetadata();
            var response = new PublishResponse(
                    event.eventId(),
                    metadata.topic(),
                    metadata.partition(),
                    metadata.offset()
            );
            return ResponseEntity.accepted().body(response);
        });
    }
}

Consumer implementation

The listener receives the complete ConsumerRecord so it can validate the key and access partition, offset, and headers. The handler must be idempotent because a record can be delivered again after a rebalance, timeout, crash, or retry.

src/main/java/com/example/kafkaorders/messaging/OrderNotificationHandler.java
package com.example.kafkaorders.messaging;

import com.example.kafkaorders.events.OrderCreatedEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

@Component
public class OrderNotificationHandler {

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

    public void handle(OrderCreatedEvent event) {
        // Replace this with an idempotent business action.
        log.info("Handled order-created event {} for order {}", event.eventId(), event.orderId());
    }
}
src/main/java/com/example/kafkaorders/messaging/OrderCreatedListener.java
package com.example.kafkaorders.messaging;

import com.example.kafkaorders.events.OrderCreatedEvent;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class OrderCreatedListener {

    private final OrderNotificationHandler handler;

    public OrderCreatedListener(OrderNotificationHandler handler) {
        this.handler = handler;
    }

    @KafkaListener(topics = "${app.kafka.orders-topic}")
    public void consume(ConsumerRecord<String, OrderCreatedEvent> record) {
        String expectedKey = record.value().orderId().toString();
        if (!expectedKey.equals(record.key())) {
            throw new IllegalArgumentException("Record key must match orderId");
        }

        handler.handle(record.value());
    }
}

Error handling, retries, and dead-letter topics

DefaultErrorHandler performs blocking retries in the listener container. The configuration below waits one second and retries twice after the original attempt. When attempts are exhausted, an explicit destination resolver sends the record to <original-topic>-dlt on the same partition. Permanent validation failures are sent directly to the DLT.

src/main/java/com/example/kafkaorders/config/KafkaErrorConfiguration.java
package com.example.kafkaorders.config;

import com.example.kafkaorders.events.OrderCreatedEvent;
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.CommonErrorHandler;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.util.backoff.FixedBackOff;

@Configuration
public class KafkaErrorConfiguration {

    @Bean
    CommonErrorHandler commonErrorHandler(
            KafkaTemplate<String, OrderCreatedEvent> kafkaTemplate
    ) {
        var recoverer = new DeadLetterPublishingRecoverer(
                kafkaTemplate,
                (record, exception) -> new TopicPartition(
                        record.topic() + "-dlt",
                        record.partition()));
        var backOff = new FixedBackOff(1_000L, 2L);
        var errorHandler = new DefaultErrorHandler(recoverer, backOff);

        errorHandler.addNotRetryableExceptions(IllegalArgumentException.class);
        return errorHandler;
    }
}
src/main/java/com/example/kafkaorders/messaging/DeadLetterMonitor.java
package com.example.kafkaorders.messaging;

import com.example.kafkaorders.events.OrderCreatedEvent;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class DeadLetterMonitor {

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

    @KafkaListener(
            topics = "${app.kafka.orders-topic}-dlt",
            groupId = "order-events-dlt-observer"
    )
    public void inspect(ConsumerRecord<String, OrderCreatedEvent> record) {
        log.error(
                "Record moved to DLT: topic={}, partition={}, offset={}, key={}, eventId={}",
                record.topic(),
                record.partition(),
                record.offset(),
                record.key(),
                record.value().eventId()
        );
    }
}

Blocking or non-blocking retries?

The handler above retries in place: the consumer thread sleeps, then tries the same record again. That is the right default, and it is also the thing that surprises teams in production, so it is worth being explicit about what it costs.

While a record is being retried, its partition makes no progress. A record that takes three attempts with a one-second wait holds its partition for roughly two extra seconds, and every record queued behind it on that partition waits too. With a slow downstream dependency and a generous backoff, blocking retries turn one bad record into partition-wide lag.

Spring Kafka's alternative is @RetryableTopic, which forwards the failed record to a separate retry topic and returns immediately. The original partition keeps moving.

Non-blocking retries with @RetryableTopic — an alternative to the handler above
package com.example.kafkaorders.messaging;

import com.example.kafkaorders.events.OrderCreatedEvent;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.BackOff;
import org.springframework.kafka.annotation.DltHandler;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.RetryableTopic;
import org.springframework.kafka.retrytopic.DltStrategy;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

@Component
public class NonBlockingOrderListener {

    @RetryableTopic(
            attempts = "4",
            backOff = @BackOff(delay = 1_000L, multiplier = 2.0, maxDelay = 8_000L),
            exclude = { IllegalArgumentException.class },
            dltStrategy = DltStrategy.FAIL_ON_ERROR,
            retryTopicSuffix = "-retry",
            dltTopicSuffix = "-dlt")
    @KafkaListener(topics = "${app.kafka.orders-topic}", groupId = "order-events-nonblocking")
    public void handle(ConsumerRecord<String, OrderCreatedEvent> record) {
        // Business logic. Throwing anything not listed in exclude triggers a retry.
    }

    @DltHandler
    public void handleDlt(
            ConsumerRecord<String, OrderCreatedEvent> record,
            @Header(KafkaHeaders.ORIGINAL_TOPIC) String originalTopic,
            @Header(KafkaHeaders.EXCEPTION_MESSAGE) String reason) {
        // Terminal handling: alert, record, or stage for replay.
    }
}

Spring Kafka creates the retry and dead-letter topics for you and applies the delay by controlling when each retry topic is consumed.

Choosing between the two retry models
Blocking (DefaultErrorHandler)Non-blocking (@RetryableTopic)
Partition progress during retryStalledContinues
Ordering within a keyPreservedLost — a retried record is reprocessed after later records
Topics createdOne DLTOne topic per retry level, plus a DLT
SuitsShort waits, order-sensitive workSlow or flaky dependencies, long backoffs
Operational surfaceSmallMore topics to monitor and reason about

Ordering is the decision, not throughput. If consumers must see events for one key in order — order created before order cancelled — non-blocking retries break that guarantee, because the retried record comes back after records that arrived later. Keep blocking retries with a short backoff for order-sensitive work, and reach for @RetryableTopic when a slow external dependency would otherwise hold up an entire partition.

This project keeps the blocking handler, because order events for one order must stay in sequence. The asynchronous Spring AI and Kafka pipeline faces the opposite trade-off: model calls take seconds, ordering between unrelated documents does not matter, and the cost of a retry is a paid API call rather than a database round trip.

HTTP-side error handling

Validation errors should return a client response. Broker timeouts and asynchronous send failures should return a stable service error and retain the underlying exception only in server logs. For business transactions that must publish exactly when a database change commits, use the transactional outbox pattern rather than a dual write from the controller.

src/main/java/com/example/kafkaorders/web/ApiError.java
package com.example.kafkaorders.web;

import java.time.Instant;

public record ApiError(String code, String message, Instant timestamp) {
}
src/main/java/com/example/kafkaorders/web/GlobalExceptionHandler.java
package com.example.kafkaorders.web;

import java.time.Instant;
import java.util.concurrent.CompletionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.kafka.KafkaException;
import org.springframework.web.bind.MethodArgumentNotValidException;
import org.springframework.web.bind.annotation.ExceptionHandler;
import org.springframework.web.bind.annotation.RestControllerAdvice;

@RestControllerAdvice
public class GlobalExceptionHandler {

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

    @ExceptionHandler(MethodArgumentNotValidException.class)
    ResponseEntity<ApiError> handleValidation(MethodArgumentNotValidException exception) {
        String message = exception.getBindingResult().getFieldErrors().stream()
                .findFirst()
                .map(error -> error.getDefaultMessage() == null
                        ? "Invalid request"
                        : error.getDefaultMessage())
                .orElse("Invalid request");

        return ResponseEntity.badRequest()
                .body(new ApiError("INVALID_REQUEST", message, Instant.now()));
    }

    @ExceptionHandler({CompletionException.class, KafkaException.class})
    ResponseEntity<ApiError> handlePublishFailure(Exception exception) {
        Throwable cause = exception instanceof CompletionException
                && exception.getCause() != null
                ? exception.getCause()
                : exception;

        log.error("Kafka publish failed", cause);
        return ResponseEntity.status(HttpStatus.SERVICE_UNAVAILABLE)
                .body(new ApiError(
                        "KAFKA_UNAVAILABLE",
                        "The event could not be published",
                        Instant.now()
                ));
    }

    @ExceptionHandler(Exception.class)
    ResponseEntity<ApiError> handleUnexpected(Exception exception) {
        log.error("Unexpected request failure", exception);
        return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
                .body(new ApiError(
                        "INTERNAL_ERROR",
                        "The request could not be completed",
                        Instant.now()
                ));
    }
}

Testing with an embedded broker

Unit-test event mapping and handler logic without Kafka. Use an embedded broker for serialisation, topic, key, and producer integration. The test below creates a real consumer with the same Jackson 3 deserializer, publishes through the application component, and verifies the stored record.

src/test/java/com/example/kafkaorders/messaging/OrderEventPublisherIntegrationTest.java
package com.example.kafkaorders.messaging;

import static org.assertj.core.api.Assertions.assertThat;

import com.example.kafkaorders.events.OrderCreatedEvent;
import java.math.BigDecimal;
import java.time.Duration;
import java.time.Instant;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.support.serializer.JacksonJsonDeserializer;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.test.annotation.DirtiesContext;

@SpringBootTest
@DirtiesContext
@EmbeddedKafka(
        partitions = 3,
        topics = {"orders.created", "orders.created-dlt"},
        bootstrapServersProperty = "spring.kafka.bootstrap-servers"
)
class OrderEventPublisherIntegrationTest {

    @Autowired
    private OrderEventPublisher publisher;

    @Autowired
    private EmbeddedKafkaBroker embeddedKafka;

    @Test
    void publishesJsonEventWithOrderIdAsKey() throws Exception {
        var consumerProperties = KafkaTestUtils.consumerProps(
                embeddedKafka.getBrokersAsString(),
                "publisher-test",
                false
        );
        consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        var valueDeserializer = new JacksonJsonDeserializer<OrderCreatedEvent>(
                OrderCreatedEvent.class,
                false
        );
        valueDeserializer.addTrustedPackages("com.example.kafkaorders.events");

        try (var consumer = new DefaultKafkaConsumerFactory<>(
                consumerProperties,
                new StringDeserializer(),
                valueDeserializer
        ).createConsumer()) {
            embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "orders.created");

            UUID orderId = UUID.randomUUID();
            var event = new OrderCreatedEvent(
                    UUID.randomUUID(),
                    orderId,
                    "customer-42",
                    new BigDecimal("49.95"),
                    Instant.now()
            );

            publisher.publish(event).get(10, TimeUnit.SECONDS);

            var record = KafkaTestUtils.getSingleRecord(
                    consumer,
                    "orders.created",
                    Duration.ofSeconds(10)
            );

            assertThat(record.key()).isEqualTo(orderId.toString());
            assertThat(record.value()).isEqualTo(event);
        }
    }
}

Add a second integration test where the handler throws a controlled exception and assert that the record reaches orders.created-dlt after the configured attempts. Also test replay tooling, schema changes, duplicate events, and broker unavailability before production.

How to run and verify the application

Build, start, publish, and inspect the topic
mvn clean verify
mvn spring-boot:run

curl --fail-with-body   --request POST   --header "Content-Type: application/json"   --data '{
    "orderId":"8f8128be-ae5a-4ec2-b86c-895882fba8af",
    "customerId":"customer-42",
    "total":49.95
  }'   http://localhost:8080/api/order-events

docker exec kafka   /opt/kafka/bin/kafka-topics.sh   --bootstrap-server localhost:9092   --describe   --topic orders.created

curl --fail http://localhost:8080/actuator/health

Confirm that the API returns HTTP 202 with a topic, partition, and offset; the application log shows the matching event ID; and the health endpoint reports the expected service state. A returned offset proves Kafka accepted the record, not that every consumer completed successfully.

Observability and security

Consumer lag is the metric that matters

Everything else is context. Consumer lag is the number of records between the newest offset on a partition and the offset your group has committed, so it answers the only question operators really have: is the application keeping up?

It is worth understanding why lag beats the obvious alternatives. Throughput looks healthy right up to the moment it does not, because a consumer that has fallen an hour behind still processes records at full speed. Processing latency measures one record, not the queue behind it. Lag measures the backlog directly, and it rises before users notice anything.

Read it per partition, not as a total. A single partition with growing lag while the others stay flat is the signature of an uneven key distribution or one slow record blocking progress — exactly the blocking-retry behaviour described above. A group-wide average hides that completely.

Two thresholds are worth alerting on separately:

  • Lag that is growing rather than merely large. A brief spike after a deployment or rebalance is normal; sustained growth means consumption is slower than production and will not recover on its own.
  • Oldest unprocessed record age, in seconds. Record count is hard to reason about across topics with different volumes, whereas "the oldest unhandled order is nine minutes old" maps directly onto a service expectation.

Actuator exposes Kafka consumer metrics through Micrometer when a metrics registry is present, and the broker reports lag per group for external monitoring. Add management.endpoints.web.exposure.include=health,info,metrics in the sample and confirm the consumer metrics appear before relying on them in production.

  • Track producer error rate and latency, consumer lag, rebalance count, processing latency, retry count, DLT volume, and oldest unprocessed event age.
  • Carry a correlation or trace identifier in record headers and logs. Avoid logging full payloads when they can contain personal or confidential data.
  • Use TLS for broker connections and SASL or mutual TLS for authentication. Apply topic-level ACLs using separate producer and consumer identities.
  • Store credentials in a secret manager and rotate them. Do not place passwords or private keys in application.yml.
  • Set message-size limits, request timeouts, retry budgets, quotas, and DLT retention deliberately.
  • Alert on sustained lag and DLT growth, but include enough context to distinguish downstream slowness from broker or deployment failures.

Production considerations

  • Event compatibility: prefer additive changes, document optional fields, and use a schema registry when contract governance requires it.
  • Idempotency: record processed event IDs or use an idempotent domain operation so redelivery does not duplicate effects.
  • Database consistency: use a transactional outbox or change-data-capture flow when a database write and event publication must be atomic from a business perspective.
  • Partition strategy: choose a key that balances ordering needs and load distribution; a hot key can overload one partition.
  • Capacity: size partitions for expected throughput and consumer parallelism, then load-test realistic payloads and downstream latency.
  • Retries: retry only transient failures. Long blocking retries reduce partition throughput; use retry topics or a delayed workflow when backoff must span minutes or hours.
  • DLT operations: define who owns review, retention, correction, replay, and audit. A DLT without an operating process is deferred data loss.
  • Shutdown and rebalances: allow graceful shutdown and keep processing below max.poll.interval.ms or pause/hand off long work safely.

Common problems and troubleshooting

SymptomLikely causeWhat to check
Producer cannot connectWrong advertised listener or bootstrap addressBroker logs, bootstrap-servers, Docker port mapping, DNS, firewall, and TLS/SASL settings
Listener receives no recordsGroup offset is already past the test record or topic name differsConsumer group, committed offsets, auto-offset-reset, topic spelling, and partition assignment
JSON deserialisation failsContract mismatch, wrong default type, or untrusted packagePayload fields, Jackson serializer classes, spring.json.value.default.type, trusted packages, and type-header policy
Messages are processed out of orderRelated records use different keys or multiple partitionsProducer key, partition count, retries, and whether the business requirement is per-key or global ordering
Consumer lag keeps growingProcessing is slower than arrival rateHandler latency, partition count, active consumers, downstream dependencies, GC, and retry volume
DLT publication failsMissing DLT, insufficient partitions, or producer permissionsCreate <topic>-dlt with at least the source partition count and grant the consumer service produce access

The same messaging concerns appear in systems that ingest device readings or connect business platforms asynchronously. The Smart Meter Management System case study and skills overview provide related context without claiming that this tutorial is the exact implementation used in those systems.

Conclusion

A reliable Kafka integration is more than a producer and an @KafkaListener. Keep the event contract explicit, let Spring Boot manage compatible client dependencies, make retries finite, route exhausted records to a dead-letter topic, and verify the behaviour with an embedded broker before deploying.

Two follow-ons build directly on this project. The asynchronous Spring AI and Kafka pipeline puts a model call inside the consumer, which changes the retry economics because every attempt costs money and takes seconds. If you are bringing an existing Boot 3 service to this baseline, the Spring Boot 3 to 4 migration guide covers the Kafka-adjacent parts of that upgrade — the Jackson 3 serializer rename shown above, the focused starters, and the Testcontainers artifact rename.

Frequently asked questions

Does an acknowledged producer send mean the event was processed?

No. It means Kafka accepted the record according to the configured acknowledgement policy. Consumer processing is separate and may happen later or fail.

How do I preserve order?

Kafka preserves order within a partition. Send related events with the same stable key so the partitioner routes them together, then avoid concurrent processing that reorders work for that key.

Should I retry every exception?

No. Retry transient failures such as temporary downstream unavailability. Validation and contract failures are usually permanent until the message or code changes, so route them for review instead of repeating them.

When should I use a schema registry?

Use one when multiple independently deployed producers and consumers need governed compatibility, discoverable schemas, and controlled evolution. JSON can still be used with a registry, depending on the platform and tooling.

Should I use `@RetryableTopic` instead of `DefaultErrorHandler`?

Decide on ordering, not throughput. @RetryableTopic keeps the partition moving but reprocesses a retried record after records that arrived later, so per-key ordering is lost. Use it when a slow dependency would otherwise stall a partition. Keep blocking retries with a short backoff when consumers depend on seeing events for one key in sequence.

Why does my `@BackOff` annotation not compile on Spring Boot 4?

Spring Kafka 4.x expects its own org.springframework.kafka.annotation.BackOff, not Spring Retry's @Backoff. Most examples online target Spring Kafka 3.x and import the wrong one. Check the import before changing the attributes.

Why is one partition lagging while the others are fine?

Usually an uneven key distribution sending disproportionate traffic to one partition, or a single record failing repeatedly and holding that partition through its blocking retries. Read lag per partition rather than as a group average, otherwise this pattern is invisible.

Do I need to change anything for Spring Boot 4 Kafka autoconfiguration?

The spring.kafka.* property namespace is unchanged, so most configuration carries over. What does change is the serializer classes — Spring Kafka 4.1 uses the Jackson 3 based JacksonJsonSerializer and JacksonJsonDeserializer, and the older JsonSerializer and JsonDeserializer are deprecated for removal. Because those names appear as property *values*, a stale one fails at runtime with a class-not-found error rather than at startup validation.

Official references

Complete implementation

Run the matching repository

The repository contains the complete project, configuration examples, tests, and operating notes used by this guide.

View the complete project on GitHub

Portfolio context

Related work and skills

Continue reading

Related technical insight

Next step

Need help with Java backend development, architecture, or AI integration?

Send a short message about the role, project, stack, and timeline, and I will get back to you.

Open to senior Java backend roles in the Netherlands and the wider EU, including hybrid and remote, plus architecture reviews and backend-led AI integration work.

Discuss a backend role or projectExplore backend case studies