Idempotency

Author: javaHelper

11-idempotency/idempotency-consumer/pom.xml

@@ -6,7 +6,7 @@
 	<parent>
 		<groupId>org.springframework.boot</groupId>
 		<artifactId>spring-boot-starter-parent</artifactId>
-		<version>2.7.5</version>
+		<version>3.3.1</version>
 		<relativePath /> <!-- lookup parent from repository -->
 	</parent>
 	<groupId>com.course.kafka</groupId>
@@ -68,4 +68,4 @@
 		</plugins>
 	</build>
 
-</project>
+</project>

11-idempotency/idempotency-consumer/src/main/java/com/course/kafka/config/CacheConfig.java

@@ -1,6 +1,7 @@
 package com.course.kafka.config;
 
 import java.time.Duration;
+import java.util.concurrent.TimeUnit;
 
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -13,11 +14,14 @@
 public class CacheConfig {
 	@Bean(name = "cachePurchaseRequest")
 	public Cache<Integer, Boolean> cachePurchaseRequest() {
-		return Caffeine.newBuilder().expireAfterWrite(Duration.ofMinutes(2)).maximumSize(1000).build();
+		//return Caffeine.newBuilder().expireAfterWrite(Duration.ofMinutes(2)).maximumSize(1000).build();
+		return Caffeine.newBuilder().expireAfterWrite(5, TimeUnit.MINUTES)
+				.maximumSize(1000).build();
 	}
 
 	@Bean(name = "cachePaymentRequest")
 	public Cache<PaymentRequestCacheKey, Boolean> cachePaymentRequest() {
-		return Caffeine.newBuilder().expireAfterWrite(Duration.ofMinutes(2)).maximumSize(1000).build();
+		return Caffeine.newBuilder().expireAfterWrite(2, TimeUnit.MINUTES)
+				.maximumSize(1000).build();
 	}
-}
+}

11-idempotency/idempotency-consumer/src/main/java/com/course/kafka/consumer/PurchaseRequestConsumer.java

@@ -33,7 +33,7 @@ private boolean isExistsInCache(int purchaseRequestId) {
 
 	@KafkaListener(topics = "t-purchase-request")
 	public void consume(String message) throws JsonMappingException, JsonProcessingException {
-		var purchaseRequest = objectMapper.readValue(message, PurchaseRequest.class);
+		PurchaseRequest purchaseRequest = objectMapper.readValue(message, PurchaseRequest.class);
 
 		var processed = isExistsInCache(purchaseRequest.getId());
 

11-idempotency/idempotency-producer/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>org.springframework.boot</groupId>
 		<artifactId>spring-boot-starter-parent</artifactId>
-		<version>2.7.5</version>
+		<version>3.3.1</version>
 		<relativePath/> <!-- lookup parent from repository -->
 	</parent>
 	<groupId>com.course.kafka</groupId>
@@ -60,4 +60,4 @@
 		</plugins>
 	</build>
 
-</project>
+</project>

11-idempotency/idempotency-producer/src/main/java/com/course/kafka/IdempotencyProducerApplication.java

@@ -24,12 +24,12 @@ public void run(String... args) throws Exception {
 		PurchaseRequest pr2 = new PurchaseRequest(5552, "PR-Second", 992, "USD");
 		PurchaseRequest pr3 = new PurchaseRequest(5553, "PR-Third", 993, "USD");
 
-		producer.send(pr1);
-		producer.send(pr2);
-		producer.send(pr3);
+		producer.sendPurchaseRequest(pr1);
+		producer.sendPurchaseRequest(pr2);
+		producer.sendPurchaseRequest(pr3);
 
 		//sending the duplicate record again
-		producer.send(pr1);
+		producer.sendPurchaseRequest(pr1);
 	}
 
-}
+}

11-idempotency/idempotency-producer/src/main/java/com/course/kafka/producer/PurchaseRequestProducer.java

@@ -17,9 +17,12 @@ public class PurchaseRequestProducer {
 	@Autowired
 	private ObjectMapper objectMapper;
 
-	public void send(PurchaseRequest purchaseRequest) throws JsonProcessingException {
-		var json = objectMapper.writeValueAsString(purchaseRequest);
-		kafkaTemplate.send("t-purchase-request", purchaseRequest.getPrNumber(), json);
+	public void sendPurchaseRequest(PurchaseRequest purchaseRequest) throws JsonProcessingException {
+		try {
+			var json = objectMapper.writeValueAsString(purchaseRequest);
+			kafkaTemplate.send("t-purchase-request", purchaseRequest.getPrNumber(), json);
+		} catch (Exception e) {
+			throw new RuntimeException(e);
+		}
 	}
-
 }

11-idempotency/idempotency-producer/src/main/resources/application.properties

@@ -1 +1 @@
-
+server.port=8085