@KafkaListener | Spring Boot Example

Key Takeaways

• @KafkaListener lets Spring manage Kafka polling and listener-container lifecycle while you write the message-handling logic.
• A basic Spring Boot consumer needs the Kafka starter, broker addresses, a consumer group, deserializers, and a listener bean—not handwritten factories.
• Consumer groups distribute partitions across consumers; adding threads beyond the available partitions doesn't add useful parallelism.
• Offset commits, retries, and dead-letter recovery need deliberate configuration. Make external side effects safe to repeat.

@KafkaListener marks a Spring bean method as a Kafka message handler. Spring Kafka connects that method to a listener container, which runs the consumer loop and invokes your code for incoming records. Spring Boot supplies the infrastructure for the basic example below through auto-configuration.

Spring Boot KafkaListener Example: Project Setup

This example uses JDK 25 and Spring Boot 4.1.1. Boot's managed dependencies supply Spring Kafka 4.1.1 and Kafka Clients 4.2.1; the local broker below uses Kafka 4.2.2. These are explicit baseline versions, not a claim that each is the newest release.

You'll need Docker and either Maven 3.6.3 or later, or Gradle 9.1 or later within Boot's supported 9.x line. Gradle's compatibility matrix lists 9.1.0 as the first version supporting both Java 25 toolchains and running Gradle on Java 25. See Boot's system requirements when adapting the build.

Maven Dependencies

Create pom.xml. The Boot parent manages dependency versions, and the Spring Boot Maven plugin provides application build and execution support.

<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>4.1.1</version>
        <relativePath/>
    </parent>
    <groupId>com.example</groupId>
    <artifactId>kafka-demo</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <properties>
        <java.version>25</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-kafka</artifactId>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

Gradle Alternative

If you prefer Gradle, use this build.gradle.kts instead of the Maven build. It uses Gradle's native platform support to import Boot's dependency-management BOM—the list of coordinated dependency versions.

plugins {
    java
    id("org.springframework.boot") version "4.1.1"
}

repositories {
    mavenCentral()
}

java {
    toolchain {
        languageVersion = JavaLanguageVersion.of(25)
    }
}

dependencies {
    implementation(platform(org.springframework.boot.gradle.plugin.SpringBootPlugin.BOM_COORDINATES))
    implementation("org.springframework.boot:spring-boot-starter-kafka")
}

Using an installed Gradle 9.1 or later within the 9.x line, generate a Gradle wrapper pinned to 9.1.0, then run the application through it:

gradle wrapper --gradle-version 9.1.0
./gradlew bootRun

Start Kafka and Create the Topic

Start the official Kafka image in one terminal. This is a disposable, single-node development broker; the published port is bound to your machine's loopback address. The application will run on your host, not inside another container.

docker run --rm --name kafka \
  -p 127.0.0.1:9092:9092 apache/kafka:4.2.2

Wait for broker startup to finish. In another terminal, create myTopic with three partitions:

docker exec kafka /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 --create --topic myTopic \
  --partitions 3 --replication-factor 1

Kafka's topic-management documentation explains these options. Replication factor one suits this one-broker demonstration, not a resilient production deployment.

Configure the Kafka Consumer

Create src/main/resources/application.yaml:

spring:
  kafka:
    bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}
    consumer:
      group-id: demo-consumers
      auto-offset-reset: earliest
      enable-auto-commit: false
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    listener:
      ack-mode: record
      concurrency: 1

app:
  kafka:
    topic: myTopic

bootstrap-servers takes broker addresses such as localhost:9092, not HTTP URLs. earliest applies when the group has no valid committed offset; it does not replay the topic on every restart. The producer settings support the dead-letter example later.

For this setup, Boot auto-configures the consumer factory and listener-container factory from spring.kafka.*. You don't need to declare either factory yourself.

Create the Listener and Run the Application

Create src/main/java/com/example/kafkademo/KafkaConsumer.java:

package com.example.kafkademo;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class KafkaConsumer {
    private static final Logger log = LoggerFactory.getLogger(KafkaConsumer.class);

    @KafkaListener(
        id = "foo",
        topics = "${app.kafka.topic}",
        groupId = "${spring.kafka.consumer.group-id}"
    )
    public void listen(String message) {
        log.info("Received message: {}", message);
    }
}

Create src/main/java/com/example/kafkademo/KafkaDemoApplication.java:

package com.example.kafkademo;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class KafkaDemoApplication {
    public static void main(String[] args) {
        SpringApplication.run(KafkaDemoApplication.class, args);
    }
}

The application class sits above the listener in the same package, so component scanning discovers it. Run from the project root:

mvn spring-boot:run
# Gradle alternative: gradle bootRun

Send a record from another terminal:

printf 'hello Kafka\n' | docker exec -i kafka \
  /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server localhost:9092 --topic myTopic

Your application log should contain:

Received message: hello Kafka

The console producer writes each input line as a separate record. Use the Kafka quick start for additional send-and-read commands.

What @KafkaListener and the Factories Actually Do

With the native client, you own subscription, the repeated poll() calls, offset coordination, and consumer shutdown. Spring Kafka moves that infrastructure into a container; your listener still owns application-specific processing.

@KafkaListener method -> endpoint -> container factory -> listener container
                                                            |
KafkaListenerEndpointRegistry -- manages lifecycle ----------+
                                                            |
                                                            v
ConsumerFactory -- creates ---------------------------> KafkaConsumer
                                                            |
                                                            v
                                                   broker / partitions

A ConsumerFactory creates Kafka consumer instances. A ConcurrentKafkaListenerContainerFactory creates concurrent containers, which delegate to one or more single-threaded KafkaMessageListenerContainer instances. The annotation doesn't independently construct a consumer or a container.

Containers created for annotated endpoints are registered with KafkaListenerEndpointRegistry, rather than being ordinary application-context beans. The registry manages their lifecycle; getListenerContainer("foo") retrieves this example's container for operations such as start() and stop(). Containers declared directly as beans aren't lifecycle-managed by that registry.

When You Need a Custom Factory

Use custom factories for genuinely different listener policies, such as record versus batch processing. If you replace Boot's default factory, its container-factory configurer lets you retain Boot's configuration before applying overrides.

package com.example.kafkademo;

import org.springframework.boot.kafka.autoconfigure.ConcurrentKafkaListenerContainerFactoryConfigurer;
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 ListenerFactoryConfig {
    @Bean
    ConcurrentKafkaListenerContainerFactory<Object, Object> kafkaListenerContainerFactory(
            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
            ConsumerFactory<Object, Object> consumerFactory) {
        var factory = new ConcurrentKafkaListenerContainerFactory<Object, Object>();
        configurer.configure(factory, consumerFactory);
        factory.setConcurrency(2);
        return factory;
    }
}

This optional class changes concurrency to two after applying Boot's settings. Leave it out of the minimal example. For more configuration context, see our Spring Boot Kafka consumer guide.

Consumer Groups, Multiple Topics, and Concurrency

Consumers with the same group ID divide the subscribed partitions among themselves. Within a conventional consumer group, each partition is assigned to one consumer at a time; a different group has independent consumption progress.

The annotation's groupId overrides the factory's group ID. Another important detail: when groupId is absent, an explicit listener id is used as the group ID unless idIsGroup = false. The original id = "foo" example therefore wasn't necessarily missing a group ID; this version makes the intended group explicit.

Set concurrency = "3" on @KafkaListener to override factory concurrency for that endpoint. Across replicas, useful parallelism for our single three-partition topic is capped at three consumers:

These limits concern ordinary partition-assigned consumers. Keep listener state thread-safe because concurrent consumers can invoke the same listener bean. See the listener thread-safety guidance and our Kafka architecture explanation.

For multiple topics, use topics = {"orders", "payments"}; for names matched by a regular expression, use topicPattern instead.

Offset Commits and Delivery Guarantees

An offset identifies a record's position within a partition. This example disables Kafka auto-commit and selects RECORD acknowledgment mode: the container commits after the listener successfully processes each record. Spring Kafka's default acknowledgment mode is BATCH, which commits after processing the records returned by a poll.

Design for at-least-once processing: a crash after a database write but before the offset commit can cause that write to run again. Use an event ID with a uniqueness constraint or another idempotency mechanism so repeating a side effect doesn't change the result twice. Kafka transactions support exactly-once Kafka read-process-write flows; the annotation alone doesn't make database writes or HTTP calls exactly-once.

Handle Exceptions, Bounded Retries, and Dead Letters

Let processing failures escape the listener so the container can apply its error policy. Catching an exception and returning normally can make a failed record look successfully processed. Boot attaches a CommonErrorHandler bean to its default factory.

For the String example, create a dead-letter topic—a topic that holds records you couldn't process:

docker exec kafka /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 --create --topic myTopic.DLT \
  --partitions 3 --replication-factor 1

Add KafkaErrorConfig.java in the application package:

package com.example.kafkademo;

import org.apache.kafka.common.TopicPartition;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.util.backoff.FixedBackOff;

@Configuration
public class KafkaErrorConfig {
    @Bean
    DefaultErrorHandler kafkaErrorHandler(KafkaTemplate<Object, Object> template) {
        var recoverer = new DeadLetterPublishingRecoverer(template,
                (record, exception) -> new TopicPartition(
                        record.topic() + ".DLT", record.partition()));
        recoverer.setFailIfSendResultIsError(true);
        return new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 2L));
    }
}

For retryable failures, this configures two retries with a one-second backoff, then dead-letter recovery: three delivery attempts in that retry sequence. Some exception types, including deserialization failures, are classified as non-retryable by default. The explicit destination resolver preserves the source partition, so provision matching partitions on the DLT.

setFailIfSendResultIsError(true) makes a failed DLT send fail recovery rather than silently count as success. Recovery failures can restart delivery and retry cycles, so the configured retry count isn't a lifetime limit during an outage. After successful recovery, the container can advance past the failed record.

To exercise this path, temporarily add the following at the beginning of listen, restart the application, and send the line fail with the earlier producer command:

// Fragment inside listen(String message), before the log statement.
if ("fail".equals(message)) {
    throw new IllegalStateException("Demonstration failure");
}

Read the recovered value with:

docker exec -i kafka /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 --topic myTopic.DLT --from-beginning

Monitor the DLT and define a repair-and-replay procedure. Moving a failed event aside also means later events can proceed without its business operation having succeeded.

JSON Payloads and Record Metadata

A String parameter receives the deserialized value, not an automatically constructed domain object. When you need keys, headers, partitions, or offsets, use ConsumerRecord<String, String>, imported from org.apache.kafka.clients.consumer.ConsumerRecord, and inspect key(), headers(), partition(), or offset().

For a separate JSON variant, remove KafkaConsumer and the String-specific KafkaErrorConfig. Add org.springframework.boot:spring-boot-starter-json to your dependencies; Boot's JSON support supplies Jackson 3. Use JacksonJsonDeserializer: the older JsonDeserializer is deprecated in favor of the Jackson 3 implementation.

Create OrderCreated.java:

package com.example.kafkademo;

public record OrderCreated(String orderId) {}

Create OrderListener.java:

package com.example.kafkademo;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class OrderListener {
    private static final Logger log = LoggerFactory.getLogger(OrderListener.class);

    @KafkaListener(id = "orders", topics = "${app.kafka.topic}",
            groupId = "${spring.kafka.consumer.group-id}")
    public void listen(OrderCreated order) {
        log.info("Received order: {}", order.orderId());
    }
}

Keep the original YAML and add these overrides in application.properties:

app.kafka.topic=orders-json
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.JacksonJsonDeserializer
spring.kafka.consumer.properties[spring.json.value.default.type]=com.example.kafkademo.OrderCreated
spring.kafka.consumer.properties[spring.json.use.type.headers]=false
spring.kafka.consumer.properties[spring.json.trusted.packages]=com.example.kafkademo

The configured target type and disabled type headers keep this consumer bound to your chosen class. ErrorHandlingDeserializer delegates JSON decoding and exposes decoding failures to container error handling instead of invoking the listener with an invalid payload.

Create and populate the separate topic, then restart the application:

docker exec kafka /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 --create --topic orders-json \
  --partitions 3 --replication-factor 1

printf '{"orderId":"order-123"}\n' | docker exec -i kafka \
  /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server localhost:9092 --topic orders-json

Production Configuration and Common Mistakes

For remote brokers, use reachable broker addresses and the cluster's required authentication. This override fragment illustrates SASL authentication over TLS; supply the complete JAAS login configuration through your deployment's secret mechanism:

# Deployment override fragment; merge with the application configuration.
spring.kafka.bootstrap-servers=${KAFKA_BOOTSTRAP_SERVERS}
spring.kafka.properties[security.protocol]=SASL_SSL
spring.kafka.properties[sasl.mechanism]=SCRAM-SHA-512
spring.kafka.properties[sasl.jaas.config]=${KAFKA_SASL_JAAS_CONFIG}

Configure certificate trust for your broker's certificate chain as needed. Don't disable hostname verification to work around certificate problems. Brokers must also advertise addresses reachable from the application's network; host-local localhost settings won't work unchanged across containers or machines.

• Don't swallow processing exceptions. Let the configured recovery policy see failures.
• Don't hand work to an executor and return immediately. In this synchronous example, listener completion controls commit eligibility.
• Don't confuse acknowledgment mode with listener shape. Batch offset commits don't require a batch listener method.
• Budget processing and retry time. Keep them within your consumer's max.poll.interval.ms budget, or choose a suitable pausing/backoff strategy.
• Test failure paths. Cover duplicate delivery, malformed JSON, DLT publishing failures, and restart behavior with Spring Kafka integration tests.
• Keep logs safe. Full payload logging belongs in this local demonstration, not an application handling sensitive data.

The container reference documents acknowledgment behavior, and the error-handling reference explains backoff strategies.

FAQ

How do I configure a Kafka listener in Spring Boot?

Add spring-boot-starter-kafka, configure spring.kafka.bootstrap-servers, a consumer group, and deserializers, then put @KafkaListener on a Spring bean method. The example uses StringDeserializer for both keys and values.

Does Spring Boot create the Kafka consumer factory automatically?

Yes. For a basic Kafka application, Spring Boot auto-configures the consumer factory and a default listener-container factory from spring.kafka properties. Define custom factories only when your application needs different behavior.

How do I set the consumer group ID for @KafkaListener?

Set groupId on the annotation, or use spring.kafka.consumer.group-id with idIsGroup set to false when the annotation has an explicit id. Without an explicit groupId, the listener id is otherwise used as the group ID.

How does @KafkaListener handle exceptions and retries?

Listener exceptions propagate to the container's error handling. Configure DefaultErrorHandler with a backoff and a recoverer to control retries and dead-letter publishing; Boot attaches a CommonErrorHandler bean to its default factory.

How many listener threads should I configure for a Kafka topic?

For ordinary partition-assigned consumers, useful parallelism in one consumer group cannot exceed the subscribed partitions. For a single three-partition topic, six configured consumers across application replicas leave at least three without assigned partitions.

Further Reading

• Spring Boot Kafka support — auto-configuration and application properties.
• @KafkaListener annotation reference — method signatures and annotation overrides.
• Listener lifecycle management — registry ownership and container operations.
• Spring Kafka error handling — retries, recovery, and dead letters.
• Serialization and deserialization — JSON types and decoding failures.
• Kafka consumer configuration — broker addresses, offsets, and polling settings.
• Exactly-once semantics — the scope of transactional Kafka processing.
• DeadLetterPublishingRecoverer API — destination routing and publication failure handling.