Spring Kafka: @KafkaListener

Spring Kafka wraps the Kafka client in a Spring-friendly programming model. The @KafkaListener annotation turns any Spring bean method into a Kafka consumer: no poll loop, no manual lifecycle management. Spring manages the consumer thread, offset commits, and reconnection automatically.

Dependency & Auto-configuration
Basic @KafkaListener
Accessing ConsumerRecord Metadata
Concurrency: Multiple Consumer Threads
Manual Acknowledgement
Batch Listener
Message Filtering

Dependency & Auto-configuration

Add spring-kafka to your project: Spring Boot auto-configures a KafkaListenerContainerFactory using properties from application.properties.

XML

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

Properties

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=my-app-group
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

Basic @KafkaListener

Annotate any Spring-managed method with @KafkaListener specifying the topic(s) to consume from.

Java

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

@Service
public class OrderConsumer {

    @KafkaListener(topics = "orders", groupId = "order-service")
    public void consume(String message) {
        System.out.println("Received: " + message);
    }

    // Listen to multiple topics
    @KafkaListener(topics = {"orders", "returns"}, groupId = "order-service")
    public void consumeMultiple(String message) {
        System.out.println("Received: " + message);
    }
}

Add @EnableKafka to a @Configuration class to activate listener processing. Spring Boot auto-configures this when spring-kafka is on the classpath.

Accessing ConsumerRecord Metadata

Accept a ConsumerRecord parameter to access partition, offset, headers, key, and timestamp.

Java

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

@Service
public class OrderConsumer {

    @KafkaListener(topics = "orders")
    public void consume(ConsumerRecord<String, String> record) {
        System.out.printf("topic=%s partition=%d offset=%d key=%s value=%s%n",
            record.topic(), record.partition(), record.offset(),
            record.key(), record.value());
    }
}

Concurrency: Multiple Consumer Threads

Set concurrency to spin up multiple consumer threads (up to the number of partitions). Each thread is an independent consumer in the same group.

Java

@KafkaListener(topics = "orders", concurrency = "3") // 3 threads, each consumes assigned partitions
public void consume(String message) {
    System.out.println(Thread.currentThread().getName() + " received: " + message);
}

Properties

# Or configure the default factory concurrency globally
spring.kafka.listener.concurrency=3

Setting concurrency higher than the number of partitions wastes threads: the extra consumers will be idle since Kafka assigns at most one consumer per partition per group.

Manual Acknowledgement

By default Spring Kafka uses auto-acknowledgement after each record is processed. For precise control, switch to MANUAL or MANUAL_IMMEDIATE mode and call ack.acknowledge() explicitly.

Properties

spring.kafka.listener.ack-mode=MANUAL_IMMEDIATE

Java

import org.springframework.kafka.support.Acknowledgment;

@KafkaListener(topics = "orders")
public void consume(String message, Acknowledgment ack) {
    try {
        processOrder(message);
        ack.acknowledge(); // commit offset only after successful processing
    } catch (Exception e) {
        // Do NOT ack — message will be redelivered
        log.error("Processing failed, will retry", e);
    }
}

Batch Listener

Consume a whole batch of records in one method call by accepting a List and enabling batch mode on the container factory.

Properties

spring.kafka.listener.type=batch

Java

import java.util.List;

@KafkaListener(topics = "orders")
public void consumeBatch(List<String> messages) {
    System.out.println("Batch of " + messages.size() + " messages received");
    messages.forEach(this::processOrder);
}

// Or with full ConsumerRecord access
@KafkaListener(topics = "orders")
public void consumeBatch(List<ConsumerRecord<String, String>> records) {
    records.forEach(r -> processOrder(r.value()));
}

Message Filtering

Use a RecordFilterStrategy on the container factory to silently skip unwanted records before they reach the listener method.

Java

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;

@Configuration
public class KafkaConsumerConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> filteringFactory(
            ConsumerFactory<String, String> consumerFactory) {

        ConcurrentKafkaListenerContainerFactory<String, String> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);

        // Skip messages where the value starts with "CANCELLED"
        factory.setRecordFilterStrategy(record ->
            record.value().startsWith("CANCELLED"));

        return factory;
    }
}

// Reference the custom factory in the listener
@KafkaListener(topics = "orders", containerFactory = "filteringFactory")
public void consume(String message) {
    // Only receives non-CANCELLED orders
}

Contents

Dependency & Auto-configuration

Basic @KafkaListener

Accessing ConsumerRecord Metadata

Concurrency: Multiple Consumer Threads

Manual Acknowledgement

Batch Listener

Message Filtering

Related articles