If you're new to Kafka, be sure to check out <a id=0 href='https://www.stackchief.com/blog/Kafka%20Architecture%20Explained'>Kafka architecture explained</a> first. If your'e new to Spring Boot Kafka consumers, be sure to check out this <a id=0 href='https://www.stackchief.com/blog/Spring%20Boot%20Kafka%20Consumer'>Spring Boot Kafka consumer example</a> first...

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-starter</artifactId>

<groupId>org.springframework.kafka</groupId>

<artifactId>spring-boot-starter-test</artifactId>

<artifactId>spring-kafka-test</artifactId>

The above dependencies include everything you need to run Spring Kafka and test the example.

spring.kafka.bootstrap-servers=localhost:9092

These are the only Kafka configuration properties we define in application.properties. While we've only specified the consumer group and broker list, you can optionally configure more properties here if you so choose...

    public void setFirstName(String firstName) {

    public void setLastName(String lastName) {

This sets up our basic User model. We need this to tell our JsonDeserializer how to deserialize the input.

import org.apache.kafka.clients.consumer.ConsumerConfig;

import org.apache.kafka.common.serialization.StringDeserializer;

import org.springframework.beans.factory.annotation.Value;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

import org.springframework.kafka.annotation.EnableKafka;

import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;

import org.springframework.kafka.config.KafkaListenerContainerFactory;

import org.springframework.kafka.core.ConsumerFactory;

import org.springframework.kafka.core.DefaultKafkaConsumerFactory;

import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;

import org.springframework.kafka.support.serializer.JsonDeserializer;

    @Value("${spring.kafka.consumer.group-id}")

    @Value("${spring.kafka.bootstrap-servers}")

public ConsumerFactory<String, User> consumerFactory() {

Map<String, Object> props = new HashMap<>();

        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);

        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);

return new DefaultKafkaConsumerFactory<>(props, new StringDeserializer(), new JsonDeserializer<>(User.class));

KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, User>>

ConcurrentKafkaListenerContainerFactory<String, User> factory =

new ConcurrentKafkaListenerContainerFactory<>();

        factory.setConsumerFactory(consumerFactory());

This is our main configuration for the Kafka client. Notice the use of the @EnableKafka to register configuration of our ConsumerFactory and ConcurrentKafkaListenerContainerFactory.

If you followed the <a id=71 href='https://www.stackchief.com/blog/Spring%20Boot%20Kafka%20Consumer'>previous example</a>, you'll notice we didn't include any of this configuration. We include these configurations in this example because we have to configure our JsonDeserializer.

By default, Kafka uses the StringDeserializer when consuming messages. Since we are consuming JSON, we must explicitly specify the Deserializer class to use for both the key and value via return new DefaultKafkaConsumerFactory<>(props, new StringDeserializer(), new JsonDeserializer<>(User.class)).

You'll also notice a reference to User. This is our model representing the object we want to read in from Kafka. Notice how we specify the User type both in defining the ConsumerFactory and the ConcurrentMessageListenerContainer.

import org.springframework.kafka.annotation.KafkaListener;

import org.springframework.stereotype.Component;

    public void processMessage(User user) {

        System.out.println("Message received by consumer 1: " + user.toString());

Spring Boot Kafka Consumer JSON example including Deserializer configuration etc.

Spring Boot Kafka Consumer JSON Example

I understand how both classes can be used to implement Kafka listeners in Spring Boot but when should you use one over the other??? Any help appreciated!

KafkaMessageListenerContainer vs ConcurrentMessageListenerContainer

KafkaMessageListenerContainer vs ConcurrentMessageListenerContainer?

1) What's an easy way to create a new topic with Spring Boot?

You can create a new topic via Spring Boot configuration by defining a NewTopic @Bean in the application context:

2) Is a KafkaAdmin bean automatically registered for you when using Spring Boot?

3) What happens when a broker is not available when starting the Spring Boot app?

The default behavior is the Spring context continues to load. You can optionally consider this behavior fatal by configuring the fatalIfBrokerNotAvailable property on the admin.

4) How is the KafkaTemplate class used in Spring Boot?

The KafkaTemplate class wraps a Kafka producer and makes it easy to send data to a topic using Spring Boot...

The message listener is an interface for handling incoming messages. There are 8 supported interfaces for message listeners. The MessageListener is called for each record returned by the underlying consumer poll() method.

6) What's the difference between KafkaMessageListenerContainer and ConcurrentMessageListenerContainer?

A KafkaMessageListenerContainer receives messages for all topics and partitions on a single thread. A ConcurrentMessageListenerContainer is a collection of KafkaMessageListenerContainer instances. It distributes message consumption across multiple threads.

7) What is the @KafkaListener annotation?

The @KafkaListener annotation can be applied to Java classes, methods, and method annotations. @KafkaListener is most commonly applied to methods, registering Kafka listener containers for those methods...

public void processMessage(String content){

    System.out.println("Message received: " + content);

In the above example, the processMessage() method is annotated with @KafkaListener. This method will be invoked when messages are read from the test topic.

Spring Boot Kafka Interview Questions, including how Spring Boot Kafka works, what different annotations do, etc

Spring Boot Kafka Interview Questions

How does Kafka consumer work?

What is Kafka consumer poll?

What is Kafka Consumer Group ID?

What is Kafka Consumer Concurrency?

If you're brand new to Kafka, be sure to check out this <a id=0 href='https://www.stackchief.com/blog/Kafka%20Architecture%20Explained'>intro to Kafka architecture</a> first.

<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">

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-starter-parent</artifactId>

<description>Demo project for Spring Boot</description>

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-starter</artifactId>

<groupId>org.springframework.kafka</groupId>

<groupId>com.fasterxml.jackson.core</groupId>

<artifactId>jackson-databind</artifactId>

<artifactId>spring-boot-starter-test</artifactId>

<artifactId>spring-kafka-test</artifactId>

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-maven-plugin</artifactId>

The pom.xml defines the dependencies we need for our Maven build. Notice how we include the dependencies needed for Spring Boot along with those needed for running and testing Kafka.

Note that the jackson-databind dependency is included to account for bugs in older versions of the Spring Kafka library. You may not need to include this depending on which version of Kafka you're using..

	public static void main(String[] args) {

		SpringApplication.run(DemoApplication.class, args);

The DemoApplication.java class runs our Spring Boot application.

Spring Boot automatically configures Kafka via @SpringBootApplication. With this annotation, Spring provides/configures a KafkaAutoConfiguration class for you.

    public void processMessage(String content){

        System.out.println("Message received: " + content);

The Consumer.java class is what actually listens for messages. Notice how we specify the topics to listen to within the @KafkaListener annotation.

By simply using the annotations provided by the Spring Boot Kafka library, we are able to easily set up a consumer for a given topic.

Thanks to auto-configuration, you only need to define a few things in application.properties to configure the consumer. Notice how our app is configured to use Kafka locally.

This example is configured to consume from a local Kafka topic.  You'll need to have Kafka installed and running locally to see messages being consumed from the Spring Boot app.

To start Kafka locally, you need to start both the Zookeeper and Kafka servers locally...

./zookeeper-server-start /usr/local/etc/kafka/zookeeper.properties

This will start the zookeeper server with the default configurations.  Please note the path to your local Kafka install may vary.

./kafka-server-start /usr/local/etc/kafka/server.properties 

This will start the Kafka server with the default configurations.  Please note the path to your local Kafka install may vary.

To test the application consumer, you'll need to produce some messages for the test topic.  To start a Kafka console producer:

./kafka-console-producer --bootstrap-server=localhost:9092 --topic test

This starts a Kafka producer for the test topic. Notice how both the bootstrap-server and topic arguments match what we've defined in our application.properties file.

Once the producer is up and running, you can send a message to the test topic like this:

If everything is working correctly, you should see the message being logged through the Spring application like this:

Spring Boot Kafka Consumer example including consumer offsets and multiple consumer examples.

Spring Boot Kafka Consumer

I am trying to use the JSON deserializer for Kafka (with Spring) and keep having issues.  Specifically it seems to throw the service into an infinite loop resulting in out of memory error...

Does anyone have issues with JSON Serializer in Kafka

Spring Boot Kafka Consumer JSON Example...

What is Kafka consumer poll?

What is Kafka Consumer Group ID?

What is Kafka Consumer Concurrency?

Spring Boot Kafka Consumer

Does anyone have issues with JSON Serializer in Kafka?

How does a consumer work in Kafka?