Skip to content
JavaAgentic

Type at least two characters. Try “RAG”, “pgvector” or “tool calling”.

Spring Kafka Integration

Spring for Apache Kafka in production: KafkaTemplate, @KafkaListener containers, JSON serialisation without trusting the wire, DefaultErrorHandler with backoff, and dead-letter topics.

Intermediate7 min readUpdated
On this page

Spring for Apache Kafka wraps the Java client in the same idioms as the rest of Spring — a template for producing, an annotation for consuming, and a container managing the polling loop. The defaults are reasonable; the ones that are not are the subject of this guide.

Key Takeaways

  • Set AckMode.MANUAL_IMMEDIATE or RECORD — auto-commit loses records on a crash.
  • Always wrap deserialisers in ErrorHandlingDeserializer, or one bad record halts the partition forever.
  • Never enable trusted.packages=* — it is a remote code execution vector.
  • DefaultErrorHandler + DeadLetterPublishingRecoverer gives you retry and DLT in three lines.
  • Use @RetryableTopic for non-blocking retries with long backoffs.

Configuration

application.yml
spring:
  kafka:
    bootstrap-servers: 'kafka-1:9092,kafka-2:9092,kafka-3:9092'
    producer:
      acks: all
      properties:
        enable.idempotence: true
        max.in.flight.requests.per.connection: 5
        linger.ms: 10
        compression.type: zstd
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
    consumer:
      group-id: billing
      # Off. Auto-commit advances offsets on a timer regardless of whether
      # processing succeeded, which silently loses records on a crash.
      enable-auto-commit: false
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      # The wrapper turns a deserialisation failure into a null payload the
      # error handler can route to the DLT, instead of an infinite loop.
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      properties:
        spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer
        # An allowlist. '*' means any class named in a message header gets
        # instantiated — a remote code execution primitive.
        spring.json.trusted.packages: 'com.acme.events'
        spring.json.use.type.headers: false
        spring.json.value.default.type: com.acme.events.OrderEvent
    listener:
      ack-mode: MANUAL_IMMEDIATE
      concurrency: 3
      poll-timeout: 3000

The trusted.packages setting deserves the warning. JsonDeserializer reads a type header from the message and instantiates that class. With *, anyone who can publish to your topic chooses which class your JVM constructs, which is the classic deserialisation attack. Pin it to your own packages, or disable type headers entirely and set a default type as above.

Two producer settings above work together in a way worth understanding rather than copying. enable.idempotence has the broker deduplicate retries using a producer id and sequence number, which turns a retried send from a probable duplicate into a no-op. It is also what makes max.in.flight.requests.per.connection: 5 safe: without idempotence, five concurrent in-flight batches can be reordered when one of them is retried, so preserving per-key order would force that value down to 1 and surrender most of the throughput. With it on, the broker restores the order and you keep both properties.

On the consumer side, the setting behind the most confusing production incident is max.poll.interval.ms, which defaults to five minutes. If handling a batch takes longer than that, the broker decides the consumer is dead and rebalances its partitions elsewhere — while the original consumer is still happily processing them. The symptoms are duplicated work and a consumer group that never settles, neither of which points at the actual cause. Either lower max.poll.records so a batch finishes well inside the interval, or raise the interval to cover your slowest realistic batch.

Producing

OrderEventPublisher.java
@Component
public class OrderEventPublisher {
 
    private final KafkaTemplate<String, OrderEvent> template;
 
    public void publish(OrderEvent event) {
        var record = new ProducerRecord<String, OrderEvent>("orders", event.orderId(), event);
        record.headers().add("eventId", event.eventId().getBytes(UTF_8));
 
        template.send(record).whenComplete((result, ex) -> {
            if (ex != null) {
                // The send failed after retries. The event is lost unless it is
                // in an outbox — which is the argument for using one.
                log.error("publish failed for {}", event.orderId(), ex);
                metrics.counter("kafka.publish.failed").increment();
            }
        });
    }
}

send is asynchronous and returns a future. Ignoring it means publish failures disappear silently — the single most common Spring Kafka mistake. Either handle the future as above, or block on it if the caller must know, or write to an outbox and let a relay handle delivery.

Consuming

OrderEventListener.java
@Component
public class OrderEventListener {
 
    private final ProcessedEventRepository processed;
    private final BillingService billing;
 
    @KafkaListener(
        topics = "orders",
        groupId = "billing",
        containerFactory = "kafkaListenerContainerFactory")
    @Transactional
    public void onOrderEvent(
            @Payload OrderEvent event,
            @Header(KafkaHeaders.RECEIVED_KEY) String key,
            @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
            @Header(KafkaHeaders.OFFSET) long offset,
            @Header(value = "eventId", required = false) String eventId,
            Acknowledgment ack) {
 
        MDC.put("correlationId", eventId);
        try {
            // At-least-once delivery guarantees duplicates eventually. The
            // unique constraint makes the check atomic; a SELECT-then-INSERT
            // has a race two concurrent consumers will find.
            processed.save(new ProcessedEvent(eventId, Instant.now()));
        } catch (DataIntegrityViolationException duplicate) {
            log.debug("event {} already processed", eventId);
            ack.acknowledge();
            return;
        }
 
        billing.handle(event);
 
        // Acknowledge only after successful processing. An exception here
        // skips the ack and the record is redelivered.
        ack.acknowledge();
        MDC.clear();
    }
}

Error handling and dead-letter topics

Bounded retries, then the dead-letter topic. Without the DLT step a poison record blocks its partition indefinitely.
KafkaErrorConfig.java
@Configuration
public class KafkaErrorConfig {
 
    @Bean
    public DefaultErrorHandler errorHandler(KafkaTemplate<String, Object> template) {
 
        var recoverer = new DeadLetterPublishingRecoverer(template,
                // Route to <topic>.DLT, keeping the original partition so
                // ordering within a key survives into the DLT.
                (record, ex) -> new TopicPartition(record.topic() + ".DLT", record.partition()));
 
        var backoff = new ExponentialBackOffWithMaxRetries(4);
        backoff.setInitialInterval(500);
        backoff.setMultiplier(2.0);
        backoff.setMaxInterval(10_000);
 
        var handler = new DefaultErrorHandler(recoverer, backoff);
 
        // These will never succeed on retry — send them straight to the DLT
        // rather than burning four attempts and ten seconds per record.
        handler.addNotRetryableExceptions(
                DeserializationException.class,
                MethodArgumentNotValidException.class,
                UnknownEventTypeException.class);
 
        handler.setRetryListeners((record, ex, attempt) ->
                log.warn("retry {} for {}-{}@{}", attempt,
                        record.topic(), record.partition(), record.offset(), ex));
 
        return handler;
    }
}

The recoverer copies the original topic, partition, offset, exception message and stack trace into headers on the DLT record. That is what makes the DLT diagnosable — without it you have a payload and no idea why it failed.

Alert on DLT depth. A dead-letter topic nobody watches is a silent data-loss mechanism, because records arrive there precisely when something needs human attention.

Non-blocking retries

The blocking retry above pauses the partition for the whole backoff. With a four-attempt exponential backoff that is over ten seconds during which no later record in that partition is processed — fine for a brief blip, unacceptable for a downstream outage lasting minutes.

NonBlockingRetry.java
@RetryableTopic(
    attempts = "4",
    backoff = @Backoff(delay = 1_000, multiplier = 3.0),
    dltStrategy = DltStrategy.FAIL_ON_ERROR,
    topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE,
    exclude = { DeserializationException.class })
@KafkaListener(topics = "orders", groupId = "billing")
public void onOrderEvent(OrderEvent event) {
    billing.handle(event);
}
 
@DltHandler
public void onDlt(OrderEvent event, @Header(KafkaHeaders.ORIGINAL_TOPIC) String topic) {
    log.error("giving up on {} from {}", event.orderId(), topic);
    alerting.raise("kafka-dlt", event.orderId());
}

Spring creates orders-retry-0, orders-retry-1 and orders-dlt. A failed record is republished to the next retry topic with a delay, and the main partition carries on immediately. The trade-off is that per-key ordering is lost for retried records — the retry lands after records that came behind it. Where ordering is essential, blocking retries are the correct choice despite the cost.

Batch consumption

BatchListener.java
@KafkaListener(topics = "metrics", batch = "true")
public void onBatch(List<ConsumerRecord<String, MetricEvent>> records, Acknowledgment ack) {
    // One database round trip for 500 records instead of 500 round trips.
    repository.saveAll(records.stream().map(r -> toEntity(r.value())).toList());
    ack.acknowledge();
}

Batching transforms throughput for write-heavy consumers. The cost is granularity: one bad record in a batch of 500 fails the whole batch, so pair it with BatchListenerFailedException naming the index of the offending record, which lets the error handler send just that one to the DLT.

Testing

OrderEventListenerTest.java
@SpringBootTest
@Testcontainers
class OrderEventListenerTest {
 
    @Container
    @ServiceConnection
    static KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("apache/kafka:3.8.0"));
 
    @Autowired KafkaTemplate<String, OrderEvent> template;
    @Autowired BillingRepository repository;
 
    @Test
    void processesEachEventExactlyOnce() {
        var event = new OrderEvent("ORD-1", "eventId-1", Money.of(1000, "EUR"));
 
        template.send("orders", event.orderId(), event);
        template.send("orders", event.orderId(), event);   // deliberate duplicate
 
        await().atMost(Duration.ofSeconds(10))
               .untilAsserted(() -> assertThat(repository.findByOrderId("ORD-1")).hasSize(1));
    }
}

Testing against a real broker is the only way to exercise rebalancing, offset commits and the error handler. EmbeddedKafka is faster but diverges from real broker behaviour in exactly the areas most likely to bite.

What to take away

Turn off auto-commit and acknowledge after processing. Wrap deserialisers so a poison record cannot halt a partition, and never trust arbitrary type headers. Configure DefaultErrorHandler with a bounded backoff and a dead-letter topic, alert on its depth, and reach for @RetryableTopic when backoffs get long enough to matter.

Frequently Asked Questions

Why do I get a deserialisation error I cannot recover from?
A record that fails to deserialise fails again on every retry, so the consumer is stuck on it forever and the partition stops advancing — a poison pill. Wrap your deserialiser in ErrorHandlingDeserializer, which converts the failure into a null payload plus a header, letting the error handler route it to the DLT instead of looping.
Should the consumer be transactional?
Only if you are writing back to Kafka in the same flow, where read-process-write transactions give you exactly-once between topics. If you are writing to a database, Kafka transactions do not extend to it — use manual acks plus an idempotent consumer instead, which is simpler and covers the real case.
How do I retry without blocking the partition?
Blocking retries with a backoff hold up every later record in that partition. For long backoffs use @RetryableTopic, which republishes the failed record to a delay topic and lets the main partition continue, then processes the retry topic separately.

Related tutorials