feat: Outbox 이벤트 실패 처리 및 재발행 로직 추가

- `OutboxEvent`에 `OutboxStatus` 및 `retryCount` 필드 추가로 상태 관리 지원.
- 실패 이벤트를 처리하기 위한 `markFailed` 메서드 구현.
- `OutboxRelay` 컴포넌트 추가로 실패 이벤트 재발행 스케줄링 및 처리 로직 구현.
- 관련 서비스 및 핸들러에 실패 상태 업데이트 로직 통합 (`markFailed` 호출 추가).

Author: yeonjiyeon

apps/commerce-api/src/main/java/com/loopers/application/event/DeadLetterQueueProcessor.java

@@ -4,15 +4,13 @@
 import com.loopers.domain.event.DomainEvent;
 import com.loopers.domain.event.FailedEvent;
 import com.loopers.infrastructure.event.FailedEventRepository;
+import java.util.List;
 import lombok.RequiredArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.annotation.Transactional;
 
-import java.util.List;
-
 @Component
 @RequiredArgsConstructor
 public class DeadLetterQueueProcessor {

apps/commerce-api/src/main/java/com/loopers/application/event/OutboxRelay.java

@@ -0,0 +1,49 @@
+package com.loopers.application.event;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.loopers.domain.event.OutboxEvent;
+import com.loopers.domain.event.OutboxStatus;
+import com.loopers.infrastructure.event.OutboxRepository;
+import java.util.List;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+
+@Component
+@RequiredArgsConstructor
+@Slf4j
+public class OutboxRelay {
+
+  private final OutboxRepository outboxRepository;
+  private final ApplicationEventPublisher eventPublisher;
+  private final ObjectMapper objectMapper;
+
+  @Scheduled(fixedDelay = 60000)
+  @Transactional
+  public void resendPendingEvents() {
+    List<OutboxEvent> pendingEvents = outboxRepository.findTop10ByStatusInAndRetryCountLessThanOrderByCreatedAtAsc(
+        List.of(OutboxStatus.INIT, OutboxStatus.FAILED), 5
+    );
+
+    if (pendingEvents.isEmpty()) return;
+
+    for (OutboxEvent outbox : pendingEvents) {
+      try {
+        Class<?> eventClass = Class.forName(outbox.getEventType());
+        Object originalEvent = objectMapper.readValue(outbox.getPayload(), eventClass);
+
+        eventPublisher.publishEvent(originalEvent);
+
+        log.info("[Outbox Relay] 이벤트 재발행 성공: {}", outbox.getEventId());
+        outbox.markPublished();
+
+      } catch (Exception e) {
+        log.error("[Outbox Relay] 재발행 실패: {}", outbox.getEventId());
+        outbox.markFailed();
+      }
+    }
+  }
+}

apps/commerce-api/src/main/java/com/loopers/application/like/event/LikeEventOutboxHandler.java

@@ -2,16 +2,15 @@
 
 import com.loopers.domain.event.OutboxService;
 import lombok.RequiredArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.event.TransactionPhase;
 import org.springframework.transaction.event.TransactionalEventListener;
 
 @Component
 @RequiredArgsConstructor
-@Slf4j
 public class LikeEventOutboxHandler {
+
   private final KafkaTemplate<Object, Object> kafkaTemplate;
   private final OutboxService outboxService;
 
@@ -24,7 +23,7 @@ public void handle(LikeCreatedEvent event) {
           if (ex == null) {
             outboxService.markPublished(event.eventId());
           } else {
-            log.error("카프카 전송 실패: {}", ex.getMessage());
+            outboxService.markFailed(event.eventId());
           }
         });
   }

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

@@ -19,6 +19,8 @@ public void handle(OrderCreatedEvent event) {
         .whenComplete((result, ex) -> {
           if (ex == null) {
             outboxService.markPublished(event.eventId());
+          } else {
+            outboxService.markFailed(event.eventId());
           }
         });
   }

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

@@ -25,6 +25,8 @@ public void handle(ProductViewEvent event) {
         .whenComplete((result, ex) -> {
           if (ex == null) {
             outboxService.markPublished(event.eventId());
+          } else {
+            outboxService.markFailed(event.eventId());
           }
         });
   }

apps/commerce-api/src/main/java/com/loopers/domain/event/OutboxEvent.java

@@ -2,6 +2,8 @@
 
 import jakarta.persistence.Column;
 import jakarta.persistence.Entity;
+import jakarta.persistence.EnumType;
+import jakarta.persistence.Enumerated;
 import jakarta.persistence.GeneratedValue;
 import jakarta.persistence.GenerationType;
 import jakarta.persistence.Id;
@@ -27,7 +29,10 @@ public class OutboxEvent {
   @Column(columnDefinition = "TEXT")
   private String payload;
 
-  private boolean published = false;
+  @Enumerated(EnumType.STRING)
+  private OutboxStatus status = OutboxStatus.INIT;
+
+  private int retryCount = 0;
   private LocalDateTime createdAt = LocalDateTime.now();
 
   public OutboxEvent(String eventId, String aggregateType, String aggregateId, String eventType, String payload) {
@@ -39,6 +44,11 @@ public OutboxEvent(String eventId, String aggregateType, String aggregateId, Str
   }
 
   public void markPublished() {
-    this.published = true;
+    this.status = OutboxStatus.PUBLISHED;
+  }
+
+  public void markFailed() {
+    this.status = OutboxStatus.FAILED;
+    this.retryCount++;
   }
 }

apps/commerce-api/src/main/java/com/loopers/domain/event/OutboxService.java

@@ -40,4 +40,11 @@ public void markPublished(String eventId) {
       outboxRepository.save(event);
     });
   }
+
+  public void markFailed(String eventId) {
+    outboxRepository.findByEventId(eventId).ifPresent(event -> {
+      event.markFailed();
+      outboxRepository.save(event);
+    });
+  }
 }

apps/commerce-api/src/main/java/com/loopers/domain/event/OutboxStatus.java

@@ -0,0 +1,5 @@
+package com.loopers.domain.event;
+
+public enum OutboxStatus {
+  INIT, PUBLISHED, FAILED
+}

apps/commerce-api/src/main/java/com/loopers/infrastructure/event/OutboxRepository.java

@@ -1,6 +1,7 @@
 package com.loopers.infrastructure.event;
 
 import com.loopers.domain.event.OutboxEvent;
+import com.loopers.domain.event.OutboxStatus;
 import java.util.List;
 import java.util.Optional;
 import org.springframework.data.jpa.repository.JpaRepository;
@@ -11,4 +12,8 @@ public interface OutboxRepository extends JpaRepository<OutboxEvent, Long> {
 
   Optional<OutboxEvent> findByEventId(String eventId);
 
+  List<OutboxEvent> findTop10ByStatusInAndRetryCountLessThanOrderByCreatedAtAsc(
+      List<OutboxStatus> statuses,
+      int retryCount
+  );
 }