[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/application/payment/PaymentFacade.java

@@ -8,6 +8,9 @@
 import com.loopers.domain.payment.PaymentApproveInfo;
 import com.loopers.domain.payment.PaymentApproveResponse;
 import com.loopers.domain.payment.PaymentApproveResponse.Meta.Result;
+import com.loopers.domain.payment.PaymentEvent.PaymentPaid;
+import com.loopers.domain.payment.PaymentEvent.PaymentFailed;
+import com.loopers.domain.payment.PaymentEventPublisher;
 import com.loopers.domain.payment.PaymentGateway;
 import com.loopers.domain.payment.PaymentService;
 import com.loopers.domain.payment.TransactionStatus;
@@ -43,6 +46,7 @@ public class PaymentFacade {
     private final PaymentGateway paymentGateway;
     private final ProductCacheService productCacheService;
     private final PointService pointService;
+    private final PaymentEventPublisher paymentEventPublisher;
 
     @Transactional
     public PaymentInfo requestPaidPayment(Long userId, Long paymentId) {
@@ -86,6 +90,7 @@ private PaymentApproveResponse approvePayment(Payment payment) {
 
     private void paymentSuccess(final Payment payment) {
         payment.paid();
+        paymentEventPublisher.publish(PaymentPaid.from(payment));
         orderService.paid(payment.getOrderId());
         Order order = orderService.findById(payment.getOrderId());
         List<OrderItem> orderItems = orderService.getOrderItemsByOrderId(order.getId());
@@ -114,6 +119,7 @@ private void paymentSuccess(final Payment payment) {
     }
     private void paymentFail(final Payment payment, final String message) {
         payment.fail("PG 승인 중 오류 발생: " + message);
+        paymentEventPublisher.publish(PaymentFailed.from(payment));
         orderService.fail(payment.getOrderId());
     }
 }

apps/commerce-api/src/main/java/com/loopers/domain/payment/PaymentEvent.java

@@ -0,0 +1,14 @@
+package com.loopers.domain.payment;
+
+public class PaymentEvent {
+    public record PaymentPaid(Payment payment) {
+        public static PaymentPaid from(final Payment payment) {
+            return new PaymentPaid(payment);
+        }
+    }
+    public record PaymentFailed(Payment payment) {
+        public static PaymentFailed from(final Payment payment) {
+            return new PaymentFailed(payment);
+        }
+    }
+}

apps/commerce-api/src/main/java/com/loopers/domain/payment/PaymentEventPublisher.java

@@ -0,0 +1,8 @@
+package com.loopers.domain.payment;
+
+import com.loopers.domain.payment.PaymentEvent.PaymentPaid;
+
+public interface PaymentEventPublisher {
+    void publish(PaymentPaid paymentCreated);
+    void publish(PaymentEvent.PaymentFailed paymentFailed);
+}

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

@@ -0,0 +1,28 @@
+package com.loopers.infrastructure.payment;
+
+import com.loopers.domain.payment.PaymentEvent.PaymentPaid;
+import com.loopers.domain.payment.PaymentEvent.PaymentFailed;
+import com.loopers.domain.payment.PaymentEventPublisher;
+import lombok.RequiredArgsConstructor;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Component;
+
+@RequiredArgsConstructor
+@Component
+public class PaymentCoreEventPublisher implements PaymentEventPublisher {
+
+    private static final String paidTopic = "payment.paid";
+    private static final String failedTopic = "payment.failed";
+    private final KafkaTemplate<Object, Object> kafkaTemplate;
+
+
+    @Override
+    public void publish(final PaymentPaid paymentCreated) {
+        kafkaTemplate.send(paidTopic, paymentCreated.payment().getId(), paymentCreated);
+    }
+
+    @Override
+    public void publish(final PaymentFailed paymentFailed) {
+        kafkaTemplate.send(failedTopic, paymentFailed.payment().getId(), paymentFailed);
+    }
+}

apps/commerce-api/src/main/java/com/loopers/interfaces/event/order/OrderEventListener.java

@@ -0,0 +1,25 @@
+package com.loopers.interfaces.event.order;
+
+import com.loopers.confg.kafka.KafkaConfig;
+import java.util.List;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.stereotype.Component;
+
+@Component
+public class OrderEventListener {
+
+    @KafkaListener(
+            topics = {"payment.paid"},
+            containerFactory = KafkaConfig.BATCH_LISTENER,
+            groupId = "order"
+    )
+    public void orderPaidListener(
+            List<ConsumerRecord<Object,Object>> messages,
+            Acknowledgment acknowledgment
+    ){
+//        messages.stream()
+        acknowledgment.acknowledge();
+    }
+}

apps/commerce-api/src/main/java/com/loopers/interfaces/event/product/ProductEventListener.java

@@ -0,0 +1,25 @@
+package com.loopers.interfaces.event.product;
+
+import com.loopers.confg.kafka.KafkaConfig;
+import java.util.List;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.stereotype.Component;
+
+@Component
+public class ProductEventListener {
+
+    @KafkaListener(
+            topics = {"payment.paid"},
+            containerFactory = KafkaConfig.BATCH_LISTENER,
+            groupId = "product"
+    )
+    public void productPaidListener(
+            List<ConsumerRecord<Object,Object>> messages,
+            Acknowledgment acknowledgment
+    ){
+//        messages.stream()
+        acknowledgment.acknowledge();
+    }
+}

modules/kafka/src/main/resources/kafka.yml

@@ -7,14 +7,17 @@ spring:
         request.timeout.ms: 20000
         retry.backoff.ms: 500
         auto:
-          create.topics.enable: false
+          create.topics.enable: true
           register.schemas: false
           offset.reset: latest
         use.latest.version: true
     producer:
       key-serializer: org.apache.kafka.common.serialization.StringSerializer
       value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
       retries: 3
+      acks: all
+      properties:
+        enable.idempotence: true
     consumer:
       group-id: loopers-default-consumer
       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer