diff --git a/src/main/java/com/michelet/inventory/application/StockCommandService.java b/src/main/java/com/michelet/inventory/application/StockCommandService.java index 2f3c242..f08a076 100644 --- a/src/main/java/com/michelet/inventory/application/StockCommandService.java +++ b/src/main/java/com/michelet/inventory/application/StockCommandService.java @@ -3,7 +3,6 @@ import com.michelet.inventory.application.dto.ProductStatusChangedEvent; import com.michelet.inventory.application.dto.StockReservedEvent; import com.michelet.inventory.application.dto.StockRestoredEvent; -import com.michelet.inventory.domain.exception.ConcurrencyFailureException; import com.michelet.inventory.domain.exception.StockNotFoundException; import com.michelet.inventory.domain.model.Product; import com.michelet.inventory.domain.model.ProductOption; @@ -14,16 +13,14 @@ import com.michelet.inventory.domain.repository.StockRepository; import com.michelet.inventory.presentation.dto.ReserveStockRequest; import com.michelet.inventory.presentation.dto.RestoreStockRequest; -import jakarta.annotation.PostConstruct; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.kafka.core.KafkaTemplate; -import org.springframework.orm.ObjectOptimisticLockingFailureException; import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.support.TransactionSynchronization; import org.springframework.transaction.support.TransactionSynchronizationManager; -import org.springframework.transaction.support.TransactionTemplate; @Slf4j @Service @@ -33,12 +30,7 @@ public class StockCommandService { private final StockRepository stockRepository; private final ProductOptionRepository productOptionRepository; private final ProductRepository productRepository; - private final KafkaTemplate kafkaTemplate; - private final TransactionTemplate transactionTemplate; // 트랜잭션을 수동으로 제어하기 위해 주입 - - @Value("${inventory.retry.max-count:3}") - private int maxRetryCount; @Value("${inventory.kafka.topic.reserved:stock.reserved}") private String topicStockReserved; @@ -49,63 +41,17 @@ public class StockCommandService { @Value("${inventory.kafka.topic.status-changed:product.status-changed}") private String topicStatusChanged; - - // 재시도 횟수 설정 오류 방어 - 최소 1회 실행 보장 - @PostConstruct - public void validateConfig() { - if (this.maxRetryCount <= 0) { - log.warn("maxRetryCount [{}] 설정이 0 이하입니다. 1로 강제 조정합니다.", this.maxRetryCount); - this.maxRetryCount = Math.max(1, this.maxRetryCount); - } - } - - // @Transactional을 제거 - AOP Self-Invocation 방지 - public void reserveStockWithRetry(ReserveStockRequest request) { - int retryCount = 0; - while (retryCount < maxRetryCount) { - try { - // 1. 매 재시도마다 독립된 새로운 트랜잭션을 연다 - StockReservedEvent event = transactionTemplate.execute(status -> reserveStockInternal(request)); - - // 2. TransactionTemplate이 에러 없이 끝나면 DB 커밋이 완료된 것 - // Fire-and-forget 방지. whenComplete 콜백을 달아 전송 실패 시 인지 및 후속 처리 가능하도록 수정 - kafkaTemplate.send(topicStockReserved, request.optionId().toString(), event) - .whenComplete((result, ex) -> { - if (ex != null) { - log.error("Kafka 메시지 발행 실패 (데이터 불일치 위험)! optionId: {}", request.optionId(), ex); - // TODO 추후 재처리 로직이나 아웃박스 패턴으로 고도화 필요 - } else { - long offset = - (result != null && result.getRecordMetadata() != null) ? result.getRecordMetadata() - .offset() : -1; - log.info("Kafka 예약 메시지 발행 성공! offset: {}", offset); - } - }); - return; - - } catch (ObjectOptimisticLockingFailureException e) { - retryCount++; - log.warn("재고 차감 동시성(낙관적 락) 충돌 발생. 재시도 횟수: {}/{}", retryCount, maxRetryCount); - if (retryCount >= maxRetryCount) { - throw new ConcurrencyFailureException(); - } - backoff("차감"); - } - } - } - - // 트랜잭션 내부에서 실행될 순수 비즈니스 로직 - private StockReservedEvent reserveStockInternal(ReserveStockRequest request) { + // while문, Thread.sleep, TransactionTemplate 모두 제거 및 순수 비즈니스 로직만 남김 + @Transactional + public void reserveStock(ReserveStockRequest request) { Stock stock = stockRepository.findById(request.optionId()) .orElseThrow(StockNotFoundException::new); // 1. 도메인 로직을 통한 3중 검증 및 차감 stock.reserve(request.quantity()); - - // 2. 가독성 위해... 어차피 더티 체킹에 의해 flush 시점에 버전 체크가 발생함 stockRepository.save(stock); - // 품절 상태 자동 전이 로직 + // 2. 품절 상태 자동 전이 로직 if (stock.getTotalQuantity() == 0) { ProductOption option = productOptionRepository.findById(stock.getOptionId()) .orElseThrow(() -> new IllegalArgumentException("옵션 정보를 찾을 수 없습니다.")); @@ -121,73 +67,26 @@ private StockReservedEvent reserveStockInternal(ReserveStockRequest request) { ProductStatusChangedEvent statusEvent = new ProductStatusChangedEvent(product.getId(), product.getStatus().name()); - - if (TransactionSynchronizationManager.isSynchronizationActive()) { - TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() { - @Override - public void afterCommit() { - // 카프카 전송 실패 시 에러 핸들링 로깅 추가 - kafkaTemplate.send(topicStatusChanged, product.getId().toString(), statusEvent) - .whenComplete((result, ex) -> { - if (ex != null) { - log.error("SOLDOUT 상태 변경 Kafka 메시지 발행 실패 (데이터 불일치 위험)! productId: {}", - product.getId(), ex); - } else { - long offset = (result != null && result.getRecordMetadata() != null) - ? result.getRecordMetadata().offset() : -1; - log.info("SOLDOUT 상태 변경 Kafka 메시지 발행 성공! productId={}, offset={}", - product.getId(), offset); - } - }); - } - }); - } + publishKafkaEvent( + topicStatusChanged, + product.getId().toString(), + statusEvent, + "SOLDOUT 상태 변경" + ); } } - // 3. Kafka 이벤트 발행 - return new StockReservedEvent( + // 3. Kafka 이벤트 발행 (DB 커밋 이후에 실행되도록 공통 메서드 분리) + StockReservedEvent event = new StockReservedEvent( stock.getOptionId(), stock.getTotalQuantity(), stock.getCurrentDailyStock() ); + publishKafkaEvent(topicStockReserved, request.optionId().toString(), event, "재고차감"); } - //복구로직 - public void restoreStockWithRetry(RestoreStockRequest request) { - int retryCount = 0; - while (retryCount < maxRetryCount) { - try { - // 1. 매 재시도마다 독립된 새로운 트랜잭션을 연다 - StockRestoredEvent event = transactionTemplate.execute(status -> restoreStockInternal(request)); - - // 2. DB 커밋 완료 후 Kafka 발행 - kafkaTemplate.send(topicStockRestored, request.optionId().toString(), event) - .whenComplete((result, ex) -> { - if (ex != null) { - log.error("Kafka 복구 메시지 발행 실패 (재고 유실 위험)! optionId: {}", request.optionId(), ex); - } else { - long offset = - (result != null && result.getRecordMetadata() != null) ? result.getRecordMetadata() - .offset() : -1; - log.info("Kafka 복구 메시지 발행 성공! offset: {}", offset); - } - }); - return; - - } catch (ObjectOptimisticLockingFailureException e) { - retryCount++; - log.warn("재고 복구 동시성(낙관적 락) 충돌 발생. 재시도 횟수: {}/{}", retryCount, maxRetryCount); - if (retryCount >= maxRetryCount) { - throw new ConcurrencyFailureException(); - } - backoff("복구"); - } - } - } - - // 트랜잭션 내부에서 실행될 순수 비즈니스 로직 - private StockRestoredEvent restoreStockInternal(RestoreStockRequest request) { + @Transactional + public void restoreStock(RestoreStockRequest request) { Stock stock = stockRepository.findById(request.optionId()) .orElseThrow(StockNotFoundException::new); @@ -197,20 +96,40 @@ private StockRestoredEvent restoreStockInternal(RestoreStockRequest request) { // 2. DB 업데이트 (더티 체킹 후 flush) stockRepository.save(stock); - // 3. Kafka 이벤트 발행용 객체 리턴 - return new StockRestoredEvent( + // 2. Kafka 이벤트 발행 + StockRestoredEvent event = new StockRestoredEvent( stock.getOptionId(), stock.getTotalQuantity(), stock.getCurrentDailyStock() ); + publishKafkaEvent(topicStockRestored, request.optionId().toString(), event, "재고복구"); } - private void backoff(String operationType) { - try { - Thread.sleep(50); - } catch (InterruptedException ie) { - Thread.currentThread().interrupt(); - throw new IllegalStateException("재고 " + operationType + " 대기 중 쓰레드가 강제 종료되었습니다.", ie); + // DB 커밋 완료 후에만 Kafka가 발행되도록 보장하는 공통 메서드 + private void publishKafkaEvent(String topic, String key, Object event, String logPrefix) { + if (TransactionSynchronizationManager.isSynchronizationActive()) { + TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() { + @Override + public void afterCommit() { + sendToKafka(topic, key, event, logPrefix); + } + }); + } else { + sendToKafka(topic, key, event, logPrefix); } } + + private void sendToKafka(String topic, String key, Object event, String logPrefix) { + kafkaTemplate.send(topic, key, event) + .whenComplete((result, ex) -> { + if (ex != null) { + log.error("{} Kafka 메시지 발행 실패 (데이터 불일치 위험)! key: {}", logPrefix, key, ex); + } else { + long offset = + (result != null && result.getRecordMetadata() != null) ? result.getRecordMetadata().offset() + : -1; + log.info("{} Kafka 메시지 발행 성공! key: {}, offset: {}", logPrefix, key, offset); + } + }); + } } diff --git a/src/main/java/com/michelet/inventory/application/StockLockFacade.java b/src/main/java/com/michelet/inventory/application/StockLockFacade.java new file mode 100644 index 0000000..b7f6630 --- /dev/null +++ b/src/main/java/com/michelet/inventory/application/StockLockFacade.java @@ -0,0 +1,70 @@ +package com.michelet.inventory.application; + +import com.michelet.inventory.presentation.dto.ReserveStockRequest; +import com.michelet.inventory.presentation.dto.RestoreStockRequest; +import java.util.concurrent.TimeUnit; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.redisson.api.RLock; +import org.redisson.api.RedissonClient; +import org.springframework.stereotype.Component; + +@Slf4j +@Component +@RequiredArgsConstructor +public class StockLockFacade { + + private final RedissonClient redissonClient; + private final StockCommandService stockCommandService; // 기존 트랜잭션 서비스 + + // 1. 재고 선점 (Order 생성 시 호출) + public void reserveStockWithLock(ReserveStockRequest request) { + RLock lock = redissonClient.getLock("stock:" + request.optionId()); + + try { + // 락 획득 시도 + // 매개변수: 최대 5초 대기, 로직이 끝날 때까지 5초마다 락 만료 시간을 계속 연장 + boolean available = lock.tryLock(5, TimeUnit.SECONDS); + if (!available) { + log.error("[StockLockFacade] 재고 선점 락 획득 실패 - OptionId: {}", request.optionId()); + throw new IllegalStateException("재고 선점 처리 중 락을 획득하지 못했습니다."); + } + + // 락 획득 성공 시 실제 비즈니스 로직(트랜잭션) 실행 + stockCommandService.reserveStock(request); + + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("재고 선점 락 대기 중 인터럽트 발생"); + } finally { + if (lock != null && lock.isHeldByCurrentThread()) { + lock.unlock(); + } + } + } + + // 2. 재고 복구 (Order 실패/취소 시 보상 트랜잭션으로 호출) + public void restoreStockWithLock(RestoreStockRequest request) { + RLock lock = redissonClient.getLock("stock:" + request.optionId()); + + try { + // 복구 로직은 선점보다 더 중요하므로 대기 시간을 넉넉히(일단 10초) 줌 + boolean available = lock.tryLock(10, TimeUnit.SECONDS); + if (!available) { + log.error("[CRITICAL] 재고 복구 락 획득 실패 (수동 복구 필요) - OptionId: {}", request.optionId()); + throw new IllegalStateException("재고 복구 처리 중 락을 획득하지 못했습니다."); + } + + // 락 획득 성공 시 복구 비즈니스 로직 실행 + stockCommandService.restoreStock(request); + + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("재고 복구 락 대기 중 인터럽트 발생"); + } finally { + if (lock != null && lock.isHeldByCurrentThread()) { + lock.unlock(); + } + } + } +} diff --git a/src/main/java/com/michelet/inventory/domain/model/Stock.java b/src/main/java/com/michelet/inventory/domain/model/Stock.java index 9adad5e..3b78125 100644 --- a/src/main/java/com/michelet/inventory/domain/model/Stock.java +++ b/src/main/java/com/michelet/inventory/domain/model/Stock.java @@ -9,19 +9,17 @@ import jakarta.persistence.Entity; import jakarta.persistence.Id; import jakarta.persistence.Table; -import jakarta.persistence.Version; import java.util.UUID; import lombok.AccessLevel; import lombok.Builder; import lombok.Getter; import lombok.NoArgsConstructor; -import org.springframework.data.domain.Persistable; @Entity(name = "p_stocks") @Table(name = "p_stocks") @Getter @NoArgsConstructor(access = AccessLevel.PROTECTED) -public class Stock extends BaseEntity implements Persistable { +public class Stock extends BaseEntity { @Id private UUID optionId; // ProductOption의 ID를 PK로 사용 @@ -38,8 +36,7 @@ public class Stock extends BaseEntity implements Persistable { @Column(nullable = false) private Integer maxLimit; - @Version // 낙관적 락용 버전 - private Long version; + // Redisson 분산 락을 사용하므로 @Version(낙관적 락) 필드 삭제 @Builder(access = AccessLevel.PRIVATE) private Stock(UUID optionId, Integer totalQuantity, Integer dailyLimit, Integer maxLimit) { @@ -94,17 +91,6 @@ private void validateReserve(Integer requestQuantity) { } } - @Override - public UUID getId() { - return optionId; - } - - @Override - public boolean isNew() { - // createdAt은 DB 저장 후에 채워지므로, 객체 생성 직후엔 version(null)으로 판단하는 것이 더 정확함 - return version == null; - } - // 복구 로직 public void restore(int quantity) { if (quantity <= 0) { diff --git a/src/main/java/com/michelet/inventory/infrastructure/config/KafkaConfig.java b/src/main/java/com/michelet/inventory/infrastructure/config/KafkaConfig.java new file mode 100644 index 0000000..34b2c40 --- /dev/null +++ b/src/main/java/com/michelet/inventory/infrastructure/config/KafkaConfig.java @@ -0,0 +1,42 @@ +package com.michelet.inventory.infrastructure.config; + +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; + +@Configuration +public class KafkaConfig { + + /** + * 카프카 컨슈머 에러 핸들러 설정 - 예외 발생 시 1초 간격으로 3번 재시도 후 DLT 토픽으로 전송 + */ + @Bean + public DefaultErrorHandler errorHandler(KafkaTemplate kafkaTemplate) { + // 1. 에러가 난 메시지를 DLT(Dead Letter Topic, 예: stock.restored.DLT)로 보내는 역할 + DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate); + + // 2. 기본 재시도 정책: 1초(1000ms) 간격으로 최대 3번 재시도 + DefaultErrorHandler errorHandler = new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 3L)); + + // 3. 특정 예외(형식이 아예 틀린 경우)는 재시도해봤자 의미 없으므로 즉시 DLT로 직행 + errorHandler.addNotRetryableExceptions(IllegalArgumentException.class); + + return errorHandler; + } + + @Bean + public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory( + ConsumerFactory consumerFactory, + DefaultErrorHandler errorHandler // 위에서 만든 에러 핸들러 주입 + ) { + ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory<>(); + factory.setConsumerFactory(consumerFactory); + factory.setCommonErrorHandler(errorHandler); // 컨슈머 팩토리에 에러 핸들러 장착 + return factory; + } +} diff --git a/src/main/java/com/michelet/inventory/infrastructure/config/RedissonConfig.java b/src/main/java/com/michelet/inventory/infrastructure/config/RedissonConfig.java new file mode 100644 index 0000000..0a44f43 --- /dev/null +++ b/src/main/java/com/michelet/inventory/infrastructure/config/RedissonConfig.java @@ -0,0 +1,38 @@ +package com.michelet.inventory.infrastructure.config; + +import org.redisson.Redisson; +import org.redisson.api.RedissonClient; +import org.redisson.config.Config; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +@Configuration +public class RedissonConfig { + + @Value("${spring.data.redis.host:127.0.0.1}") + private String redisHost; + + @Value("${spring.data.redis.port:6379}") + private int redisPort; + + // 나중에 비밀번호 생길 때를 대비한 변수 추가 (기본값은 비어있음) + @Value("${spring.data.redis.password:}") + private String redisPassword; + + @Bean + public RedissonClient redissonClient() { + Config config = new Config(); + // 단일 Redis 서버 기준 설정 + String address = "redis://" + redisHost + ":" + redisPort; + + config.useSingleServer().setAddress(address); + + // 비밀번호가 세팅되어 있을 때만 적용 + if (redisPassword != null && !redisPassword.isBlank()) { + config.useSingleServer().setPassword(redisPassword); + } + + return Redisson.create(config); + } +} diff --git a/src/main/java/com/michelet/inventory/infrastructure/messaging/OrderEventConsumer.java b/src/main/java/com/michelet/inventory/infrastructure/messaging/OrderEventConsumer.java new file mode 100644 index 0000000..1760ea3 --- /dev/null +++ b/src/main/java/com/michelet/inventory/infrastructure/messaging/OrderEventConsumer.java @@ -0,0 +1,49 @@ +package com.michelet.inventory.infrastructure.messaging; + +import com.michelet.inventory.application.StockLockFacade; +import com.michelet.inventory.infrastructure.messaging.dto.StockRestoreMessage; +import com.michelet.inventory.presentation.dto.RestoreStockRequest; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.stereotype.Component; + +@Slf4j +@Component +@RequiredArgsConstructor +public class OrderEventConsumer { + + private final StockLockFacade stockLockFacade; + + @KafkaListener( + topics = "${inventory.kafka.topic.restored:stock.restored}", + groupId = "${spring.kafka.consumer.group-id:inventory-service-consumer}" + ) + public void consumeStockRestoredEvent(StockRestoreMessage payload) { + if (payload == null) { + log.error("[Kafka Consumer] 잘못된 복구 이벤트 수신: payload가 null입니다 (Tombstone 메시지일 가능성)."); + // IllegalArgumentException을 던지면 아래 catch 블록에서 잡지 않고 밖으로 던져져서 DLT로 직행함 + throw new IllegalArgumentException("재고 복구 이벤트 파싱 오류: payload는 null일 수 없습니다."); + } + + log.info("[Kafka Consumer] 재고 복구 이벤트 수신: optionId={}, quantity={}", payload.optionId(), payload.quantity()); + + try { + // 파사드를 통해 분산 락을 걸고 안전하게 재고 복구 로직 실행 + // TODO(`#26`): reservationId를 Kafka 페이로드에 포함시켜 멱등성 키로 활용 + RestoreStockRequest request = new RestoreStockRequest(payload.optionId(), payload.quantity(), null); + stockLockFacade.restoreStockWithLock(request); + + log.info("[Kafka Consumer] 재고 복구 완료! optionId: {}, quantity: {}", payload.optionId(), payload.quantity()); + + } catch (IllegalArgumentException e) { + // 데이터 검증 실패나 비즈니스 룰 위반 시 -> 즉각 DLT로 직행하도록 원본 예외 던짐 + log.error("[Kafka Consumer] 비즈니스/검증 룰 위반 에러 (DLT 직행 대상). optionId: {}", payload.optionId(), e); + throw e; + } catch (Exception e) { + // 락 대기 시간 초과 등 일시적 장애 -> 재시도(Retry)를 위해 RuntimeException으로 래핑 + log.error("[Kafka Consumer] 재고 복구 이벤트 처리 중 일시적 에러 발생 (재시도 대상). optionId: {}", payload.optionId(), e); + throw new RuntimeException("재고 복구 컨슈머 처리 실패", e); + } + } +} diff --git a/src/main/java/com/michelet/inventory/infrastructure/messaging/dto/StockRestoreMessage.java b/src/main/java/com/michelet/inventory/infrastructure/messaging/dto/StockRestoreMessage.java new file mode 100644 index 0000000..43d3767 --- /dev/null +++ b/src/main/java/com/michelet/inventory/infrastructure/messaging/dto/StockRestoreMessage.java @@ -0,0 +1,18 @@ +package com.michelet.inventory.infrastructure.messaging.dto; + +import java.util.UUID; + +// 오더 서비스가 발행한 이벤트를 읽어들이기 위한 수신 전용 DTO +public record StockRestoreMessage( + UUID optionId, + Integer quantity +) { + public StockRestoreMessage { + if (optionId == null) { + throw new IllegalArgumentException("재고 복구 이벤트 파싱 오류: optionId는 null일 수 없습니다."); + } + if (quantity == null || quantity <= 0) { + throw new IllegalArgumentException("재고 복구 이벤트 파싱 오류: quantity는 1 이상이어야 합니다."); + } + } +} diff --git a/src/main/java/com/michelet/inventory/infrastructure/repository/JpaStockRepository.java b/src/main/java/com/michelet/inventory/infrastructure/repository/JpaStockRepository.java index fcf4d60..7b974c3 100644 --- a/src/main/java/com/michelet/inventory/infrastructure/repository/JpaStockRepository.java +++ b/src/main/java/com/michelet/inventory/infrastructure/repository/JpaStockRepository.java @@ -12,8 +12,7 @@ public interface JpaStockRepository extends JpaRepository { // dailyLimit, totalQuantity 중 작은 값으로 리셋함 @Modifying(clearAutomatically = true) @Query("UPDATE p_stocks s " + - "SET s.currentDailyStock = LEAST(s.dailyLimit, s.totalQuantity), " + - " s.version = s.version + 1 " + + "SET s.currentDailyStock = LEAST(s.dailyLimit, s.totalQuantity) " + "WHERE s.currentDailyStock <> LEAST(s.dailyLimit, s.totalQuantity)") int resetDailyStock(); } diff --git a/src/main/java/com/michelet/inventory/presentation/InternalStockController.java b/src/main/java/com/michelet/inventory/presentation/InternalStockController.java index 238eb1f..bdc5c67 100644 --- a/src/main/java/com/michelet/inventory/presentation/InternalStockController.java +++ b/src/main/java/com/michelet/inventory/presentation/InternalStockController.java @@ -1,7 +1,7 @@ package com.michelet.inventory.presentation; import com.michelet.common.response.ApiResponse; -import com.michelet.inventory.application.StockCommandService; +import com.michelet.inventory.application.StockLockFacade; import com.michelet.inventory.presentation.dto.ReserveStockRequest; import com.michelet.inventory.presentation.dto.RestoreStockRequest; import jakarta.validation.Valid; @@ -17,13 +17,13 @@ @RequiredArgsConstructor public class InternalStockController { - private final StockCommandService stockCommandService; + private final StockLockFacade stockLockFacade; @PostMapping("/reserve") public ResponseEntity> reserveStock( @RequestBody @Valid ReserveStockRequest request ) { - stockCommandService.reserveStockWithRetry(request); + stockLockFacade.reserveStockWithLock(request); return ResponseEntity.ok(ApiResponse.ok(null)); } @@ -31,7 +31,7 @@ public ResponseEntity> reserveStock( public ResponseEntity> restoreStock( @RequestBody @Valid RestoreStockRequest request ) { - stockCommandService.restoreStockWithRetry(request); + stockLockFacade.restoreStockWithLock(request); return ResponseEntity.ok(ApiResponse.ok(null)); } } diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index c2fba32..df685ff 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -22,6 +22,20 @@ spring: max.in.flight.requests.per.connection: 5 retry.backoff.ms: 1000 max.block.ms: 3000 + consumer: + group-id: inventory-service-consumer + auto-offset-reset: earliest + key-deserializer: org.apache.kafka.common.serialization.StringDeserializer + # 에러 시 무한 루프 방지를 위한 ErrorHandlingDeserializer 래퍼 적용 + value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer + properties: + # 실제 역직렬화를 수행할 델리게이트 클래스 지정 + spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer + spring.json.trusted.packages: "com.michelet.inventory.infrastructure.messaging.dto" + spring.json.use.type.headers: true + # 서로 다른 패키지의 이벤트를 1:1로 매핑 + spring.json.type.mapping: "com.michelet.order.application.dto.StockRestoreEventPayload:com.michelet.inventory.infrastructure.messaging.dto.StockRestoreMessage" + server: port: 19900 diff --git a/src/test/java/com/michelet/inventory/application/StockCommandServiceTest.java b/src/test/java/com/michelet/inventory/application/StockCommandServiceTest.java index f0e4fc5..ecbe9c5 100644 --- a/src/test/java/com/michelet/inventory/application/StockCommandServiceTest.java +++ b/src/test/java/com/michelet/inventory/application/StockCommandServiceTest.java @@ -3,7 +3,6 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.BDDMockito.given; import static org.mockito.Mockito.times; @@ -12,12 +11,13 @@ import com.michelet.inventory.application.dto.StockReservedEvent; import com.michelet.inventory.application.dto.StockRestoredEvent; -import com.michelet.inventory.domain.exception.ConcurrencyFailureException; import com.michelet.inventory.domain.exception.MaxLimitExceededException; import com.michelet.inventory.domain.exception.OutOfStockException; import com.michelet.inventory.domain.exception.SoldOutException; import com.michelet.inventory.domain.exception.StockNotFoundException; import com.michelet.inventory.domain.model.Stock; +import com.michelet.inventory.domain.repository.ProductOptionRepository; +import com.michelet.inventory.domain.repository.ProductRepository; import com.michelet.inventory.domain.repository.StockRepository; import com.michelet.inventory.presentation.dto.ReserveStockRequest; import com.michelet.inventory.presentation.dto.RestoreStockRequest; @@ -34,10 +34,7 @@ import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; import org.springframework.kafka.core.KafkaTemplate; -import org.springframework.orm.ObjectOptimisticLockingFailureException; import org.springframework.test.util.ReflectionTestUtils; -import org.springframework.transaction.support.TransactionCallback; -import org.springframework.transaction.support.TransactionTemplate; @ExtendWith(MockitoExtension.class) class StockCommandServiceTest { @@ -48,11 +45,15 @@ class StockCommandServiceTest { @Mock private StockRepository stockRepository; + // SOLDOUT 자동 전이 처리를 위한 Repository Mock 추가 @Mock - private KafkaTemplate kafkaTemplate; + private ProductOptionRepository productOptionRepository; + + @Mock + private ProductRepository productRepository; @Mock - private TransactionTemplate transactionTemplate; + private KafkaTemplate kafkaTemplate; // 카프카로 전송된 이벤트를 낚아채서 내부 값을 검증하기 위한 Captor @Captor @@ -63,13 +64,9 @@ class StockCommandServiceTest { @BeforeEach void setUp() { - given(transactionTemplate.execute(any())).willAnswer(invocation -> { - TransactionCallback action = invocation.getArgument(0); - return action.doInTransaction(null); - }); - ReflectionTestUtils.setField(stockCommandService, "maxRetryCount", 3); ReflectionTestUtils.setField(stockCommandService, "topicStockReserved", "stock.reserved"); ReflectionTestUtils.setField(stockCommandService, "topicStockRestored", "stock.restored"); + ReflectionTestUtils.setField(stockCommandService, "topicStatusChanged", "product.status-changed"); } @Test @@ -81,11 +78,11 @@ void reserveStock_Success() { ReserveStockRequest request = new ReserveStockRequest(optionId, 2, null); given(stockRepository.findById(optionId)).willReturn(Optional.of(stock)); - given(kafkaTemplate.send(eq("stock.reserved"), eq(optionId.toString()), any(StockReservedEvent.class))) + given(kafkaTemplate.send(eq("stock.reserved"), eq(optionId.toString()), any())) .willReturn(CompletableFuture.completedFuture(null)); // when - stockCommandService.reserveStockWithRetry(request); + stockCommandService.reserveStock(request); // then verify(stockRepository, times(1)).save(any(Stock.class)); @@ -108,7 +105,7 @@ void reserveStock_Fail_NotEnoughDailyStock() { given(stockRepository.findById(optionId)).willReturn(Optional.of(stock)); // when & then - assertThatThrownBy(() -> stockCommandService.reserveStockWithRetry(request)) + assertThatThrownBy(() -> stockCommandService.reserveStock(request)) .isInstanceOf(OutOfStockException.class); // 예외가 터졌으므로 Kafka 전송은 절대 일어나지 않아야 함 @@ -127,7 +124,7 @@ void reserveStock_Fail_NotEnoughTotalQuantity() { given(stockRepository.findById(optionId)).willReturn(Optional.of(stock)); // when & then - assertThatThrownBy(() -> stockCommandService.reserveStockWithRetry(request)) + assertThatThrownBy(() -> stockCommandService.reserveStock(request)) .isInstanceOf(SoldOutException.class); verifyNoInteractions(kafkaTemplate); @@ -145,76 +142,14 @@ void reserveStock_Fail_MaxLimitExceeded() { given(stockRepository.findById(optionId)).willReturn(Optional.of(stock)); // when & then - assertThatThrownBy(() -> stockCommandService.reserveStockWithRetry(request)) + assertThatThrownBy(() -> stockCommandService.reserveStock(request)) .isInstanceOf(MaxLimitExceededException.class); verifyNoInteractions(kafkaTemplate); } - // N번 시도 후 성공하는 낙관적 락 재시도 테스트 - @Test - @DisplayName("재시도 성공: 낙관적 락 충돌이 발생해도 3회 이내면 재시도하여 성공한다.") - void reserveStock_Retry_Success() { - // given - UUID optionId = UUID.randomUUID(); - ReserveStockRequest request = new ReserveStockRequest(optionId, 2, null); - - // DB에서 최신 데이터를 다시 읽어오는 동작을 시뮬레이션 (매 호출마다 새로운 Stock 객체 반환) - given(stockRepository.findById(optionId)).willAnswer(invocation -> - Optional.of(Stock.create(optionId, 100, 50, 10)) - ); - - // 첫 번째, 두 번째는 예외 발생시키고 세 번째에 정상 통과 - given(stockRepository.save(any(Stock.class))) - .willThrow(new ObjectOptimisticLockingFailureException(Stock.class.getName(), optionId)) - .willThrow(new ObjectOptimisticLockingFailureException(Stock.class.getName(), optionId)) - .willReturn(null); - - given(kafkaTemplate.send(anyString(), anyString(), any())) - .willReturn(CompletableFuture.completedFuture(null)); - - // when - stockCommandService.reserveStockWithRetry(request); - - // then - findById가 매 재시도마다 호출되었는지(총 3회) 검증 - verify(stockRepository, times(3)).findById(optionId); - verify(stockRepository, times(3)).save(any(Stock.class)); - - // 카프카 페이로드 내부 필드 검증 - verify(kafkaTemplate, times(1)).send(eq("stock.reserved"), eq(optionId.toString()), eventCaptor.capture()); - StockReservedEvent capturedEvent = eventCaptor.getValue(); - assertThat(capturedEvent.optionId()).isEqualTo(optionId); - assertThat(capturedEvent.totalQuantity()).isEqualTo(98); - assertThat(capturedEvent.currentDailyStock()).isEqualTo(48); - } - - // 최대 재시도 횟수 초과 실패 테스트 @Test - @DisplayName("재시도 실패: 3회를 초과하여 낙관적 락 충돌이 발생하면 예외를 던진다.") - void reserveStock_Retry_Fail_MaxAttempts() { - // given - UUID optionId = UUID.randomUUID(); - ReserveStockRequest request = new ReserveStockRequest(optionId, 2, null); - - given(stockRepository.findById(optionId)).willAnswer(invocation -> - Optional.of(Stock.create(optionId, 100, 50, 10)) - ); - - // 항상 충돌 발생 - given(stockRepository.save(any(Stock.class))) - .willThrow(new ObjectOptimisticLockingFailureException(Stock.class.getName(), optionId)); - - // when & then: 3번 시도 후 ConcurrencyFailureException - assertThatThrownBy(() -> stockCommandService.reserveStockWithRetry(request)) - .isInstanceOf(ConcurrencyFailureException.class); - - verify(stockRepository, times(3)).findById(optionId); - verify(stockRepository, times(3)).save(any(Stock.class)); // 정확히 3번 시도됨 - verifyNoInteractions(kafkaTemplate); // 실패했으므로 카프카 메시지 발행 안 됨 - } - - @Test - @DisplayName("성공: 재고 복구 경로 정상 동작 및 Kafka 발행 테스트 (total/daily 모두 검증)") + @DisplayName("성공: 재고 복구 경로 정상 동작 및 Kafka 발행 테스트") void restoreStock_Success() { UUID optionId = UUID.randomUUID(); // 1. 초기 재고 100개, 일일 재고 50개 생성 @@ -227,11 +162,11 @@ void restoreStock_Success() { RestoreStockRequest request = new RestoreStockRequest(optionId, 2, null); given(stockRepository.findById(optionId)).willReturn(Optional.of(stock)); - given(kafkaTemplate.send(eq("stock.restored"), eq(optionId.toString()), any(StockRestoredEvent.class))) + given(kafkaTemplate.send(eq("stock.restored"), eq(optionId.toString()), any())) .willReturn(CompletableFuture.completedFuture(null)); // when - stockCommandService.restoreStockWithRetry(request); + stockCommandService.restoreStock(request); // then verify(stockRepository, times(1)).save(any(Stock.class)); @@ -246,29 +181,6 @@ void restoreStock_Success() { assertThat(event.currentDailyStock()).isEqualTo(50); } - @Test - @DisplayName("재시도 성공: 복구 시 낙관적 락 충돌이 발생해도 재시도하여 성공한다.") - void restoreStock_Retry_Success() { - UUID optionId = UUID.randomUUID(); - RestoreStockRequest request = new RestoreStockRequest(optionId, 2, null); - - given(stockRepository.findById(optionId)).willAnswer(invocation -> - Optional.of(Stock.create(optionId, 100, 50, 10)) - ); - - given(stockRepository.save(any(Stock.class))) - .willThrow(new ObjectOptimisticLockingFailureException(Stock.class.getName(), optionId)) - .willReturn(null); // 두 번째 시도 성공 - - given(kafkaTemplate.send(anyString(), anyString(), any())) - .willReturn(CompletableFuture.completedFuture(null)); - - stockCommandService.restoreStockWithRetry(request); - - verify(stockRepository, times(2)).findById(optionId); - verify(stockRepository, times(2)).save(any(Stock.class)); - } - @Test @DisplayName("실패: 재고 복구 시 존재하지 않는 옵션 예외") void restoreStock_Fail_NotFound() { @@ -277,7 +189,7 @@ void restoreStock_Fail_NotFound() { given(stockRepository.findById(optionId)).willReturn(Optional.empty()); - assertThatThrownBy(() -> stockCommandService.restoreStockWithRetry(request)) + assertThatThrownBy(() -> stockCommandService.restoreStock(request)) .isInstanceOf(StockNotFoundException.class); verifyNoInteractions(kafkaTemplate); diff --git a/src/test/java/com/michelet/inventory/application/StockConcurrencyIntegrationTest.java b/src/test/java/com/michelet/inventory/application/StockConcurrencyIntegrationTest.java index a3ee1ba..d6b4ee4 100644 --- a/src/test/java/com/michelet/inventory/application/StockConcurrencyIntegrationTest.java +++ b/src/test/java/com/michelet/inventory/application/StockConcurrencyIntegrationTest.java @@ -27,6 +27,7 @@ import org.springframework.test.context.DynamicPropertyRegistry; import org.springframework.test.context.DynamicPropertySource; import org.springframework.test.context.bean.override.mockito.MockitoBean; +import org.testcontainers.containers.GenericContainer; import org.testcontainers.containers.PostgreSQLContainer; import org.testcontainers.junit.jupiter.Container; import org.testcontainers.junit.jupiter.Testcontainers; @@ -39,16 +40,25 @@ class StockConcurrencyIntegrationTest { @Container static PostgreSQLContainer postgres = new PostgreSQLContainer<>("postgres:15-alpine"); + // Redisson 분산 락 테스트를 위해 Redis 컨테이너 추가 + @Container + static GenericContainer redis = new GenericContainer<>("redis:7.0-alpine").withExposedPorts(6379); + @DynamicPropertySource static void configureProperties(DynamicPropertyRegistry registry) { registry.add("spring.datasource.url", postgres::getJdbcUrl); registry.add("spring.datasource.username", postgres::getUsername); registry.add("spring.datasource.password", postgres::getPassword); registry.add("spring.datasource.driver-class-name", postgres::getDriverClassName); + + // Redis 설정 주입 + registry.add("spring.data.redis.host", redis::getHost); + registry.add("spring.data.redis.port", () -> redis.getMappedPort(6379)); } + // 서비스가 아닌 '락 파사드(Facade)'를 주입받아 동시성 테스트 진행 @Autowired - private StockCommandService stockCommandService; + private StockLockFacade stockLockFacade; // 조회 로직을 위한 도메인 레포지토리 @Autowired @@ -85,7 +95,6 @@ void tearDown() { @Test @DisplayName("동시성: 10명이 동시에 1개씩 재고 차감을 시도하여, 성공한 횟수만큼 정확히 재고가 줄어든다.") void reserveStock_Concurrency() throws InterruptedException { - // given: 100명이 한번에 몰리면 낙관적락 재시도 3회로 감당이 안되어 모두 실패하므로, 10명으로 조정하여 검증 int threadCount = 10; ExecutorService executorService = Executors.newFixedThreadPool(10); // 모든 쓰레드가 동시에 출발하도록 제어하는 startLatch @@ -103,11 +112,10 @@ void reserveStock_Concurrency() throws InterruptedException { try { // 모든 작업 쓰레드는 여기서 대기하며 신호를 기다림 startLatch.await(); - - stockCommandService.reserveStockWithRetry(request); + // 락 획득 메서드 호출 + stockLockFacade.reserveStockWithLock(request); successCount.incrementAndGet(); // 에러 없이 통과하면 성공 횟수 1 증가 } catch (Exception e) { - // 낙관적 락 재시도 3회 모두 실패한 스레드들은 예외를 던지며 이곳으로 옴 System.out.println("차감 실패: " + e.getMessage()); } finally { doneLatch.countDown(); @@ -117,15 +125,13 @@ void reserveStock_Concurrency() throws InterruptedException { // 대기 중이던 쓰레드를 동시에 실행 시작 (출발 신호) startLatch.countDown(); - // 스레드 자원 누수를 막기 위해 무조건 shutdown 로직이 실행되도록 try-finally 적용 try { boolean completed = doneLatch.await(30, TimeUnit.SECONDS); - assertThat(completed).withFailMessage("쓰레드 작업이 지정된 시간 내에 완료되지 않았습니다.").isTrue(); + assertThat(completed).withFailMessage("쓰레드 작업 타임아웃").isTrue(); } finally { executorService.shutdown(); if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) { executorService.shutdownNow(); - assertThat(false).withFailMessage("ExecutorService가 정상적으로 종료되지 않았습니다.").isTrue(); } }