Spring Kafka: Listeners, Templates, and Error Handling

A practical guide to building production Kafka producers and consumers with Spring Boot 3 and Spring Kafka.

August 14, 2026

backend-engineerspring-kafkaproducerconsumererror-handlingserialization

Spring Kafka Integration

Spring Kafka wraps the raw Kafka client with the same declarative, configuration-driven style Spring developers already know from JPA and Spring MVC. This guide covers how to wire up KafkaTemplate for producing, @KafkaListener for consuming, how serialization decisions ripple through your whole pipeline, and the error-handling machinery — DefaultErrorHandler, retry topics, and dead-letter topics — that keeps a Spring Boot Kafka application resilient in production.

1. Dependencies and Baseline Setup

xml

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka-test</artifactId>
    <scope>test</scope>
</dependency>

Spring Boot 3's autoconfiguration wires up ProducerFactory, ConsumerFactory, KafkaTemplate, and a default ConcurrentKafkaListenerContainerFactory from application.yml properties alone — you only need explicit @Bean definitions when you need to customize behavior beyond what properties expose (custom error handlers, manual ack modes, non-default serializers per listener).

yaml

spring:
  kafka:
    bootstrap-servers: broker1:9092,broker2:9092,broker3:9092
    client-id: order-service
    producer:
      acks: all
      retries: 2147483647
      properties:
        enable.idempotence: true
        linger.ms: 5
        compression.type: lz4
        delivery.timeout.ms: 120000
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
    consumer:
      group-id: order-service
      auto-offset-reset: earliest
      enable-auto-commit: false
      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.trusted.packages: "com.company.orders.events"
        max.poll.records: 500
    listener:
      ack-mode: manual_immediate
      concurrency: 3

💡

ErrorHandlingDeserializer wrapping the real deserializer is a production essential, not an optional nicety. Without it, a malformed message (bad JSON, schema mismatch) throws during deserialization before your @KafkaListener method is even invoked — and by default that exception can crash the container's poll loop entirely, repeatedly, on the same poisoned offset. ErrorHandlingDeserializer catches the failure and hands control to your configured error handler instead.

2. Producing with KafkaTemplate

KafkaTemplate is the primary API for sending records — thread-safe, backed by a pooled Producer instance, and returning a CompletableFuture for the send result (Spring Kafka 3.x moved off the older ListenableFuture).

java

@Service
public class OrderEventPublisher {
 
    private static final Logger log = LoggerFactory.getLogger(OrderEventPublisher.class);
    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;
 
    public OrderEventPublisher(KafkaTemplate<String, OrderEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }
 
    public CompletableFuture<SendResult<String, OrderEvent>> publish(OrderEvent event) {
        ProducerRecord<String, OrderEvent> record =
            new ProducerRecord<>("orders", event.orderId(), event);
        record.headers().add("event-type", "OrderCreated".getBytes(StandardCharsets.UTF_8));
 
        return kafkaTemplate.send(record)
            .whenComplete((result, ex) -> {
                if (ex != null) {
                    log.error("Failed to publish order {}", event.orderId(), ex);
                } else {
                    log.debug("Order {} -> partition {} offset {}",
                        event.orderId(),
                        result.getRecordMetadata().partition(),
                        result.getRecordMetadata().offset());
                }
            });
    }
}

Synchronous send when you need the guarantee before returning

java

public void publishAndWait(OrderEvent event) {
    try {
        SendResult<String, OrderEvent> result = kafkaTemplate
            .send("orders", event.orderId(), event)
            .get(5, TimeUnit.SECONDS);  // blocks — use sparingly, on critical paths only
        log.info("Confirmed at offset {}", result.getRecordMetadata().offset());
    } catch (TimeoutException | ExecutionException | InterruptedException e) {
        throw new EventPublishException("Failed to publish order " + event.orderId(), e);
    }
}

⚠️

Blocking on .get() for every send defeats most of the throughput benefit of Kafka's async batching. Reserve synchronous sends for genuinely critical paths (e.g., "must be durably queued before returning 200 to the caller") and prefer the async callback style for everything else, including bulk/background publishing.

Transactional producer (for read-process-write pipelines)

yaml

spring:
  kafka:
    producer:
      transaction-id-prefix: order-service-tx-

java

@Transactional("kafkaTransactionManager")
public void processAndForward(OrderEvent input) {
    EnrichedOrderEvent enriched = enrich(input);
    kafkaTemplate.send("orders-enriched", enriched.orderId(), enriched);
    // If this throws, the send above is rolled back — never committed to the log
    auditLog.record(enriched);
}

3. Consuming with @KafkaListener

@KafkaListener is the declarative annotation-driven consumer entry point, backed by a ConcurrentMessageListenerContainer under the hood.

java

@Component
public class OrderEventListener {
 
    private static final Logger log = LoggerFactory.getLogger(OrderEventListener.class);
    private final OrderService orderService;
 
    public OrderEventListener(OrderService orderService) {
        this.orderService = orderService;
    }
 
    @KafkaListener(
        topics = "orders",
        groupId = "order-processing-service",
        containerFactory = "kafkaListenerContainerFactory"
    )
    public void onOrderEvent(
            @Payload OrderEvent event,
            @Header(KafkaHeaders.RECEIVED_KEY) String key,
            @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
            @Header(KafkaHeaders.OFFSET) long offset,
            Acknowledgment ack) {
 
        log.info("Processing order {} [partition={}, offset={}]", key, partition, offset);
        orderService.process(event);
        ack.acknowledge();
    }
}

Batch listeners for higher throughput

java

@KafkaListener(
    topics = "orders",
    groupId = "order-analytics-service",
    containerFactory = "batchListenerContainerFactory"
)
public void onOrderBatch(List<OrderEvent> events, Acknowledgment ack) {
    analyticsService.processBatch(events);
    ack.acknowledge(); // acknowledges the whole batch
}

java

@Bean
public ConcurrentKafkaListenerContainerFactory<String, OrderEvent> batchListenerContainerFactory(
        ConsumerFactory<String, OrderEvent> consumerFactory) {
    ConcurrentKafkaListenerContainerFactory<String, OrderEvent> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory);
    factory.setBatchListener(true);
    factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
    return factory;
}

✅

Batch listeners trade per-message error granularity for throughput — a single bad record in the batch requires care (usually a BatchListenerFailedException naming the specific failed index, so Spring Kafka's error handler can seek back to just that record rather than reprocessing or dead-lettering the whole batch).

Listener lifecycle and graceful shutdown

💡

Spring Boot's default shutdown handling stops Kafka listener containers before the application context closes, allowing in-flight records to finish and the consumer to leave the group cleanly — this avoids an unnecessary rebalance delay (waiting out session.timeout.ms) on every routine deploy. Ensure server.shutdown=graceful and a reasonable spring.lifecycle.timeout-per-shutdown-phase are set so this window isn't cut short by your orchestrator's SIGKILL grace period.

4. Serialization: JSON and Avro

Spring Kafka's JsonSerializer/JsonDeserializer (backed by Jackson) is the most common starting point — no schema registry required, human-readable payloads, easy debugging.

java

public record OrderEvent(
    String eventId,
    String orderId,
    String customerId,
    BigDecimal amount,
    Instant occurredAt
) {}

yaml

spring:
  kafka:
    consumer:
      properties:
        spring.json.trusted.packages: "com.company.orders.events"
        spring.json.value.default.type: com.company.orders.events.OrderEvent
        spring.json.use.type.headers: false

⚠️

spring.json.trusted.packages is a security control, not boilerplate. JsonDeserializer by default embeds the fully-qualified Java class name in message headers and will deserialize to any class on the classpath unless you restrict trusted packages — an attacker-controlled or misconfigured producer could otherwise trigger deserialization of arbitrary types. Never set this to "*" in production.

JSON vs Avro (with Schema Registry)

Aspect	JSON (Jackson)	Avro (Confluent Schema Registry)
Schema enforcement	None by default — structural drift is silent	Enforced at produce/consume time via registry
Payload size	Larger (field names repeated per message)	Compact binary encoding
Schema evolution	Manual discipline, no tooling support	Registry enforces compatibility rules (backward/forward)
Tooling overhead	None — works out of the box	Requires running/operating a Schema Registry service
Debuggability	Human-readable, easy to inspect with CLI tools	Requires registry-aware tooling to decode
Best for	Small teams, internal topics, early-stage systems	Multi-team organizations, contracts across team boundaries

✅

If more than one team produces or consumes a topic, a schema registry (Avro or Protobuf with Confluent Schema Registry, or equivalent) earns its operational overhead quickly — it turns "someone changed the event shape and broke three downstream consumers" from a runtime incident into a registry-enforced compile/produce-time rejection.

5. Error Handling with DefaultErrorHandler

DefaultErrorHandler is the container-level error handling mechanism that replaced the older SeekToCurrentErrorHandler/ErrorHandler interfaces in Spring Kafka 2.8+. It controls what happens when a listener method throws.

java

@Configuration
public class KafkaErrorHandlingConfig {
 
    @Bean
    public DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> kafkaTemplate) {
        DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(
            kafkaTemplate,
            (record, ex) -> new TopicPartition(record.topic() + ".DLT", record.partition())
        );
 
        FixedBackOff backOff = new FixedBackOff(2_000L, 3L); // 3 retries, 2s apart
 
        DefaultErrorHandler handler = new DefaultErrorHandler(recoverer, backOff);
        handler.addNotRetryableExceptions(
            DeserializationException.class,
            MethodArgumentNotValidException.class
        );
        handler.setRetryListeners((record, ex, deliveryAttempt) ->
            log.warn("Retry attempt {} for record at offset {}: {}",
                deliveryAttempt, record.offset(), ex.getMessage()));
        return handler;
    }
 
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, OrderEvent> kafkaListenerContainerFactory(
            ConsumerFactory<String, OrderEvent> consumerFactory,
            DefaultErrorHandler errorHandler) {
        ConcurrentKafkaListenerContainerFactory<String, OrderEvent> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setCommonErrorHandler(errorHandler);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        return factory;
    }
}

Non-blocking retries with @RetryableTopic

For failures that need a delay before retrying without blocking the partition from processing other keys in the meantime, Spring Kafka's @RetryableTopic creates and manages retry topics automatically.

java

@RetryableTopic(
    attempts = "4",
    backoff = @Backoff(delay = 1000, multiplier = 2.0, maxDelay = 30000),
    dltStrategy = DltStrategy.FAIL_ON_ERROR,
    include = { TransientDownstreamException.class }
)
@KafkaListener(topics = "orders", groupId = "order-processing-service")
public void onOrderEvent(OrderEvent event) {
    orderService.process(event);
}
 
@DltHandler
public void handleDlt(OrderEvent event, @Header(KafkaHeaders.EXCEPTION_MESSAGE) String message) {
    log.error("Order {} exhausted retries: {}", event.orderId(), message);
    incidentService.raise(event, message);
}

@RetryableTopic transparently creates orders-retry-0, orders-retry-1, ... (or a single retry topic depending on configuration) and an orders-dlt topic, routing failed messages through them with the configured backoff — without blocking the original partition's consumption of subsequent messages.

Approach	Blocks partition during backoff?	Best for
`DefaultErrorHandler` + `FixedBackOff`	Yes — retries happen in the poll loop	Short backoffs (sub-second to a few seconds)
`@RetryableTopic`	No — retry happens via a separate topic/consumer	Longer backoffs (seconds to minutes) where blocking would stall unrelated keys

🚨

Blocking retries inside the poll loop hold up every other message in that partition, not just the failing one — because Kafka only tracks a single offset position per partition. A transient downstream outage with a 30-second blocking backoff and 4 retries can stall an entire partition for two minutes. This is exactly the scenario @RetryableTopic exists to avoid.

6. Testing with EmbeddedKafka

java

@SpringBootTest
@EmbeddedKafka(partitions = 3, topics = { "orders", "orders.DLT" })
class OrderEventListenerTest {
 
    @Autowired
    private KafkaTemplate<String, OrderEvent> kafkaTemplate;
 
    @MockBean
    private OrderService orderService;
 
    @Test
    void consumesAndProcessesOrderEvent() {
        OrderEvent event = new OrderEvent("evt-1", "order-42", "cust-1",
            new BigDecimal("99.99"), Instant.now());
 
        kafkaTemplate.send("orders", event.orderId(), event);
 
        await().atMost(Duration.ofSeconds(5))
            .untilAsserted(() -> verify(orderService).process(event));
    }
}

💡

@EmbeddedKafka spins up a real, in-process Kafka broker for the test — this exercises actual serialization, partitioning, and consumer group behavior, catching issues that pure unit tests with mocked KafkaTemplate would miss (deserializer misconfiguration, header propagation, partition assignment).

7. Production Observations

ErrorHandlingDeserializer + not-retryable exception lists are the two settings most commonly missing in Spring Kafka setups that later suffer a poison-message incident.
Manual ack mode should be the default for any consumer whose processing has side effects with real consequences (charges, notifications, state mutations) — auto-commit's at-most-once lean is rarely what you actually want.
concurrency in the container factory is capped by partition count — setting it higher than available partitions (accounting for other pod replicas in the same group) wastes threads.
Non-blocking retry topics add operational surface area — more topics to monitor, more consumer groups, more lag dashboards. Reserve them for genuinely long backoffs; short ones are fine as blocking retries.
spring.json.trusted.packages is a security setting, not boilerplate to copy-paste with * — restrict it to your actual event package.
Graceful shutdown matters for deploy-time rebalance cost — verify your container orchestrator's SIGTERM grace period is long enough for in-flight records to finish and the consumer to leave the group cleanly.

Key takeaways

KafkaTemplate is thread-safe and async by default; reserve blocking .get() sends for genuinely critical paths.
Always wrap real deserializers in ErrorHandlingDeserializer so malformed messages route to your error handler instead of crashing the poll loop.
Manual acknowledgement (MANUAL_IMMEDIATE) after successful processing is the standard production ack mode for business-critical consumers.
DefaultErrorHandler with a DeadLetterPublishingRecoverer and a not-retryable exception list is the baseline error-handling setup for every consumer.
Use @RetryableTopic for backoffs long enough that blocking the partition would meaningfully delay unrelated messages; use blocking DefaultErrorHandler retries for short, sub-second backoffs.
spring.json.trusted.packages must be explicitly scoped — an unrestricted JSON deserializer is a deserialization security risk.
Test with @EmbeddedKafka for behavior that depends on real serialization, partitioning, and consumer group mechanics — mocked KafkaTemplate unit tests won't catch these classes of bugs.

Interview Questions

What does ErrorHandlingDeserializer do, and why is it necessary even if your producer always sends well-formed messages?
How does DefaultErrorHandler differ from the older SeekToCurrentErrorHandler? What responsibilities does it combine?
When would you choose @RetryableTopic over blocking retries configured on DefaultErrorHandler?
Why is committing offsets in MANUAL_IMMEDIATE mode after processing preferred over relying on Spring Kafka's auto-commit default?
What security risk does an unrestricted spring.json.trusted.packages setting introduce?
How would you configure a transactional producer in Spring Kafka for a read-process-write pipeline?
What's the difference between a batch listener and a record listener in Spring Kafka? What error-handling complication does batching introduce?
How does @EmbeddedKafka differ from mocking KafkaTemplate in a unit test? What classes of bugs does it catch that mocks won't?
Why does Spring Kafka's default graceful shutdown behavior matter for rebalance cost during deploys?
Compare JSON and Avro serialization for a Kafka-based system shared across multiple teams. When does the schema registry overhead pay off?
How would you size concurrency in ConcurrentKafkaListenerContainerFactory relative to topic partitions and other consumer instances?
What happens if a listener method throws an exception that's in the DefaultErrorHandler's not-retryable list?
Describe how DeadLetterPublishingRecoverer decides which topic and partition to publish a failed record to.
Why might blocking retries inside the poll loop be dangerous for a high-throughput, multi-key partition, even with a short backoff?

Kafka Reliability and Scaling: Replication, Rebalancing, and DLQs

Redis Core Concepts: Data Structures, TTLs, and Persistence