Kafka with Spring Boot
Intermediatespring-kafka auto-configures KafkaTemplate for production and a ConcurrentKafkaListenerContainerFactory for consumption; set bootstrap-servers in application.properties.
Overview
Spring for Apache Kafka (spring-kafka) provides first-class Kafka integration for Spring Boot. Auto-configuration wires a KafkaTemplate for publishing and a ConcurrentKafkaListenerContainerFactory for consuming based on application.properties settings. For production, you must tune consumer and producer factories beyond the defaults: configure deserialiser error handling (ErrorHandlingDeserializer), set concurrency to match partition count, choose manual acknowledgement mode for at-least-once delivery, and enable publisher confirms for reliable publishing. Spring Boot 3 + spring-kafka 3.x uses the modern ContainerCustomizer and ObservationRegistry APIs; Micrometer traces are automatically propagated through Kafka headers.
Application.properties + auto-configured producer
Spring Boot auto-configures KafkaTemplate with these properties. Serialisation, acks, retries, and idempotence are the critical production settings.
# application.properties — producer
spring.kafka.bootstrap-servers=kafka-1:9092,kafka-2:9092,kafka-3:9092
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
# Reliability settings
spring.kafka.producer.acks=all # wait for all ISR replicas
spring.kafka.producer.retries=3
spring.kafka.producer.properties.enable.idempotence=true
# Throughput
spring.kafka.producer.batch-size=65536 # 64 KB batch
spring.kafka.producer.properties.linger.ms=5
# Sending via auto-configured KafkaTemplate
@Service
public class OrderProducer {
@Autowired KafkaTemplate<String, OrderEvent> template;
public void publish(OrderEvent event) {
CompletableFuture<SendResult<String, OrderEvent>> future =
template.send("order-events", event.getOrderId(), event);
future.whenComplete((result, ex) -> {
if (ex != null) {
log.error("Send failed", ex);
} else {
log.info("Sent to partition {} offset {}",
result.getRecordMetadata().partition(),
result.getRecordMetadata().offset());
}
});
}
}Consumer factory with manual ack and error handling
The auto-configured consumer factory is sufficient for demos but production requires: manual ack mode, concurrency tuned to partition count, and ErrorHandlingDeserializer to prevent poison-message crashes.
# application.properties — consumer
spring.kafka.consumer.group-id=order-processor
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer= org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
spring.kafka.consumer.properties.spring.deserializer.value.delegate.class= org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.properties.spring.json.trusted.packages=com.example.events
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.listener.ack-mode=MANUAL_IMMEDIATE
# Custom container factory for production
@Configuration
public class KafkaConsumerConfig {
@Bean
public ConcurrentKafkaListenerContainerFactory<String, OrderEvent>
kafkaListenerContainerFactory(ConsumerFactory<String, OrderEvent> cf) {
ConcurrentKafkaListenerContainerFactory<String, OrderEvent> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(cf);
factory.setConcurrency(3); // match partition count / N
factory.getContainerProperties().setAckMode(AckMode.MANUAL_IMMEDIATE);
// Retry 3x with 1s backoff, then send to DLT
factory.setCommonErrorHandler(
new DefaultErrorHandler(
new DeadLetterPublishingRecoverer(kafkaTemplate),
new FixedBackOff(1000L, 3)));
return factory;
}
}@KafkaListener with manual acknowledgement
Manual ack gives explicit control over offset commits. Always ack in a finally block or use try-catch to ensure the consumer does not stall on uncaught exceptions.
@Component
public class OrderEventConsumer {
@KafkaListener(
topics = "order-events",
groupId = "order-processor",
concurrency = "3", // 3 consumer threads
containerFactory = "kafkaListenerContainerFactory"
)
public void handle(
@Payload OrderEvent event,
@Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
@Header(KafkaHeaders.OFFSET) long offset,
Acknowledgment ack) {
log.info("Processing order {} from partition {} offset {}",
event.getOrderId(), partition, offset);
try {
orderService.process(event);
ack.acknowledge(); // commit offset only on success
} catch (RetryableException e) {
// Do NOT ack: let DefaultErrorHandler retry
throw e;
} catch (FatalException e) {
ack.acknowledge(); // ack to skip; DLT recoverer published it
}
}
// Batch listener variant
@KafkaListener(topics = "audit-events", batch = "true")
public void handleBatch(List<AuditEvent> events, Acknowledgment ack) {
auditService.saveAll(events);
ack.acknowledge();
}
}Key Points to Remember
- 1spring-kafka auto-configures KafkaTemplate and a listener container factory from application.properties.
- 2Use ErrorHandlingDeserializer to prevent a malformed message from crashing the entire consumer thread.
- 3Set listener concurrency to match the number of partitions divided by expected instance count.
- 4MANUAL_IMMEDIATE ack mode commits the offset only after your handler confirms success — essential for at-least-once delivery.
- 5DefaultErrorHandler replaces the old SeekToCurrentErrorHandler (spring-kafka 2.8+); use DeadLetterPublishingRecoverer for DLT routing.
- 6Enable spring.kafka.producer.properties.enable.idempotence=true to prevent duplicate publishes on producer retries.
Interview Questions
Sign in to ask AriaWhat is the difference between AckMode.BATCH and AckMode.MANUAL_IMMEDIATE in spring-kafka?
How does ErrorHandlingDeserializer prevent a poison message from crashing the consumer?
How would you configure a dead-letter topic (DLT) in Spring Kafka?
Why should listener concurrency match partition count and what happens if it exceeds it?
How does spring-kafka propagate Micrometer tracing context through Kafka headers?
Ask Aria about Kafka with Spring Boot
Your personal AI tutor — ask anything about this concept
Revision Status
Personal Notes
Sign in to save personal notes for this topic.
Discussion
Sign in to join the discussion.