fix: 재고 차감 시나리오 예외 처리

product/build.gradle

@@ -1,4 +1,4 @@
-plugins {
+    plugins {
     id 'java'
     id 'org.springframework.boot' version '3.4.4'
     id 'io.spring.dependency-management' version '1.1.7'
@@ -45,7 +45,6 @@ dependencies {
 
     testAnnotationProcessor 'org.mapstruct:mapstruct-processor:1.5.3.Final'
     testImplementation 'org.springframework.kafka:spring-kafka-test'
-    testImplementation 'org.springframework.kafka:spring-kafka-test'
     testImplementation 'org.springframework.boot:spring-boot-starter-test'
     testImplementation 'com.h2database:h2'
     testImplementation 'org.testcontainers:testcontainers:1.19.3'

product/src/main/java/com/sangyunpark/product/application/StockService.java

@@ -4,6 +4,7 @@
 import com.sangyunpark.product.constant.ErrorCode;
 import com.sangyunpark.product.exception.BusinessException;
 import com.sangyunpark.product.infrastructure.kafka.StockEventProducer;
+import com.sangyunpark.product.infrastructure.redis.OrderDuplicationRepository;
 import com.sangyunpark.product.infrastructure.redis.StockRedisRepository;
 import com.sangyunpark.product.infrastructure.repository.StockJpaRepository;
 import lombok.RequiredArgsConstructor;
@@ -20,13 +21,19 @@ public class StockService {
     private final StockJpaRepository stockJpaRepository;
     private final StockRedisRepository stockRedisRepository;
     private final StockEventProducer stockEventProducer;
+    private final OrderDuplicationRepository orderDuplicationRepository;
 
     public void decreaseStockAndPublish(final Long productId, final Long quantity, final Long orderId) {
 
-        if(!stockJpaRepository.existsById(productId)) {
+        if(!orderDuplicationRepository.saveIfAbsent(orderId, Duration.ofSeconds(30L))) {
+            log.info("이미 처리된 주문 orderId: {}", orderId);
+            return;
+        }
+
+        if(!stockRedisRepository.isExisted(productId)) {
             Long dbQuantity = stockJpaRepository.findQuantityByProductId(productId)
                     .orElseThrow(() -> new BusinessException(ErrorCode.PRODUCT_NOT_FOUND));
-            stockRedisRepository.setIfAbsentWithTTL(productId, dbQuantity, Duration.ofSeconds(30));
+            stockRedisRepository.setIfAbsentWithTTL(productId, dbQuantity, Duration.ofMinutes(3));
         }
 
         final Long remainStock =  stockRedisRepository.decrease(productId, quantity);

product/src/main/java/com/sangyunpark/product/config/KafkaConsumerConfig.java

@@ -1,19 +1,22 @@
 package com.sangyunpark.product.config;
 
 import com.sangyunpark.product.application.event.StockDeductedEvent;
+import com.sangyunpark.product.global.ExponentialBackOffWithJitter;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
 import org.springframework.kafka.listener.DefaultErrorHandler;
-import org.springframework.util.backoff.FixedBackOff;
 
+@Slf4j
 @Configuration
-public class KafkaConsumerConfig {
+public class    KafkaConsumerConfig {
 
-    final String DLT = ".DLT";
+    private final String DLT = ".DLT";
+    private final int MAX_RETRY_COUNT = 5;
 
     @Bean
     public ConcurrentKafkaListenerContainerFactory<String, StockDeductedEvent> kafkaListenerContainerFactory(
@@ -33,6 +36,14 @@ public DefaultErrorHandler errorHandler(KafkaTemplate<String, StockDeductedEvent
         DeadLetterPublishingRecoverer recover = new DeadLetterPublishingRecoverer(kafkaTemplate,
                 (record, ex) -> new org.apache.kafka.common.TopicPartition(record.topic() + DLT, record.partition()));
 
-        return new DefaultErrorHandler(recover, new FixedBackOff(1000L, 3L));
+        ExponentialBackOffWithJitter backOff = new ExponentialBackOffWithJitter(1000L, 2.0, 16000L, MAX_RETRY_COUNT);
+
+        DefaultErrorHandler errorHandler = new DefaultErrorHandler(recover, backOff);
+
+        errorHandler.setRetryListeners((record, ex, deliveryAttempt) -> {
+            log.warn("카프카 컨슈머 재시도 횟수 {} 기록 내용: {}", deliveryAttempt, record);
+        });
+
+        return errorHandler;
     }
 }

product/src/main/java/com/sangyunpark/product/global/ExponentialBackOffWithJitter.java

@@ -0,0 +1,44 @@
+package com.sangyunpark.product.global;
+
+import java.util.Random;
+import org.springframework.util.backoff.BackOff;
+import org.springframework.util.backoff.BackOffExecution;
+
+public class ExponentialBackOffWithJitter implements BackOff {
+
+    private final double multiplier;
+    private final long maxElapsedTime;
+    private final int maxAttempts;
+
+    private long currentInterval;
+    private int attemptCount;
+    private long elapsedTime;
+    private final Random random = new Random();
+
+    public ExponentialBackOffWithJitter(long initialInterval, double multiplier, long maxElapsedTime, int maxAttempts) {
+        this.maxAttempts= maxAttempts;
+        this.multiplier = multiplier;
+        this.maxElapsedTime = maxElapsedTime;
+        this.currentInterval = initialInterval;
+        this.elapsedTime = 0;
+    }
+
+    @Override
+    public BackOffExecution start() {
+        return () -> {
+            if (elapsedTime >= maxElapsedTime || attemptCount >= maxAttempts) {
+                return BackOffExecution.STOP;
+            }
+
+            long next = currentInterval;
+            long jittered = (next > 0) ? random.nextLong(next + 1) : 0;
+
+            currentInterval = Math.min((long) (currentInterval * multiplier), Long.MAX_VALUE / 2);
+
+            elapsedTime += jittered;
+            attemptCount++;
+
+            return jittered;
+        };
+    }
+}

product/src/main/java/com/sangyunpark/product/infrastructure/kafka/StockEventConsumer.java

@@ -3,8 +3,6 @@
 import com.sangyunpark.product.application.event.StockDeductedEvent;
 import com.sangyunpark.product.constant.ErrorCode;
 import com.sangyunpark.product.exception.BusinessException;
-import com.sangyunpark.product.infrastructure.redis.OrderDuplicationRepository;
-import com.sangyunpark.product.infrastructure.redis.StockRedisRepository;
 import com.sangyunpark.product.infrastructure.repository.StockJpaRepository;
 import jakarta.transaction.Transactional;
 import lombok.RequiredArgsConstructor;
@@ -21,24 +19,14 @@ public class StockEventConsumer {
     private final String GROUP_ID = "product-service";
 
     private final StockJpaRepository stockJpaRepository;
-    private final OrderDuplicationRepository orderDuplicationRepository;
-    private final StockRedisRepository stockRedisRepository;
 
     @Transactional
     @KafkaListener(topics = TOPIC, groupId = GROUP_ID)
     public void consumeStockDeductedEvent(final StockDeductedEvent event) {
 
-        if(orderDuplicationRepository.isAlreadyProcessed(event.orderId())) {
-            log.info("이미 처리된 주문 orderId: {}", event.orderId());
-            stockRedisRepository.increase(event.productId(), event.quantity());
-            return;
-        }
-
         int updatedRows = stockJpaRepository.decreaseStock(event.productId(), event.quantity());
         if(updatedRows == 0) {
             throw new BusinessException(ErrorCode.STOCK_NOT_ENOUGH);
         }
-
-        orderDuplicationRepository.saveProcessed(event.orderId());
     }
 }

product/src/main/java/com/sangyunpark/product/infrastructure/kafka/StockEventProducer.java

@@ -1,7 +1,6 @@
 package com.sangyunpark.product.infrastructure.kafka;
 
 import com.sangyunpark.product.application.event.StockDeductedEvent;
-import com.sangyunpark.product.infrastructure.redis.StockRedisRepository;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.kafka.core.KafkaTemplate;
@@ -13,15 +12,11 @@
 public class StockEventProducer {
 
     private final String TOPIC = "stock.deducted";
-
-    private final StockRedisRepository stockRedisRepository;
     private final KafkaTemplate<String, StockDeductedEvent> kafkaTemplate;
 
     public void sendStockDeductedEvent(final StockDeductedEvent stockDeductedEvent) {
-
-        kafkaTemplate.send(TOPIC, stockDeductedEvent)
+        kafkaTemplate.send(TOPIC, String.valueOf(stockDeductedEvent.productId()), stockDeductedEvent)
                 .exceptionally(ex -> {
-                    stockRedisRepository.increase(stockDeductedEvent.productId(), stockDeductedEvent.quantity());
                     log.error("Kafka 메시지 전송 실패로 재고 복구 수행");
                     return null;
                 });

product/src/main/java/com/sangyunpark/product/infrastructure/kafka/dlt/StockDLTListener.java

@@ -1,16 +1,22 @@
 package com.sangyunpark.product.infrastructure.kafka.dlt;
 
 import com.sangyunpark.product.application.event.StockDeductedEvent;
+import com.sangyunpark.product.infrastructure.redis.StockRedisRepository;
+import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.stereotype.Component;
 
 @Slf4j
 @Component
+@RequiredArgsConstructor
 public class StockDLTListener {
 
+    private final StockRedisRepository stockRedisRepository;
+
     @KafkaListener(topics = "stock.deducted.DLT", groupId = "product-service-dlt")
     public void listenStockDLT(final StockDeductedEvent event) {
-        log.error("DLT 수신 - event: {}", event);
+        log.error("Dead Letter 수신 - 재고 복구 진행. event: {}", event);
+        stockRedisRepository.increase(event.productId(), event.quantity());
     }
 }

product/src/main/java/com/sangyunpark/product/infrastructure/redis/OrderDuplicationRepository.java

@@ -13,11 +13,11 @@ public class OrderDuplicationRepository {
     private final StringRedisTemplate stringRedisTemplate;
     private final String PREFIX = "stock:deducted:";
 
-    public boolean isAlreadyProcessed(final Long orderId) {
-        return Boolean.TRUE.equals(stringRedisTemplate.hasKey(PREFIX + orderId));
+    private String getKey(final Long orderId) {
+        return PREFIX + orderId;
     }
 
-    public void saveProcessed(final Long orderId) {
-        stringRedisTemplate.opsForValue().set(PREFIX + orderId, "done", Duration.ofMinutes(10));
+    public boolean saveIfAbsent(final Long orderId, final Duration ttl) {
+        return Boolean.TRUE.equals(stringRedisTemplate.opsForValue().setIfAbsent(getKey(orderId), "done", ttl));
     }
 }

product/src/main/java/com/sangyunpark/product/infrastructure/redis/StockRedisRepository.java

@@ -49,6 +49,11 @@ public Long decrease(final Long productId, final Long amount) {
         );
     }
 
+    public boolean isExisted(final Long productId) {
+        String key = getKey(productId);
+        return Boolean.TRUE.equals(redisTemplate.hasKey(key));
+    }
+
     public void increase(final Long productId, final Long amount) {
         redisTemplate.opsForValue().increment(getKey(productId), amount);
     }

product/src/test/java/com/sangyunpark/product/application/StockServiceTest.java

@@ -4,6 +4,7 @@
 import com.sangyunpark.product.constant.ErrorCode;
 import com.sangyunpark.product.exception.BusinessException;
 import com.sangyunpark.product.infrastructure.kafka.StockEventProducer;
+import com.sangyunpark.product.infrastructure.redis.OrderDuplicationRepository;
 import com.sangyunpark.product.infrastructure.redis.StockRedisRepository;
 import com.sangyunpark.product.infrastructure.repository.StockJpaRepository;
 import org.junit.jupiter.api.DisplayName;
@@ -13,11 +14,11 @@
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 
+import java.time.Duration;
 import java.util.Optional;
 
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
@@ -37,22 +38,29 @@ class StockServiceTest {
     @Mock
     private StockEventProducer stockEventProducer;
 
+    @Mock
+    private OrderDuplicationRepository orderDuplicationRepository;
+
     private final String ERROR_CODE = "errorCode";
 
     @Test
     @DisplayName("Redis 캐시 없음 → DB에서 조회 후 캐시에 저장 + 재고 차감 성공")
     void 캐시없음_정상차감() {
         // given
         Long productId = 1L, orderId = 100L, quantity = 2L;
-        given(stockJpaRepository.existsById(productId)).willReturn(false);
+        Duration duplicationTtl = Duration.ofSeconds(30L);
+        Duration cacheTtl = Duration.ofMinutes(3L);
+
+        given(orderDuplicationRepository.saveIfAbsent(orderId, duplicationTtl)).willReturn(true);
+        given(stockRedisRepository.isExisted(productId)).willReturn(false); //
         given(stockJpaRepository.findQuantityByProductId(productId)).willReturn(Optional.of(10L));
         given(stockRedisRepository.decrease(productId, quantity)).willReturn(8L);
 
         // when
         stockService.decreaseStockAndPublish(productId, quantity, orderId);
 
         // then
-        verify(stockRedisRepository).setIfAbsentWithTTL(eq(productId), eq(10L), any());
+        verify(stockRedisRepository).setIfAbsentWithTTL(productId, 10L, cacheTtl); //
         verify(stockEventProducer).sendStockDeductedEvent(
                 new StockDeductedEvent(orderId, productId, quantity)
         );
@@ -63,14 +71,17 @@ class StockServiceTest {
     void 캐시있음_정상차감() {
         // given
         Long productId = 2L, orderId = 200L, quantity = 3L;
-        given(stockJpaRepository.existsById(productId)).willReturn(true);
+        Duration duration = Duration.ofSeconds(30L);
+
+        given(orderDuplicationRepository.saveIfAbsent(orderId, duration)).willReturn(true); // ✅ 고쳐야 할 부분
+        given(stockRedisRepository.isExisted(productId)).willReturn(true);
         given(stockRedisRepository.decrease(productId, quantity)).willReturn(7L);
 
         // when
         stockService.decreaseStockAndPublish(productId, quantity, orderId);
 
         // then
-        verify(stockRedisRepository, never()).setIfAbsentWithTTL(any(), any(), any());
+        verify(stockRedisRepository, never()).setIfAbsentWithTTL(any(), any(), any()); // 캐시가 이미 존재하므로 호출 X
         verify(stockEventProducer).sendStockDeductedEvent(
                 new StockDeductedEvent(orderId, productId, quantity)
         );
@@ -81,7 +92,9 @@ class StockServiceTest {
     void 재고부족_예외() {
         // given
         Long productId = 3L, orderId = 300L, quantity = 20L;
-        given(stockJpaRepository.existsById(productId)).willReturn(true);
+
+        given(orderDuplicationRepository.saveIfAbsent(orderId, Duration.ofSeconds(30L))).willReturn(true);
+        given(stockRedisRepository.isExisted(productId)).willReturn(true);
         given(stockRedisRepository.decrease(productId, quantity)).willReturn(-1L);
 
         // expect
@@ -95,7 +108,9 @@ class StockServiceTest {
     void 상품없음_예외() {
         // given
         Long productId = 99L, orderId = 400L, quantity = 1L;
-        given(stockJpaRepository.existsById(productId)).willReturn(false);
+
+        given(orderDuplicationRepository.saveIfAbsent(orderId, Duration.ofSeconds(30L))).willReturn(true);
+        given(stockRedisRepository.isExisted(productId)).willReturn(false);
         given(stockJpaRepository.findQuantityByProductId(productId)).willReturn(Optional.empty());
 
         // expect