feat/116 :: Order 주문 서비스 구현

services/auth/src/main/java/kr/magicbox/auth/adapter/in/kafka/UserEventKafkaListener.java

@@ -12,7 +12,9 @@
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.springframework.kafka.annotation.DltHandler;
 import org.springframework.kafka.annotation.KafkaListener;
+import kr.magicbox.auth.global.exception.BusinessException;
 import org.springframework.kafka.annotation.RetryableTopic;
+import org.springframework.kafka.retrytopic.DltStrategy;
 import org.springframework.stereotype.Component;
 
 @Slf4j
@@ -25,15 +27,15 @@ public class UserEventKafkaListener {
     private final AuthInboxRepository authInboxRepository;
 
     @Idempotent
-    @RetryableTopic
+    @RetryableTopic(dltStrategy = DltStrategy.FAIL_ON_ERROR, dltTopicSuffix = "-dlt", exclude = {BusinessException.class})
     @KafkaListener(topics = "outbox.event.user-withdrawn", groupId = "auth-service")
     public void handleUserWithdrawnEvent(ConsumerRecord<String, UserWithdrawnEvent> consumerRecord) {
         UserWithdrawnEvent event = consumerRecord.value();
         handleUserWithdrawnUseCase.handleUserWithdrawn(UserId.of(event.userId()));
     }
 
     @Idempotent
-    @RetryableTopic
+    @RetryableTopic(dltStrategy = DltStrategy.FAIL_ON_ERROR, dltTopicSuffix = "-dlt", exclude = {BusinessException.class})
     @KafkaListener(topics = "outbox.event.user-banned", groupId = "auth-service")
     public void handleUserBannedEvent(ConsumerRecord<String, UserBannedEvent> consumerRecord) {
         UserBannedEvent event = consumerRecord.value();

services/auth/src/main/java/kr/magicbox/auth/adapter/in/kafka/aop/IdempotentAspect.java

@@ -65,9 +65,14 @@ public Object around(ProceedingJoinPoint pjp) {
                     .build());
             try {
                 pjp.proceed();
+            } catch (Error e) {
+                throw e;
+            } catch (RuntimeException e) {
+                status.setRollbackOnly();
+                throw e;
             } catch (Throwable e) {
                 status.setRollbackOnly();
-                throw new RuntimeException(e);
+                throw new IllegalStateException(e);
             }
             inbox.markProcessed();
             return null;

services/auth/src/main/java/kr/magicbox/auth/adapter/out/communication/grpc/UserGrpcAdapter.java

@@ -16,6 +16,7 @@
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.grpc.client.GrpcChannelFactory;
 import org.springframework.stereotype.Component;
+import java.util.concurrent.TimeUnit;
 
 import java.util.concurrent.TimeUnit;
 

services/auth/src/main/java/kr/magicbox/auth/adapter/out/communication/grpc/UserStatusGrpcAdapter.java

@@ -10,6 +10,7 @@
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.grpc.client.GrpcChannelFactory;
 import org.springframework.stereotype.Component;
+import java.util.concurrent.TimeUnit;
 
 import java.util.concurrent.TimeUnit;
 

services/auth/src/main/resources/application-dev.yml

@@ -131,5 +131,9 @@ security:
     ips:
       - ${TRUSTED_IP_GATEWAY}
 
+logging:
+  level:
+    org.springframework.web: INFO
+    org.springframework.data.redis: INFO
 inbox:
   max-event-age-minutes: 5

services/auth/src/main/resources/application-prod.yml

@@ -105,7 +105,7 @@ security:
       - ${TRUSTED_IP_GATEWAY}
 
 inbox:
-  max-event-age-minutes: 60
+  max-event-age-minutes: 5
 
 logging:
   level:

services/creator/src/main/java/kr/magicbox/creator/adapter/in/kafka/UserEventKafkaListener.java

@@ -12,7 +12,9 @@
 import org.springframework.kafka.annotation.DltHandler;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.annotation.RetryableTopic;
+import org.springframework.kafka.retrytopic.DltStrategy;
 import org.springframework.stereotype.Component;
+import kr.magicbox.creator.global.exception.BusinessException;
 
 @Slf4j
 @Component
@@ -24,15 +26,15 @@ public class UserEventKafkaListener {
     private final CreatorInboxRepository creatorInboxRepository;
 
     @Idempotent
-    @RetryableTopic
+    @RetryableTopic(dltStrategy = DltStrategy.FAIL_ON_ERROR, dltTopicSuffix = "-dlt", exclude = {kr.magicbox.creator.global.exception.BusinessException.class})
     @KafkaListener(topics = "outbox.event.user-withdrawn", groupId = "creator-service")
     public void handleUserWithdrawnEvent(ConsumerRecord<String, UserWithdrawnEvent> consumerRecord) {
         log.info("[Inbox] user-withdrawn 이벤트 수신. eventId={}", consumerRecord.key());
         handleUserWithdrawnUseCase.handleUserWithdrawn(consumerRecord.value().userId());
     }
 
     @Idempotent
-    @RetryableTopic
+    @RetryableTopic(dltStrategy = DltStrategy.FAIL_ON_ERROR, dltTopicSuffix = "-dlt", exclude = {kr.magicbox.creator.global.exception.BusinessException.class})
     @KafkaListener(topics = "outbox.event.user-banned", groupId = "creator-service")
     public void handleUserBannedEvent(ConsumerRecord<String, UserBannedEvent> consumerRecord) {
         log.info("[Inbox] user-banned 이벤트 수신. eventId={}", consumerRecord.key());

services/creator/src/main/java/kr/magicbox/creator/adapter/in/kafka/aop/IdempotentAspect.java

@@ -65,9 +65,14 @@ public Object around(ProceedingJoinPoint pjp) {
                     .build());
             try {
                 pjp.proceed();
+            } catch (Error e) {
+                throw e;
+            } catch (RuntimeException e) {
+                status.setRollbackOnly();
+                throw e;
             } catch (Throwable e) {
                 status.setRollbackOnly();
-                throw new RuntimeException(e);
+                throw new IllegalStateException(e);
             }
             inbox.markProcessed();
             return null;

services/creator/src/main/resources/application-prod.yml

@@ -69,4 +69,4 @@ security:
       - ${TRUSTED_IP_GATEWAY}
 
 inbox:
-  max-event-age-minutes: 60
+  max-event-age-minutes: 5

services/general-goods/src/main/java/kr/magicbox/generalgoods/adapter/in/kafka/CreatorEventKafkaListener.java

@@ -5,12 +5,14 @@
 import kr.magicbox.generalgoods.application.dto.command.HandleCreatorRevokedCommand;
 import kr.magicbox.generalgoods.adapter.out.persistence.repository.GeneralGoodsInboxRepository;
 import kr.magicbox.generalgoods.application.port.in.HandleCreatorRevokedUseCase;
+import kr.magicbox.generalgoods.global.exception.BusinessException;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.springframework.kafka.annotation.DltHandler;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.annotation.RetryableTopic;
+import org.springframework.kafka.retrytopic.DltStrategy;
 import org.springframework.stereotype.Component;
 
 @Slf4j
@@ -20,8 +22,7 @@ public class CreatorEventKafkaListener {
     private final HandleCreatorRevokedUseCase handleCreatorRevokedUseCase;
     private final GeneralGoodsInboxRepository generalGoodsInboxRepository;
 
-    @Idempotent
-    @RetryableTopic
+    @RetryableTopic(dltStrategy = DltStrategy.FAIL_ON_ERROR, dltTopicSuffix = "-dlt", exclude = {BusinessException.class})
     @KafkaListener(topics = "outbox.event.creator-revoked", groupId = "general-goods-service")
     public void handleCreatorRevokedEvent(ConsumerRecord<String, CreatorRevokedEvent> consumerRecord) {
         CreatorRevokedEvent event = consumerRecord.value();

services/general-goods/src/main/java/kr/magicbox/generalgoods/adapter/in/kafka/aop/IdempotentAspect.java

@@ -65,9 +65,14 @@ public Object around(ProceedingJoinPoint pjp) {
                     .build());
             try {
                 pjp.proceed();
+            } catch (Error e) {
+                throw e;
+            } catch (RuntimeException e) {
+                status.setRollbackOnly();
+                throw e;
             } catch (Throwable e) {
                 status.setRollbackOnly();
-                throw new RuntimeException(e);
+                throw new IllegalStateException(e);
             }
             inbox.markProcessed();
             return null;

services/general-goods/src/main/java/kr/magicbox/generalgoods/adapter/out/communication/grpc/CreatorGrpcAdapter.java

@@ -15,6 +15,7 @@
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
+import java.util.concurrent.TimeUnit;
 
 import java.util.concurrent.TimeUnit;
 
@@ -31,7 +32,8 @@ public CreatorId getCreatorId(UserId userId) {
                 .setUserId(userId.value())
                 .build();
 
-        CreatorServiceGrpc.CreatorServiceBlockingStub stub = CreatorServiceGrpc.newBlockingStub(creatorManagedChannel)
+        ManagedChannel channel = grpcChannelFactory.createChannel(ServiceHost.CREATOR.getHostName());
+        GeneralGoodsServiceGrpc.GeneralGoodsServiceBlockingStub stub = GeneralGoodsServiceGrpc.newBlockingStub(channel)
                 .withDeadlineAfter(2, TimeUnit.SECONDS);
         GetCreatorIdByUserIdResponse response = stub.getCreatorIdByUserId(request);
 

services/general-goods/src/main/resources/application-prod.yml

@@ -47,7 +47,7 @@ security:
       - ${TRUSTED_IP_GATEWAY}
 
 inbox:
-  max-event-age-minutes: 60
+  max-event-age-minutes: 5
 
 resilience4j:
   circuitbreaker:

services/order/build.gradle

@@ -18,6 +18,9 @@ dependencies {
     implementation 'org.springframework.boot:spring-boot-starter-security'
     implementation 'org.springframework.grpc:spring-grpc-client-spring-boot-starter'
     implementation 'org.springframework.cloud:spring-cloud-starter-circuitbreaker-resilience4j'
+    implementation 'net.javacrumbs.shedlock:shedlock-spring:6.9.2'
+    implementation 'net.javacrumbs.shedlock:shedlock-provider-redisson:6.9.2'
+    implementation 'org.redisson:redisson-spring-boot-starter:3.45.1'
     runtimeOnly 'com.mysql:mysql-connector-j'
 
     testImplementation 'org.springframework.kafka:spring-kafka-test'

services/order/src/main/java/kr/magicbox/order/adapter/in/kafka/DeliveryEventKafkaListener.java

@@ -0,0 +1,42 @@
+package kr.magicbox.order.adapter.in.kafka;
+
+import kr.magicbox.order.adapter.in.kafka.annotation.Idempotent;
+import kr.magicbox.order.adapter.in.kafka.event.DeliveryCompletedEvent;
+import kr.magicbox.order.adapter.in.kafka.event.DeliveryStartedEvent;
+import kr.magicbox.order.application.port.in.HandleDeliveryCompletedUseCase;
+import kr.magicbox.order.application.port.in.HandleDeliveryStartedUseCase;
+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.annotation.RetryableTopic;
+import org.springframework.kafka.retrytopic.DltStrategy;
+import org.springframework.stereotype.Component;
+import kr.magicbox.order.global.exception.BusinessException;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class DeliveryEventKafkaListener {
+
+    private final HandleDeliveryStartedUseCase handleDeliveryStartedUseCase;
+    private final HandleDeliveryCompletedUseCase handleDeliveryCompletedUseCase;
+
+    @Idempotent
+    @RetryableTopic(dltStrategy = DltStrategy.FAIL_ON_ERROR, dltTopicSuffix = "-dlt", exclude = {kr.magicbox.order.global.exception.BusinessException.class})
+    @KafkaListener(topics = "outbox.event.delivery-started", groupId = "order-service")
+    public void handleDeliveryStarted(ConsumerRecord<String, DeliveryStartedEvent> consumerRecord) {
+        log.info("[Inbox] delivery.started 이벤트 수신. eventId={}", consumerRecord.key());
+        DeliveryStartedEvent event = consumerRecord.value();
+        handleDeliveryStartedUseCase.handleDeliveryStarted(event.orderId(), event.orderLineId());
+    }
+
+    @Idempotent
+    @RetryableTopic(dltStrategy = DltStrategy.FAIL_ON_ERROR, dltTopicSuffix = "-dlt", exclude = {kr.magicbox.order.global.exception.BusinessException.class})
+    @KafkaListener(topics = "outbox.event.delivery-completed", groupId = "order-service")
+    public void handleDeliveryCompleted(ConsumerRecord<String, DeliveryCompletedEvent> consumerRecord) {
+        log.info("[Inbox] delivery.completed 이벤트 수신. eventId={}", consumerRecord.key());
+        DeliveryCompletedEvent event = consumerRecord.value();
+        handleDeliveryCompletedUseCase.handleDeliveryCompleted(event.orderId(), event.orderLineId());
+    }
+}

services/order/src/main/java/kr/magicbox/order/adapter/in/kafka/KafkaConfiguration.java

@@ -0,0 +1,22 @@
+package kr.magicbox.order.adapter.in.kafka;
+
+import kr.magicbox.order.adapter.in.kafka.properties.InboxProperties;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.annotation.EnableKafkaRetryTopic;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+
+@EnableKafkaRetryTopic
+@Configuration
+@EnableConfigurationProperties(InboxProperties.class)
+public class KafkaConfiguration {
+
+    @Bean
+    public ThreadPoolTaskScheduler taskScheduler() {
+        ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
+        scheduler.setPoolSize(1);
+        scheduler.setThreadNamePrefix("kafka-retry-");
+        return scheduler;
+    }
+}

services/order/src/main/java/kr/magicbox/order/adapter/in/kafka/OrderStateKafkaListener.java

@@ -0,0 +1,29 @@
+package kr.magicbox.order.adapter.in.kafka;
+
+import kr.magicbox.order.adapter.in.kafka.annotation.Idempotent;
+import kr.magicbox.order.adapter.in.kafka.event.OrderPrepareEventDto;
+import kr.magicbox.order.application.port.in.HandleOrderPrepareUseCase;
+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.annotation.RetryableTopic;
+import org.springframework.kafka.retrytopic.DltStrategy;
+import org.springframework.stereotype.Component;
+import kr.magicbox.order.global.exception.BusinessException;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class OrderStateKafkaListener {
+
+    private final HandleOrderPrepareUseCase handleOrderPrepareUseCase;
+
+    @Idempotent
+    @RetryableTopic(dltStrategy = DltStrategy.FAIL_ON_ERROR, dltTopicSuffix = "-dlt", exclude = {kr.magicbox.order.global.exception.BusinessException.class})
+    @KafkaListener(topics = "outbox.event.order-prepare", groupId = "order-service")
+    public void handleOrderPrepare(ConsumerRecord<String, OrderPrepareEventDto> consumerRecord) {
+        log.info("[Inbox] order.prepare 이벤트 수신. eventId={}", consumerRecord.key());
+        handleOrderPrepareUseCase.handleOrderPrepare(consumerRecord.value().orderId());
+    }
+}

services/order/src/main/java/kr/magicbox/order/adapter/in/kafka/PaymentEventKafkaListener.java

@@ -0,0 +1,62 @@
+package kr.magicbox.order.adapter.in.kafka;
+
+import kr.magicbox.order.adapter.in.kafka.annotation.Idempotent;
+import kr.magicbox.order.adapter.in.kafka.event.PaymentCancelFailedEvent;
+import kr.magicbox.order.adapter.in.kafka.event.PaymentCancelSucceededEvent;
+import kr.magicbox.order.adapter.in.kafka.event.PaymentFailedEvent;
+import kr.magicbox.order.adapter.in.kafka.event.PaymentSucceededEvent;
+import kr.magicbox.order.application.port.in.HandlePaymentCancelFailedUseCase;
+import kr.magicbox.order.application.port.in.HandlePaymentCancelSucceededUseCase;
+import kr.magicbox.order.application.port.in.HandlePaymentFailedUseCase;
+import kr.magicbox.order.application.port.in.HandlePaymentSucceededUseCase;
+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.annotation.RetryableTopic;
+import org.springframework.kafka.retrytopic.DltStrategy;
+import org.springframework.stereotype.Component;
+import kr.magicbox.order.global.exception.BusinessException;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class PaymentEventKafkaListener {
+
+    private final HandlePaymentSucceededUseCase handlePaymentSucceededUseCase;
+    private final HandlePaymentFailedUseCase handlePaymentFailedUseCase;
+    private final HandlePaymentCancelSucceededUseCase handlePaymentCancelSucceededUseCase;
+    private final HandlePaymentCancelFailedUseCase handlePaymentCancelFailedUseCase;
+
+    @Idempotent
+    @RetryableTopic(dltStrategy = DltStrategy.FAIL_ON_ERROR, dltTopicSuffix = "-dlt", exclude = {kr.magicbox.order.global.exception.BusinessException.class})
+    @KafkaListener(topics = "outbox.event.payment-succeeded", groupId = "order-service")
+    public void handlePaymentSucceeded(ConsumerRecord<String, PaymentSucceededEvent> consumerRecord) {
+        log.info("[Inbox] payment.succeeded 이벤트 수신. eventId={}", consumerRecord.key());
+        handlePaymentSucceededUseCase.handlePaymentSucceeded(consumerRecord.value().orderId());
+    }
+
+    @Idempotent
+    @RetryableTopic(dltStrategy = DltStrategy.FAIL_ON_ERROR, dltTopicSuffix = "-dlt", exclude = {kr.magicbox.order.global.exception.BusinessException.class})
+    @KafkaListener(topics = "outbox.event.payment-failed", groupId = "order-service")
+    public void handlePaymentFailed(ConsumerRecord<String, PaymentFailedEvent> consumerRecord) {
+        log.info("[Inbox] payment.failed 이벤트 수신. eventId={}", consumerRecord.key());
+        handlePaymentFailedUseCase.handlePaymentFailed(consumerRecord.value().orderId());
+    }
+
+    @Idempotent
+    @RetryableTopic(dltStrategy = DltStrategy.FAIL_ON_ERROR, dltTopicSuffix = "-dlt", exclude = {kr.magicbox.order.global.exception.BusinessException.class})
+    @KafkaListener(topics = "outbox.event.payment-cancel-succeeded", groupId = "order-service")
+    public void handlePaymentCancelSucceeded(ConsumerRecord<String, PaymentCancelSucceededEvent> consumerRecord) {
+        log.info("[Inbox] payment.cancel.succeeded 이벤트 수신. eventId={}", consumerRecord.key());
+        handlePaymentCancelSucceededUseCase.handlePaymentCancelSucceeded(consumerRecord.value().orderId());
+    }
+
+    @Idempotent
+    @RetryableTopic(dltStrategy = DltStrategy.FAIL_ON_ERROR, dltTopicSuffix = "-dlt", exclude = {kr.magicbox.order.global.exception.BusinessException.class})
+    @KafkaListener(topics = "outbox.event.payment-cancel-failed", groupId = "order-service")
+    public void handlePaymentCancelFailed(ConsumerRecord<String, PaymentCancelFailedEvent> consumerRecord) {
+        log.info("[Inbox] payment.cancel.failed 이벤트 수신. eventId={}", consumerRecord.key());
+        handlePaymentCancelFailedUseCase.handlePaymentCancelFailed(consumerRecord.value().orderId());
+    }
+}