Spring Boot Kafka Consumer JSON Example

Key Takeaways

• Use Spring Kafka dependencies to integrate Kafka with Spring Boot.
• Configure your ConsumerFactory and ListenerContainerFactory for JSON deserialization.
• Set up ErrorHandlingDeserializer to manage deserialization errors efficiently.
• Create a User model for message deserialization into objects.

Spring Boot Kafka Consumer JSON Example

pom.xml

<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>
<dependency>
  <groupId>org.springframework.kafka</groupId>
  <artifactId>spring-kafka-test</artifactId>
  <scope>test</scope>
</dependency>

These dependencies are essential for building and testing a Spring Boot application with Kafka integration, offering all necessary components for Kafka operations.

application.properties

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=myGroup

Here, only the essential Kafka properties are specified. You can add more configurations as needed for different Kafka options.

User.java

package model;
public class User {
    private String firstName;
    private String lastName;
    public String getFirstName() {
        return firstName;
    }
    public void setFirstName(String firstName) {
        this.firstName = firstName;
    }
    public String getLastName() {
        return lastName;
    }
    public void setLastName(String lastName) {
        this.lastName = lastName;
    }
}

This User model is crucial for telling the JsonDeserializer how to convert JSON into Java objects.

KafkaConfig.java

package com.example.demo.config;
import model.User;
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;
import java.util.HashMap;
import java.util.Map;

@EnableKafka
@Configuration
public class KafkaConfig {
    @Value("${spring.kafka.consumer.group-id}")
    private String groupId;

    @Value("${spring.kafka.bootstrap-servers}")
    private String brokers;

    @Bean
    public ConsumerFactory consumerFactory() {
        Map 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));
    }

    @Bean
    KafkaListenerContainerFactory> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}

This configuration class sets up Kafka consumers with @EnableKafka, preparing to handle User type messages.

Consumer.java

package com.example.demo.messaging;
import model.User;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class Consumer {
    @KafkaListener(topics = "test")
    public void processMessage(User user) {
        System.out.println("Message received by consumer 1: " + user.toString());
    }
}

This setup uses @KafkaListener to consume messages from the "test" topic, processing them directly into User objects.

Publishing JSON messages to the topic

kafka-console-producer json from file

If you're running Kafka locally, use the kafka-console-producer to publish messages from a JSON file like this:

./kafka-console-producer --bootstrap-server localhost:9092 --topic test < ~/path/to/test/file/testUser.json

The "Infinite Loop" problem with invalid messages

If a message can't be deserialized, you might encounter an "infinite loop" of exceptions:

Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition test-0 at offset 87. If needed, please seek past the record to continue consumption.

This occurs when deserialization issues aren't caught early in the Kafka consumer cycle.

How to fix the infinite loop problem with JSON and Kafka consumers

Use ErrorHandlingDeserializer to avoid these issues:

@Bean
public ConsumerFactory consumerFactory() {
    Map props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
    props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class);
    props.put(JsonDeserializer.VALUE_DEFAULT_TYPE, User.class);
    return new DefaultKafkaConsumerFactory<>(props);
}

The error handling setup above helps catch and manage deserialization errors without halting the consumer process.

FAQ

Why use JsonDeserializer with Kafka?

JsonDeserializer allows you to automatically convert JSON data into Java objects, simplifying data handling in your application.

What's the role of ErrorHandlingDeserializer?

ErrorHandlingDeserializer captures deserialization errors, preventing the consumer from crashing and allowing graceful handling of bad records.

How do I handle multiple consumer groups?

Define additional ConsumerFactory configurations, each with a distinct group id, to handle different consumer groups efficiently.