Kafka — Idempotency

In this tutorial, we’ll see how to make the consumer idempotent. Although messages coming are duplicate the consumer will not process it.

Create Topic

kafka-topics --bootstrap-server localhost:9092 --create --partitions 1 --replication-factor 1 --topic t-purchase-request
Created topic t-purchase-request.

Producer

PurchaseRequest.java

package com.course.kafka.entity;

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;

@AllArgsConstructor
@NoArgsConstructor
@Data
@Builder
public class PurchaseRequest {
   private int id;
   private String prNumber;
   private int amount;
   private String currency;
}

PurchaseRequestProducer.java

package com.course.kafka.producer;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import com.course.kafka.entity.PurchaseRequest;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;

@Service
public class PurchaseRequestProducer {
   @Autowired
   private KafkaTemplate<String, String> kafkaTemplate;

   @Autowired
   private ObjectMapper objectMapper;

   public void send(PurchaseRequest purchaseRequest) throws JsonProcessingException {
      var json = objectMapper.writeValueAsString(purchaseRequest);
      
      kafkaTemplate.send("t-purchase-request", purchaseRequest.getPrNumber(), json);
   }
}

MainApp.java

package com.course.kafka;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;

import com.course.kafka.entity.PurchaseRequest;
import com.course.kafka.producer.PurchaseRequestProducer;

@SpringBootApplication
@EnableScheduling
public class KafkaProducerApplication implements CommandLineRunner {

   public static void main(String[] args) {
      SpringApplication.run(KafkaProducerApplication.class, args);
   }
   
   @Autowired
   private PurchaseRequestProducer producer;

   @Override
   public void run(String... args) throws Exception {
      var pr1 = PurchaseRequest.builder().id(5551).prNumber("PR-First").amount(991).currency("USD").build();
      var pr2 = PurchaseRequest.builder().id(5552).prNumber("PR-Second").amount(992).currency("USD").build();
      var pr3 = PurchaseRequest.builder().id(5553).prNumber("PR-Third").amount(993).currency("USD").build();
      
      producer.send(pr1);
      producer.send(pr2);
      producer.send(pr3);
      
      producer.send(pr1);  //send the same message again!
   }
}

Consumer.java

Add following dependency

<dependency>
   <groupId>com.github.ben-manes.caffeine</groupId>
   <artifactId>caffeine</artifactId>
   <version>3.1.0</version>
</dependency>

PurchaseRequestConsumer.java

package com.course.kafka.consumer;

import java.util.Optional;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

import com.course.kafka.entity.PurchaseRequest;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonMappingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.github.benmanes.caffeine.cache.Cache;

@Service
public class PurchaseRequestConsumer {

   private static final Logger LOG = LoggerFactory.getLogger(PurchaseRequestConsumer.class);
   
   @Autowired
   private ObjectMapper objectMapper;
   
   @Autowired
   @Qualifier("cachePurchaseRequest")
   private Cache<Integer, Boolean> cache;
   
   private boolean isExistsInCache(int purchaseRequestId) {
      return Optional.ofNullable(cache.getIfPresent(purchaseRequestId)).orElse(false);
   }
   
   @KafkaListener(topics = "t-purchase-request")
   public void consume(String message) throws JsonMappingException, JsonProcessingException {
      var purchaseRequest = objectMapper.readValue(message, PurchaseRequest.class);
      
      var processed = isExistsInCache(purchaseRequest.getId());
      
      if (processed) {
         return;
      }
      
      LOG.info("---Processing {}", purchaseRequest);
      
      cache.put(purchaseRequest.getId(), true);
   }
}

CacheConfig.java

package com.course.kafka.config;

import java.time.Duration;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;

@Configuration
public class CacheConfig {

   @Bean(name = "cachePurchaseRequest")
   public Cache<Integer, Boolean> cachePurchaseRequest() {
      return Caffeine.newBuilder().expireAfterWrite(Duration.ofMinutes(2)).maximumSize(1000).build();
   }
}

From the output we conclude that although producer sends the message twice, consumer will not consume it.

Although if I run the Producer again and send the same messages, it will not process it, your topic will have 8 entries, but consumer has process only 3.