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.

--

--

Get the Medium app

A button that says 'Download on the App Store', and if clicked it will lead you to the iOS App store
A button that says 'Get it on, Google Play', and if clicked it will lead you to the Google Play store