@KafkaListener
Intermediate@KafkaListener annotates a method to consume records from one or more topics; supports batch consumption, error handlers, and retry with @RetryableTopic for DLT routing.
Overview
@KafkaListener is the Spring Kafka annotation that wires a method to a Kafka consumer. Spring manages the consumer lifecycle (start, stop, rebalance), the poll loop, deserialization, thread management, and error handling — you just write the business logic. The annotation supports single-record and batch modes, regex-based topic subscriptions, topic patterns, concurrency (multiple consumer threads), manual acknowledgement, and error handlers. @RetryableTopic (Spring Kafka 2.7+) adds retry with exponential back-off and automatic Dead Letter Topic routing with almost no configuration.
@KafkaListener — Common Configurations
The annotation parameters control topics, group ID, concurrency, and container factory. Method parameters are resolved by Spring from the record's key, value, headers, and metadata.
// 1. Single-record listener — one ConsumerRecord per call
@Component
public class OrderConsumer {
// topics: list of topic names
// groupId: overrides spring.kafka.consumer.group-id for this listener
// concurrency: starts N concurrent consumer threads (up to partition count)
@KafkaListener(
topics = {"orders", "order-corrections"},
groupId = "order-processor",
concurrency = "3"
)
public void onOrder(
ConsumerRecord<String, Order> record, // full record with headers
@Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
@Header(KafkaHeaders.OFFSET) long offset) {
log.info("Processing order {} from partition {} offset {}",
record.key(), partition, offset);
orderService.process(record.value());
}
// 2. Simplified — just the value (Spring resolves via type)
@KafkaListener(topics = "payments", groupId = "payment-service")
public void onPayment(Payment payment) {
paymentService.handle(payment);
}
// 3. Regex subscription — all topics matching the pattern
@KafkaListener(topicPattern = "order.*", groupId = "analytics")
public void onAnyOrderEvent(ConsumerRecord<String, String> record) {
analyticsService.ingest(record);
}
}Batch Listener — Process Multiple Records per Poll
Set the listener container factory's batchListener=true (or spring.kafka.listener.type=batch) to receive a whole batch of records per call. Useful for bulk DB inserts or downstream HTTP batching. Manual ack in batch mode can acknowledge the entire batch or individual offsets.
# application.yml — enable batch listener
spring:
kafka:
listener:
type: batch
ack-mode: MANUAL_IMMEDIATE
@Component
public class BatchOrderConsumer {
@KafkaListener(topics = "orders", groupId = "batch-processor")
public void onOrders(
List<ConsumerRecord<String, Order>> records,
Acknowledgment ack) {
log.info("Processing batch of {} orders", records.size());
// Bulk insert — much faster than one-by-one
List<Order> orders = records.stream()
.map(ConsumerRecord::value)
.collect(Collectors.toList());
orderRepository.saveAll(orders);
ack.acknowledge(); // acknowledge the full batch
}
}@RetryableTopic — Automatic Retry with Back-off and DLT
@RetryableTopic (Spring Kafka 2.7+) is the modern way to add retry logic to @KafkaListener. It creates retry topics automatically and routes records through them with configurable back-off. After exhausting retries, records go to a .DLT topic. No manual retry loop or DLQ wiring needed.
@Component
public class PaymentConsumer {
// @RetryableTopic creates these topics automatically:
// payments-retry-0 (immediate retry)
// payments-retry-1 (10s delay)
// payments-retry-2 (20s delay)
// payments.DLT (after 3 retries, sent here)
@RetryableTopic(
attempts = "4", // 1 original + 3 retries
backoff = @Backoff(
delay = 1_000, // 1s first retry
multiplier = 2, // 2s, 4s, 8s...
maxDelay = 10_000 // cap at 10s
),
include = TransientDataException.class, // only retry these
exclude = PermanentException.class, // skip retries for these
dltTopicSuffix = ".DLT",
autoCreateTopics = "true"
)
@KafkaListener(topics = "payments", groupId = "payment-service")
public void onPayment(Payment payment) {
paymentService.process(payment); // throws TransientDataException on DB hiccup
}
// DLT listener — runs after all retries exhausted
@DltHandler
public void onDlt(Payment payment, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
log.error("Payment {} permanently failed from topic {}", payment.getId(), topic);
alertService.notify(payment);
}
}Key Points to Remember
- 1@KafkaListener manages the consumer lifecycle, poll loop, and deserialization — you write only business logic.
- 2concurrency = "N" starts N consumer threads per listener; effective up to the partition count of the topic.
- 3Batch mode (type: batch) delivers a List<ConsumerRecord<K,V>> per call — much more efficient for bulk DB writes.
- 4@RetryableTopic creates retry topics with exponential back-off and a DLT automatically — the modern replacement for manual retry loops.
- 5Use @DltHandler in the same class to handle records that exhausted all retries (alert, persist, replay later).
- 6Inject Acknowledgment + Channel for manual ack mode; inject @Header(KafkaHeaders.RECEIVED_PARTITION) / OFFSET for record metadata.
Interview Questions
Sign in to ask AriaHow does @KafkaListener work under the hood — who manages the poll loop?
When would you use batch mode for a @KafkaListener?
How does @RetryableTopic implement retry with back-off?
What is the difference between @RetryableTopic and a manual retry loop inside a listener?
How do you handle messages that should be retried for transient errors but immediately DLT'd for permanent errors?
Ask Aria about @KafkaListener
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.