DEV Community: Someswar Duddi

Integration testing Spring Boot with Testcontainers, PostgreSQL, Kafka and JUnit 5

Someswar Duddi — Fri, 16 Oct 2026 14:37:18 +0000

Mocks are great for unit tests, but they can't tell you whether your SQL works on PostgreSQL, whether your JSON survives a round trip through Kafka, or whether your listener is wired to the right topic. H2 and embedded Kafka get closer, but they aren't the things you run in production.

Testcontainers starts real dependencies in Docker for the duration of a test run. Since Spring Boot 3.1, @ServiceConnection wires them into your application automatically, so the setup is now a handful of lines.

This post tests a small service that consumes OrderPlaced events from Kafka and stores them in PostgreSQL. The examples use Spring Boot 3.5 and the Testcontainers 1.x version it manages. You'll need Docker (or a compatible runtime) available wherever the tests run.

Dependencies

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-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>junit-jupiter</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>postgresql</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>kafka</artifactId>
    <scope>test</scope>
</dependency>

Boot's dependency management covers all of these, plus Awaitility (pulled in by spring-boot-starter-test), which we'll use for asynchronous assertions.

The code under test

An event, an entity, a repository and a listener:

public record OrderPlaced(String orderId, String customerId, BigDecimal total, Instant placedAt) {}

@Entity
@Table(name = "orders")
public class Order {

    @Id
    private String id;
    private String customerId;
    private BigDecimal total;
    private Instant placedAt;

    protected Order() {}

    public Order(String id, String customerId, BigDecimal total, Instant placedAt) {
        this.id = id;
        this.customerId = customerId;
        this.total = total;
        this.placedAt = placedAt;
    }

    public String getId() { return id; }
    public String getCustomerId() { return customerId; }
    public BigDecimal getTotal() { return total; }
    public Instant getPlacedAt() { return placedAt; }
}

public interface OrderRepository extends JpaRepository<Order, String> {

    List<Order> findByCustomerIdOrderByPlacedAtDesc(String customerId);
}

@Component
public class OrderPlacedListener {

    private final OrderRepository orders;

    public OrderPlacedListener(OrderRepository orders) {
        this.orders = orders;
    }

    @KafkaListener(topics = "orders.placed", groupId = "order-projection")
    public void on(OrderPlaced event) {
        if (orders.existsById(event.orderId())) {
            return; // at-least-once delivery: ignore duplicates
        }
        orders.save(new Order(event.orderId(), event.customerId(), event.total(), event.placedAt()));
    }
}

The Kafka settings in application.yml are the usual JSON serialiser and deserialiser, with auto-offset-reset: earliest so the listener sees messages sent before its partitions were assigned. Notice there's no bootstrap-servers or datasource URL in the test setup below; that's what @ServiceConnection replaces.

Option 1: containers in the test class

The most direct approach declares containers as static fields:

@SpringBootTest(properties = "spring.jpa.hibernate.ddl-auto=create-drop")
@Testcontainers
class OrderPlacedListenerIT {

    @Container
    @ServiceConnection
    static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres:16-alpine");

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

    @Autowired
    KafkaTemplate<String, OrderPlaced> kafkaTemplate;

    @Autowired
    OrderRepository orders;

    @Test
    void storesOrderWhenEventArrives() {
        var event = new OrderPlaced("ord-1", "cust-42", new BigDecimal("19.99"),
                Instant.parse("2026-01-15T10:00:00Z"));

        kafkaTemplate.send("orders.placed", event.orderId(), event);

        await().atMost(Duration.ofSeconds(15)).untilAsserted(() ->
                assertThat(orders.findById("ord-1"))
                        .hasValueSatisfying(order -> {
                            assertThat(order.getCustomerId()).isEqualTo("cust-42");
                            assertThat(order.getTotal()).isEqualByComparingTo("19.99");
                        }));
    }
}

KafkaContainer here is org.testcontainers.kafka.KafkaContainer, which runs the official apache/kafka image in KRaft mode (no ZooKeeper). Spring Boot 3.4 and later support it with @ServiceConnection; on older Boot versions, use the Confluent-based org.testcontainers.containers.KafkaContainer instead.

How the pieces fit:

@Testcontainers and @Container hand lifecycle to the JUnit 5 extension. Static fields mean the containers start once for the class, not once per test.
@ServiceConnection tells Boot to derive spring.datasource.* and spring.kafka.bootstrap-servers from the running containers. Previously you'd write a @DynamicPropertySource method for this; that still works, and you'll need it for containers Boot doesn't recognise.
Awaitility handles the asynchrony. The listener runs on a different thread, so asserting immediately after send would be flaky. untilAsserted retries the AssertJ block until it passes or the timeout expires. Avoid Thread.sleep; it's both slower and less reliable.

ddl-auto=create-drop keeps the example short. If you use Flyway or Liquibase, drop that property and let migrations run against the container. Testing your real migrations on real PostgreSQL is one of the best reasons to use Testcontainers at all.

Option 2: a shared test configuration

Option 1 starts fresh containers for every test class. Once you have a dozen integration test classes, that adds up. The approach Spring Initializr generates is to declare containers as beans:

@TestConfiguration(proxyBeanMethods = false)
public class TestcontainersConfiguration {

    @Bean
    @ServiceConnection
    PostgreSQLContainer<?> postgresContainer() {
        return new PostgreSQLContainer<>(DockerImageName.parse("postgres:16-alpine"));
    }

    @Bean
    @ServiceConnection
    KafkaContainer kafkaContainer() {
        return new KafkaContainer(DockerImageName.parse("apache/kafka:3.8.0"));
    }
}

Test classes then import it:

@SpringBootTest(properties = "spring.jpa.hibernate.ddl-auto=create-drop")
@Import(TestcontainersConfiguration.class)
class OrderQueriesIT {

    @Autowired
    OrderRepository orders;

    @Test
    void returnsCustomerOrdersNewestFirst() {
        orders.save(new Order("q-1", "cust-7", new BigDecimal("5.00"), Instant.parse("2026-01-01T09:00:00Z")));
        orders.save(new Order("q-2", "cust-7", new BigDecimal("8.00"), Instant.parse("2026-01-02T09:00:00Z")));

        assertThat(orders.findByCustomerIdOrderByPlacedAtDesc("cust-7"))
                .extracting(Order::getId)
                .containsExactly("q-2", "q-1");
    }
}

Spring starts the containers with the application context and stops them when the context closes. Because Spring's test framework caches contexts, every class with the same configuration reuses the same context and the same containers.

The trade-off is that shared containers mean shared state. Tests must not depend on an empty database or topic. The habits that keep this manageable:

Use unique IDs per test (as above: q-1, ord-1, or random UUIDs) and assert on those, not on table counts.
Clean up explicitly where needed, for example with @Sql scripts or a @BeforeEach that deletes rows the test owns.
Keep context configuration identical across classes. A stray @MockitoBean or different properties creates a new context, and with it new containers.

Bonus: the same containers for local development

The shared configuration can also run the application locally against throwaway containers. Add a main class under src/test/java:

public class TestOrderServiceApplication {

    public static void main(String[] args) {
        SpringApplication.from(OrderServiceApplication::main)
                .with(TestcontainersConfiguration.class)
                .run(args);
    }
}

Run it from your IDE (or ./mvnw spring-boot:test-run) and you get PostgreSQL and Kafka without a separate docker-compose.yml to keep in sync.

Things to watch

Pin image versions. postgres:latest will change under you. Match the major version you run in production.
CI needs Docker. Most hosted runners have it. Some locked-down environments don't, so check before you commit to this approach.
Start-up time is real. Expect several seconds for Kafka and PostgreSQL to start. Shared containers keep that a one-off cost per run rather than per class.
Keep unit tests as unit tests. Integration tests are slower and should cover wiring, SQL, serialisation and configuration, not every branch of your business logic.

Summary

@ServiceConnection removes the property plumbing: declare a container and Boot connects to it.
Use Awaitility for anything asynchronous, such as Kafka listeners.
Move containers into a shared @TestConfiguration once you have more than a couple of test classes, and write tests that tolerate shared state.
Test against the same database engine and migrations you run in production.

Hardening an AWS EKS cluster: IRSA, NetworkPolicies, restricted Pod Security and Secrets Manager via CSI

Someswar Duddi — Fri, 16 Oct 2026 14:31:57 +0000

A freshly created EKS cluster works, but it's permissive. Every pod can reach every other pod, workloads can run as root, and the easy way to give an application AWS access is to attach permissions to the node role, which hands them to every pod on that node.

This post works through four controls that close those gaps, using a single example workload: an orders-api deployment in the orders namespace that needs a database password from AWS Secrets Manager.

IRSA — per-workload IAM roles instead of node-wide permissions.
NetworkPolicies — default-deny traffic, then allow only what's needed.
Pod Security Admission (restricted) — block privileged and root containers at admission time.
Secrets Store CSI driver — mount Secrets Manager secrets as files, rather than copying them into Kubernetes Secrets by hand.

The examples assume eksctl, kubectl, helm and the AWS CLI, and a cluster called prod-eu in eu-west-2.

1. IAM Roles for Service Accounts (IRSA)

IRSA lets a Kubernetes service account assume an IAM role. The pod receives a projected, short-lived token, and the AWS SDK exchanges it for temporary credentials. No access keys, and no permissions on the node.

First, register the cluster's OIDC issuer with IAM (once per cluster):

eksctl utils associate-iam-oidc-provider --cluster prod-eu --region eu-west-2 --approve

Next, a least-privilege policy. This one can read exactly one secret:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["secretsmanager:GetSecretValue", "secretsmanager:DescribeSecret"],
      "Resource": "arn:aws:secretsmanager:eu-west-2:123456789012:secret:prod/orders/db-??????"
    }
  ]
}

The -?????? suffix matters: Secrets Manager appends six random characters to every secret ARN. If the secret is encrypted with a customer-managed KMS key, the role also needs kms:Decrypt on that key.

Then create the role and service account together:

aws iam create-policy --policy-name orders-api-secrets \
  --policy-document file://orders-api-secrets.json

eksctl create iamserviceaccount \
  --cluster prod-eu --region eu-west-2 \
  --namespace orders --name orders-api \
  --role-name orders-api-irsa \
  --attach-policy-arn arn:aws:iam::123456789012:policy/orders-api-secrets \
  --approve

eksctl creates a role whose trust policy only accepts tokens for that specific service account:

"Condition": {
  "StringEquals": {
    "oidc.eks.eu-west-2.amazonaws.com/id/EXAMPLE0123456789:sub": "system:serviceaccount:orders:orders-api",
    "oidc.eks.eu-west-2.amazonaws.com/id/EXAMPLE0123456789:aud": "sts.amazonaws.com"
  }
}

and annotates the service account with eks.amazonaws.com/role-arn. If you manage the service account in Git rather than letting eksctl own it, add that annotation yourself.

Two follow-ups make IRSA meaningful:

Strip the node role back to what the kubelet and CNI need. Anything extra there is available to every pod.
Block pods from the instance metadata service so they can't fall back to node credentials. Set the IMDSv2 hop limit to 1 on your launch template (or disablePodIMDS: true in an eksctl node group).

EKS Pod Identity is the newer alternative and avoids per-cluster OIDC setup and trust-policy editing. IRSA still works everywhere, including self-managed tooling that predates Pod Identity, which is why it's shown here; the rest of this post works the same with either.

2. NetworkPolicies

Kubernetes allows all pod-to-pod traffic by default. NetworkPolicies change that, but only if something enforces them. The Amazon VPC CNI can do this natively; enable it on the add-on:

aws eks update-addon --cluster-name prod-eu --region eu-west-2 \
  --addon-name vpc-cni \
  --configuration-values '{"enableNetworkPolicy": "true"}'

Start with default-deny for the namespace, in both directions:

apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: default-deny-all
  namespace: orders
spec:
  podSelector: {}
  policyTypes: ["Ingress", "Egress"]

Everything in orders is now isolated, including DNS, which is the usual first surprise. Allow it back:

apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: allow-dns
  namespace: orders
spec:
  podSelector: {}
  policyTypes: ["Egress"]
  egress:
    - to:
        - namespaceSelector:
            matchLabels:
              kubernetes.io/metadata.name: kube-system
          podSelector:
            matchLabels:
              k8s-app: kube-dns
      ports:
        - { protocol: UDP, port: 53 }
        - { protocol: TCP, port: 53 }

Then open the specific paths orders-api needs: inbound from the ingress controller, and outbound to the database subnet.

apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: orders-api
  namespace: orders
spec:
  podSelector:
    matchLabels:
      app: orders-api
  policyTypes: ["Ingress", "Egress"]
  ingress:
    - from:
        - namespaceSelector:
            matchLabels:
              kubernetes.io/metadata.name: ingress-nginx
      ports:
        - { protocol: TCP, port: 8080 }
  egress:
    - to:
        - ipBlock:
            cidr: 10.0.64.0/20   # private database subnets
      ports:
        - { protocol: TCP, port: 5432 }

The trade-off is operational: every new dependency now needs a policy change, and a missing rule shows up as a timeout rather than a clear error. Roll out per namespace, and keep policies next to the application's manifests so they change together.

3. Pod Security Admission: restricted

Pod Security Admission is built into Kubernetes and is configured with namespace labels. The restricted profile rejects pods that run as root, allow privilege escalation, keep Linux capabilities, or skip a seccomp profile.

Turn on warn and audit first to see what would break, then enforce:

kubectl label namespace orders \
  pod-security.kubernetes.io/enforce=restricted \
  pod-security.kubernetes.io/enforce-version=latest \
  pod-security.kubernetes.io/warn=restricted \
  pod-security.kubernetes.io/audit=restricted \
  --overwrite

A deployment that passes looks like this. The securityContext blocks are the part that matters:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: orders-api
  namespace: orders
spec:
  replicas: 2
  selector:
    matchLabels: { app: orders-api }
  template:
    metadata:
      labels: { app: orders-api }
    spec:
      serviceAccountName: orders-api
      securityContext:
        runAsNonRoot: true
        runAsUser: 10001
        seccompProfile: { type: RuntimeDefault }
      containers:
        - name: app
          image: 123456789012.dkr.ecr.eu-west-2.amazonaws.com/orders-api:1.4.2
          ports: [{ containerPort: 8080 }]
          env:
            - name: SPRING_CONFIG_IMPORT
              value: "optional:configtree:/mnt/secrets/"
          securityContext:
            allowPrivilegeEscalation: false
            readOnlyRootFilesystem: true
            capabilities: { drop: ["ALL"] }
          volumeMounts:
            - { name: tmp, mountPath: /tmp }
            - { name: db-secrets, mountPath: /mnt/secrets, readOnly: true }
      volumes:
        - name: tmp
          emptyDir: {}
        - name: db-secrets
          csi:
            driver: secrets-store.csi.k8s.io
            readOnly: true
            volumeAttributes:
              secretProviderClass: orders-db

readOnlyRootFilesystem isn't required by restricted, but it's cheap. A Spring Boot app with embedded Tomcat needs a writable /tmp, hence the emptyDir.

Leave kube-system and other system namespaces at privileged or baseline; CNI, CSI and logging agents legitimately need host access.

4. Secrets Manager via the Secrets Store CSI driver

The CSI driver mounts secrets into the pod at start-up, fetched using the pod's own IAM identity, which is the IRSA role from step 1. Install the driver and the AWS provider:

helm repo add secrets-store-csi-driver https://kubernetes-sigs.github.io/secrets-store-csi-driver/charts
helm repo add aws-secrets-manager https://aws.github.io/secrets-store-csi-driver-provider-aws

helm install csi-secrets-store secrets-store-csi-driver/secrets-store-csi-driver \
  --namespace kube-system

helm install secrets-provider-aws aws-secrets-manager/secrets-store-csi-driver-provider-aws \
  --namespace kube-system

Then describe which secret to mount. The secret prod/orders/db holds JSON with username and password keys; jmesPath splits them into separate files:

apiVersion: secrets-store.csi.x-k8s.io/v1
kind: SecretProviderClass
metadata:
  name: orders-db
  namespace: orders
spec:
  provider: aws
  parameters:
    objects: |
      - objectName: "prod/orders/db"
        objectType: "secretsmanager"
        jmesPath:
          - path: username
            objectAlias: spring.datasource.username
          - path: password
            objectAlias: spring.datasource.password

The aliases are deliberate. With SPRING_CONFIG_IMPORT=optional:configtree:/mnt/secrets/, Spring Boot treats each file name as a property name and its contents as the value, so the datasource is configured with no code and no environment variables holding secrets.

Trade-offs to know about:

Rotation isn't automatic end to end. The driver can refresh mounted files (enableSecretRotation=true on the Helm chart), but Spring reads config trees at start-up. You'll need a rolling restart after rotation, or a connection pool that re-reads credentials.
Syncing to Kubernetes Secrets (secretObjects, with syncSecret.enabled=true) is available if you need environment variables, but it puts the value back in etcd. If you do, make sure envelope encryption with KMS is enabled on the cluster.
Pods won't start if the secret can't be fetched. That's usually what you want, but it makes Secrets Manager and STS part of your start-up path.

Summary

None of these controls is exotic, and together they remove the most common EKS footguns:

IRSA (or Pod Identity) gives each workload its own narrowly scoped IAM role; strip the node role and block IMDS so pods can't bypass it.
Default-deny NetworkPolicies, with explicit allows for DNS and real dependencies.
restricted Pod Security on application namespaces, rolled out via warn and audit first.
Secrets mounted from Secrets Manager using the workload's own identity, with a clear plan for rotation.

Event-driven Spring Boot 3 microservices with Kafka: producers, consumers, retries and dead-letter topics

Someswar Duddi — Fri, 16 Oct 2026 14:31:23 +0000

Most Kafka tutorials stop at the happy path: send a message, receive a message, done. The interesting part of running event-driven services is what happens when a consumer can't process a record. Without a plan, one bad message can block a partition, or get silently dropped.

This post builds a small producer and consumer with Spring Boot 3 and Spring Kafka, then adds retries with back-off and a dead-letter topic (DLT). The example is an orders service publishing OrderPlaced events and an inventory service consuming them.

Dependencies

You only need the web starter (for the producer endpoint) and Spring Kafka. Spring Boot manages the versions.

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

The event

Keep events small, immutable and explicit. A Java record does the job:

package com.example.orders;

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

public record OrderPlaced(String orderId, String customerId, BigDecimal total, Instant placedAt) {}

Configuration

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all
      properties:
        enable.idempotence: true
        spring.json.add.type.headers: false
    consumer:
      group-id: inventory-service
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      properties:
        spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer
        spring.json.value.default.type: com.example.orders.OrderPlaced
        spring.json.use.type.headers: false
        spring.json.trusted.packages: com.example.orders

A few choices worth calling out:

acks: all plus idempotence means the broker only confirms a write once all in-sync replicas have it, and producer retries can't create duplicates on the broker. It costs a little latency; for business events that's usually the right trade.
Type headers are off. By default JsonSerializer adds a header with the producer's Java class name. That couples the consumer to the producer's package structure. Turning it off and setting a default type on the consumer keeps the services independent.
ErrorHandlingDeserializer wraps the JSON deserialiser. Without it, a malformed payload throws inside the Kafka client before your code runs, and the container retries the same poll forever. With it, the failure is handed to the error handler like any other exception.

Topics

In development it's handy to let the application create its topics. In production you'll probably manage them with Terraform or similar, but the shape is the same.

@Configuration
public class TopicConfig {

    @Bean
    NewTopic ordersPlaced() {
        return TopicBuilder.name("orders.placed").partitions(6).replicas(3).build();
    }

    @Bean
    NewTopic ordersPlacedDlt() {
        return TopicBuilder.name("orders.placed.DLT").partitions(6).replicas(3).build();
    }
}

The DLT has the same partition count on purpose. By default, the dead-letter recoverer sends a failed record to the same partition number on <topic>.DLT. If the DLT has fewer partitions, publishing fails.

The producer

@Service
public class OrderEventPublisher {

    private static final Logger log = LoggerFactory.getLogger(OrderEventPublisher.class);
    static final String TOPIC = "orders.placed";

    private final KafkaTemplate<String, OrderPlaced> kafkaTemplate;

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

    public CompletableFuture<SendResult<String, OrderPlaced>> publish(OrderPlaced event) {
        return kafkaTemplate.send(TOPIC, event.orderId(), event)
                .whenComplete((result, ex) -> {
                    if (ex != null) {
                        log.error("Failed to publish order {}", event.orderId(), ex);
                    } else {
                        log.debug("Published order {} to partition {} at offset {}",
                                event.orderId(),
                                result.getRecordMetadata().partition(),
                                result.getRecordMetadata().offset());
                    }
                });
    }
}

The key is the order ID. Kafka only guarantees ordering within a partition, and the key decides the partition, so every event for one order lands in order on one partition.

In Spring Kafka 3.x, send returns a CompletableFuture. Don't ignore it. If the caller needs to know the event was written (for example, before returning 201 Created), wait on it with a timeout. If you need the database write and the event to succeed or fail together, look at the transactional outbox pattern; a Kafka send after a database commit isn't atomic.

The consumer

@Component
public class OrderPlacedListener {

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

    private final StockReservationService reservations;

    public OrderPlacedListener(StockReservationService reservations) {
        this.reservations = reservations;
    }

    @KafkaListener(topics = "orders.placed")
    public void onOrderPlaced(OrderPlaced event) {
        log.info("Reserving stock for order {}", event.orderId());
        reservations.reserveFor(event);
    }
}

The listener throws if anything goes wrong. That matters: you want the error handler to decide what happens next, not a try/catch that logs and moves on.

Delivery is at-least-once, so the same event can arrive twice (after a rebalance, or a retry that actually succeeded the first time). reserveFor should be idempotent, for example by recording processed order IDs under a unique constraint.

Retries and the dead-letter topic

Spring Boot automatically applies a CommonErrorHandler bean to the listener container factory, so this is all the wiring you need:

@Configuration
public class KafkaErrorHandlingConfig {

    @Bean
    DefaultErrorHandler kafkaErrorHandler(ProducerFactory<?, ?> producerFactory) {
        // A dedicated template for the DLT: raw bytes for records that failed
        // deserialisation, JSON for everything else.
        Map<String, Object> props = new HashMap<>(producerFactory.getConfigurationProperties());
        var serializer = new DelegatingByTypeSerializer(Map.<Class<?>, Serializer<?>>of(
                byte[].class, new ByteArraySerializer(),
                OrderPlaced.class, new JsonSerializer<OrderPlaced>()));
        var dltTemplate = new KafkaTemplate<>(
                new DefaultKafkaProducerFactory<>(props, new StringSerializer(), serializer));

        var recoverer = new DeadLetterPublishingRecoverer(dltTemplate);

        var backOff = new ExponentialBackOffWithMaxRetries(4);
        backOff.setInitialInterval(1_000L);
        backOff.setMultiplier(2.0);
        backOff.setMaxInterval(10_000L);

        var handler = new DefaultErrorHandler(recoverer, backOff);
        handler.addNotRetryableExceptions(InvalidOrderException.class);
        return handler;
    }
}

What happens now when reserveFor throws:

The handler retries the record in place: after 1s, 2s, 4s and 8s (capped at 10s).
If it still fails, DeadLetterPublishingRecoverer publishes it to orders.placed.DLT, same partition, with headers describing the original topic, partition, offset and exception.
The offset is committed and the consumer moves on.

Exceptions listed as not retryable skip straight to step 2. Retrying a validation failure four times just wastes 15 seconds. Spring Kafka already treats DeserializationException and a few others as fatal, so a malformed payload goes straight to the DLT as its original bytes. That's why the DLT template needs ByteArraySerializer for byte[].

The trade-off: blocking retries

These retries are blocking. While a record is backing off, the rest of that partition waits. That's what keeps ordering intact, and for short, transient failures (a brief database blip) it's the right default.

If a downstream dependency can be down for minutes, blocking retries will stall the partition and can push you past max.poll.interval.ms, causing a rebalance. Keep total back-off well below that limit (five minutes by default). For longer outages, Spring Kafka's @RetryableTopic provides non-blocking retries through intermediate retry topics. The cost is that retried records are processed out of order relative to newer ones, so only use it where per-key ordering doesn't matter.

Inspecting the DLT

A dead-letter topic that nobody reads is just a slower way of dropping messages. At a minimum, consume it and log or alert:

@Component
public class OrderPlacedDltListener {

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

    @KafkaListener(
            topics = "orders.placed.DLT",
            groupId = "inventory-service-dlt",
            properties = "value.deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer")
    public void onDeadLetter(ConsumerRecord<String, byte[]> record) {
        log.error("Dead-lettered record key={} originalOffset={} reason={} payload={}",
                record.key(),
                header(record, KafkaHeaders.DLT_ORIGINAL_OFFSET),
                header(record, KafkaHeaders.DLT_EXCEPTION_MESSAGE),
                new String(record.value(), StandardCharsets.UTF_8));
    }

    private static String header(ConsumerRecord<?, ?> record, String name) {
        Header header = record.headers().lastHeader(name);
        if (header == null) {
            return "n/a";
        }
        if (KafkaHeaders.DLT_ORIGINAL_OFFSET.equals(name)) {
            return Long.toString(ByteBuffer.wrap(header.value()).getLong());
        }
        return new String(header.value(), StandardCharsets.UTF_8);
    }
}

The DLT listener reads raw bytes so it never fails on a payload it can't parse; that's the whole reason the record is there. The original offset header is a big-endian long, not text, hence the special case.

From here, the usual next step is a small replay tool: once the bug or outage is fixed, read records from the DLT and publish them back to the original topic.

Summary

Key events by the entity ID so per-entity ordering holds.
Use acks=all and idempotent producers for business events.
Wrap deserialisers in ErrorHandlingDeserializer so poison messages don't loop forever.
Use DefaultErrorHandler with bounded exponential back-off, mark validation errors as non-retryable, and recover to a DLT.
Make consumers idempotent; at-least-once means duplicates will happen.
Monitor the DLT and have a way to replay it.