Merge pull request #34 from yeonsu00/refactor/week8

yeonsu00 · web-flow · commit 3fa6cf615d20 · 2025-12-23T16:10:10.000+09:00
Refactor/week8
diff --git a/apps/commerce-api/src/main/java/com/loopers/domain/outbox/OutboxService.java b/apps/commerce-api/src/main/java/com/loopers/domain/outbox/OutboxService.java
@@ -4,6 +4,7 @@
 import com.loopers.support.error.CoreException;
 import com.loopers.support.error.ErrorType;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.kafka.core.KafkaTemplate;
@@ -18,7 +19,7 @@ public class OutboxService {
     private static final int MAX_RETRIES = 3;
 
     private final OutboxRepository outboxRepository;
-    private final KafkaTemplate<String, Object> kafkaTemplate;
+    private final KafkaTemplate<String, String> kafkaTemplate;
     private final ObjectMapper objectMapper;
 
     @Transactional
@@ -36,19 +37,16 @@ public void publishPendingOutboxes(int batchSize) {
         List<Outbox> pendingOutboxes = outboxRepository.findPendingOutboxes(batchSize);
 
         for (Outbox outbox : pendingOutboxes) {
-            kafkaTemplate
-                    .send(
-                            outbox.getTopic(),
-                            outbox.getPartitionKey(),
-                            outbox.getPayload()
-                    )
-                    .whenComplete((result, exception) -> {
-                        if (exception == null) {
-                            outboxRepository.markAsPublished(outbox.getId());
-                        } else {
-                            outboxRepository.markAsFailed(outbox.getId(), exception.getMessage());
-                        }
-                    });
+            try {
+                kafkaTemplate.send(
+                        outbox.getTopic(),
+                        outbox.getPartitionKey(),
+                        outbox.getPayload()
+                ).get();
+                outboxRepository.markAsPublished(outbox.getId());
+            } catch (Exception e) {
+                outboxRepository.markAsFailed(outbox.getId(), e.getMessage());
+            }
         }
     }
 
@@ -62,19 +60,16 @@ public void retryFailedOutboxes(int batchSize) {
                 continue;
             }
 
-            kafkaTemplate
-                    .send(
-                            outbox.getTopic(),
-                            outbox.getPartitionKey(),
-                            outbox.getPayload()
-                    )
-                    .whenComplete((result, exception) -> {
-                        if (exception == null) {
-                            outboxRepository.markAsPublished(outbox.getId());
-                        } else {
-                            outboxRepository.markAsFailed(outbox.getId(), exception.getMessage());
-                        }
-                    });
+            try {
+                kafkaTemplate.send(
+                        outbox.getTopic(),
+                        outbox.getPartitionKey(),
+                        outbox.getPayload()
+                ).get(5, TimeUnit.SECONDS);
+                outboxRepository.markAsPublished(outbox.getId());
+            } catch (Exception e) {
+                outboxRepository.markAsFailed(outbox.getId(), e.getMessage());
+            }
         }
     }
 }
diff --git a/apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxJpaRepository.java b/apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxJpaRepository.java
@@ -3,17 +3,18 @@
 import com.loopers.domain.outbox.Outbox;
 import com.loopers.domain.outbox.OutboxStatus;
 import java.util.List;
+import org.springframework.data.domain.Pageable;
 import org.springframework.data.jpa.repository.JpaRepository;
 import org.springframework.data.jpa.repository.Modifying;
 import org.springframework.data.jpa.repository.Query;
 import org.springframework.data.repository.query.Param;
 
 public interface OutboxJpaRepository extends JpaRepository<Outbox, Long> {
     @Query("SELECT o FROM Outbox o WHERE o.status = :status ORDER BY o.createdAt ASC")
-    List<Outbox> findByStatusOrderByCreatedAtAsc(@Param("status") OutboxStatus status);
+    List<Outbox> findByStatusOrderByCreatedAtAsc(@Param("status") OutboxStatus status, Pageable pageable);
 
     @Query("SELECT o FROM Outbox o WHERE o.status = :status AND o.retryCount < :maxRetries ORDER BY o.createdAt ASC")
-    List<Outbox> findFailedOutboxesForRetry(@Param("status") OutboxStatus status, @Param("maxRetries") int maxRetries);
+    List<Outbox> findFailedOutboxesForRetry(@Param("status") OutboxStatus status, @Param("maxRetries") int maxRetries, Pageable pageable);
 
     @Modifying(clearAutomatically = true)
     @Query("UPDATE Outbox o SET o.status = 'PUBLISHED', o.errorMessage = NULL WHERE o.id = :id")
diff --git a/apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxRepositoryImpl.java b/apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxRepositoryImpl.java
@@ -5,7 +5,9 @@
 import com.loopers.domain.outbox.OutboxStatus;
 import java.util.List;
 import lombok.RequiredArgsConstructor;
+import org.springframework.data.domain.PageRequest;
 import org.springframework.stereotype.Repository;
+import org.springframework.transaction.annotation.Transactional;
 
 @RequiredArgsConstructor
 @Repository
@@ -20,26 +22,22 @@ public void saveOutbox(Outbox outbox) {
 
     @Override
     public List<Outbox> findPendingOutboxes(int limit) {
-        return outboxJpaRepository.findByStatusOrderByCreatedAtAsc(OutboxStatus.PENDING)
-                .stream()
-                .limit(limit)
-                .toList();
+        return outboxJpaRepository.findByStatusOrderByCreatedAtAsc(OutboxStatus.PENDING, PageRequest.of(0, limit));
     }
 
     @Override
     public List<Outbox> findFailedAndRetryableOutboxes(int limit, int maxRetries) {
-        return outboxJpaRepository.findFailedOutboxesForRetry(OutboxStatus.FAILED, maxRetries)
-                .stream()
-                .limit(limit)
-                .toList();
+        return outboxJpaRepository.findFailedOutboxesForRetry(OutboxStatus.FAILED, maxRetries, PageRequest.of(0, limit));
     }
 
     @Override
+    @Transactional
     public void markAsPublished(Long id) {
         outboxJpaRepository.updateOutboxPublishedStatus(id);
     }
 
     @Override
+    @Transactional
     public void markAsFailed(Long id, String message) {
         outboxJpaRepository.updateOutboxFailedStatusAndErrorMessageAndRetryCount(id, message);
     }
diff --git a/apps/commerce-api/src/main/java/com/loopers/interfaces/scheduler/OutboxRelayScheduler.java b/apps/commerce-api/src/main/java/com/loopers/interfaces/scheduler/OutboxRelayScheduler.java
@@ -2,24 +2,38 @@
 
 import com.loopers.domain.outbox.OutboxService;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 
 @RequiredArgsConstructor
 @Component
+@Slf4j
 public class OutboxRelayScheduler {
 
     private final OutboxService outboxService;
     private static final int BATCH_SIZE = 100;
 
     @Scheduled(fixedDelay = 3000)
     public void relayOutboxEvents() {
-        outboxService.publishPendingOutboxes(BATCH_SIZE);
+        try {
+            log.debug("Outbox 이벤트 릴레이 시작 - batchSize: {}", BATCH_SIZE);
+            outboxService.publishPendingOutboxes(BATCH_SIZE);
+            log.debug("Outbox 이벤트 릴레이 완료");
+        } catch (Exception e) {
+            log.error("Outbox 이벤트 릴레이 중 오류 발생", e);
+        }
     }
 
     @Scheduled(fixedDelay = 5000)
     public void retryFailedOutboxEvents() {
-        outboxService.retryFailedOutboxes(BATCH_SIZE);
+        try {
+            log.debug("실패한 Outbox 이벤트 재시도 시작 - batchSize: {}", BATCH_SIZE);
+            outboxService.retryFailedOutboxes(BATCH_SIZE);
+            log.debug("실패한 Outbox 이벤트 재시도 완료");
+        } catch (Exception e) {
+            log.error("실패한 Outbox 이벤트 재시도 중 오류 발생", e);
+        }
     }
 }
 
diff --git a/apps/commerce-api/src/test/java/com/loopers/support/IntegrationTest.java b/apps/commerce-api/src/test/java/com/loopers/support/IntegrationTest.java
@@ -6,5 +6,5 @@
 public abstract class IntegrationTest {
 
     @MockitoBean
-    protected KafkaTemplate<String, Object> kafkaTemplate;
+    protected KafkaTemplate<String, String> kafkaTemplate;
 }
diff --git a/apps/commerce-streamer/src/main/java/com/loopers/domain/eventhandled/EventHandledService.java b/apps/commerce-streamer/src/main/java/com/loopers/domain/eventhandled/EventHandledService.java
@@ -1,6 +1,7 @@
 package com.loopers.domain.eventhandled;
 
 import lombok.RequiredArgsConstructor;
+import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
@@ -17,10 +18,8 @@ public boolean isAlreadyHandled(String eventId) {
 
     @Transactional
     public void markAsHandled(String eventId, String eventType, String aggregateKey) {
-        if (!eventHandledRepository.existsByEventId(eventId)) {
-            EventHandled eventHandled = EventHandled.create(eventId, eventType, aggregateKey);
-            eventHandledRepository.saveEventHandled(eventHandled);
-        }
+        EventHandled eventHandled = EventHandled.create(eventId, eventType, aggregateKey);
+        eventHandledRepository.saveEventHandled(eventHandled);
     }
 }
 
diff --git a/apps/commerce-streamer/src/main/java/com/loopers/domain/metrics/ProductMetricsRepository.java b/apps/commerce-streamer/src/main/java/com/loopers/domain/metrics/ProductMetricsRepository.java
@@ -6,5 +6,7 @@ public interface ProductMetricsRepository {
     void saveProductMetrics(ProductMetrics productMetrics);
 
     Optional<ProductMetrics> findByProductId(Long productId);
+
+    int incrementLikeCount(Long productId);
 }
 
diff --git a/apps/commerce-streamer/src/main/java/com/loopers/domain/metrics/ProductMetricsService.java b/apps/commerce-streamer/src/main/java/com/loopers/domain/metrics/ProductMetricsService.java
@@ -1,34 +1,42 @@
 package com.loopers.domain.metrics;
 
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
 @RequiredArgsConstructor
 @Service
+@Slf4j
 public class ProductMetricsService {
 
     private final ProductMetricsRepository productMetricsRepository;
 
     @Transactional
     public void incrementLikeCount(Long productId) {
-        ProductMetrics metrics = productMetricsRepository.findByProductId(productId)
-                .orElseGet(() -> {
-                    ProductMetrics newMetrics = ProductMetrics.create(productId);
-                    productMetricsRepository.saveProductMetrics(newMetrics);
-                    return newMetrics;
-                });
-        metrics.incrementLikeCount();
-        productMetricsRepository.saveProductMetrics(metrics);
+        int updatedRows = productMetricsRepository.incrementLikeCount(productId);
+
+        if (updatedRows == 0) {
+            try {
+                ProductMetrics newMetrics = ProductMetrics.create(productId);
+                newMetrics.incrementLikeCount();
+                productMetricsRepository.saveProductMetrics(newMetrics);
+            } catch (DataIntegrityViolationException e) {
+                productMetricsRepository.incrementLikeCount(productId);
+            }
+        }
     }
 
     @Transactional
     public void decrementLikeCount(Long productId) {
-        productMetricsRepository.findByProductId(productId)
-                .ifPresent(metrics -> {
-                    metrics.decrementLikeCount();
-                    productMetricsRepository.saveProductMetrics(metrics);
+        ProductMetrics productMetrics = productMetricsRepository.findByProductId(productId)
+                .orElseGet(() -> {
+                    log.warn("좋아요 취소 시 메트릭이 존재하지 않음: productId={}", productId);
+                    return ProductMetrics.create(productId);
                 });
+        productMetrics.decrementLikeCount();
+        productMetricsRepository.saveProductMetrics(productMetrics);
     }
 
     @Transactional
diff --git a/apps/commerce-streamer/src/main/java/com/loopers/infrastructure/eventhandled/EventHandledRepositoryImpl.java b/apps/commerce-streamer/src/main/java/com/loopers/infrastructure/eventhandled/EventHandledRepositoryImpl.java
@@ -3,17 +3,24 @@
 import com.loopers.domain.eventhandled.EventHandled;
 import com.loopers.domain.eventhandled.EventHandledRepository;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.stereotype.Repository;
 
 @RequiredArgsConstructor
 @Repository
+@Slf4j
 public class EventHandledRepositoryImpl implements EventHandledRepository {
 
     private final EventHandledJpaRepository eventHandledJpaRepository;
 
     @Override
     public void saveEventHandled(EventHandled eventHandled) {
-        eventHandledJpaRepository.save(eventHandled);
+        try {
+            eventHandledJpaRepository.save(eventHandled);
+        } catch (DataIntegrityViolationException e) {
+            log.debug("중복 이벤트 감지로 인해 처리를 생략함: {}", eventHandled.getEventId());
+        }
     }
 
     @Override
diff --git a/apps/commerce-streamer/src/main/java/com/loopers/infrastructure/metrics/ProductMetricsJpaRepository.java b/apps/commerce-streamer/src/main/java/com/loopers/infrastructure/metrics/ProductMetricsJpaRepository.java
@@ -3,8 +3,15 @@
 import com.loopers.domain.metrics.ProductMetrics;
 import java.util.Optional;
 import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Modifying;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.query.Param;
 
 public interface ProductMetricsJpaRepository extends JpaRepository<ProductMetrics, Long> {
     Optional<ProductMetrics> findByProductId(Long productId);
+
+    @Modifying
+    @Query("UPDATE ProductMetrics m SET m.likeCount = m.likeCount + 1 WHERE m.productId = :productId")
+    int incrementLikeCount(@Param("productId") Long productId);
 }
 
diff --git a/apps/commerce-streamer/src/main/java/com/loopers/infrastructure/metrics/ProductMetricsRepositoryImpl.java b/apps/commerce-streamer/src/main/java/com/loopers/infrastructure/metrics/ProductMetricsRepositoryImpl.java
@@ -21,5 +21,10 @@ public void saveProductMetrics(ProductMetrics productMetrics) {
     public Optional<ProductMetrics> findByProductId(Long productId) {
         return productMetricsJpaRepository.findByProductId(productId);
     }
+
+    @Override
+    public int incrementLikeCount(Long productId) {
+        return productMetricsJpaRepository.incrementLikeCount(productId);
+    }
 }
 
diff --git a/apps/commerce-streamer/src/main/java/com/loopers/interfaces/consumer/DemoKafkaConsumer.java b/apps/commerce-streamer/src/main/java/com/loopers/interfaces/consumer/DemoKafkaConsumer.java
@@ -1,6 +1,6 @@
 package com.loopers.interfaces.consumer;
 
-import com.loopers.confg.kafka.KafkaConfig;
+import com.loopers.config.kafka.KafkaConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.support.Acknowledgment;
diff --git a/apps/commerce-streamer/src/main/java/com/loopers/interfaces/consumer/MetricsConsumer.java b/apps/commerce-streamer/src/main/java/com/loopers/interfaces/consumer/MetricsConsumer.java
@@ -1,7 +1,7 @@
 package com.loopers.interfaces.consumer;
 
 import com.loopers.application.kafka.KafkaEvent;
-import com.loopers.confg.kafka.KafkaConfig;
+import com.loopers.config.kafka.KafkaConfig;
 import com.loopers.domain.cache.ProductCacheService;
 import com.loopers.domain.eventhandled.EventHandledService;
 import com.loopers.domain.metrics.ProductMetricsService;
diff --git a/modules/kafka/src/main/java/com/loopers/application/kafka/KafkaEvent.java b/modules/kafka/src/main/java/com/loopers/application/kafka/KafkaEvent.java
@@ -3,14 +3,15 @@
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import java.time.ZonedDateTime;
 import java.util.List;
+import java.util.UUID;
 
 @JsonTypeInfo(use = JsonTypeInfo.Id.CLASS, property = "@class")
 public interface KafkaEvent {
     String eventId();
     ZonedDateTime timestamp();
 
     static String generateEventId(String eventType, String aggregateKey) {
-        return String.format("%s-%s-%d", eventType, aggregateKey, System.currentTimeMillis());
+        return String.format("%s-%s-%s", eventType, aggregateKey, UUID.randomUUID());
     }
 
     sealed interface OrderEvent extends KafkaEvent permits
diff --git a/modules/kafka/src/main/java/com/loopers/config/kafka/KafkaConfig.java b/modules/kafka/src/main/java/com/loopers/config/kafka/KafkaConfig.java
@@ -1,4 +1,4 @@
-package com.loopers.confg.kafka;
+package com.loopers.config.kafka;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.kafka.clients.consumer.ConsumerConfig;

Original file line number	Diff line number	Diff line change
`@@ -6,5 +6,5 @@`
`6`	`6`	`public abstract class IntegrationTest {`
`7`	`7`
`8`	`8`	`@MockitoBean`
`9`		`- protected KafkaTemplate<String, Object> kafkaTemplate;`
	`9`	`+ protected KafkaTemplate<String, String> kafkaTemplate;`
`10`	`10`	`}`
Original file line number	Diff line number	Diff line change
`@@ -6,5 +6,7 @@ public interface ProductMetricsRepository {`
`6`	`6`	`void saveProductMetrics(ProductMetrics productMetrics);`
`7`	`7`
`8`	`8`	`Optional<ProductMetrics> findByProductId(Long productId);`
	`9`	`+`
	`10`	`+ int incrementLikeCount(Long productId);`
`9`	`11`	`}`
`10`	`12`