[250619] 상품도메인 - 주문오케스트레이터에서 발행한 이벤트 소비 / imageRepository 오류 수정/ 카프카설정 변경

Author: jamiewell

src/main/java/com/ecommerce/config/KafkaConfig.java

@@ -27,7 +27,7 @@ public class KafkaConfig {
     private static final String ORDER_EVENTS_TOPIC = "order-events";
     private static final String PAYMENT_EVENTS_TOPIC = "payment-events";
     private static final String DELIVERY_EVENTS_TOPIC = "delivery-events";
-    @Value("${spring.kafka.bootstrap-server}")
+    @Value("${spring.kafka.bootstrap-servers[0]}")
     private String bootStrapServers;
     @Value("${spring.kafka.consumer.group-id:ecommerce-service}")
     private String defaultGroupId;

src/main/java/com/ecommerce/product/dto/request/StockDeductRequest.java

@@ -1,14 +1,14 @@
 package com.ecommerce.product.dto.request;
 
+import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
-import lombok.NonNull;
 
-import java.math.BigDecimal;
 import java.util.UUID;
 
 @Getter
 @NoArgsConstructor
+@AllArgsConstructor
 public class StockDeductRequest {
 
     private UUID productUUID;

src/main/java/com/ecommerce/product/event/ActiveStockEventConsumer.java

@@ -1,4 +1,77 @@
 package com.ecommerce.product.event;
 
+import com.ecommerce.product.dto.request.StockDeductRequest;
+import com.ecommerce.product.productService.ProductService;
+import com.ecommerce.proto.StockDeductEvent;
+import com.ecommerce.proto.StockRestoredEvent;
+import com.google.protobuf.InvalidProtocolBufferException;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+import java.util.UUID;
+import java.util.stream.Collectors;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
 public class ActiveStockEventConsumer {
-}
+
+    private final ProductService productService;
+
+    @KafkaListener(topics = "order-event", groupId = "product-service")
+    public void consumeOrderEvents(ConsumerRecord<String, byte[]> record, Acknowledgment ack) {
+        try {
+            String key = record.key();
+            byte[] value = record.value();
+            log.info("Payment consume: order key={}", key);
+            if ("stock.deduction.request".equals(key)) {
+                operateStockDeduction(value);
+            } else if ("stock.restore.request".equals(key)) {
+                operateStockRestoration(value);
+            }
+            ack.acknowledge();
+        } catch (Exception e) {
+            log.error("Stock error : consume order event {}", e.getMessage());
+        }
+    }
+
+    private void operateStockDeduction(byte[] value) {
+        try {
+            StockDeductEvent deductEvent = StockDeductEvent.parseFrom(value);
+            UUID orderUUID = UUID.fromString(deductEvent.getOrderUuid());
+
+            List<StockDeductRequest> deductReqLit = deductEvent.getItemsList().stream()
+                    .map(item -> new StockDeductRequest(UUID.fromString(item.getProductUuid()), item.getDeductQuantity()))
+                    .collect(Collectors.toList());
+            productService.deductStocks(deductReqLit, orderUUID);
+
+        } catch (InvalidProtocolBufferException e) {
+            log.error("Parsing deduction error: {}", e.getMessage());
+        } catch (RuntimeException e) {
+            log.error("Stock deduction error: {}", e.getMessage());
+        }
+    }
+
+    private void operateStockRestoration(byte[] value) {
+        try {
+            StockRestoredEvent restoreEvent = StockRestoredEvent.parseFrom(value);
+            UUID orderUUID = UUID.fromString(restoreEvent.getOrderUuid());
+
+            List<StockDeductRequest> restoreReqList = restoreEvent.getItemsList().stream()
+                    .map(item -> new StockDeductRequest(UUID.fromString(item.getProductUuid()), item.getRestoredQuantity()))
+                    .collect(Collectors.toList());
+            productService.restoreStocks(restoreReqList, orderUUID);
+
+        } catch (InvalidProtocolBufferException e) {
+            log.error("Parsing restoration error: {}", e.getMessage());
+        } catch (RuntimeException e) {
+            log.error("Stock restoration error: {}", e.getMessage());
+        }
+    }
+
+}

src/main/java/com/ecommerce/product/productRepository/ProductImageRepository.java

@@ -8,6 +8,6 @@
 
 public interface ProductImageRepository extends JpaRepository<ProductImage, Long> {
 
-    List<ProductImage> findByProductNotDeleted(Product product);
+    List<ProductImage> findByProductAndIsDeletedFalse(Product product);
 
 }

src/main/java/com/ecommerce/product/productService/ProductService.java

@@ -84,7 +84,7 @@ public Product modifyProduct(String sellerId, String productId, ProductModifyReq
                 .orElseThrow(() -> new IllegalArgumentException("Cannot find products"));
 
         // 기존 이미지 소프트 딜리트 처리
-        List<ProductImage> images = imageRepository.findByProductNotDeleted(product);
+        List<ProductImage> images = imageRepository.findByProductAndIsDeletedFalse(product);
         images.forEach(ProductImage::softDelete);
 
         // 새 이미지들 저장
@@ -184,7 +184,7 @@ public StockDeductResult deductStock(UUID productUUID, int quantity, UUID orderU
         }
     }
 
-    // 여러 상품 재고 차감
+    // 여러 상품 재고 차감(벌크+Dirty check)
     public void deductStocks(List<StockDeductRequest> requests, UUID orderUUID) {
         List<StockDeductResult> results = new ArrayList<>();
 
@@ -196,7 +196,7 @@ public void deductStocks(List<StockDeductRequest> requests, UUID orderUUID) {
         Map<UUID, Product> productMap = products.stream()
                 .collect(Collectors.toMap(Product::getProductUUID, Function.identity()));
 
-        for(StockDeductRequest request : requests) {
+        for (StockDeductRequest request : requests) {
             Product product = productMap.get(request.getProductUUID());
 
             boolean deductSucceed = product.deductStock(request.getQuantity());
@@ -230,7 +230,7 @@ public void restoreStock(UUID orderUUID, UUID productUUID, int quantity) {
 
             eventPublisher.publishStockRestored(orderUUID, productUUID, quantity, product.getStockQuantity());
 
-        }catch (Exception e) {
+        } catch (Exception e) {
             log.error("Fail to restore stock: productUUID={}, error={}", productUUID, e.getMessage());
             throw new RuntimeException("Fail to restore stock");
         }

src/main/resources/application.yaml

@@ -7,15 +7,15 @@ spring:
     username: root
     password: User#1234
   jpa:
-    generate-ddl: false
+    generate-ddl: true
     hibernate:
-      ddl-auto: none
+      ddl-auto: update
       show_sql: true
       format_sql: true
       use_sql_comments: true
   kafka:
     bootstrap-servers:
-      - localhost:9093
+      - localhost:9092
     consumer:
       group-id: integration-test-group
       auto-offset-reset: earliest