From 643def4062ceeb32e4619b6e6af269f503dcc907 Mon Sep 17 00:00:00 2001 From: ji-circle Date: Tue, 12 May 2026 17:57:03 +0900 Subject: [PATCH 1/6] =?UTF-8?q?refactor:=20redisson=20=EB=B6=84=EC=82=B0?= =?UTF-8?q?=20lock=EC=9C=BC=EB=A1=9C=20=EB=B3=80=EA=B2=BD=20-=20Stock.java?= =?UTF-8?q?=EC=97=90=EC=84=9C=20@Version=20=EC=82=AD=EC=A0=9C=20-=20StockC?= =?UTF-8?q?ommandService.java=EC=97=90=EC=84=9C=20while,=20sleep,=20Transa?= =?UTF-8?q?ctionTemplate=20=EB=8B=A4=20=EA=B1=B7=EC=96=B4=EB=82=B4?= =?UTF-8?q?=EA=B3=A0=20=EC=88=9C=EC=88=98=20@Transactional=20=EB=B9=84?= =?UTF-8?q?=EC=A6=88=EB=8B=88=EC=8A=A4=20=EB=A1=9C=EC=A7=81=EB=A7=8C=20?= =?UTF-8?q?=EB=82=A8=EA=B9=80=20-=20StockLockFacade.java=EC=97=90=EC=84=9C?= =?UTF-8?q?=20tryLock=EA=B3=BC=20finally=20unlock=EC=9D=84=20=EC=9D=B4?= =?UTF-8?q?=EC=9A=A9=ED=95=B4=20DB=20=ED=8A=B8=EB=9E=9C=EC=9E=AD=EC=85=98?= =?UTF-8?q?=20=EB=B0=96=EC=97=90=EC=84=9C=20=EB=9D=BD=EC=9D=84=20=EC=A0=9C?= =?UTF-8?q?=EC=96=B4=ED=95=98=EB=8F=84=EB=A1=9D=20=EC=88=98=EC=A0=95=20-?= =?UTF-8?q?=20=EC=BB=A8=ED=8A=B8=EB=A1=A4=EB=9F=AC=EA=B0=80=20Facade?= =?UTF-8?q?=EB=A5=BC=20=ED=98=B8=EC=B6=9C=ED=95=98=EB=8F=84=EB=A1=9D=20?= =?UTF-8?q?=EB=B3=80=EA=B2=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../application/StockCommandService.java | 169 +++++------------- .../application/StockLockFacade.java | 70 ++++++++ .../inventory/domain/model/Stock.java | 18 +- .../infrastructure/config/RedissonConfig.java | 27 +++ .../repository/JpaStockRepository.java | 3 +- .../presentation/InternalStockController.java | 8 +- .../application/StockCommandServiceTest.java | 124 ++----------- .../StockConcurrencyIntegrationTest.java | 22 ++- 8 files changed, 180 insertions(+), 261 deletions(-) create mode 100644 src/main/java/com/michelet/inventory/application/StockLockFacade.java create mode 100644 src/main/java/com/michelet/inventory/infrastructure/config/RedissonConfig.java diff --git a/src/main/java/com/michelet/inventory/application/StockCommandService.java b/src/main/java/com/michelet/inventory/application/StockCommandService.java index 2f3c242..6a89619 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..24a5234 --- /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초 대기, 락 획득 시 3초간 점유 후 자동 해제(데드락 방지) + boolean available = lock.tryLock(5, 3, 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, 5, 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/RedissonConfig.java b/src/main/java/com/michelet/inventory/infrastructure/config/RedissonConfig.java new file mode 100644 index 0000000..aee632c --- /dev/null +++ b/src/main/java/com/michelet/inventory/infrastructure/config/RedissonConfig.java @@ -0,0 +1,27 @@ +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; + + @Bean + public RedissonClient redissonClient() { + Config config = new Config(); + // 단일 Redis 서버 기준 설정 + config.useSingleServer() + .setAddress("redis://" + redisHost + ":" + redisPort); + return Redisson.create(config); + } +} 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/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(); } } From 492146f4e4bfa2d2d5f94bc62271d8e2fb31eb00 Mon Sep 17 00:00:00 2001 From: ji-circle Date: Tue, 12 May 2026 18:26:10 +0900 Subject: [PATCH 2/6] =?UTF-8?q?fix:=20leaseTime=EC=9D=84=20=EB=B9=BC?= =?UTF-8?q?=EC=84=9C=20Redisson=20Watchdog=EC=9D=B4=20=EB=A1=9C=EC=A7=81?= =?UTF-8?q?=EC=9D=B4=20=EB=81=9D=EB=82=A0=20=EB=95=8C=EA=B9=8C=EC=A7=80=20?= =?UTF-8?q?n=EC=B4=88=EB=A7=88=EB=8B=A4=20=EA=B3=84=EC=86=8D=20=EC=88=98?= =?UTF-8?q?=EB=AA=85=EC=9D=84=20=EC=97=B0=EC=9E=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../com/michelet/inventory/application/StockLockFacade.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/main/java/com/michelet/inventory/application/StockLockFacade.java b/src/main/java/com/michelet/inventory/application/StockLockFacade.java index 24a5234..b7f6630 100644 --- a/src/main/java/com/michelet/inventory/application/StockLockFacade.java +++ b/src/main/java/com/michelet/inventory/application/StockLockFacade.java @@ -23,8 +23,8 @@ public void reserveStockWithLock(ReserveStockRequest request) { try { // 락 획득 시도 - // 매개변수: 최대 5초 대기, 락 획득 시 3초간 점유 후 자동 해제(데드락 방지) - boolean available = lock.tryLock(5, 3, TimeUnit.SECONDS); + // 매개변수: 최대 5초 대기, 로직이 끝날 때까지 5초마다 락 만료 시간을 계속 연장 + boolean available = lock.tryLock(5, TimeUnit.SECONDS); if (!available) { log.error("[StockLockFacade] 재고 선점 락 획득 실패 - OptionId: {}", request.optionId()); throw new IllegalStateException("재고 선점 처리 중 락을 획득하지 못했습니다."); @@ -49,7 +49,7 @@ public void restoreStockWithLock(RestoreStockRequest request) { try { // 복구 로직은 선점보다 더 중요하므로 대기 시간을 넉넉히(일단 10초) 줌 - boolean available = lock.tryLock(10, 5, TimeUnit.SECONDS); + boolean available = lock.tryLock(10, TimeUnit.SECONDS); if (!available) { log.error("[CRITICAL] 재고 복구 락 획득 실패 (수동 복구 필요) - OptionId: {}", request.optionId()); throw new IllegalStateException("재고 복구 처리 중 락을 획득하지 못했습니다."); From 079e80f26f9fc54b4ba370e237b4d1c668a3fa33 Mon Sep 17 00:00:00 2001 From: ji-circle Date: Tue, 12 May 2026 19:03:05 +0900 Subject: [PATCH 3/6] =?UTF-8?q?fix:=20redissonconfig=20pw=EB=8C=80?= =?UTF-8?q?=EB=B9=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../infrastructure/config/RedissonConfig.java | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/src/main/java/com/michelet/inventory/infrastructure/config/RedissonConfig.java b/src/main/java/com/michelet/inventory/infrastructure/config/RedissonConfig.java index aee632c..0a44f43 100644 --- a/src/main/java/com/michelet/inventory/infrastructure/config/RedissonConfig.java +++ b/src/main/java/com/michelet/inventory/infrastructure/config/RedissonConfig.java @@ -16,12 +16,23 @@ public class RedissonConfig { @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 서버 기준 설정 - config.useSingleServer() - .setAddress("redis://" + redisHost + ":" + redisPort); + String address = "redis://" + redisHost + ":" + redisPort; + + config.useSingleServer().setAddress(address); + + // 비밀번호가 세팅되어 있을 때만 적용 + if (redisPassword != null && !redisPassword.isBlank()) { + config.useSingleServer().setPassword(redisPassword); + } + return Redisson.create(config); } } From 8963c23c637ec40880bbb6f60d5d1ca28d4143c0 Mon Sep 17 00:00:00 2001 From: ji-circle Date: Tue, 12 May 2026 20:08:47 +0900 Subject: [PATCH 4/6] =?UTF-8?q?feat:=20=EC=A3=BC=EB=AC=B8=EC=9C=BC?= =?UTF-8?q?=EB=A1=9C=EB=B6=80=ED=84=B0=EC=9D=98=20=EC=B9=B4=ED=94=84?= =?UTF-8?q?=EC=B9=B4=20=EC=88=98=EC=8B=A0=20=EC=97=B0=EA=B2=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../application/StockCommandService.java | 4 +- .../infrastructure/config/KafkaConfig.java | 42 +++++++++++++++++++ .../messaging/OrderEventConsumer.java | 42 +++++++++++++++++++ .../messaging/dto/StockRestoreMessage.java | 10 +++++ src/main/resources/application.yml | 14 +++++++ 5 files changed, 110 insertions(+), 2 deletions(-) create mode 100644 src/main/java/com/michelet/inventory/infrastructure/config/KafkaConfig.java create mode 100644 src/main/java/com/michelet/inventory/infrastructure/messaging/OrderEventConsumer.java create mode 100644 src/main/java/com/michelet/inventory/infrastructure/messaging/dto/StockRestoreMessage.java diff --git a/src/main/java/com/michelet/inventory/application/StockCommandService.java b/src/main/java/com/michelet/inventory/application/StockCommandService.java index 6a89619..f08a076 100644 --- a/src/main/java/com/michelet/inventory/application/StockCommandService.java +++ b/src/main/java/com/michelet/inventory/application/StockCommandService.java @@ -82,7 +82,7 @@ public void reserveStock(ReserveStockRequest request) { stock.getTotalQuantity(), stock.getCurrentDailyStock() ); - publishKafkaEvent(topicStockReserved, request.optionId().toString(), event, "예약"); + publishKafkaEvent(topicStockReserved, request.optionId().toString(), event, "재고차감"); } @Transactional @@ -102,7 +102,7 @@ public void restoreStock(RestoreStockRequest request) { stock.getTotalQuantity(), stock.getCurrentDailyStock() ); - publishKafkaEvent(topicStockRestored, request.optionId().toString(), event, "복구"); + publishKafkaEvent(topicStockRestored, request.optionId().toString(), event, "재고복구"); } // DB 커밋 완료 후에만 Kafka가 발행되도록 보장하는 공통 메서드 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/messaging/OrderEventConsumer.java b/src/main/java/com/michelet/inventory/infrastructure/messaging/OrderEventConsumer.java new file mode 100644 index 0000000..b427bac --- /dev/null +++ b/src/main/java/com/michelet/inventory/infrastructure/messaging/OrderEventConsumer.java @@ -0,0 +1,42 @@ +package com.michelet.inventory.infrastructure.messaging; + +import com.fasterxml.jackson.databind.ObjectMapper; +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; + private final ObjectMapper objectMapper; + + @KafkaListener( + topics = "${inventory.kafka.topic.restored:stock.restored}", + groupId = "${spring.kafka.consumer.group-id:inventory-service-consumer}" + ) + public void consumeStockRestoredEvent(String message) { + log.info("[Kafka Consumer] 재고 복구 이벤트 수신: {}", message); + + try { + // 1. DTO를 사용하여 JSON 파싱 + StockRestoreMessage payload = objectMapper.readValue(message, StockRestoreMessage.class); + + // 2. 파사드를 통해 분산 락을 걸고 안전하게 재고 복구 로직 실행 + RestoreStockRequest request = new RestoreStockRequest(payload.optionId(), payload.quantity(), null); + stockLockFacade.restoreStockWithLock(request); + + log.info("[Kafka Consumer] 재고 복구 완료! optionId: {}, quantity: {}", payload.optionId(), payload.quantity()); + + } catch (Exception e) { + log.error("[Kafka Consumer] 재고 복구 이벤트 처리 중 에러 발생! 메시지: {}", message, 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..b180267 --- /dev/null +++ b/src/main/java/com/michelet/inventory/infrastructure/messaging/dto/StockRestoreMessage.java @@ -0,0 +1,10 @@ +package com.michelet.inventory.infrastructure.messaging.dto; + +import java.util.UUID; + +// 오더 서비스가 발행한 이벤트를 읽어들이기 위한 수신 전용 DTO +public record StockRestoreMessage( + UUID optionId, + Integer quantity +) { +} 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 From 8dbf7163d80faafa33e73db2885e304863746794 Mon Sep 17 00:00:00 2001 From: ji-circle Date: Tue, 12 May 2026 23:53:19 +0900 Subject: [PATCH 5/6] =?UTF-8?q?fix:=20=EC=BD=94=EB=93=9C=EB=9E=98=EB=B9=97?= =?UTF-8?q?=20-=20=EC=88=98=EB=8F=99=20JSON=20=ED=8C=8C=EC=8B=B1=20?= =?UTF-8?q?=EC=A0=9C=EA=B1=B0=20=EB=B0=8F=20Spring=20Kafka=20=EC=9E=90?= =?UTF-8?q?=EB=8F=99=20=EC=97=AD=EC=A7=81=EB=A0=AC=ED=99=94=20=ED=99=9C?= =?UTF-8?q?=EC=9A=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../messaging/OrderEventConsumer.java | 19 ++++++++++--------- .../messaging/dto/StockRestoreMessage.java | 8 ++++++++ 2 files changed, 18 insertions(+), 9 deletions(-) diff --git a/src/main/java/com/michelet/inventory/infrastructure/messaging/OrderEventConsumer.java b/src/main/java/com/michelet/inventory/infrastructure/messaging/OrderEventConsumer.java index b427bac..fd3df07 100644 --- a/src/main/java/com/michelet/inventory/infrastructure/messaging/OrderEventConsumer.java +++ b/src/main/java/com/michelet/inventory/infrastructure/messaging/OrderEventConsumer.java @@ -1,6 +1,5 @@ package com.michelet.inventory.infrastructure.messaging; -import com.fasterxml.jackson.databind.ObjectMapper; import com.michelet.inventory.application.StockLockFacade; import com.michelet.inventory.infrastructure.messaging.dto.StockRestoreMessage; import com.michelet.inventory.presentation.dto.RestoreStockRequest; @@ -15,27 +14,29 @@ public class OrderEventConsumer { private final StockLockFacade stockLockFacade; - private final ObjectMapper objectMapper; @KafkaListener( topics = "${inventory.kafka.topic.restored:stock.restored}", groupId = "${spring.kafka.consumer.group-id:inventory-service-consumer}" ) - public void consumeStockRestoredEvent(String message) { - log.info("[Kafka Consumer] 재고 복구 이벤트 수신: {}", message); + public void consumeStockRestoredEvent(StockRestoreMessage payload) { + log.info("[Kafka Consumer] 재고 복구 이벤트 수신: optionId={}, quantity={}", payload.optionId(), payload.quantity()); try { - // 1. DTO를 사용하여 JSON 파싱 - StockRestoreMessage payload = objectMapper.readValue(message, StockRestoreMessage.class); - - // 2. 파사드를 통해 분산 락을 걸고 안전하게 재고 복구 로직 실행 + // 파사드를 통해 분산 락을 걸고 안전하게 재고 복구 로직 실행 + // 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) { - log.error("[Kafka Consumer] 재고 복구 이벤트 처리 중 에러 발생! 메시지: {}", message, e); + // DB 락 타임아웃 등 일시적인 장애는 DefaultErrorHandler가 재시도할 수 있도록 래핑하여 던짐 + 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 index b180267..43d3767 100644 --- a/src/main/java/com/michelet/inventory/infrastructure/messaging/dto/StockRestoreMessage.java +++ b/src/main/java/com/michelet/inventory/infrastructure/messaging/dto/StockRestoreMessage.java @@ -7,4 +7,12 @@ 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 이상이어야 합니다."); + } + } } From 831ba2753d7f1b8821918119b0254d9773061d60 Mon Sep 17 00:00:00 2001 From: ji-circle Date: Wed, 13 May 2026 00:54:45 +0900 Subject: [PATCH 6/6] =?UTF-8?q?fix:=20=EC=BD=94=EB=93=9C=EB=9E=98=EB=B9=97?= =?UTF-8?q?=20-=20payload=20null=20=EA=B0=80=EB=93=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../infrastructure/messaging/OrderEventConsumer.java | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/src/main/java/com/michelet/inventory/infrastructure/messaging/OrderEventConsumer.java b/src/main/java/com/michelet/inventory/infrastructure/messaging/OrderEventConsumer.java index fd3df07..1760ea3 100644 --- a/src/main/java/com/michelet/inventory/infrastructure/messaging/OrderEventConsumer.java +++ b/src/main/java/com/michelet/inventory/infrastructure/messaging/OrderEventConsumer.java @@ -20,6 +20,12 @@ public class OrderEventConsumer { 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 { @@ -31,11 +37,11 @@ public void consumeStockRestoredEvent(StockRestoreMessage payload) { log.info("[Kafka Consumer] 재고 복구 완료! optionId: {}, quantity: {}", payload.optionId(), payload.quantity()); } catch (IllegalArgumentException e) { - // 데이터 형식이 잘못된 경우 무의미한 재시도를 막고 즉시 DLT로 보내기 위해 원래 에러를 던짐 + // 데이터 검증 실패나 비즈니스 룰 위반 시 -> 즉각 DLT로 직행하도록 원본 예외 던짐 log.error("[Kafka Consumer] 비즈니스/검증 룰 위반 에러 (DLT 직행 대상). optionId: {}", payload.optionId(), e); throw e; } catch (Exception e) { - // DB 락 타임아웃 등 일시적인 장애는 DefaultErrorHandler가 재시도할 수 있도록 래핑하여 던짐 + // 락 대기 시간 초과 등 일시적 장애 -> 재시도(Retry)를 위해 RuntimeException으로 래핑 log.error("[Kafka Consumer] 재고 복구 이벤트 처리 중 일시적 에러 발생 (재시도 대상). optionId: {}", payload.optionId(), e); throw new RuntimeException("재고 복구 컨슈머 처리 실패", e); }