Messaging with Apache Kafka

This section documents Spring Boot 4.1.x and Spring Framework 7.0.x on the Java 17/21+ baseline, as described by the official Spring Boot reference documentation and each sub-project’s own site (Spring Data, Spring for Apache Kafka, Micrometer, Project Reactor, springdoc-openapi, Spring gRPC, and the others listed in this section’s Bibliography) — which are the references these pages are written and verified against.

This content was generated with the assistance of AI and should be verified against those official docs before being relied on in production. Spring Boot and its ecosystem continue to evolve; the examples here target the current 4.1.x / 7.0.x releases.

This section’s bibliography lists the reference material consulted while preparing these pages.

Spring for Apache Kafka layers KafkaTemplate and @KafkaListener over the native Kafka Java client, and Spring Boot auto-configures the underlying ProducerFactory and ConsumerFactory beans from a handful of spring.kafka.* properties. This page covers producing, consuming, error handling, and the auto-configured factories; for Avro payloads, Schema Registry, and AsyncAPI contracts see API-First: Messaging.

Dependency and baseline configuration

Add spring-kafka (Spring Boot manages its version via the dependency BOM):

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

A minimal application.yml pointing at a broker and naming a consumer group:

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: orders-service
      auto-offset-reset: earliest
    producer:
      acks: all

spring.kafka.bootstrap-servers seeds both the producer and consumer factories unless overridden per-side under spring.kafka.producer. / spring.kafka.consumer.. See Spring Boot’s Kafka reference for the full property list.

Producing with KafkaTemplate

Spring Boot auto-configures a ProducerFactory<Object, Object> bean and a matching KafkaTemplate<Object, Object> bean from those properties — no manual @Bean definitions are needed for the common case:

@Service
public class OrderEventPublisher {

    private final KafkaTemplate<String, OrderPlacedEvent> kafkaTemplate;

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

    public void publish(OrderPlacedEvent event) {
        // key = order id, so all events for one order land on the same partition
        kafkaTemplate.send("orders.placed", event.orderId(), event);
    }

    public void publishAndHandle(OrderPlacedEvent event) {
        kafkaTemplate.send("orders.placed", event.orderId(), event)
                .whenComplete((result, ex) -> {
                    if (ex != null) {
                        log.error("failed to publish order {}", event.orderId(), ex);
                    } else {
                        log.debug("published to partition {} offset {}",
                                result.getRecordMetadata().partition(),
                                result.getRecordMetadata().offset());
                    }
                });
    }
}

send returns a CompletableFuture<SendResult<K, V>>; a synchronous send (rarely desirable on a request thread) is kafkaTemplate.send(…​).get(timeout, unit). The topic key determines the partition: records with the same key always land on the same partition, which preserves per-key ordering.

Transactional producers

Wrapping several sends — or a send plus a database write via ChainedKafkaTransactionManager — in one Kafka transaction requires a transaction-id-prefix:

spring:
  kafka:
    producer:
      transaction-id-prefix: orders-tx-
@Service
public class TransactionalOrderPublisher {

    private final KafkaTemplate<String, OrderPlacedEvent> kafkaTemplate;

    public TransactionalOrderPublisher(KafkaTemplate<String, OrderPlacedEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    @Transactional
    public void publishBoth(OrderPlacedEvent placed, OrderShippedEvent shipped) {
        // both records commit or both roll back as a single Kafka transaction
        kafkaTemplate.send("orders.placed", placed.orderId(), placed);
        kafkaTemplate.send("orders.shipped", shipped.orderId(), shipped);
    }
}

Setting transaction-id-prefix makes Spring Boot register a KafkaTransactionManager bean and switch KafkaTemplate into transactional mode; @Transactional on the calling method then delimits the transaction boundary (executeInTransaction is the lower-level, non-annotation equivalent). Consumers that should only see committed records must set isolation-level: read_committed. See the Spring for Apache Kafka reference for exactly-once semantics across a consume-transform-produce chain.

Consuming with @KafkaListener

The simplest listener needs only a topic and the consumer group configured above:

@Component
public class OrderEventListener {

    @KafkaListener(topics = "orders.placed", groupId = "orders-service")
    public void onOrderPlaced(OrderPlacedEvent event) {
        log.info("received order {}", event.orderId());
    }
}

The record key, headers, and partition/offset metadata are available by adding parameters annotated with @Header:

@KafkaListener(topics = "orders.placed")
public void onOrderPlaced(
        OrderPlacedEvent event,
        @Header(KafkaHeaders.RECEIVED_KEY) String key,
        @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
        @Header(KafkaHeaders.OFFSET) long offset) {
    log.info("order {} (key={}) from partition {} offset {}", event.orderId(), key, partition, offset);
}

Manual and batch acknowledgment

By default Spring commits offsets automatically after each successful listener invocation (AckMode.BATCH/container-managed). For at-least-once processing where the application must control exactly when an offset is safe to commit, switch to MANUAL or MANUAL_IMMEDIATE:

spring:
  kafka:
    listener:
      ack-mode: manual
      type: single   # or "batch" to receive List<ConsumerRecord<...>> per invocation
@KafkaListener(topics = "orders.placed", groupId = "orders-service")
public void onOrderPlaced(OrderPlacedEvent event, Acknowledgment acknowledgment) {
    orderProcessor.process(event);
    acknowledgment.acknowledge();   // commit only after processing succeeds
}

With type: batch, the listener receives a List<OrderPlacedEvent> (or List<ConsumerRecord<K, V>>) per poll and acknowledges the whole batch at once:

@KafkaListener(topics = "orders.placed", groupId = "orders-service", containerFactory = "batchFactory")
public void onOrderPlacedBatch(List<OrderPlacedEvent> events, Acknowledgment acknowledgment) {
    orderProcessor.processAll(events);
    acknowledgment.acknowledge();
}

Error handlers, retry, and dead-letter topics

A DefaultErrorHandler bean controls what happens when a listener throws: it retries a fixed number of times with backoff, then routes the failed record to a dead-letter topic (DLT) via a DeadLetterPublishingRecoverer:

@Configuration
public class KafkaErrorHandlingConfig {

    @Bean
    public DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> kafkaTemplate) {
        var recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate,
                (record, ex) -> new TopicPartition(record.topic() + ".DLT", record.partition()));

        // retry 3 times, 1s apart, then publish to "<topic>.DLT"
        var errorHandler = new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 3L));

        // exceptions that should never be retried (e.g. deserialization/validation failures)
        errorHandler.addNotRetryableExceptions(IllegalArgumentException.class);
        return errorHandler;
    }
}

Spring Boot picks up a single DefaultErrorHandler (or CommonErrorHandler) bean and wires it into the auto-configured listener container factory automatically — no extra property is required. A listener can also be routed straight to a DLT method for local handling instead of a separate topic:

@KafkaListener(topics = "orders.placed", groupId = "orders-service")
public void onOrderPlaced(OrderPlacedEvent event) {
    orderProcessor.process(event);
}

@DltHandler
public void onOrdersPlacedDlt(OrderPlacedEvent event, @Header(KafkaHeaders.EXCEPTION_MESSAGE) String reason) {
    log.error("order {} sent to DLT: {}", event.orderId(), reason);
}

Serializers, deserializers, and the auto-configured ConsumerFactory

Spring Boot builds the ConsumerFactory<Object, Object> bean from spring.kafka.consumer.*, defaulting key and value deserialization to StringDeserializer unless overridden. For JSON payloads, point the value (de)serializer at Spring Kafka’s own JSON support and, on the consumer side, tell it which Java type to materialize:

spring:
  kafka:
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: "com.example.orders.events"
        spring.json.value.default.type: com.example.orders.events.OrderPlacedEvent

A DefaultKafkaConsumerFactory (or DefaultKafkaProducerFactory) bean can be declared explicitly when per-consumer-group settings or an ErrorHandlingDeserializer wrapper are needed — the latter catches deserialization exceptions and hands them to the DefaultErrorHandler/DLT pipeline above instead of killing the container:

@Bean
public ConsumerFactory<String, OrderPlacedEvent> orderConsumerFactory(KafkaProperties properties) {
    Map<String, Object> props = properties.buildConsumerProperties(null);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
    props.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS, StringDeserializer.class);
    props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class);
    return new DefaultKafkaConsumerFactory<>(props);
}

Registering a custom ConsumerFactory/ProducerFactory bean overrides Spring Boot’s auto-configured one, so apply overrides through KafkaProperties/spring.kafka.* (as above) rather than hand-rolling the whole map whenever the defaults are otherwise adequate. Avro payloads validated against a Confluent Schema Registry use KafkaAvroSerializer/KafkaAvroDeserializer instead — see API-First: Messaging.

Producer-to-consumer flow

sequenceDiagram participant P as Producer bean participant KT as KafkaTemplate participant T as Topic (partitioned) participant CG as Consumer group participant L as @KafkaListener P->>KT: send("orders.placed", key, event) KT->>T: append record to partition = hash(key) T-->>CG: broker assigns partition to a group member CG->>L: poll() delivers record alt processing succeeds L-->>CG: acknowledgment.acknowledge() / auto-commit CG-->>T: commit offset else processing fails after retries L-->>CG: DefaultErrorHandler exhausts FixedBackOff CG->>T: DeadLetterPublishingRecoverer publishes to orders.placed.DLT CG-->>T: commit original offset (record considered handled) end

Further reading