Blog

Spring Boot Kafka Consumer

Key Takeaways

  • Use @KafkaListener on a Spring bean and let Spring Boot configure the listener infrastructure.
  • Consumers in one group divide partitions; different groups read independently using their own offsets.
  • Listener concurrency creates additional consumer instances, but partition assignments limit useful parallelism.
  • Configure offset commits and failure recovery explicitly. Processing a record and committing its offset are separate operations.

To build a Spring Boot Kafka consumer, add Kafka support, configure your brokers and deserializers, and annotate a component method with @KafkaListener. Spring Boot supplies the default listener container factory; your configuration determines group membership and acknowledgment behavior.

This example starts with plain-text messages, then covers four multi-consumer arrangements, offsets, retries, and JSON. If partitions and brokers are unfamiliar, read the Kafka architecture explanation first.

Spring Boot Kafka Consumer Example

Version baseline and project structure

The example targets Spring Boot 4.1.1, its managed Spring Kafka 4.1.1 and Kafka client 4.2.1, Java 25, and an Apache Kafka 4.3.1 broker. Install JDK 25, Maven 3.6.3 or later, and Docker before starting. These are explicit example versions, not a claim that every component is the latest patch.

pom.xml
src/main/java/com/example/demo/DemoApplication.java
src/main/java/com/example/demo/messaging/Consumer.java
src/main/resources/application.properties

pom.xml

For this Boot generation, use spring-boot-starter-kafka. The Boot parent manages the dependency versions, so don't independently pin spring-kafka or kafka-clients without a compatibility reason.

<project>
  <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>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>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-test</artifactId>
      <scope>test</scope>
    </dependency>
  </dependencies>
  <build>
    <plugins>
      <plugin>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-maven-plugin</artifactId>
      </plugin>
    </plugins>
  </build>
</project>

DemoApplication.java

package com.example.demo;

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

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

Consumer.java

package com.example.demo.messaging;

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

@Component
public class Consumer {
    @KafkaListener(topics = "test")
    public void processMessage(String content) {
        System.out.println("Message received: " + content);
    }
}

The listener uses the configured consumer group because the annotation doesn't specify groupId. For additional annotation options, see the Spring Boot KafkaListener example.

application.properties

spring.kafka.bootstrap-servers=${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}
spring.kafka.consumer.group-id=myGroup
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.enable-auto-commit=false
spring.kafka.listener.ack-mode=record
spring.kafka.listener.missing-topics-fatal=true
spring.kafka.consumer.properties[group.protocol]=classic
spring.kafka.consumer.properties[partition.assignment.strategy]=org.apache.kafka.clients.consumer.RangeAssignor

Both deserializers turn Kafka's stored bytes into strings. This example deliberately selects container-managed, per-record commits and an explicit classic-protocol assignor instead of relying on assignment defaults.

auto-offset-reset=earliest applies when a partition has no valid committed offset for the group. It does not rewind an existing group on every restart.

Running the Example

Start Kafka in KRaft mode

Kafka 4.0 removed ZooKeeper mode. Kafka 4.x uses KRaft, Kafka's built-in metadata-management architecture; the old ZooKeeper startup commands are no longer applicable.

Start the official JVM-based image:

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

The image's default setup is a single combined broker/controller node. The following commands use the tools installed under /opt/kafka/bin inside that container, so you don't need a separate Kafka installation on your host.

After Kafka starts, create both topics used by the examples. If startup fails, inspect docker logs kafka before continuing.

for topic in test test2; do
  docker exec kafka /opt/kafka/bin/kafka-topics.sh \
    --bootstrap-server localhost:9092 \
    --create --if-not-exists --topic "$topic" \
    --partitions 3 --replication-factor 1
done

Each topic now has three partitions and one replica. Topic creation is explicit; the application doesn't depend on automatic topic creation.

Run the application and produce messages

From the directory containing pom.xml, run:

mvn spring-boot:run

The Maven plugin's spring-boot:run goal starts the application. In another terminal, start the console producer:

docker exec -it kafka /opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server localhost:9092 --topic test

Type Hello world and press Enter. The producer prompt and expected application output look like this:

Producer:
> Hello world

Application:
Message received: Hello world

Spring Boot Kafka Multiple Consumer Example

Use each variation separately. The Java blocks below are method fragments that replace the methods inside the existing Consumer class; keep its package, imports, and @Component annotation.

1) Multiple Consumers in the Same Consumer Group

// Fragment inside Consumer: replace processMessage.
@KafkaListener(topics = "test", groupId = "myGroup", concurrency = "2")
public void processMessage(String content) {
    System.out.println("Message received: " + content);
}

concurrency = "2" creates two child listener containers, each with a Kafka consumer. With three partitions, a typical assignment is two partitions to one consumer and one to the other. Each partition belongs to at most one member of this group at a time.

Two application instances with concurrency two create four group members, but only three can own partitions of this topic. Concurrency doesn't create partitions or guarantee equal processing load. The same listener bean is invoked concurrently, so keep it stateless or make shared state thread-safe.

2) Multiple Consumer Groups Reading from the Same Topic

// Fragments inside Consumer: replace its existing listener methods.
@KafkaListener(topics = "test", groupId = "myGroup")
public void processMessage(String content) {
    System.out.println("myGroup: " + content);
}

@KafkaListener(topics = "test", groupId = "anotherGroup")
public void processMessage2(String content) {
    System.out.println("anotherGroup: " + content);
}

These groups read independently, subject to their subscriptions, offsets, retention, and successful processing. A new group using earliest can read retained history; an existing group resumes from its own committed offsets. This is independent consumption, not broadcast delivery to every member of one group.

One possible assignment for topic test:

myGroup                         anotherGroup
  consumer A <- P0, P1            consumer C <- P0, P1, P2
  consumer B <- P2

Partition ownership is divided  Independent offsets and consumption
within this group.              for the same topic.

3) Multiple Consumer Groups Reading from Different Topics

// Fragments inside Consumer: replace its existing listener methods.
@KafkaListener(topics = "test", groupId = "myGroup")
public void processMessage(String content) {
    System.out.println("test / myGroup: " + content);
}

@KafkaListener(topics = "test2", groupId = "anotherGroup")
public void processMessage2(String content) {
    System.out.println("test2 / anotherGroup: " + content);
}

Here the subscriptions and groups are separate: one listener processes test, the other test2. Produce to test2 by changing the console producer's --topic argument.

4) Multiple Consumers in the Same Group Reading from Different Topics

// Fragment inside Consumer: replace its existing listener methods.
@KafkaListener(topics = {"test", "test2"},
        groupId = "myGroup", concurrency = "2")
public void processMessage(String content) {
    System.out.println("Message received: " + content);
}

Both consumers subscribe to both topics. Partition assignment determines which topic partitions each consumer receives; concurrency does not mean one consumer per topic.

Range versus round-robin assignment

The configured RangeAssignor assigns partitions per topic. For two consumers and two three-partition topics with identical subscriptions, it produces ownership counts of four and two. RoundRobinAssignor produces three and three in this arrangement. Neither balances message rates or processing cost.

To try round-robin assignment, replace the assignment property on every member of this example group:

spring.kafka.consumer.properties[partition.assignment.strategy]=org.apache.kafka.clients.consumer.RoundRobinAssignor

Round-robin assignment can still be uneven when subscriptions differ. This setting belongs to the classic group protocol; the newer consumer protocol uses server-side assignors instead.

Inspect assignments and consumer lag

docker exec kafka /opt/kafka/bin/kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 --group myGroup \
  --describe --members --verbose

docker exec kafka /opt/kafka/bin/kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 --group myGroup --describe

The first command shows members and their assigned partitions. The second shows committed offsets, log-end offsets, and lag: how far the group's committed progress trails the log. Run them after the group stabilizes rather than inferring assignments from console output alone.

Consumer Offsets and Acknowledgments

An offset identifies a position within a partition. A committed offset records where the group should resume; committing 43 means the next position to read is 43, not that record 43 has already been processed.

Read offset 42 -> finish business operation -> commit offset 43
                            |
                            +-- crash before commit: 42 may be replayed

Processing before committing permits duplicate processing after a crash. Committing before durable work finishes can instead skip unfinished work after restart. Make side effects idempotent—safe to repeat—and don't treat an offset commit as a transaction with your database or HTTP service.

In Spring Kafka 4.1.1, the container default is BATCH, and Kafka auto-commit is disabled unless explicitly enabled. This example explicitly sets enable-auto-commit=false and ack-mode=record: for normal successful delivery, the container commits after the record listener returns. Error recovery can also advance offsets, as discussed next.

AckModeNontransactional commit behavior
RECORDCommit after each record is processed by the listener.
BATCHCommit after all records from the previous poll are processed, even with a record listener.
MANUALThe listener acknowledges; commits then follow batch semantics.
MANUAL_IMMEDIATECommit immediately when acknowledged on the consumer thread.

These are the relevant acknowledgment modes for this tutorial. They don't override Kafka auto-commit when it is enabled.

An explicit manual acknowledgment variant

For this alternative, change spring.kafka.listener.ack-mode to manual-immediate, keep auto-commit disabled, and replace Consumer.java with:

package com.example.demo.messaging;

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
public class Consumer {
    @KafkaListener(topics = "test")
    public void processMessage(String content, Acknowledgment acknowledgment) {
        System.out.println("Message received: " + content);
        acknowledgment.acknowledge();
    }
}

The acknowledgment happens synchronously on the listener's consumer thread. Put your durable business operation before it; the print statement only illustrates that boundary. Manual acknowledgment does not remove the crash window between a side effect and its offset commit.

Failure Handling, Retries, and Dead-Letter Topics

If processMessage throws, the container delegates to its error handler. Spring Kafka's default DefaultErrorHandler uses nine retries with no delay before its default recoverer logs the failed record; some nonretryable failures bypass retries. Logging and moving on is not durable failure recovery.

The following optional recipe is for the original String listener with RECORD acknowledgment, not the manual or JSON variants. Create a dead-letter topic, a separate topic holding records that couldn't be processed:

docker exec kafka /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 --create --if-not-exists \
  --topic test-dlt --partitions 3 --replication-factor 1

Add producer serializers to application.properties:

spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

Add src/main/java/com/example/demo/ErrorHandlingConfig.java:

package com.example.demo;

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 ErrorHandlingConfig {
    @Bean
    DefaultErrorHandler errorHandler(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));
    }
}

Boot associates this error-handler bean with its default factory. Retryable failures get two retries, one second apart, before recovery publishes to test-dlt; failures classified as nonretryable go directly to recovery. The resolver preserves the original partition, so provision at least as many dead-letter partitions as source partitions. Create test2-dlt too if you enable the second topic.

setFailIfSendResultIsError(true) makes failed publication throw instead of reporting successful recovery. After successful recovery, the RECORD-mode container can commit past the failed record. This nontransactional recipe can still produce duplicates across crashes, so alert on dead-letter traffic and establish a replay procedure.

To exercise recovery, temporarily throw IllegalStateException when the input equals fail, then publish that value. Never catch an exception and return normally unless you intend the container to treat the record as successfully processed.

Consuming JSON Instead of Strings

A String listener doesn't need JSON mapping. A typed JSON listener needs Jackson on the classpath and an explicitly configured deserializer or converter; arbitrary JSON is not automatically converted just because Spring Kafka is present. For this Spring Kafka generation, use JacksonJsonDeserializer, which uses Jackson 3.

Start from a separate copy of the original project without the String-specific error-handler recipe. Add this Boot-managed dependency inside dependencies:

<dependency>
  <groupId>tools.jackson.core</groupId>
  <artifactId>jackson-databind</artifactId>
</dependency>

The tools.jackson.core coordinates are managed by this Boot baseline.

Create src/main/java/com/example/demo/messaging/OrderEvent.java:

package com.example.demo.messaging;

public record OrderEvent(String orderId, int totalCents) {}

Replace Consumer.java with:

package com.example.demo.messaging;

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

@Component
public class Consumer {
    @KafkaListener(topics = "orders", groupId = "ordersGroup")
    public void processMessage(OrderEvent event) {
        System.out.println("Order received: " + event.orderId());
    }
}

Keep the original common properties, replace the value-deserializer setting, and add:

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.demo.messaging.OrderEvent
spring.kafka.consumer.properties[spring.json.use.type.headers]=false
spring.kafka.consumer.properties[spring.json.trusted.packages]=com.example.demo.messaging

The configured target type is OrderEvent, and producer type headers are ignored. ErrorHandlingDeserializer delegates parsing and makes deserialization failures available to the container error handler rather than your record listener.

Create the topic, restart the application, then send JSON:

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

printf '%s\n' '{"orderId":"A-100","totalCents":1250}' | \
  docker exec -i kafka /opt/kafka/bin/kafka-console-producer.sh \
    --bootstrap-server localhost:9092 --topic orders

Choose a recovery policy before production. JSON dead-letter publishing must support both normal objects and the raw byte[] supplied for deserialization failures; don't reuse the String-only producer configuration unchanged. See the Spring Boot Kafka consumer JSON example for a focused walkthrough.

Common Mistakes

  • Expecting replay after setting earliest: existing committed offsets still take precedence.
  • Adding threads to fix a hot partition: one group's partition still has only one owning consumer.
  • Returning before asynchronous work finishes: automatic container acknowledgment can get ahead of that work.
  • Assuming retries preserve failed messages forever: configure and monitor recovery instead of relying on the default logging recoverer.

Production Checklist

  • Secure connections: externalize broker addresses, use TLS and your cluster's authentication and authorization settings, and obtain credentials from secret management. Kafka supports encrypted connections, SASL authentication, and access controls.
  • Bound processing time: keep poll processing within max.poll.interval.ms; reduce max.poll.records when slow work makes polls too large.
  • Shut down cleanly: let Spring stop the listener containers and budget deployment termination time for processing, commits, and consumer closure.
  • Observe actual consumption: monitor lag, listener failures, rebalances, dead-letter traffic, and whether containers remain running. Spring Kafka provides Micrometer listener timing and observation support; process health alone isn't enough.
  • Test failure boundaries: cover restarts, duplicate side effects, malformed payloads, and unavailable dead-letter destinations. Use the Spring Kafka testing examples as a starting point.

FAQ

How do I create a Kafka consumer in Spring Boot?

Add Kafka support, configure bootstrap servers, a consumer group, and deserializers, then annotate a Spring component method with @KafkaListener. Spring Boot supplies a default listener container factory when you haven't defined one.

What does concurrency do in @KafkaListener?

Concurrency creates multiple child listener containers and Kafka consumer instances for that listener. A three-partition topic can have at most three actively assigned consumers within one group, regardless of how many application instances or listener threads you start.

Can multiple Kafka consumer groups read the same topic?

Yes. Each group maintains independent offsets and can consume the topic's retained records according to its subscriptions and starting position. Members within one group divide partition ownership rather than each receiving every record.

How are offsets committed in Spring Kafka?

With Kafka auto-commit disabled, the listener container's AckMode controls commit behavior. RECORD commits after successful record processing, BATCH commits after a poll's records are processed, and manual modes let the listener acknowledge explicitly; configured error recovery can also advance offsets.

Can Spring Boot Kafka run with Kafka in KRaft mode?

Yes. Configure the Kafka broker bootstrap addresses as usual; the consumer does not connect to ZooKeeper. Kafka 4.0 removed ZooKeeper mode, and Kafka 4.x runs in KRaft mode.

Further Reading

Written by Sam Erickson

Software engineer. I write every article on StackChief myself. Author of Mastering the Tech Interview. More about me

Related reading