Key Takeaways
- Spring Kafka simplifies setting up consumers using annotations.
- Adjust consumer settings through simple configuration properties.
- Different multi-consumer configurations support various use cases.
Spring Boot Kafka Consumer Example
pom.xml
<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>3.2.0</version>
<relativePath />
</parent>
<groupId>com.example</groupId>
<artifactId>demo</artifactId>
<version>0.0.1-SNAPSHOT</version>
<properties>
<java.version>17</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-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>
The pom.xml defines our Maven build dependencies. Spring Boot and Kafka libraries are included, but note that the additional jackson-databind dependency is no longer necessary in recent Spring Kafka versions.
DemoApplication.java
@SpringBootApplication
public class DemoApplication {
public static void main(String[] args) {
SpringApplication.run(DemoApplication.class, args);
}
}
The DemoApplication.java class initiates our Spring Boot application. The @SpringBootApplication annotation automatically configures essential components like KafkaAutoConfiguration.
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 Consumer.java class listens for messages using the @KafkaListener annotation, specifying topics directly within it.
application.properties
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=myGroup
Your configuration in application.properties stays minimal thanks to Spring Boot's auto-configuration capabilities. Here, we're running Kafka locally.
Running the Example
You'll need a local Kafka setup to consume messages. Here’s how to start Kafka services:
Starting Kafka from the CLI:
Begin by starting Zookeeper and Kafka servers locally.
./zookeeper-server-start.sh /usr/local/etc/kafka/zookeeper.properties
Ensure the path to your installation is correct for starting Zookeeper.
./kafka-server-start.sh /usr/local/etc/kafka/server.properties
This launches the Kafka server with default settings.
Producing Messages from the CLI:
To test, produce some messages for the "test" topic:
./kafka-console-producer --bootstrap-server=localhost:9092 --topic test
Send messages like this:
> Hello world
If all goes well, you’ll see:
Message received: Hello world
Spring Boot Kafka Multiple Consumer Example
Spring Kafka supports multiple consumer configurations within the same application.
1) Multiple Consumers in the Same Consumer Group
@Component
public class Consumer {
@KafkaListener(topics = "test", concurrency = "2", groupId = "myGroup")
public void processMessage(String content) {
System.out.println("Message received: " + content);
}
}
This example sets up a single consumer with two members by setting concurrency = "2". Each member will cooperatively consume messages.
2) Multiple Consumer Groups Reading from the Same Topic
@Component
public class Consumer {
@KafkaListener(topics = "test", groupId = "myGroup")
public void processMessage(String content) {
System.out.println("Message received by consumer 1: " + content);
}
@KafkaListener(topics = "test", groupId = "anotherGroup")
public void processMessage2(String content) {
System.out.println("Message received by consumer 2: " + content);
}
}
Here, two consumers read from the same topic with different groupIds, causing each message to be processed by both consumers.
3) Multiple Consumer Groups Reading from Different Topics
@Component
public class Consumer {
@KafkaListener(topics = "test", groupId = "myGroup")
public void processMessage(String content) {
System.out.println("Message received by consumer 1: " + content);
}
@KafkaListener(topics = "test2", groupId = "anotherGroup")
public void processMessage2(String content) {
System.out.println("Message received by consumer 2: " + content);
}
}
This setup separates the consumer groups and topics completely, allowing distinct message processing paths.
4) Multiple Consumers in the Same Group Reading from Different Topics
@Component
public class Consumer {
@KafkaListener(topics = {"test", "test2"}, groupId = "myGroup", concurrency = "2")
public void processMessage(String content) {
System.out.println("Message received by consumer 1: " + content);
}
}
Using a single group with multiple topics, this configuration distributes the load across topics and members, which may require custom partition assignment strategies like RoundRobinAssignor for balanced distribution.
FAQ
How do I handle JSON data in Kafka messages?
Recent versions of Spring Kafka handle JSON out of the box. You only need Jackson dependencies if you're doing custom serialization or using specific JSON features.
Why aren't all partitions equally assigned?
The default PartitionAssignor is RangeAssignor, which may not distribute partitions equally. Use RoundRobinAssignor for even distribution if needed.
Can I run Kafka without Zookeeper?
As of Kafka 3.0, Zookeeper-free mode became available, though it’s mainly suited for simple, controlled environments. Most production tasks still rely on Zookeeper.
