Spring for Kafka: Producer/Consumer with Offset Management

1. The Big Picture: Producer and Consumer with Spring

Before we write any code, let's visualize the system we are about to build.

The spring-kafka project provides a powerful abstraction over the native Kafka client libraries, making it straightforward to integrate Kafka into a Spring application using familiar concepts like dependency injection and annotations.

2. Setting Up Your Kafka Environment

To develop our application, we first need a running Kafka instance. The easiest way to get one up and running locally is with Docker.

With Kafka running and a demo topic created, we're ready to build our Spring Boot application.

3. Implementing the Kafka Producer

Our first task is to create a component that can send messages to our demo topic. In Spring for Kafka, the primary tool for this is the KafkaTemplate.

Configuration

Spring Boot's auto-configuration makes setting up a producer incredibly simple. First, you'll need the spring-kafka dependency. In a Gradle project, your build.gradle would include:

implementation 'org.springframework.kafka:spring-kafka'

Next, you need to provide the producer configuration in your application.yml:

spring:
  kafka:
    producer:
      bootstrap-servers: localhost:9092
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

• bootstrap-servers: The address of your Kafka broker(s).
• key-serializer / value-serializer: The classes used to convert your message's key and value from Java objects into bytes before sending them over the network. For now, we'll just send simple strings.

Behind the scenes, Spring Boot uses these properties to auto-configure a ProducerFactory and a KafkaTemplate bean for you.

Sending Messages with KafkaTemplate

Now you can inject the auto-configured KafkaTemplate into any Spring bean (like a service or a controller) and use it to send messages.

As the resource explains, the send method returns a CompletableFuture. While you can call .get() to block and wait for the result, this is an anti-pattern that negates the high-throughput nature of Kafka. The best practice is to handle the result asynchronously.

Here's a complete example of a simple producer service:

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import java.util.concurrent.CompletableFuture;

@Service
public class KafkaMessageProducer {

    private static final Logger log = LoggerFactory.getLogger(KafkaMessageProducer.class);
    private final KafkaTemplate<String, String> kafkaTemplate;

    public KafkaMessageProducer(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void sendMessage(String topic, String message) {
        CompletableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, message);
        future.whenComplete((result, ex) -> {
            if (ex == null) {
                log.info("Sent message=[{}] with offset=[{}]", message, result.getRecordMetadata().offset());
            } else {
                log.error("Unable to send message=[{}] due to : {}", message, ex.getMessage());
            }
        });
    }
}

You can now create a simple CommandLineRunner or a REST controller to call this sendMessage method and publish messages to your demo topic.

4. Implementing the Kafka Consumer

Now for the other side: a component that listens for messages on the demo topic. This is where the @KafkaListener annotation comes in.

Configuration

First, let's configure the consumer in application.yml.

Your application.yml should look like this:

spring:
  kafka:
    consumer:
      bootstrap-servers: localhost:9092
      group-id: demo-group
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

• group-id: This is a crucial property. It identifies the consumer group this instance belongs to. All consumers with the same group-id will work together to process messages from a topic.
• auto-offset-reset: Defines what the consumer should do when there is no initial offset in Kafka for its group (e.g., when the group starts for the first time).
  • earliest: Automatically reset the offset to the earliest offset. The group will start reading from the beginning of the topic.
  • latest: Automatically reset the offset to the latest offset (the default). The group will only see new messages produced after it started.

Creating the Listener

With the configuration in place, creating the listener is as simple as annotating a method with @KafkaListener. You'll also need to add @EnableKafka to one of your @Configuration classes to enable the detection of these listeners.

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class KafkaMessageConsumer {

    private static final Logger log = LoggerFactory.getLogger(KafkaMessageConsumer.class);

    @KafkaListener(topics = "demo", groupId = "demo-group")
    public void listen(String message) {
        log.info("Received message: {}", message);
        // Business logic to process the message would go here.
    }
}

Now, if you run your application and use the producer to send a message, you should see the "Received message" log from your consumer.

5. Mastering Offset Management for Reliability

So far, our consumer works, but it's operating under a default setting that is risky for production systems: automatic offset committing. This is a critical topic for interviews.

The Problem with Auto-Commit

By default, the Kafka consumer polls for messages and commits the last received offset back to Kafka in the background periodically (controlled by auto.commit.interval.ms). This can lead to two failure scenarios:

1. Data Loss: Your consumer processes a message successfully, but then crashes before the background commit happens. When it restarts, it will re-process the same message. This is at-least-once delivery. More dangerously, if the offset is committed before you successfully process the message, and your application crashes, the message is lost forever. This is at-most-once delivery.
2. Lack of Control: You have no fine-grained control over when the commit happens. If processing a message involves multiple steps (e.g., database write, calling another API), you want to commit the offset only after all steps are complete.

To solve this, we must take control of offset management ourselves.

Implementing Manual Acknowledgment

Spring for Kafka provides an elegant solution using manual acknowledgment. This gives you precise control over when a message is considered "done".

Let's walk through the steps to implement this:

Step 1: Update application.yml

Disable auto-commit and tell Spring you want to handle acknowledgments manually.

spring:
  kafka:
    consumer:
      # ... other properties
      enable-auto-commit: false
    listener:
      ack-mode: MANUAL_IMMEDIATE

(Note: The resource uses MANUAL, but MANUAL_IMMEDIATE is often preferred as it commits the offset immediately upon calling acknowledge()).

Step 2: Update the Listener Method

Modify your listener method to accept an Acknowledgment object. Then, call its acknowledge() method after you have successfully processed the message.

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
public class KafkaMessageConsumer {

    private static final Logger log = LoggerFactory.getLogger(KafkaMessageConsumer.class);

    @KafkaListener(topics = "demo", groupId = "demo-group")
    public void listen(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) {
        try {
            log.info("Received message: {}", record.value());
            
            // Simulate business logic
            // e.g., saveToDatabase(record.value());
            
            // If we get here, processing was successful.
            log.info("Processing successful, acknowledging the message.");
            acknowledgment.acknowledge(); // This commits the offset to Kafka.

        } catch (Exception e) {
            log.error("Failed to process message: {}. Not acknowledging.", record.value(), e);
            // We do NOT call acknowledge(). When the consumer restarts or after a timeout,
            // Kafka will redeliver this message because its offset was not committed.
        }
    }
}

By wrapping your logic in a try-catch block and only calling acknowledgment.acknowledge() on success, you build a robust consumer that guarantees at-least-once processing. If your application crashes mid-process, the offset is never committed, and the message will be redelivered upon restart.

Conclusion

In this lesson, you've bridged the gap from theory to practice by building a fully functional Kafka producer and consumer with Spring Boot. More importantly, you've tackled the critical concept of offset management, a cornerstone of reliable message processing.

Key Takeaways:

• Producer: Use the KafkaTemplate provided by Spring for Kafka to send messages. Handle the CompletableFuture result asynchronously for best performance.
• Consumer: Use the @KafkaListener annotation to create a message-driven POJO.
• Configuration: Spring Boot makes configuration easy through application.yml, handling the creation of ProducerFactory, ConsumerFactory, and KafkaTemplate beans.
• Offset Management: The default auto-commit behavior is risky. For reliable, at-least-once processing, disable auto-commit and use ack-mode: MANUAL_IMMEDIATE, explicitly calling acknowledgment.acknowledge() upon successful message processing.

Preview of the Next Lesson:
So far, we've only sent simple String messages. Real-world systems exchange complex objects with structured data (e.g., User, Order). What happens when the structure of these objects needs to change? In our next lesson, you will explain the role of a schema registry and implement event versioning with Avro for schema evolution. This is a key technique for maintaining compatibility between producers and consumers in a large, evolving microservices landscape.