[volume - 8] Decoupling with Kafka

apps/commerce-api/build.gradle.kts

@@ -2,6 +2,7 @@ dependencies {
     // add-ons
     implementation(project(":modules:jpa"))
     implementation(project(":modules:redis"))
+    implementation(project(":modules:kafka"))
     implementation(project(":supports:jackson"))
     implementation(project(":supports:logging"))
     implementation(project(":supports:monitoring"))

apps/commerce-api/src/main/java/com/loopers/CommerceApiApplication.java

@@ -1,12 +1,12 @@
 package com.loopers;
 
 import jakarta.annotation.PostConstruct;
+import java.util.TimeZone;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.context.properties.ConfigurationPropertiesScan;
 import org.springframework.cloud.openfeign.EnableFeignClients;
 import org.springframework.scheduling.annotation.EnableScheduling;
-import java.util.TimeZone;
 
 @EnableFeignClients
 @EnableScheduling

apps/commerce-api/src/main/java/com/loopers/application/dataplatform/DataPlatformEventHandler.java

@@ -5,16 +5,15 @@
 import com.loopers.infrastructure.dataplatform.DataPlatformClient;
 import com.loopers.infrastructure.dataplatform.DataPlatformOrderRequest;
 import com.loopers.infrastructure.dataplatform.DataPlatformPaymentRequest;
+import java.time.ZoneOffset;
+import java.util.stream.Collectors;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.event.TransactionPhase;
 import org.springframework.transaction.event.TransactionalEventListener;
 
-import java.time.ZoneOffset;
-import java.util.stream.Collectors;
-
 /**
  * 데이터 플랫폼 이벤트 핸들러
  * - 주문/결제 데이터를 외부 데이터 플랫폼으로 비동기 전송

apps/commerce-api/src/main/java/com/loopers/application/kafka/EventKafkaProducer.java

@@ -0,0 +1,70 @@
+package com.loopers.application.kafka;
+
+import com.loopers.domain.outbox.EventOutbox;
+import java.util.concurrent.CompletableFuture;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.SendResult;
+import org.springframework.stereotype.Component;
+
+/**
+ * Kafka 이벤트 발행
+ * - Outbox 이벤트를 Kafka로 전송
+ * - Topic은 aggregateType에 따라 결정
+ * - PartitionKey는 aggregateId로 설정 (순서 보장)
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class EventKafkaProducer {
+
+    private final KafkaTemplate<Object, Object> kafkaTemplate;
+
+    @Value("${kafka.topics.catalog-events}")
+    private String catalogEventsTopic;
+
+    @Value("${kafka.topics.order-events}")
+    private String orderEventsTopic;
+
+    /**
+     * Outbox 이벤트를 Kafka로 발행
+     *
+     * @param outbox Outbox 이벤트
+     * @return CompletableFuture<SendResult>
+     */
+    public CompletableFuture<SendResult<Object, Object>> publish(EventOutbox outbox) {
+        String topic = getTopicByAggregateType(outbox.getAggregateType());
+        String partitionKey = outbox.getAggregateId();  // 순서 보장을 위한 Partition Key
+
+        log.info("Kafka 발행 시작 - topic: {}, key: {}, eventType: {}",
+            topic, partitionKey, outbox.getEventType());
+
+        return kafkaTemplate.send(topic, partitionKey, outbox.getPayload())
+            .thenApply(result -> {
+                log.info("Kafka 발행 성공 - topic: {}, partition: {}, offset: {}, eventId: {}",
+                    topic,
+                    result.getRecordMetadata().partition(),
+                    result.getRecordMetadata().offset(),
+                    outbox.getId());
+                return result;
+            })
+            .exceptionally(ex -> {
+                log.error("Kafka 발행 실패 - topic: {}, key: {}, eventId: {}, error: {}",
+                    topic, partitionKey, outbox.getId(), ex.getMessage(), ex);
+                throw new RuntimeException("Kafka 발행 실패", ex);
+            });
+    }
+
+    /**
+     * AggregateType에 따라 Topic 결정
+     */
+    private String getTopicByAggregateType(String aggregateType) {
+        return switch (aggregateType.toUpperCase()) {
+            case "ORDER", "PAYMENT" -> orderEventsTopic;
+            case "PRODUCT", "LIKE" -> catalogEventsTopic;
+            default -> throw new IllegalArgumentException("알 수 없는 AggregateType: " + aggregateType);
+        };
+    }
+}

apps/commerce-api/src/main/java/com/loopers/application/order/OrderFacade.java

@@ -3,6 +3,7 @@
 import com.loopers.application.coupon.CouponService;
 import com.loopers.application.order.OrderCommand.OrderItemRequest;
 import com.loopers.application.outbox.OutboxEventService;
+import com.loopers.application.product.ProductCacheService;
 import com.loopers.domain.coupon.UserCoupon;
 import com.loopers.domain.order.Order;
 import com.loopers.domain.order.OrderItem;
@@ -38,6 +39,7 @@ public class OrderFacade {
     private final PointService pointService;
     private final CouponService couponService;
     private final OutboxEventService outboxEventService;
+    private final ProductCacheService productCacheService;
 
     @Transactional
     public OrderInfo createOrder(String userId, OrderCommand.Create command) {
@@ -92,6 +94,12 @@ private Order createOrderWithItems(String userId, List<OrderItemRequest> orderIt
             Product product = productMap.get(request.productId());
             product.deductStock(request.quantity());
             order.addOrderItem(OrderItem.from(product, request.quantity()));
+
+            // 재고 소진 시 캐시 갱신
+            if (product.getStock() == 0) {
+                log.info("재고 소진 - 캐시 갱신 - productId: {}", product.getId());
+                productCacheService.evictCache(product.getId());
+            }
         }
 
         order.calculateTotalAmount();

apps/commerce-api/src/main/java/com/loopers/application/outbox/OutboxEventPublisher.java

@@ -5,6 +5,7 @@
 import java.util.List;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 
@@ -16,6 +17,7 @@
 @Slf4j
 @Component
 @RequiredArgsConstructor
+@ConditionalOnProperty(name = "outbox.publisher.enabled", havingValue = "true", matchIfMissing = true)
 public class OutboxEventPublisher {
 
     private final EventOutboxRepository outboxRepository;

apps/commerce-api/src/main/java/com/loopers/application/outbox/OutboxEventService.java

@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.loopers.application.kafka.EventKafkaProducer;
 import com.loopers.domain.outbox.EventOutbox;
 import com.loopers.domain.outbox.EventOutboxRepository;
 import lombok.RequiredArgsConstructor;
@@ -23,6 +24,7 @@ public class OutboxEventService {
 
     private final EventOutboxRepository outboxRepository;
     private final ApplicationEventPublisher eventPublisher;
+    private final EventKafkaProducer kafkaProducer;
     private final ObjectMapper objectMapper;
 
     /**
@@ -55,24 +57,36 @@ public void saveEvent(String aggregateType, String aggregateId,
     }
 
     /**
-     * Outbox에서 이벤트를 읽어 실제 발행
+     * Outbox에서 이벤트를 읽어 Kafka로 발행
      * - 별도 트랜잭션에서 실행
+     * - Kafka 발행 성공 시 Outbox 상태를 PUBLISHED로 변경
      */
     @Transactional(propagation = Propagation.REQUIRES_NEW)
     public void publishEvent(EventOutbox outbox) {
         try {
-            // 이벤트 타입에 따라 실제 이벤트 객체로 변환
-            Object event = deserializeEvent(outbox);
-
-            // Spring Event 발행
-            eventPublisher.publishEvent(event);
-
-            // 발행 성공 표시
-            outbox.markAsPublished();
-            outboxRepository.save(outbox);
-
-            log.info("Outbox 이벤트 발행 완료 - id: {}, type: {}",
-                outbox.getId(), outbox.getEventType());
+            // Kafka로 발행
+            kafkaProducer.publish(outbox)
+                .thenAccept(result -> {
+                    // 발행 성공 시 Outbox 상태 업데이트
+                    outbox.markAsPublished();
+                    outboxRepository.save(outbox);
+
+                    log.info("Outbox → Kafka 발행 완료 - outboxId: {}, eventType: {}, offset: {}",
+                        outbox.getId(),
+                        outbox.getEventType(),
+                        result.getRecordMetadata().offset());
+                })
+                .exceptionally(ex -> {
+                    // 발행 실패 시 Outbox 상태 업데이트
+                    outbox.markAsFailed(ex.getMessage());
+                    outboxRepository.save(outbox);
+
+                    log.error("Outbox → Kafka 발행 실패 - outboxId: {}, error: {}",
+                        outbox.getId(), ex.getMessage(), ex);
+
+                    return null;
+                })
+                .join();  // 동기 대기 (트랜잭션 내에서 완료 보장)
 
         } catch (Exception e) {
             log.error("Outbox 이벤트 발행 실패 - id: {}, error: {}",

apps/commerce-api/src/main/java/com/loopers/application/product/ProductCacheService.java

@@ -1,13 +1,13 @@
 package com.loopers.application.product;
 
+import java.time.Duration;
+import java.util.Optional;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.cache.annotation.CacheEvict;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Service;
 
-import java.time.Duration;
-import java.util.Optional;
-
 @Slf4j
 @Service
 @RequiredArgsConstructor
@@ -119,4 +119,27 @@ public CacheStats getCacheStats(Long productId) {
     }
 
     public record CacheStats(boolean viewCountCached, boolean likeCountCached) {}
+
+    /**
+     * 특정 상품의 Spring Cache를 삭제
+     *
+     * 사용 시나리오:
+     * - 재고가 소진되었을 때 호출
+     * - 다음 조회 시 DB에서 최신 데이터를 가져와 캐시에 저장
+     *
+     * @param productId 캐시를 삭제할 상품 ID
+     */
+    @CacheEvict(value = "product", key = "#productId")
+    public void evictCache(Long productId) {
+        log.info("상품 캐시 삭제 (재고 소진) - productId: {}", productId);
+    }
+
+    /**
+     * 모든 상품 캐시를 삭제
+     * - 대량 업데이트 등 특수한 경우에 사용
+     */
+    @CacheEvict(value = "product", allEntries = true)
+    public void evictAllCache() {
+        log.info("전체 상품 캐시 삭제");
+    }
 }

apps/commerce-api/src/main/java/com/loopers/application/user/UserFacade.java

@@ -2,8 +2,6 @@
 
 import com.loopers.domain.user.User;
 import com.loopers.domain.user.UserService;
-import com.loopers.support.error.CoreException;
-import com.loopers.support.error.ErrorType;
 import java.math.BigDecimal;
 import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Component;

apps/commerce-api/src/main/java/com/loopers/config/CacheConfig.java

@@ -3,6 +3,9 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializationFeature;
 import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
 import org.springframework.cache.CacheManager;
 import org.springframework.cache.annotation.EnableCaching;
 import org.springframework.context.annotation.Bean;
@@ -14,10 +17,6 @@
 import org.springframework.data.redis.serializer.RedisSerializationContext;
 import org.springframework.data.redis.serializer.StringRedisSerializer;
 
-import java.time.Duration;
-import java.util.HashMap;
-import java.util.Map;
-
 @Configuration
 @EnableCaching
 public class CacheConfig {

apps/commerce-api/src/main/java/com/loopers/infrastructure/dataplatform/DataPlatformClientConfig.java

@@ -3,9 +3,8 @@
 import feign.Logger;
 import feign.Request;
 import feign.Retryer;
-import org.springframework.context.annotation.Bean;
-
 import java.util.concurrent.TimeUnit;
+import org.springframework.context.annotation.Bean;
 
 /**
  * DataPlatform Client 설정

apps/commerce-api/src/main/java/com/loopers/infrastructure/example/ExampleRepositoryImpl.java

@@ -2,11 +2,10 @@
 
 import com.loopers.domain.example.ExampleModel;
 import com.loopers.domain.example.ExampleRepository;
+import java.util.Optional;
 import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Component;
 
-import java.util.Optional;
-
 @RequiredArgsConstructor
 @Component
 public class ExampleRepositoryImpl implements ExampleRepository {

apps/commerce-api/src/main/java/com/loopers/infrastructure/payment/PgPaymentRequest.java

@@ -1,6 +1,5 @@
 package com.loopers.infrastructure.payment;
 
-import java.math.BigDecimal;
 import lombok.Builder;
 
 @Builder

apps/commerce-api/src/main/java/com/loopers/interfaces/api/ApiControllerAdvice.java

@@ -5,6 +5,10 @@
 import com.fasterxml.jackson.databind.exc.MismatchedInputException;
 import com.loopers.support.error.CoreException;
 import com.loopers.support.error.ErrorType;
+import java.util.Arrays;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.http.ResponseEntity;
 import org.springframework.http.converter.HttpMessageNotReadableException;
@@ -15,11 +19,6 @@
 import org.springframework.web.server.ServerWebInputException;
 import org.springframework.web.servlet.resource.NoResourceFoundException;
 
-import java.util.Arrays;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-import java.util.stream.Collectors;
-
 @RestControllerAdvice
 @Slf4j
 public class ApiControllerAdvice {

apps/commerce-api/src/main/java/com/loopers/interfaces/api/payment/PaymentApi.java

@@ -4,7 +4,6 @@
 import com.loopers.domain.payment.Payment;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.tags.Tag;
-import java.math.BigDecimal;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.http.ResponseEntity;

apps/commerce-api/src/main/resources/application.yml

@@ -21,11 +21,18 @@ spring:
     import:
       - jpa.yml
       - redis.yml
+      - kafka.yml
       - logging.yml
       - monitoring.yml
       - feign.yml
       - resilience4j.yml
 
+# Kafka Topics
+kafka:
+  topics:
+    catalog-events: catalog-events
+    order-events: order-events
+
 springdoc:
   use-fqn: true
   swagger-ui:
@@ -37,6 +44,11 @@ spring:
     activate:
       on-profile: local, test
 
+# Outbox Publisher 비활성화 (테스트 환경)
+outbox:
+  publisher:
+    enabled: false
+
 ---
 spring:
   config: