thelinuxvault blog

Spring Boot | How to Consume JSON Messages using Apache Kafka

In modern distributed systems, message queuing is a crucial component. Apache Kafka is a popular open-source distributed streaming platform. Spring Boot provides excellent support for integrating with Kafka. In this blog, we'll explore how to consume JSON messages using Spring Boot and Apache Kafka. We'll cover the setup, configuration, and implementation details.

2026-06

Table of Contents#

  1. Prerequisites
  2. Project Setup
  3. Kafka Configuration in Spring Boot
  4. Creating a JSON Consumer
  5. Handling JSON Messages
  6. Best Practices
  7. Example Usage
  8. References

1. Prerequisites#

  • Java Development Kit (JDK): Make sure you have a compatible JDK installed (e.g., Java 8 or higher).
  • Apache Kafka: Install and run a Kafka broker. You can download it from the official Kafka website (https://kafka.apache.org/downloads).
  • Spring Boot: Familiarity with Spring Boot concepts and a basic Spring Boot project setup.

2. Project Setup#

  • Create a Spring Boot Project: You can use Spring Initializr (https://start.spring.io/) to create a new Spring Boot project. Include the following dependencies:

    • spring - boot - starter - web: For basic web functionality (optional, depending on your needs).
    • spring - kafka: The Spring Kafka integration library.
    • jackson - databind: For handling JSON serialization/deserialization.

  • Build Tool: If using Maven, your pom.xml will have entries like:

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
    </dependency>
</dependencies>

If using Gradle, your build.gradle will have:

dependencies {
    implementation 'org.springframework.boot:spring-boot-starter-web'
    implementation 'org.springframework.kafka:spring-kafka'
    implementation 'com.fasterxml.jackson.core:jackson-databind'
}

3. Kafka Configuration in Spring Boot#

  • application.properties (or application.yml):
    • Configure the Kafka bootstrap servers. For example, in application.properties:
spring.kafka.bootstrap-servers=localhost:9092
- If you want to configure consumer - specific properties like group id, auto offset reset, etc. For example:

spring.kafka.consumer.group-id=my - consumer - group
spring.kafka.consumer.auto-offset-reset=earliest
- In `application.yml`:

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: my - consumer - group
      auto-offset-reset: earliest

4. Creating a JSON Consumer#

  • Create a Java Class:
    • Let's assume we have a simple JSON object representing a User with id and name fields.
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
 
@Component
public class JsonConsumer {
 
    private final ObjectMapper objectMapper = new ObjectMapper();
 
    @KafkaListener(topics = "my - json - topic", groupId = "my - consumer - group")
    public void consume(ConsumerRecord<String, String> record) {
        try {
            User user = objectMapper.readValue(record.value(), User.class);
            System.out.println("Consumed user: " + user);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
 
class User {
    private Long id;
    private String name;
 
    // Getters and setters
    public Long getId() {
        return id;
    }
 
    public void setId(Long id) {
        this.id = id;
    }
 
    public String getName() {
        return name;
    }
 
    public void setName(String name) {
        this.name = name;
    }
 
    @Override
    public String toString() {
        return "User{" +
                "id=" + id +
                ", name='" + name + '\'' +
                '}';
    }
}

In the above code: - The @KafkaListener annotation is used to define a Kafka consumer. The topics attribute specifies the topic to listen to, and groupId is the consumer group. - The ObjectMapper from Jackson is used to deserialize the JSON string (from the Kafka message value) into a User object.

5. Handling JSON Messages#

  • Error Handling:
    • As shown in the consume method, it's important to handle exceptions during JSON deserialization. You can log the error, retry the operation (with appropriate backoff strategies if using a retry mechanism), or mark the message as dead - letter (by sending it to a dead - letter queue topic).
  • Message Acknowledgment:
    • By default, Spring Kafka auto - commits the offsets. But in some cases, you may want to control the offset commit manually. You can use the Acknowledgment object (available as a method parameter in the @KafkaListener method) to commit the offset. For example:
@KafkaListener(topics = "my - json - topic", groupId = "my - consumer - group")
public void consume(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) {
    try {
        User user = objectMapper.readValue(record.value(), User.class);
        System.out.println("Consumed user: " + user);
        acknowledgment.acknowledge();
    } catch (Exception e) {
        e.printStackTrace();
    }
}

6. Best Practices#

  • Idempotency:
    • Ensure that your consumer is idempotent. If a message is redelivered (due to a consumer failure or other reasons), the consumer should be able to handle it without causing duplicate side - effects. This can be achieved by using message keys (if applicable) and checking for already processed messages (e.g., using a database or cache to track processed message IDs).
  • Monitoring and Logging:
    • Use Spring Boot's built - in logging capabilities (e.g., Logback or Log4j) to log important events like message consumption, errors, and offset commits. Also, integrate with monitoring tools (e.g., Prometheus and Grafana) to monitor Kafka consumer metrics like lag, throughput, etc.
  • Scalability:
    • If you need to scale your consumer, use multiple consumer instances in the same consumer group. Kafka will distribute the partitions among the consumers in the group. Make sure your application is stateless (or can handle state in a distributed way) when scaling.

7. Example Usage#

  • Produce a JSON Message:
    • You can use a Kafka producer (either in a separate Spring Boot application or in a test class) to produce a JSON message to the my - json - topic. For example:
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
 
@Component
public class JsonProducer {
 
    private final KafkaTemplate<String, String> kafkaTemplate;
    private final ObjectMapper objectMapper = new ObjectMapper();
 
    @Autowired
    public JsonProducer(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }
 
    public void produce() {
        User user = new User();
        user.setId(1L);
        user.setName("John Doe");
        try {
            String json = objectMapper.writeValueAsString(user);
            kafkaTemplate.send("my - json - topic", json);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
  • Run the Application:
    • Start the Spring Boot application. The consumer will start listening to the my - json - topic. When the producer sends a message, the consumer will deserialize the JSON and process it.

8. References#