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: allplus 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
JsonSerializeradds 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. -
ErrorHandlingDeserializerwraps 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,
DeadLetterPublishingRecovererpublishes it toorders.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=alland idempotent producers for business events. - Wrap deserialisers in
ErrorHandlingDeserializerso poison messages don't loop forever. - Use
DefaultErrorHandlerwith 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.
Top comments (0)