>) (CompletableFuture)
+ CompletableFuture.completedFuture(null);
+ }
+}
diff --git a/apps/commerce-streamer/src/main/java/com/loopers/application/eventhandled/EventHandledService.java b/apps/commerce-streamer/src/main/java/com/loopers/application/eventhandled/EventHandledService.java
new file mode 100644
index 000000000..bbb016ee2
--- /dev/null
+++ b/apps/commerce-streamer/src/main/java/com/loopers/application/eventhandled/EventHandledService.java
@@ -0,0 +1,64 @@
+package com.loopers.application.eventhandled;
+
+import com.loopers.domain.eventhandled.EventHandled;
+import com.loopers.domain.eventhandled.EventHandledRepository;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+/**
+ * 이벤트 처리 기록 서비스.
+ *
+ * Kafka Consumer에서 이벤트의 멱등성을 보장하기 위한 서비스입니다.
+ * 이벤트 처리 전 `eventId`가 이미 처리되었는지 확인하고,
+ * 처리되지 않은 경우에만 처리 기록을 저장합니다.
+ *
+ *
+ * @author Loopers
+ * @version 1.0
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class EventHandledService {
+
+ private final EventHandledRepository eventHandledRepository;
+
+ /**
+ * 이벤트가 이미 처리되었는지 확인합니다.
+ *
+ * @param eventId 이벤트 ID
+ * @return 이미 처리된 경우 true, 그렇지 않으면 false
+ */
+ @Transactional(readOnly = true)
+ public boolean isAlreadyHandled(String eventId) {
+ return eventHandledRepository.existsByEventId(eventId);
+ }
+
+ /**
+ * 이벤트 처리 기록을 저장합니다.
+ *
+ * UNIQUE 제약조건 위반 시 예외를 발생시킵니다.
+ * 이는 동시성 상황에서 중복 처리를 방지하기 위한 것입니다.
+ *
+ *
+ * @param eventId 이벤트 ID
+ * @param eventType 이벤트 타입 (예: "LikeAdded", "OrderCreated")
+ * @param topic Kafka 토픽 이름
+ * @throws org.springframework.dao.DataIntegrityViolationException 이미 처리된 이벤트인 경우
+ */
+ @Transactional
+ public void markAsHandled(String eventId, String eventType, String topic) {
+ try {
+ EventHandled eventHandled = new EventHandled(eventId, eventType, topic);
+ eventHandledRepository.save(eventHandled);
+ log.debug("이벤트 처리 기록 저장: eventId={}, eventType={}, topic={}",
+ eventId, eventType, topic);
+ } catch (org.springframework.dao.DataIntegrityViolationException e) {
+ // UNIQUE 제약조건 위반 = 이미 처리됨 (멱등성 보장)
+ log.warn("이벤트가 이미 처리되었습니다: eventId={}", eventId);
+ throw e;
+ }
+ }
+}
diff --git a/apps/commerce-streamer/src/main/java/com/loopers/application/metrics/ProductMetricsService.java b/apps/commerce-streamer/src/main/java/com/loopers/application/metrics/ProductMetricsService.java
new file mode 100644
index 000000000..98227105d
--- /dev/null
+++ b/apps/commerce-streamer/src/main/java/com/loopers/application/metrics/ProductMetricsService.java
@@ -0,0 +1,165 @@
+package com.loopers.application.metrics;
+
+import com.loopers.domain.metrics.ProductMetrics;
+import com.loopers.domain.metrics.ProductMetricsRepository;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.dao.DataIntegrityViolationException;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+/**
+ * 상품 메트릭 집계 서비스.
+ *
+ * Kafka Consumer에서 이벤트를 수취하여 상품 메트릭을 집계합니다.
+ * 좋아요 수, 판매량, 상세 페이지 조회 수 등을 upsert합니다.
+ *
+ *
+ * 도메인 분리 근거:
+ *
+ * - 외부 시스템(데이터 플랫폼, 분석 시스템)을 위한 메트릭 집계
+ * - Product 도메인의 핵심 비즈니스 로직과는 분리된 관심사
+ * - Kafka Consumer를 통한 이벤트 기반 집계
+ *
+ *
+ *
+ * @author Loopers
+ * @version 1.0
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class ProductMetricsService {
+
+ private final ProductMetricsRepository productMetricsRepository;
+
+ /**
+ * 좋아요 수를 증가시킵니다.
+ *
+ * 이벤트의 버전을 기준으로 최신 이벤트만 반영합니다.
+ *
+ *
+ * @param productId 상품 ID
+ * @param eventVersion 이벤트의 버전
+ */
+ @Transactional
+ public void incrementLikeCount(Long productId, Long eventVersion) {
+ ProductMetrics metrics = findOrCreate(productId);
+
+ // 버전 비교: 이벤트가 최신이 아니면 스킵
+ if (!metrics.shouldUpdate(eventVersion)) {
+ log.debug("오래된 이벤트 스킵: productId={}, eventVersion={}, metricsVersion={}",
+ productId, eventVersion, metrics.getVersion());
+ return;
+ }
+
+ metrics.incrementLikeCount();
+ productMetricsRepository.save(metrics);
+ log.debug("좋아요 수 증가: productId={}, likeCount={}", productId, metrics.getLikeCount());
+ }
+
+ /**
+ * 좋아요 수를 감소시킵니다.
+ *
+ * 이벤트의 버전을 기준으로 최신 이벤트만 반영합니다.
+ *
+ *
+ * @param productId 상품 ID
+ * @param eventVersion 이벤트의 버전
+ */
+ @Transactional
+ public void decrementLikeCount(Long productId, Long eventVersion) {
+ ProductMetrics metrics = findOrCreate(productId);
+
+ // 버전 비교: 이벤트가 최신이 아니면 스킵
+ if (!metrics.shouldUpdate(eventVersion)) {
+ log.debug("오래된 이벤트 스킵: productId={}, eventVersion={}, metricsVersion={}",
+ productId, eventVersion, metrics.getVersion());
+ return;
+ }
+
+ metrics.decrementLikeCount();
+ productMetricsRepository.save(metrics);
+ log.debug("좋아요 수 감소: productId={}, likeCount={}", productId, metrics.getLikeCount());
+ }
+
+ /**
+ * 판매량을 증가시킵니다.
+ *
+ * 이벤트의 버전을 기준으로 최신 이벤트만 반영합니다.
+ *
+ *
+ * @param productId 상품 ID
+ * @param quantity 판매 수량
+ * @param eventVersion 이벤트의 버전
+ */
+ @Transactional
+ public void incrementSalesCount(Long productId, Integer quantity, Long eventVersion) {
+ ProductMetrics metrics = findOrCreate(productId);
+
+ // 버전 비교: 이벤트가 최신이 아니면 스킵
+ if (!metrics.shouldUpdate(eventVersion)) {
+ log.debug("오래된 이벤트 스킵: productId={}, eventVersion={}, metricsVersion={}",
+ productId, eventVersion, metrics.getVersion());
+ return;
+ }
+
+ metrics.incrementSalesCount(quantity);
+ productMetricsRepository.save(metrics);
+ log.debug("판매량 증가: productId={}, quantity={}, salesCount={}",
+ productId, quantity, metrics.getSalesCount());
+ }
+
+ /**
+ * 상세 페이지 조회 수를 증가시킵니다.
+ *
+ * 이벤트의 버전을 기준으로 최신 이벤트만 반영합니다.
+ *
+ *
+ * @param productId 상품 ID
+ * @param eventVersion 이벤트의 버전
+ */
+ @Transactional
+ public void incrementViewCount(Long productId, Long eventVersion) {
+ ProductMetrics metrics = findOrCreate(productId);
+
+ // 버전 비교: 이벤트가 최신이 아니면 스킵
+ if (!metrics.shouldUpdate(eventVersion)) {
+ log.debug("오래된 이벤트 스킵: productId={}, eventVersion={}, metricsVersion={}",
+ productId, eventVersion, metrics.getVersion());
+ return;
+ }
+
+ metrics.incrementViewCount();
+ productMetricsRepository.save(metrics);
+ log.debug("조회 수 증가: productId={}, viewCount={}", productId, metrics.getViewCount());
+ }
+
+ /**
+ * 상품 메트릭을 조회하거나 없으면 생성합니다.
+ *
+ * 비관적 락을 사용하여 동시성 제어를 보장합니다.
+ * 신규 생성 시 동시 삽입으로 인한 unique constraint violation을 처리합니다.
+ *
+ *
+ * @param productId 상품 ID
+ * @return ProductMetrics 인스턴스
+ */
+ private ProductMetrics findOrCreate(Long productId) {
+ return productMetricsRepository
+ .findByProductIdForUpdate(productId)
+ .orElseGet(() -> {
+ try {
+ ProductMetrics newMetrics = new ProductMetrics(productId);
+ return productMetricsRepository.save(newMetrics);
+ } catch (DataIntegrityViolationException e) {
+ // 동시 삽입 시 재조회
+ log.debug("동시 삽입 감지, 재조회: productId={}", productId);
+ return productMetricsRepository
+ .findByProductIdForUpdate(productId)
+ .orElseThrow(() -> new IllegalStateException(
+ "ProductMetrics 생성 실패: productId=" + productId));
+ }
+ });
+ }
+}
diff --git a/apps/commerce-streamer/src/main/java/com/loopers/domain/event/LikeEvent.java b/apps/commerce-streamer/src/main/java/com/loopers/domain/event/LikeEvent.java
new file mode 100644
index 000000000..f806ffea1
--- /dev/null
+++ b/apps/commerce-streamer/src/main/java/com/loopers/domain/event/LikeEvent.java
@@ -0,0 +1,37 @@
+package com.loopers.domain.event;
+
+import java.time.LocalDateTime;
+
+/**
+ * 좋아요 이벤트 DTO.
+ *
+ * Kafka에서 수신한 좋아요 이벤트를 파싱하기 위한 DTO입니다.
+ * 주의: 이 클래스는 commerce-api의 LikeEvent와 동일한 구조를 가진 DTO입니다.
+ * 향후 공유 모듈로 분리하는 것을 고려해야 합니다.
+ *
+ *
+ * @author Loopers
+ * @version 1.0
+ */
+public class LikeEvent {
+
+ /**
+ * 좋아요 추가 이벤트.
+ */
+ public record LikeAdded(
+ Long userId,
+ Long productId,
+ LocalDateTime occurredAt
+ ) {
+ }
+
+ /**
+ * 좋아요 취소 이벤트.
+ */
+ public record LikeRemoved(
+ Long userId,
+ Long productId,
+ LocalDateTime occurredAt
+ ) {
+ }
+}
diff --git a/apps/commerce-streamer/src/main/java/com/loopers/domain/event/OrderEvent.java b/apps/commerce-streamer/src/main/java/com/loopers/domain/event/OrderEvent.java
new file mode 100644
index 000000000..eacbbc19a
--- /dev/null
+++ b/apps/commerce-streamer/src/main/java/com/loopers/domain/event/OrderEvent.java
@@ -0,0 +1,40 @@
+package com.loopers.domain.event;
+
+import java.time.LocalDateTime;
+import java.util.List;
+
+/**
+ * 주문 이벤트 DTO.
+ *
+ * Kafka에서 수신한 주문 이벤트를 파싱하기 위한 DTO입니다.
+ * 주의: 이 클래스는 commerce-api의 OrderEvent와 동일한 구조를 가진 DTO입니다.
+ * 향후 공유 모듈로 분리하는 것을 고려해야 합니다.
+ *
+ *
+ * @author Loopers
+ * @version 1.0
+ */
+public class OrderEvent {
+
+ /**
+ * 주문 생성 이벤트.
+ */
+ public record OrderCreated(
+ Long orderId,
+ Long userId,
+ String couponCode,
+ Integer subtotal,
+ Long usedPointAmount,
+ List orderItems,
+ LocalDateTime createdAt
+ ) {
+ /**
+ * 주문 아이템 정보.
+ */
+ public record OrderItemInfo(
+ Long productId,
+ Integer quantity
+ ) {
+ }
+ }
+}
diff --git a/apps/commerce-streamer/src/main/java/com/loopers/domain/event/ProductEvent.java b/apps/commerce-streamer/src/main/java/com/loopers/domain/event/ProductEvent.java
new file mode 100644
index 000000000..4bd7f3587
--- /dev/null
+++ b/apps/commerce-streamer/src/main/java/com/loopers/domain/event/ProductEvent.java
@@ -0,0 +1,27 @@
+package com.loopers.domain.event;
+
+import java.time.LocalDateTime;
+
+/**
+ * 상품 이벤트 DTO.
+ *
+ * Kafka에서 수신한 상품 이벤트를 파싱하기 위한 DTO입니다.
+ * 주의: 이 클래스는 commerce-api의 ProductEvent와 동일한 구조를 가진 DTO입니다.
+ * 향후 공유 모듈로 분리하는 것을 고려해야 합니다.
+ *
+ *
+ * @author Loopers
+ * @version 1.0
+ */
+public class ProductEvent {
+
+ /**
+ * 상품 상세 페이지 조회 이벤트.
+ */
+ public record ProductViewed(
+ Long productId,
+ Long userId,
+ LocalDateTime occurredAt
+ ) {
+ }
+}
diff --git a/apps/commerce-streamer/src/main/java/com/loopers/domain/eventhandled/EventHandled.java b/apps/commerce-streamer/src/main/java/com/loopers/domain/eventhandled/EventHandled.java
new file mode 100644
index 000000000..b280fb891
--- /dev/null
+++ b/apps/commerce-streamer/src/main/java/com/loopers/domain/eventhandled/EventHandled.java
@@ -0,0 +1,62 @@
+package com.loopers.domain.eventhandled;
+
+import jakarta.persistence.*;
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+
+import java.time.LocalDateTime;
+
+/**
+ * 이벤트 처리 기록 엔티티.
+ *
+ * Kafka Consumer에서 처리한 이벤트의 멱등성을 보장하기 위한 엔티티입니다.
+ * `eventId`를 Primary Key로 사용하여 중복 처리를 방지합니다.
+ *
+ *
+ * 멱등성 보장:
+ *
+ * - 동일한 `eventId`를 가진 이벤트는 한 번만 처리됩니다
+ * - UNIQUE 제약조건으로 데이터베이스 레벨에서 중복 방지
+ * - 이벤트 처리 전 `eventId` 존재 여부를 확인하여 중복 처리 방지
+ *
+ *
+ *
+ * @author Loopers
+ * @version 1.0
+ */
+@Entity
+@Table(name = "event_handled", indexes = {
+ @Index(name = "idx_handled_at", columnList = "handled_at")
+})
+@NoArgsConstructor(access = AccessLevel.PROTECTED)
+@Getter
+public class EventHandled {
+
+ @Id
+ @Column(name = "event_id", nullable = false, length = 255)
+ private String eventId;
+
+ @Column(name = "event_type", nullable = false, length = 100)
+ private String eventType;
+
+ @Column(name = "topic", nullable = false, length = 255)
+ private String topic;
+
+ @Column(name = "handled_at", nullable = false)
+ private LocalDateTime handledAt;
+
+ /**
+ * EventHandled 인스턴스를 생성합니다.
+ *
+ * @param eventId 이벤트 ID (UUID)
+ * @param eventType 이벤트 타입 (예: "LikeAdded", "OrderCreated")
+ * @param topic Kafka 토픽 이름
+ */
+ public EventHandled(String eventId, String eventType, String topic) {
+ this.eventId = eventId;
+ this.eventType = eventType;
+ this.topic = topic;
+ this.handledAt = LocalDateTime.now();
+ }
+}
diff --git a/apps/commerce-streamer/src/main/java/com/loopers/domain/eventhandled/EventHandledRepository.java b/apps/commerce-streamer/src/main/java/com/loopers/domain/eventhandled/EventHandledRepository.java
new file mode 100644
index 000000000..536ddbd63
--- /dev/null
+++ b/apps/commerce-streamer/src/main/java/com/loopers/domain/eventhandled/EventHandledRepository.java
@@ -0,0 +1,40 @@
+package com.loopers.domain.eventhandled;
+
+import java.util.Optional;
+
+/**
+ * EventHandled 엔티티에 대한 저장소 인터페이스.
+ *
+ * 이벤트 처리 기록의 영속성 계층과의 상호작용을 정의합니다.
+ * DIP를 준수하여 도메인 레이어에서 인터페이스를 정의합니다.
+ *
+ *
+ * @author Loopers
+ * @version 1.0
+ */
+public interface EventHandledRepository {
+
+ /**
+ * 이벤트 처리 기록을 저장합니다.
+ *
+ * @param eventHandled 저장할 이벤트 처리 기록
+ * @return 저장된 이벤트 처리 기록
+ */
+ EventHandled save(EventHandled eventHandled);
+
+ /**
+ * 이벤트 ID로 처리 기록을 조회합니다.
+ *
+ * @param eventId 이벤트 ID
+ * @return 조회된 처리 기록을 담은 Optional
+ */
+ Optional findByEventId(String eventId);
+
+ /**
+ * 이벤트 ID가 이미 처리되었는지 확인합니다.
+ *
+ * @param eventId 이벤트 ID
+ * @return 이미 처리된 경우 true, 그렇지 않으면 false
+ */
+ boolean existsByEventId(String eventId);
+}
diff --git a/apps/commerce-streamer/src/main/java/com/loopers/domain/metrics/ProductMetrics.java b/apps/commerce-streamer/src/main/java/com/loopers/domain/metrics/ProductMetrics.java
new file mode 100644
index 000000000..f552b355c
--- /dev/null
+++ b/apps/commerce-streamer/src/main/java/com/loopers/domain/metrics/ProductMetrics.java
@@ -0,0 +1,127 @@
+package com.loopers.domain.metrics;
+
+import jakarta.persistence.*;
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+
+import java.time.LocalDateTime;
+
+/**
+ * 상품 메트릭 집계 엔티티.
+ *
+ * Kafka Consumer에서 이벤트를 수취하여 집계한 메트릭을 저장합니다.
+ * 좋아요 수, 판매량, 상세 페이지 조회 수 등을 관리합니다.
+ *
+ *
+ * 도메인 분리 근거:
+ *
+ * - 외부 시스템(데이터 플랫폼, 분석 시스템)을 위한 메트릭 집계
+ * - Product 도메인의 핵심 비즈니스 로직과는 분리된 관심사
+ * - Kafka Consumer를 통한 이벤트 기반 집계
+ *
+ *
+ *
+ * @author Loopers
+ * @version 1.0
+ */
+@Entity
+@Table(name = "product_metrics")
+@NoArgsConstructor(access = AccessLevel.PROTECTED)
+@Getter
+public class ProductMetrics {
+
+ @Id
+ @Column(name = "product_id")
+ private Long productId;
+
+ @Column(name = "like_count", nullable = false)
+ private Long likeCount;
+
+ @Column(name = "sales_count", nullable = false)
+ private Long salesCount;
+
+ @Column(name = "view_count", nullable = false)
+ private Long viewCount;
+
+ @Column(name = "version", nullable = false)
+ private Long version;
+
+ @Column(name = "updated_at", nullable = false)
+ private LocalDateTime updatedAt;
+
+ /**
+ * ProductMetrics 인스턴스를 생성합니다.
+ *
+ * @param productId 상품 ID
+ */
+ public ProductMetrics(Long productId) {
+ this.productId = productId;
+ this.likeCount = 0L;
+ this.salesCount = 0L;
+ this.viewCount = 0L;
+ this.version = 0L;
+ this.updatedAt = LocalDateTime.now();
+ }
+
+ /**
+ * 좋아요 수를 증가시킵니다.
+ */
+ public void incrementLikeCount() {
+ this.likeCount++;
+ this.version++;
+ this.updatedAt = LocalDateTime.now();
+ }
+
+ /**
+ * 좋아요 수를 감소시킵니다.
+ */
+ public void decrementLikeCount() {
+ if (this.likeCount > 0) {
+ this.likeCount--;
+ this.version++;
+ this.updatedAt = LocalDateTime.now();
+ }
+ }
+
+ /**
+ * 판매량을 증가시킵니다.
+ *
+ * @param quantity 판매 수량
+ */
+ public void incrementSalesCount(Integer quantity) {
+ if (quantity != null && quantity > 0) {
+ this.salesCount += quantity;
+ this.version++;
+ this.updatedAt = LocalDateTime.now();
+ }
+ }
+
+ /**
+ * 상세 페이지 조회 수를 증가시킵니다.
+ */
+ public void incrementViewCount() {
+ this.viewCount++;
+ this.version++;
+ this.updatedAt = LocalDateTime.now();
+ }
+
+ /**
+ * 이벤트의 버전을 기준으로 메트릭을 업데이트해야 하는지 확인합니다.
+ *
+ * 이벤트의 `version`이 메트릭의 `version`보다 크면 업데이트합니다.
+ * 이를 통해 오래된 이벤트가 최신 메트릭을 덮어쓰는 것을 방지합니다.
+ *
+ *
+ * @param eventVersion 이벤트의 버전
+ * @return 업데이트해야 하면 true, 그렇지 않으면 false
+ */
+ public boolean shouldUpdate(Long eventVersion) {
+ if (eventVersion == null) {
+ // 이벤트에 버전 정보가 없으면 업데이트 (하위 호환성)
+ return true;
+ }
+ // 이벤트 버전이 메트릭 버전보다 크면 업데이트
+ return eventVersion > this.version;
+ }
+}
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
new file mode 100644
index 000000000..4ffe5938e
--- /dev/null
+++ b/apps/commerce-streamer/src/main/java/com/loopers/domain/metrics/ProductMetricsRepository.java
@@ -0,0 +1,58 @@
+package com.loopers.domain.metrics;
+
+import java.util.Optional;
+
+/**
+ * ProductMetrics 엔티티에 대한 저장소 인터페이스.
+ *
+ * 상품 메트릭 집계 데이터의 영속성 계층과의 상호작용을 정의합니다.
+ * DIP를 준수하여 도메인 레이어에서 인터페이스를 정의합니다.
+ *
+ *
+ * 도메인 분리 근거:
+ *
+ * - Metric 도메인은 외부 시스템 연동을 위한 별도 관심사
+ * - Product 도메인의 핵심 비즈니스 로직과는 분리
+ *
+ *
+ *
+ * @author Loopers
+ * @version 1.0
+ */
+public interface ProductMetricsRepository {
+
+ /**
+ * 상품 메트릭을 저장합니다.
+ *
+ * @param productMetrics 저장할 상품 메트릭
+ * @return 저장된 상품 메트릭
+ */
+ ProductMetrics save(ProductMetrics productMetrics);
+
+ /**
+ * 상품 ID로 메트릭을 조회합니다.
+ *
+ * @param productId 상품 ID
+ * @return 조회된 메트릭을 담은 Optional
+ */
+ Optional findByProductId(Long productId);
+
+ /**
+ * 상품 ID로 메트릭을 조회합니다. (비관적 락)
+ *
+ * Upsert 시 동시성 제어를 위해 사용합니다.
+ *
+ *
+ * Lock 전략:
+ *
+ * - PESSIMISTIC_WRITE: SELECT ... FOR UPDATE 사용
+ * - Lock 범위: PK(productId) 기반 조회로 해당 행만 락 (최소화)
+ * - 사용 목적: 메트릭 집계 시 Lost Update 방지
+ *
+ *
+ *
+ * @param productId 상품 ID
+ * @return 조회된 메트릭을 담은 Optional
+ */
+ Optional findByProductIdForUpdate(Long productId);
+}
diff --git a/apps/commerce-streamer/src/main/java/com/loopers/infrastructure/eventhandled/EventHandledJpaRepository.java b/apps/commerce-streamer/src/main/java/com/loopers/infrastructure/eventhandled/EventHandledJpaRepository.java
new file mode 100644
index 000000000..f3aefc464
--- /dev/null
+++ b/apps/commerce-streamer/src/main/java/com/loopers/infrastructure/eventhandled/EventHandledJpaRepository.java
@@ -0,0 +1,31 @@
+package com.loopers.infrastructure.eventhandled;
+
+import com.loopers.domain.eventhandled.EventHandled;
+import org.springframework.data.jpa.repository.JpaRepository;
+
+import java.util.Optional;
+
+/**
+ * EventHandled 엔티티에 대한 JPA Repository.
+ *
+ * @author Loopers
+ * @version 1.0
+ */
+public interface EventHandledJpaRepository extends JpaRepository {
+
+ /**
+ * 이벤트 ID로 처리 기록을 조회합니다.
+ *
+ * @param eventId 이벤트 ID
+ * @return 조회된 처리 기록을 담은 Optional
+ */
+ Optional findByEventId(String eventId);
+
+ /**
+ * 이벤트 ID가 이미 처리되었는지 확인합니다.
+ *
+ * @param eventId 이벤트 ID
+ * @return 이미 처리된 경우 true, 그렇지 않으면 false
+ */
+ boolean existsByEventId(String eventId);
+}
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
new file mode 100644
index 000000000..95dfc6b06
--- /dev/null
+++ b/apps/commerce-streamer/src/main/java/com/loopers/infrastructure/eventhandled/EventHandledRepositoryImpl.java
@@ -0,0 +1,39 @@
+package com.loopers.infrastructure.eventhandled;
+
+import com.loopers.domain.eventhandled.EventHandled;
+import com.loopers.domain.eventhandled.EventHandledRepository;
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Repository;
+
+import java.util.Optional;
+
+/**
+ * EventHandledRepository의 구현체.
+ *
+ * JPA를 사용하여 EventHandled 엔티티의 영속성을 관리합니다.
+ *
+ *
+ * @author Loopers
+ * @version 1.0
+ */
+@Repository
+@RequiredArgsConstructor
+public class EventHandledRepositoryImpl implements EventHandledRepository {
+
+ private final EventHandledJpaRepository jpaRepository;
+
+ @Override
+ public EventHandled save(EventHandled eventHandled) {
+ return jpaRepository.save(eventHandled);
+ }
+
+ @Override
+ public Optional findByEventId(String eventId) {
+ return jpaRepository.findByEventId(eventId);
+ }
+
+ @Override
+ public boolean existsByEventId(String eventId) {
+ return jpaRepository.existsByEventId(eventId);
+ }
+}
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
new file mode 100644
index 000000000..e54cb6aef
--- /dev/null
+++ b/apps/commerce-streamer/src/main/java/com/loopers/infrastructure/metrics/ProductMetricsJpaRepository.java
@@ -0,0 +1,40 @@
+package com.loopers.infrastructure.metrics;
+
+import com.loopers.domain.metrics.ProductMetrics;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Lock;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.query.Param;
+
+import jakarta.persistence.LockModeType;
+import java.util.Optional;
+
+/**
+ * ProductMetrics JPA Repository.
+ *
+ * 상품 메트릭 집계 데이터를 관리합니다.
+ *
+ */
+public interface ProductMetricsJpaRepository extends JpaRepository {
+
+ /**
+ * 상품 ID로 메트릭을 조회합니다.
+ *
+ * @param productId 상품 ID
+ * @return 조회된 메트릭을 담은 Optional
+ */
+ Optional findByProductId(Long productId);
+
+ /**
+ * 상품 ID로 메트릭을 조회합니다. (비관적 락)
+ *
+ * Upsert 시 동시성 제어를 위해 사용합니다.
+ *
+ *
+ * @param productId 상품 ID
+ * @return 조회된 메트릭을 담은 Optional
+ */
+ @Lock(LockModeType.PESSIMISTIC_WRITE)
+ @Query("SELECT pm FROM ProductMetrics pm WHERE pm.productId = :productId")
+ Optional findByProductIdForUpdate(@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
new file mode 100644
index 000000000..253da5917
--- /dev/null
+++ b/apps/commerce-streamer/src/main/java/com/loopers/infrastructure/metrics/ProductMetricsRepositoryImpl.java
@@ -0,0 +1,49 @@
+package com.loopers.infrastructure.metrics;
+
+import com.loopers.domain.metrics.ProductMetrics;
+import com.loopers.domain.metrics.ProductMetricsRepository;
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Component;
+
+import java.util.Optional;
+
+/**
+ * ProductMetricsRepository의 JPA 구현체.
+ *
+ * Spring Data JPA를 활용하여 ProductMetrics 엔티티의
+ * 영속성 작업을 처리합니다.
+ *
+ *
+ * @author Loopers
+ * @version 1.0
+ */
+@Component
+@RequiredArgsConstructor
+public class ProductMetricsRepositoryImpl implements ProductMetricsRepository {
+
+ private final ProductMetricsJpaRepository productMetricsJpaRepository;
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public ProductMetrics save(ProductMetrics productMetrics) {
+ return productMetricsJpaRepository.save(productMetrics);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Optional findByProductId(Long productId) {
+ return productMetricsJpaRepository.findByProductId(productId);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Optional findByProductIdForUpdate(Long productId) {
+ return productMetricsJpaRepository.findByProductIdForUpdate(productId);
+ }
+}
diff --git a/apps/commerce-streamer/src/main/java/com/loopers/interfaces/consumer/ProductMetricsConsumer.java b/apps/commerce-streamer/src/main/java/com/loopers/interfaces/consumer/ProductMetricsConsumer.java
new file mode 100644
index 000000000..2811056d9
--- /dev/null
+++ b/apps/commerce-streamer/src/main/java/com/loopers/interfaces/consumer/ProductMetricsConsumer.java
@@ -0,0 +1,430 @@
+package com.loopers.interfaces.consumer;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.loopers.application.eventhandled.EventHandledService;
+import com.loopers.application.metrics.ProductMetricsService;
+import com.loopers.confg.kafka.KafkaConfig;
+import com.loopers.domain.event.LikeEvent;
+import com.loopers.domain.event.OrderEvent;
+import com.loopers.domain.event.ProductEvent;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.header.Header;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.stereotype.Component;
+
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+
+/**
+ * 상품 메트릭 집계 Kafka Consumer.
+ *
+ * Kafka에서 이벤트를 수취하여 상품 메트릭을 집계합니다.
+ * 좋아요 수, 판매량, 상세 페이지 조회 수 등을 `product_metrics` 테이블에 upsert합니다.
+ *
+ *
+ * 처리 이벤트:
+ *
+ * - like-events: LikeAdded, LikeRemoved (좋아요 수 집계)
+ * - order-events: OrderCreated (판매량 집계)
+ * - product-events: ProductViewed (조회 수 집계)
+ *
+ *
+ *
+ * Manual Ack:
+ *
+ * - 이벤트 처리 성공 후 수동으로 커밋하여 At Most Once 보장
+ * - 에러 발생 시 커밋하지 않아 재처리 가능
+ *
+ *
+ *
+ * @author Loopers
+ * @version 1.0
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class ProductMetricsConsumer {
+
+ private final ProductMetricsService productMetricsService;
+ private final EventHandledService eventHandledService;
+ private final ObjectMapper objectMapper;
+
+ private static final String EVENT_ID_HEADER = "eventId";
+ private static final String EVENT_TYPE_HEADER = "eventType";
+ private static final String VERSION_HEADER = "version";
+
+ /**
+ * like-events 토픽을 구독하여 좋아요 수를 집계합니다.
+ *
+ * 멱등성 처리:
+ *
+ * - Kafka 메시지 헤더에서 `eventId`를 추출
+ * - 이미 처리된 이벤트는 스킵하여 중복 처리 방지
+ * - 처리 후 `event_handled` 테이블에 기록
+ *
+ *
+ *
+ * @param records Kafka 메시지 레코드 목록
+ * @param acknowledgment 수동 커밋을 위한 Acknowledgment
+ */
+ @KafkaListener(
+ topics = "like-events",
+ containerFactory = KafkaConfig.BATCH_LISTENER
+ )
+ public void consumeLikeEvents(
+ List> records,
+ Acknowledgment acknowledgment
+ ) {
+ try {
+ for (ConsumerRecord record : records) {
+ try {
+ String eventId = extractEventId(record);
+ if (eventId == null) {
+ log.warn("eventId가 없는 메시지는 건너뜁니다: offset={}, partition={}",
+ record.offset(), record.partition());
+ continue;
+ }
+
+ // 멱등성 체크: 이미 처리된 이벤트는 스킵
+ if (eventHandledService.isAlreadyHandled(eventId)) {
+ log.debug("이미 처리된 이벤트 스킵: eventId={}", eventId);
+ continue;
+ }
+
+ Object value = record.value();
+ String eventType;
+
+ // 버전 추출 (헤더에서)
+ Long eventVersion = extractVersion(record);
+
+ // Spring Kafka가 자동으로 역직렬화한 경우
+ if (value instanceof LikeEvent.LikeAdded) {
+ LikeEvent.LikeAdded event = (LikeEvent.LikeAdded) value;
+ productMetricsService.incrementLikeCount(
+ event.productId(),
+ eventVersion
+ );
+ eventType = "LikeAdded";
+ } else if (value instanceof LikeEvent.LikeRemoved) {
+ LikeEvent.LikeRemoved event = (LikeEvent.LikeRemoved) value;
+ productMetricsService.decrementLikeCount(
+ event.productId(),
+ eventVersion
+ );
+ eventType = "LikeRemoved";
+ } else {
+ // JSON 문자열인 경우 이벤트 타입 헤더로 구분
+ String eventTypeHeader = extractEventType(record);
+ if ("LikeRemoved".equals(eventTypeHeader)) {
+ LikeEvent.LikeRemoved event = parseLikeRemovedEvent(value);
+ productMetricsService.decrementLikeCount(
+ event.productId(),
+ eventVersion
+ );
+ eventType = "LikeRemoved";
+ } else {
+ // 기본값은 LikeAdded
+ LikeEvent.LikeAdded event = parseLikeEvent(value);
+ productMetricsService.incrementLikeCount(
+ event.productId(),
+ eventVersion
+ );
+ eventType = "LikeAdded";
+ }
+ }
+
+ // 이벤트 처리 기록 저장
+ eventHandledService.markAsHandled(eventId, eventType, "like-events");
+ } catch (org.springframework.dao.DataIntegrityViolationException e) {
+ // UNIQUE 제약조건 위반 = 동시성 상황에서 이미 처리됨 (정상)
+ log.debug("동시성 상황에서 이미 처리된 이벤트: offset={}, partition={}",
+ record.offset(), record.partition());
+ } catch (Exception e) {
+ log.error("좋아요 이벤트 처리 실패: offset={}, partition={}",
+ record.offset(), record.partition(), e);
+ // 개별 이벤트 처리 실패는 로그만 기록하고 계속 진행
+ }
+ }
+
+ // 모든 이벤트 처리 완료 후 수동 커밋
+ acknowledgment.acknowledge();
+ log.debug("좋아요 이벤트 처리 완료: count={}", records.size());
+ } catch (Exception e) {
+ log.error("좋아요 이벤트 배치 처리 실패: count={}", records.size(), e);
+ // 에러 발생 시 커밋하지 않음 (재처리 가능)
+ throw e;
+ }
+ }
+
+ /**
+ * order-events 토픽을 구독하여 판매량을 집계합니다.
+ *
+ * 멱등성 처리:
+ *
+ * - Kafka 메시지 헤더에서 `eventId`를 추출
+ * - 이미 처리된 이벤트는 스킵하여 중복 처리 방지
+ * - 처리 후 `event_handled` 테이블에 기록
+ *
+ *
+ *
+ * @param records Kafka 메시지 레코드 목록
+ * @param acknowledgment 수동 커밋을 위한 Acknowledgment
+ */
+ @KafkaListener(
+ topics = "order-events",
+ containerFactory = KafkaConfig.BATCH_LISTENER
+ )
+ public void consumeOrderEvents(
+ List> records,
+ Acknowledgment acknowledgment
+ ) {
+ try {
+ for (ConsumerRecord record : records) {
+ try {
+ String eventId = extractEventId(record);
+ if (eventId == null) {
+ log.warn("eventId가 없는 메시지는 건너뜁니다: offset={}, partition={}",
+ record.offset(), record.partition());
+ continue;
+ }
+
+ // 멱등성 체크: 이미 처리된 이벤트는 스킵
+ if (eventHandledService.isAlreadyHandled(eventId)) {
+ log.debug("이미 처리된 이벤트 스킵: eventId={}", eventId);
+ continue;
+ }
+
+ Object value = record.value();
+ OrderEvent.OrderCreated event = parseOrderCreatedEvent(value);
+
+ // 버전 추출 (헤더에서)
+ Long eventVersion = extractVersion(record);
+
+ // 주문 아이템별로 판매량 집계
+ for (OrderEvent.OrderCreated.OrderItemInfo item : event.orderItems()) {
+ productMetricsService.incrementSalesCount(
+ item.productId(),
+ item.quantity(),
+ eventVersion
+ );
+ }
+
+ // 이벤트 처리 기록 저장
+ eventHandledService.markAsHandled(eventId, "OrderCreated", "order-events");
+ } catch (org.springframework.dao.DataIntegrityViolationException e) {
+ // UNIQUE 제약조건 위반 = 동시성 상황에서 이미 처리됨 (정상)
+ log.debug("동시성 상황에서 이미 처리된 이벤트: offset={}, partition={}",
+ record.offset(), record.partition());
+ } catch (Exception e) {
+ log.error("주문 이벤트 처리 실패: offset={}, partition={}",
+ record.offset(), record.partition(), e);
+ // 개별 이벤트 처리 실패는 로그만 기록하고 계속 진행
+ }
+ }
+
+ // 모든 이벤트 처리 완료 후 수동 커밋
+ acknowledgment.acknowledge();
+ log.debug("주문 이벤트 처리 완료: count={}", records.size());
+ } catch (Exception e) {
+ log.error("주문 이벤트 배치 처리 실패: count={}", records.size(), e);
+ // 에러 발생 시 커밋하지 않음 (재처리 가능)
+ throw e;
+ }
+ }
+
+ /**
+ * Kafka 메시지 값을 LikeAdded 이벤트로 파싱합니다.
+ *
+ * @param value Kafka 메시지 값
+ * @return 파싱된 LikeAdded 이벤트
+ */
+ private LikeEvent.LikeAdded parseLikeEvent(Object value) {
+ try {
+ // JSON 문자열인 경우 파싱
+ String json = value instanceof String ? (String) value : objectMapper.writeValueAsString(value);
+ return objectMapper.readValue(json, LikeEvent.LikeAdded.class);
+ } catch (Exception e) {
+ throw new RuntimeException("LikeAdded 이벤트 파싱 실패", e);
+ }
+ }
+
+ /**
+ * Kafka 메시지 값을 LikeRemoved 이벤트로 파싱합니다.
+ *
+ * @param value Kafka 메시지 값
+ * @return 파싱된 LikeRemoved 이벤트
+ */
+ private LikeEvent.LikeRemoved parseLikeRemovedEvent(Object value) {
+ try {
+ // JSON 문자열인 경우 파싱
+ String json = value instanceof String ? (String) value : objectMapper.writeValueAsString(value);
+ return objectMapper.readValue(json, LikeEvent.LikeRemoved.class);
+ } catch (Exception e) {
+ throw new RuntimeException("LikeRemoved 이벤트 파싱 실패", e);
+ }
+ }
+
+ /**
+ * product-events 토픽을 구독하여 조회 수를 집계합니다.
+ *
+ * 멱등성 처리:
+ *
+ * - Kafka 메시지 헤더에서 `eventId`를 추출
+ * - 이미 처리된 이벤트는 스킵하여 중복 처리 방지
+ * - 처리 후 `event_handled` 테이블에 기록
+ *
+ *
+ *
+ * @param records Kafka 메시지 레코드 목록
+ * @param acknowledgment 수동 커밋을 위한 Acknowledgment
+ */
+ @KafkaListener(
+ topics = "product-events",
+ containerFactory = KafkaConfig.BATCH_LISTENER
+ )
+ public void consumeProductEvents(
+ List> records,
+ Acknowledgment acknowledgment
+ ) {
+ try {
+ for (ConsumerRecord record : records) {
+ try {
+ String eventId = extractEventId(record);
+ if (eventId == null) {
+ log.warn("eventId가 없는 메시지는 건너뜁니다: offset={}, partition={}",
+ record.offset(), record.partition());
+ continue;
+ }
+
+ // 멱등성 체크: 이미 처리된 이벤트는 스킵
+ if (eventHandledService.isAlreadyHandled(eventId)) {
+ log.debug("이미 처리된 이벤트 스킵: eventId={}", eventId);
+ continue;
+ }
+
+ Object value = record.value();
+ ProductEvent.ProductViewed event = parseProductViewedEvent(value);
+
+ // 버전 추출 (헤더에서)
+ Long eventVersion = extractVersion(record);
+
+ productMetricsService.incrementViewCount(
+ event.productId(),
+ eventVersion
+ );
+
+ // 이벤트 처리 기록 저장
+ eventHandledService.markAsHandled(eventId, "ProductViewed", "product-events");
+ } catch (org.springframework.dao.DataIntegrityViolationException e) {
+ // UNIQUE 제약조건 위반 = 동시성 상황에서 이미 처리됨 (정상)
+ log.debug("동시성 상황에서 이미 처리된 이벤트: offset={}, partition={}",
+ record.offset(), record.partition());
+ } catch (Exception e) {
+ log.error("상품 조회 이벤트 처리 실패: offset={}, partition={}",
+ record.offset(), record.partition(), e);
+ // 개별 이벤트 처리 실패는 로그만 기록하고 계속 진행
+ }
+ }
+
+ // 모든 이벤트 처리 완료 후 수동 커밋
+ acknowledgment.acknowledge();
+ log.debug("상품 조회 이벤트 처리 완료: count={}", records.size());
+ } catch (Exception e) {
+ log.error("상품 조회 이벤트 배치 처리 실패: count={}", records.size(), e);
+ // 에러 발생 시 커밋하지 않음 (재처리 가능)
+ throw e;
+ }
+ }
+
+ /**
+ * Kafka 메시지 값을 OrderCreated 이벤트로 파싱합니다.
+ *
+ * @param value Kafka 메시지 값
+ * @return 파싱된 OrderCreated 이벤트
+ */
+ private OrderEvent.OrderCreated parseOrderCreatedEvent(Object value) {
+ try {
+ if (value instanceof OrderEvent.OrderCreated) {
+ return (OrderEvent.OrderCreated) value;
+ }
+
+ // JSON 문자열인 경우 파싱
+ String json = value instanceof String ? (String) value : objectMapper.writeValueAsString(value);
+ return objectMapper.readValue(json, OrderEvent.OrderCreated.class);
+ } catch (Exception e) {
+ throw new RuntimeException("OrderCreated 이벤트 파싱 실패", e);
+ }
+ }
+
+ /**
+ * Kafka 메시지 값을 ProductViewed 이벤트로 파싱합니다.
+ *
+ * @param value Kafka 메시지 값
+ * @return 파싱된 ProductViewed 이벤트
+ */
+ private ProductEvent.ProductViewed parseProductViewedEvent(Object value) {
+ try {
+ if (value instanceof ProductEvent.ProductViewed) {
+ return (ProductEvent.ProductViewed) value;
+ }
+
+ // JSON 문자열인 경우 파싱
+ String json = value instanceof String ? (String) value : objectMapper.writeValueAsString(value);
+ return objectMapper.readValue(json, ProductEvent.ProductViewed.class);
+ } catch (Exception e) {
+ throw new RuntimeException("ProductViewed 이벤트 파싱 실패", e);
+ }
+ }
+
+ /**
+ * Kafka 메시지 헤더에서 eventId를 추출합니다.
+ *
+ * @param record Kafka 메시지 레코드
+ * @return eventId (없으면 null)
+ */
+ private String extractEventId(ConsumerRecord record) {
+ Header header = record.headers().lastHeader(EVENT_ID_HEADER);
+ if (header != null && header.value() != null) {
+ return new String(header.value(), StandardCharsets.UTF_8);
+ }
+ return null;
+ }
+
+ /**
+ * Kafka 메시지 헤더에서 eventType을 추출합니다.
+ *
+ * @param record Kafka 메시지 레코드
+ * @return eventType (없으면 null)
+ */
+ private String extractEventType(ConsumerRecord record) {
+ Header header = record.headers().lastHeader(EVENT_TYPE_HEADER);
+ if (header != null && header.value() != null) {
+ return new String(header.value(), StandardCharsets.UTF_8);
+ }
+ return null;
+ }
+
+ /**
+ * Kafka 메시지 헤더에서 version을 추출합니다.
+ *
+ * @param record Kafka 메시지 레코드
+ * @return version (없으면 null)
+ */
+ private Long extractVersion(ConsumerRecord record) {
+ Header header = record.headers().lastHeader(VERSION_HEADER);
+ if (header != null && header.value() != null) {
+ try {
+ String versionStr = new String(header.value(), StandardCharsets.UTF_8);
+ return Long.parseLong(versionStr);
+ } catch (NumberFormatException e) {
+ log.warn("버전 헤더 파싱 실패: offset={}, partition={}",
+ record.offset(), record.partition());
+ return null;
+ }
+ }
+ return null;
+ }
+}
diff --git a/apps/commerce-streamer/src/test/java/com/loopers/application/eventhandled/EventHandledServiceTest.java b/apps/commerce-streamer/src/test/java/com/loopers/application/eventhandled/EventHandledServiceTest.java
new file mode 100644
index 000000000..77d7efcd9
--- /dev/null
+++ b/apps/commerce-streamer/src/test/java/com/loopers/application/eventhandled/EventHandledServiceTest.java
@@ -0,0 +1,96 @@
+package com.loopers.application.eventhandled;
+
+import com.loopers.domain.eventhandled.EventHandled;
+import com.loopers.domain.eventhandled.EventHandledRepository;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.dao.DataIntegrityViolationException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
+
+/**
+ * EventHandledService 테스트.
+ */
+@ExtendWith(MockitoExtension.class)
+class EventHandledServiceTest {
+
+ @Mock
+ private EventHandledRepository eventHandledRepository;
+
+ @InjectMocks
+ private EventHandledService eventHandledService;
+
+ @DisplayName("처리되지 않은 이벤트는 false를 반환한다.")
+ @Test
+ void isAlreadyHandled_returnsFalse_whenNotHandled() {
+ // arrange
+ String eventId = "test-event-id";
+ when(eventHandledRepository.existsByEventId(eventId)).thenReturn(false);
+
+ // act
+ boolean result = eventHandledService.isAlreadyHandled(eventId);
+
+ // assert
+ assertThat(result).isFalse();
+ verify(eventHandledRepository).existsByEventId(eventId);
+ }
+
+ @DisplayName("이미 처리된 이벤트는 true를 반환한다.")
+ @Test
+ void isAlreadyHandled_returnsTrue_whenAlreadyHandled() {
+ // arrange
+ String eventId = "test-event-id";
+ when(eventHandledRepository.existsByEventId(eventId)).thenReturn(true);
+
+ // act
+ boolean result = eventHandledService.isAlreadyHandled(eventId);
+
+ // assert
+ assertThat(result).isTrue();
+ verify(eventHandledRepository).existsByEventId(eventId);
+ }
+
+ @DisplayName("처리되지 않은 이벤트는 정상적으로 저장된다.")
+ @Test
+ void markAsHandled_savesSuccessfully_whenNotHandled() {
+ // arrange
+ String eventId = "test-event-id";
+ String eventType = "LikeAdded";
+ String topic = "like-events";
+
+ EventHandled savedEventHandled = new EventHandled(eventId, eventType, topic);
+ when(eventHandledRepository.save(any(EventHandled.class))).thenReturn(savedEventHandled);
+
+ // act
+ eventHandledService.markAsHandled(eventId, eventType, topic);
+
+ // assert
+ verify(eventHandledRepository).save(any(EventHandled.class));
+ }
+
+ @DisplayName("이미 처리된 이벤트는 DataIntegrityViolationException을 발생시킨다.")
+ @Test
+ void markAsHandled_throwsException_whenAlreadyHandled() {
+ // arrange
+ String eventId = "test-event-id";
+ String eventType = "LikeAdded";
+ String topic = "like-events";
+
+ when(eventHandledRepository.save(any(EventHandled.class)))
+ .thenThrow(new DataIntegrityViolationException("UNIQUE constraint violation"));
+
+ // act & assert
+ assertThatThrownBy(() ->
+ eventHandledService.markAsHandled(eventId, eventType, topic)
+ ).isInstanceOf(DataIntegrityViolationException.class);
+
+ verify(eventHandledRepository).save(any(EventHandled.class));
+ }
+}
diff --git a/apps/commerce-streamer/src/test/java/com/loopers/application/metrics/ProductMetricsServiceTest.java b/apps/commerce-streamer/src/test/java/com/loopers/application/metrics/ProductMetricsServiceTest.java
new file mode 100644
index 000000000..e8064e333
--- /dev/null
+++ b/apps/commerce-streamer/src/test/java/com/loopers/application/metrics/ProductMetricsServiceTest.java
@@ -0,0 +1,217 @@
+package com.loopers.application.metrics;
+
+import com.loopers.domain.metrics.ProductMetrics;
+import com.loopers.domain.metrics.ProductMetricsRepository;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.util.Optional;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
+
+/**
+ * ProductMetricsService 테스트.
+ */
+@ExtendWith(MockitoExtension.class)
+class ProductMetricsServiceTest {
+
+ @Mock
+ private ProductMetricsRepository productMetricsRepository;
+
+ @InjectMocks
+ private ProductMetricsService productMetricsService;
+
+ @DisplayName("좋아요 수를 증가시킬 수 있다.")
+ @Test
+ void canIncrementLikeCount() {
+ // arrange
+ Long productId = 1L;
+ ProductMetrics existingMetrics = new ProductMetrics(productId);
+ existingMetrics.incrementLikeCount(); // 초기값: 1
+
+ when(productMetricsRepository.findByProductIdForUpdate(productId))
+ .thenReturn(Optional.of(existingMetrics));
+ when(productMetricsRepository.save(any(ProductMetrics.class)))
+ .thenAnswer(invocation -> invocation.getArgument(0));
+
+ // act
+ productMetricsService.incrementLikeCount(productId, existingMetrics.getVersion() + 1L);
+
+ // assert
+ assertThat(existingMetrics.getLikeCount()).isEqualTo(2L);
+ verify(productMetricsRepository).findByProductIdForUpdate(productId);
+ verify(productMetricsRepository).save(existingMetrics);
+ }
+
+ @DisplayName("좋아요 수를 감소시킬 수 있다.")
+ @Test
+ void canDecrementLikeCount() {
+ // arrange
+ Long productId = 1L;
+ ProductMetrics existingMetrics = new ProductMetrics(productId);
+ existingMetrics.incrementLikeCount(); // 초기값: 1
+
+ when(productMetricsRepository.findByProductIdForUpdate(productId))
+ .thenReturn(Optional.of(existingMetrics));
+ when(productMetricsRepository.save(any(ProductMetrics.class)))
+ .thenAnswer(invocation -> invocation.getArgument(0));
+
+ // act
+ productMetricsService.decrementLikeCount(productId, existingMetrics.getVersion() + 1L);
+
+ // assert
+ assertThat(existingMetrics.getLikeCount()).isEqualTo(0L);
+ verify(productMetricsRepository).findByProductIdForUpdate(productId);
+ verify(productMetricsRepository).save(existingMetrics);
+ }
+
+ @DisplayName("판매량을 증가시킬 수 있다.")
+ @Test
+ void canIncrementSalesCount() {
+ // arrange
+ Long productId = 1L;
+ Integer quantity = 5;
+ ProductMetrics existingMetrics = new ProductMetrics(productId);
+
+ when(productMetricsRepository.findByProductIdForUpdate(productId))
+ .thenReturn(Optional.of(existingMetrics));
+ when(productMetricsRepository.save(any(ProductMetrics.class)))
+ .thenAnswer(invocation -> invocation.getArgument(0));
+
+ // act
+ productMetricsService.incrementSalesCount(productId, quantity, existingMetrics.getVersion() + 1L);
+
+ // assert
+ assertThat(existingMetrics.getSalesCount()).isEqualTo(5L);
+ verify(productMetricsRepository).findByProductIdForUpdate(productId);
+ verify(productMetricsRepository).save(existingMetrics);
+ }
+
+ @DisplayName("조회 수를 증가시킬 수 있다.")
+ @Test
+ void canIncrementViewCount() {
+ // arrange
+ Long productId = 1L;
+ ProductMetrics existingMetrics = new ProductMetrics(productId);
+
+ when(productMetricsRepository.findByProductIdForUpdate(productId))
+ .thenReturn(Optional.of(existingMetrics));
+ when(productMetricsRepository.save(any(ProductMetrics.class)))
+ .thenAnswer(invocation -> invocation.getArgument(0));
+
+ // act
+ productMetricsService.incrementViewCount(productId, existingMetrics.getVersion() + 1L);
+
+ // assert
+ assertThat(existingMetrics.getViewCount()).isEqualTo(1L);
+ verify(productMetricsRepository).findByProductIdForUpdate(productId);
+ verify(productMetricsRepository).save(existingMetrics);
+ }
+
+ @DisplayName("메트릭이 없으면 새로 생성한다.")
+ @Test
+ void createsNewMetrics_whenNotExists() {
+ // arrange
+ Long productId = 1L;
+ Long eventVersion = 1L; // 새로 생성된 메트릭의 버전(0)보다 큰 버전
+
+ when(productMetricsRepository.findByProductIdForUpdate(productId))
+ .thenReturn(Optional.empty());
+ when(productMetricsRepository.save(any(ProductMetrics.class)))
+ .thenAnswer(invocation -> invocation.getArgument(0));
+
+ // act
+ productMetricsService.incrementLikeCount(productId, eventVersion);
+
+ // assert
+ verify(productMetricsRepository).findByProductIdForUpdate(productId);
+ // findOrCreate에서 1번, incrementLikeCount에서 1번 총 2번 호출됨
+ verify(productMetricsRepository, atLeast(1)).save(any(ProductMetrics.class));
+ }
+
+ @DisplayName("판매량 증가 시 null이나 0 이하의 수량은 무시된다.")
+ @Test
+ void ignoresInvalidQuantity_whenIncrementingSalesCount() {
+ // arrange
+ Long productId = 1L;
+ ProductMetrics existingMetrics = new ProductMetrics(productId);
+ Long initialSalesCount = existingMetrics.getSalesCount();
+ Long initialVersion = existingMetrics.getVersion();
+
+ when(productMetricsRepository.findByProductIdForUpdate(productId))
+ .thenReturn(Optional.of(existingMetrics));
+ when(productMetricsRepository.save(any(ProductMetrics.class)))
+ .thenAnswer(invocation -> invocation.getArgument(0));
+
+ // act
+ productMetricsService.incrementSalesCount(productId, null, existingMetrics.getVersion() + 1L);
+ productMetricsService.incrementSalesCount(productId, 0, existingMetrics.getVersion() + 1L);
+ productMetricsService.incrementSalesCount(productId, -1, existingMetrics.getVersion() + 1L);
+
+ // assert
+ // 유효하지 않은 수량은 무시되므로 값이 변경되지 않음
+ assertThat(existingMetrics.getSalesCount()).isEqualTo(initialSalesCount);
+ assertThat(existingMetrics.getVersion()).isEqualTo(initialVersion);
+ // save()는 호출되지만 메트릭 값은 변경되지 않음
+ verify(productMetricsRepository, times(3)).findByProductIdForUpdate(productId);
+ verify(productMetricsRepository, times(3)).save(existingMetrics);
+ }
+
+ @DisplayName("오래된 이벤트는 스킵하여 메트릭을 업데이트하지 않는다.")
+ @Test
+ void skipsOldEvent_whenEventIsOlderThanMetrics() {
+ // arrange
+ Long productId = 1L;
+ ProductMetrics existingMetrics = new ProductMetrics(productId);
+ existingMetrics.incrementLikeCount(); // 초기값: 1, version = 1
+
+ Long oldEventVersion = existingMetrics.getVersion() - 1L; // 이전 버전 이벤트
+
+ Long initialLikeCount = existingMetrics.getLikeCount();
+ Long initialVersion = existingMetrics.getVersion();
+
+ when(productMetricsRepository.findByProductIdForUpdate(productId))
+ .thenReturn(Optional.of(existingMetrics));
+
+ // act
+ productMetricsService.incrementLikeCount(productId, oldEventVersion);
+
+ // assert
+ // 오래된 이벤트는 스킵되므로 값이 변경되지 않음
+ assertThat(existingMetrics.getLikeCount()).isEqualTo(initialLikeCount);
+ assertThat(existingMetrics.getVersion()).isEqualTo(initialVersion);
+ verify(productMetricsRepository).findByProductIdForUpdate(productId);
+ verify(productMetricsRepository, never()).save(any(ProductMetrics.class));
+ }
+
+ @DisplayName("최신 이벤트는 메트릭을 업데이트한다.")
+ @Test
+ void updatesMetrics_whenEventIsNewerThanMetrics() {
+ // arrange
+ Long productId = 1L;
+ ProductMetrics existingMetrics = new ProductMetrics(productId);
+ existingMetrics.incrementLikeCount(); // 초기값: 1, version = 1
+
+ Long newEventVersion = existingMetrics.getVersion() + 1L; // 최신 버전 이벤트
+
+ when(productMetricsRepository.findByProductIdForUpdate(productId))
+ .thenReturn(Optional.of(existingMetrics));
+ when(productMetricsRepository.save(any(ProductMetrics.class)))
+ .thenAnswer(invocation -> invocation.getArgument(0));
+
+ // act
+ productMetricsService.incrementLikeCount(productId, newEventVersion);
+
+ // assert
+ // 최신 이벤트는 반영됨
+ assertThat(existingMetrics.getLikeCount()).isEqualTo(2L);
+ verify(productMetricsRepository).findByProductIdForUpdate(productId);
+ verify(productMetricsRepository).save(existingMetrics);
+ }
+}
diff --git a/apps/commerce-streamer/src/test/java/com/loopers/domain/metrics/ProductMetricsTest.java b/apps/commerce-streamer/src/test/java/com/loopers/domain/metrics/ProductMetricsTest.java
new file mode 100644
index 000000000..6fab02bfb
--- /dev/null
+++ b/apps/commerce-streamer/src/test/java/com/loopers/domain/metrics/ProductMetricsTest.java
@@ -0,0 +1,155 @@
+package com.loopers.domain.metrics;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.time.LocalDateTime;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class ProductMetricsTest {
+
+ @DisplayName("ProductMetrics는 상품 ID로 생성되며 초기값이 0으로 설정된다.")
+ @Test
+ void createsProductMetricsWithInitialValues() {
+ // arrange
+ Long productId = 1L;
+
+ // act
+ ProductMetrics metrics = new ProductMetrics(productId);
+
+ // assert
+ assertThat(metrics.getProductId()).isEqualTo(productId);
+ assertThat(metrics.getLikeCount()).isEqualTo(0L);
+ assertThat(metrics.getSalesCount()).isEqualTo(0L);
+ assertThat(metrics.getViewCount()).isEqualTo(0L);
+ assertThat(metrics.getVersion()).isEqualTo(0L);
+ assertThat(metrics.getUpdatedAt()).isNotNull();
+ }
+
+ @DisplayName("좋아요 수를 증가시킬 수 있다.")
+ @Test
+ void canIncrementLikeCount() throws InterruptedException {
+ // arrange
+ ProductMetrics metrics = new ProductMetrics(1L);
+ Long initialLikeCount = metrics.getLikeCount();
+ Long initialVersion = metrics.getVersion();
+ LocalDateTime initialUpdatedAt = metrics.getUpdatedAt();
+
+ // act
+ Thread.sleep(1); // 시간 차이를 보장하기 위한 작은 지연
+ metrics.incrementLikeCount();
+
+ // assert
+ assertThat(metrics.getLikeCount()).isEqualTo(initialLikeCount + 1);
+ assertThat(metrics.getVersion()).isEqualTo(initialVersion + 1);
+ assertThat(metrics.getUpdatedAt()).isAfter(initialUpdatedAt);
+ }
+
+ @DisplayName("좋아요 수를 감소시킬 수 있다.")
+ @Test
+ void canDecrementLikeCount() {
+ // arrange
+ ProductMetrics metrics = new ProductMetrics(1L);
+ metrics.incrementLikeCount(); // 먼저 증가시킴
+ Long initialLikeCount = metrics.getLikeCount();
+ Long initialVersion = metrics.getVersion();
+
+ // act
+ metrics.decrementLikeCount();
+
+ // assert
+ assertThat(metrics.getLikeCount()).isEqualTo(initialLikeCount - 1);
+ assertThat(metrics.getVersion()).isEqualTo(initialVersion + 1);
+ }
+
+ @DisplayName("좋아요 수가 0일 때 감소해도 음수가 되지 않는다 (멱등성 보장).")
+ @Test
+ void preventsNegativeLikeCount_whenDecrementingFromZero() {
+ // arrange
+ ProductMetrics metrics = new ProductMetrics(1L);
+ assertThat(metrics.getLikeCount()).isEqualTo(0L);
+ Long initialVersion = metrics.getVersion();
+
+ // act
+ metrics.decrementLikeCount();
+
+ // assert
+ assertThat(metrics.getLikeCount()).isEqualTo(0L);
+ assertThat(metrics.getVersion()).isEqualTo(initialVersion); // version도 변경되지 않음
+ }
+
+ @DisplayName("판매량을 증가시킬 수 있다.")
+ @Test
+ void canIncrementSalesCount() {
+ // arrange
+ ProductMetrics metrics = new ProductMetrics(1L);
+ Long initialSalesCount = metrics.getSalesCount();
+ Long initialVersion = metrics.getVersion();
+ Integer quantity = 5;
+
+ // act
+ metrics.incrementSalesCount(quantity);
+
+ // assert
+ assertThat(metrics.getSalesCount()).isEqualTo(initialSalesCount + quantity);
+ assertThat(metrics.getVersion()).isEqualTo(initialVersion + 1);
+ }
+
+ @DisplayName("판매량 증가 시 null이나 0 이하의 수량은 무시된다.")
+ @Test
+ void ignoresInvalidQuantity_whenIncrementingSalesCount() {
+ // arrange
+ ProductMetrics metrics = new ProductMetrics(1L);
+ Long initialSalesCount = metrics.getSalesCount();
+ Long initialVersion = metrics.getVersion();
+
+ // act
+ metrics.incrementSalesCount(null);
+ metrics.incrementSalesCount(0);
+ metrics.incrementSalesCount(-1);
+
+ // assert
+ assertThat(metrics.getSalesCount()).isEqualTo(initialSalesCount);
+ assertThat(metrics.getVersion()).isEqualTo(initialVersion); // version도 변경되지 않음
+ }
+
+ @DisplayName("상세 페이지 조회 수를 증가시킬 수 있다.")
+ @Test
+ void canIncrementViewCount() throws InterruptedException {
+ // arrange
+ ProductMetrics metrics = new ProductMetrics(1L);
+ Long initialViewCount = metrics.getViewCount();
+ Long initialVersion = metrics.getVersion();
+ LocalDateTime initialUpdatedAt = metrics.getUpdatedAt();
+
+ // act
+ Thread.sleep(1); // 시간 차이를 보장하기 위한 작은 지연
+ metrics.incrementViewCount();
+
+ // assert
+ assertThat(metrics.getViewCount()).isEqualTo(initialViewCount + 1);
+ assertThat(metrics.getVersion()).isEqualTo(initialVersion + 1);
+ assertThat(metrics.getUpdatedAt()).isAfter(initialUpdatedAt);
+ }
+
+ @DisplayName("여러 메트릭을 연속으로 업데이트할 수 있다.")
+ @Test
+ void canUpdateMultipleMetrics() {
+ // arrange
+ ProductMetrics metrics = new ProductMetrics(1L);
+
+ // act
+ metrics.incrementLikeCount();
+ metrics.incrementLikeCount();
+ metrics.incrementSalesCount(10);
+ metrics.incrementViewCount();
+ metrics.decrementLikeCount();
+
+ // assert
+ assertThat(metrics.getLikeCount()).isEqualTo(1L);
+ assertThat(metrics.getSalesCount()).isEqualTo(10L);
+ assertThat(metrics.getViewCount()).isEqualTo(1L);
+ assertThat(metrics.getVersion()).isEqualTo(5L); // 5번 업데이트됨
+ }
+}
diff --git a/apps/commerce-streamer/src/test/java/com/loopers/interfaces/consumer/ProductMetricsConsumerIntegrationTest.java b/apps/commerce-streamer/src/test/java/com/loopers/interfaces/consumer/ProductMetricsConsumerIntegrationTest.java
new file mode 100644
index 000000000..c43519195
--- /dev/null
+++ b/apps/commerce-streamer/src/test/java/com/loopers/interfaces/consumer/ProductMetricsConsumerIntegrationTest.java
@@ -0,0 +1,116 @@
+package com.loopers.interfaces.consumer;
+
+import com.loopers.domain.event.LikeEvent;
+import com.loopers.testcontainers.KafkaTestContainersConfig;
+import com.loopers.utils.KafkaCleanUp;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.TopicPartition;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Import;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.test.context.ActiveProfiles;
+
+import java.time.LocalDateTime;
+import java.util.Collections;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * ProductMetricsConsumer 통합 테스트.
+ *
+ * 실제 Kafka를 사용하여 이벤트 처리 동작을 검증합니다.
+ *
+ *
+ * Kafka 컨테이너:
+ * {@link KafkaTestContainersConfig}가 테스트 실행 시 자동으로 Kafka 컨테이너를 시작합니다.
+ *
+ */
+@SpringBootTest
+@ActiveProfiles("test")
+@Import(KafkaTestContainersConfig.class)
+class ProductMetricsConsumerIntegrationTest {
+
+ @Autowired
+ private KafkaCleanUp kafkaCleanUp;
+
+ @Autowired
+ private KafkaTemplate kafkaTemplate;
+
+ @Autowired
+ private KafkaProperties kafkaProperties;
+
+ @BeforeEach
+ void setUp() {
+ // 테스트 전에 토픽의 모든 메시지 삭제 및 재생성
+ kafkaCleanUp.resetAllTestTopics();
+ }
+
+ /**
+ * offset.reset: latest 설정이 제대로 적용되는지 확인하는 테스트.
+ *
+ * 테스트 목적:
+ * kafka.yml에 설정된 `offset.reset: latest`가 실제로 동작하는지 검증합니다.
+ *
+ *
+ * 동작 원리:
+ * 1. 이전 메시지를 Kafka에 발행 (이 메시지는 나중에 읽히지 않아야 함)
+ * 2. Consumer Group을 삭제하여 offset 정보 제거
+ * 3. 새로운 메시지를 Kafka에 발행
+ * 4. 새로운 Consumer Group으로 Consumer를 시작
+ * 5. offset.reset: latest 설정으로 인해 Consumer는 최신 메시지(새로운 메시지)부터 읽기 시작해야 함
+ *
+ *
+ * 검증 내용:
+ * - Consumer의 현재 position이 최신 offset(endOffset)과 같거나 가까운지 확인
+ * - 이는 Consumer가 이전 메시지를 건너뛰고 최신 메시지부터 읽기 시작했다는 의미
+ *
+ */
+ @DisplayName("offset.reset: latest 설정이 적용되어 새로운 Consumer Group은 최신 메시지만 읽는다.")
+ @Test
+ void offsetResetLatest_shouldOnlyReadLatestMessages() throws Exception {
+ // 이 메시지는 나중에 Consumer가 읽지 않아야 함 (offset.reset: latest 때문)
+ String topic = "like-events";
+ String partitionKey = "product-1";
+ LikeEvent.LikeAdded oldMessage = new LikeEvent.LikeAdded(100L, 1L, LocalDateTime.now());
+ kafkaTemplate.send(topic, partitionKey, oldMessage).get();
+
+ // Consumer Group을 삭제하면 offset 정보가 사라짐
+ // 다음에 같은 Consumer Group으로 시작할 때 offset.reset 설정이 적용됨
+ String testGroupId = "test-offset-reset-" + System.currentTimeMillis();
+ kafkaCleanUp.resetConsumerGroup(testGroupId);
+
+ // 이 메시지는 Consumer가 읽어야 함 (최신 메시지이므로)
+ LikeEvent.LikeAdded newMessage = new LikeEvent.LikeAdded(200L, 1L, LocalDateTime.now());
+ kafkaTemplate.send(topic, partitionKey, newMessage).get();
+
+ // 프로젝트의 kafka.yml 설정을 사용하여 Consumer 생성
+ // 이 설정에는 offset.reset: latest가 포함되어 있음
+ Map consumerProps = kafkaProperties.buildConsumerProperties();
+ consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, testGroupId);
+
+ try (KafkaConsumer consumer = new KafkaConsumer<>(consumerProps)) {
+ // 특정 파티션에 할당 (테스트용)
+ TopicPartition partition = new TopicPartition(topic, 0);
+ consumer.assign(Collections.singletonList(partition));
+
+ // endOffset: 토픽의 마지막 메시지 다음 offset (현재는 2개 메시지가 있으므로 2)
+ // currentPosition: Consumer가 현재 읽을 위치 (offset.reset: latest면 endOffset과 같아야 함)
+ Long endOffset = consumer.endOffsets(Collections.singletonList(partition)).get(partition);
+ long currentPosition = consumer.position(partition);
+
+ // offset.reset: latest 설정이 적용되었다면:
+ // - currentPosition은 endOffset과 같거나 가까워야 함
+ // - 이는 Consumer가 이전 메시지(oldMessage)를 건너뛰고 최신 메시지(newMessage)부터 읽기 시작했다는 의미
+ // 예: endOffset=2, currentPosition=2 → 이전 메시지(offset 0)를 건너뛰고 최신 메시지(offset 1)부터 시작
+ assertThat(currentPosition)
+ .isGreaterThanOrEqualTo(endOffset);
+ }
+ }
+}
diff --git a/apps/commerce-streamer/src/test/java/com/loopers/interfaces/consumer/ProductMetricsConsumerTest.java b/apps/commerce-streamer/src/test/java/com/loopers/interfaces/consumer/ProductMetricsConsumerTest.java
new file mode 100644
index 000000000..bf5306797
--- /dev/null
+++ b/apps/commerce-streamer/src/test/java/com/loopers/interfaces/consumer/ProductMetricsConsumerTest.java
@@ -0,0 +1,393 @@
+package com.loopers.interfaces.consumer;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.loopers.application.eventhandled.EventHandledService;
+import com.loopers.application.metrics.ProductMetricsService;
+import com.loopers.domain.event.LikeEvent;
+import com.loopers.domain.event.OrderEvent;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.header.internals.RecordHeader;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.apache.kafka.common.record.TimestampType;
+
+import java.util.Optional;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.dao.DataIntegrityViolationException;
+import org.springframework.kafka.support.Acknowledgment;
+
+import java.nio.charset.StandardCharsets;
+import java.time.LocalDateTime;
+import java.util.List;
+
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+
+/**
+ * ProductMetricsConsumer 테스트.
+ */
+@ExtendWith(MockitoExtension.class)
+class ProductMetricsConsumerTest {
+
+ @Mock
+ private ProductMetricsService productMetricsService;
+
+ @Mock
+ private EventHandledService eventHandledService;
+
+ @Mock
+ private ObjectMapper objectMapper;
+
+ @Mock
+ private Acknowledgment acknowledgment;
+
+ @InjectMocks
+ private ProductMetricsConsumer productMetricsConsumer;
+
+ @DisplayName("LikeAdded 이벤트를 처리할 수 있다.")
+ @Test
+ void canConsumeLikeAddedEvent() {
+ // arrange
+ String eventId = "test-event-id";
+ Long productId = 1L;
+ Long userId = 100L;
+ LikeEvent.LikeAdded event = new LikeEvent.LikeAdded(userId, productId, LocalDateTime.now());
+
+ Headers headers = new RecordHeaders();
+ headers.add(new RecordHeader("eventId", eventId.getBytes(StandardCharsets.UTF_8)));
+ headers.add(new RecordHeader("version", "1".getBytes(StandardCharsets.UTF_8)));
+
+ ConsumerRecord record = new ConsumerRecord<>(
+ "like-events", 0, 0L, 0L, TimestampType.CREATE_TIME, 0, 0, "key", event, headers, Optional.empty()
+ );
+ List> records = List.of(record);
+
+ when(eventHandledService.isAlreadyHandled(eventId)).thenReturn(false);
+
+ // act
+ productMetricsConsumer.consumeLikeEvents(records, acknowledgment);
+
+ // assert
+ verify(eventHandledService).isAlreadyHandled(eventId);
+ verify(productMetricsService).incrementLikeCount(eq(productId), eq(1L));
+ verify(eventHandledService).markAsHandled(eventId, "LikeAdded", "like-events");
+ verify(acknowledgment).acknowledge();
+ }
+
+ @DisplayName("LikeRemoved 이벤트를 처리할 수 있다.")
+ @Test
+ void canConsumeLikeRemovedEvent() {
+ // arrange
+ String eventId = "test-event-id-2";
+ Long productId = 1L;
+ Long userId = 100L;
+ LikeEvent.LikeRemoved event = new LikeEvent.LikeRemoved(userId, productId, LocalDateTime.now());
+
+ Headers headers = new RecordHeaders();
+ headers.add(new RecordHeader("eventId", eventId.getBytes(StandardCharsets.UTF_8)));
+ headers.add(new RecordHeader("version", "2".getBytes(StandardCharsets.UTF_8)));
+
+ ConsumerRecord record = new ConsumerRecord<>(
+ "like-events", 0, 0L, 0L, TimestampType.CREATE_TIME, 0, 0, "key", event, headers, Optional.empty()
+ );
+ List> records = List.of(record);
+
+ when(eventHandledService.isAlreadyHandled(eventId)).thenReturn(false);
+
+ // act
+ productMetricsConsumer.consumeLikeEvents(records, acknowledgment);
+
+ // assert
+ verify(eventHandledService).isAlreadyHandled(eventId);
+ verify(productMetricsService).decrementLikeCount(eq(productId), eq(2L));
+ verify(eventHandledService).markAsHandled(eventId, "LikeRemoved", "like-events");
+ verify(acknowledgment).acknowledge();
+ }
+
+ @DisplayName("OrderCreated 이벤트를 처리할 수 있다.")
+ @Test
+ void canConsumeOrderCreatedEvent() {
+ // arrange
+ String eventId = "test-event-id-3";
+ Long orderId = 1L;
+ Long userId = 100L;
+ Long productId1 = 1L;
+ Long productId2 = 2L;
+
+ List orderItems = List.of(
+ new OrderEvent.OrderCreated.OrderItemInfo(productId1, 3),
+ new OrderEvent.OrderCreated.OrderItemInfo(productId2, 2)
+ );
+
+ OrderEvent.OrderCreated event = new OrderEvent.OrderCreated(
+ orderId, userId, null, 10000, 0L, orderItems, LocalDateTime.now()
+ );
+
+ Headers headers = new RecordHeaders();
+ headers.add(new RecordHeader("eventId", eventId.getBytes(StandardCharsets.UTF_8)));
+ headers.add(new RecordHeader("version", "3".getBytes(StandardCharsets.UTF_8)));
+
+ ConsumerRecord record = new ConsumerRecord<>(
+ "order-events", 0, 0L, 0L, TimestampType.CREATE_TIME, 0, 0, "key", event, headers, Optional.empty()
+ );
+ List> records = List.of(record);
+
+ when(eventHandledService.isAlreadyHandled(eventId)).thenReturn(false);
+
+ // act
+ productMetricsConsumer.consumeOrderEvents(records, acknowledgment);
+
+ // assert
+ verify(eventHandledService).isAlreadyHandled(eventId);
+ verify(productMetricsService).incrementSalesCount(eq(productId1), eq(3), eq(3L));
+ verify(productMetricsService).incrementSalesCount(eq(productId2), eq(2), eq(3L));
+ verify(eventHandledService).markAsHandled(eventId, "OrderCreated", "order-events");
+ verify(acknowledgment).acknowledge();
+ }
+
+ @DisplayName("배치로 여러 이벤트를 처리할 수 있다.")
+ @Test
+ void canConsumeMultipleEvents() {
+ // arrange
+ String eventId1 = "test-event-id-4";
+ String eventId2 = "test-event-id-5";
+ Long productId = 1L;
+ Long userId = 100L;
+
+ LikeEvent.LikeAdded event1 = new LikeEvent.LikeAdded(userId, productId, LocalDateTime.now());
+ LikeEvent.LikeRemoved event2 = new LikeEvent.LikeRemoved(userId, productId, LocalDateTime.now());
+
+ Headers headers1 = new RecordHeaders();
+ headers1.add(new RecordHeader("eventId", eventId1.getBytes(StandardCharsets.UTF_8)));
+ headers1.add(new RecordHeader("version", "4".getBytes(StandardCharsets.UTF_8)));
+ Headers headers2 = new RecordHeaders();
+ headers2.add(new RecordHeader("eventId", eventId2.getBytes(StandardCharsets.UTF_8)));
+ headers2.add(new RecordHeader("version", "5".getBytes(StandardCharsets.UTF_8)));
+
+ List> records = List.of(
+ new ConsumerRecord<>("like-events", 0, 0L, 0L, TimestampType.CREATE_TIME, 0, 0, "key", event1, headers1, Optional.empty()),
+ new ConsumerRecord<>("like-events", 0, 1L, 0L, TimestampType.CREATE_TIME, 0, 0, "key", event2, headers2, Optional.empty())
+ );
+
+ when(eventHandledService.isAlreadyHandled(eventId1)).thenReturn(false);
+ when(eventHandledService.isAlreadyHandled(eventId2)).thenReturn(false);
+
+ // act
+ productMetricsConsumer.consumeLikeEvents(records, acknowledgment);
+
+ // assert
+ verify(eventHandledService).isAlreadyHandled(eventId1);
+ verify(eventHandledService).isAlreadyHandled(eventId2);
+ verify(productMetricsService).incrementLikeCount(eq(productId), eq(4L));
+ verify(productMetricsService).decrementLikeCount(eq(productId), eq(5L));
+ verify(eventHandledService).markAsHandled(eventId1, "LikeAdded", "like-events");
+ verify(eventHandledService).markAsHandled(eventId2, "LikeRemoved", "like-events");
+ verify(acknowledgment, times(1)).acknowledge();
+ }
+
+ @DisplayName("개별 이벤트 처리 실패 시에도 배치 처리를 계속한다.")
+ @Test
+ void continuesProcessing_whenIndividualEventFails() {
+ // arrange
+ String eventId1 = "test-event-id-6";
+ String eventId2 = "test-event-id-7";
+ Long productId = 1L;
+ Long userId = 100L;
+
+ LikeEvent.LikeAdded validEvent = new LikeEvent.LikeAdded(userId, productId, LocalDateTime.now());
+ Object invalidEvent = "invalid-event";
+
+ Headers headers1 = new RecordHeaders();
+ headers1.add(new RecordHeader("eventId", eventId1.getBytes(StandardCharsets.UTF_8)));
+ headers1.add(new RecordHeader("version", "6".getBytes(StandardCharsets.UTF_8)));
+ Headers headers2 = new RecordHeaders();
+ headers2.add(new RecordHeader("eventId", eventId2.getBytes(StandardCharsets.UTF_8)));
+ headers2.add(new RecordHeader("version", "7".getBytes(StandardCharsets.UTF_8)));
+
+ when(eventHandledService.isAlreadyHandled(eventId1)).thenReturn(false);
+ when(eventHandledService.isAlreadyHandled(eventId2)).thenReturn(false);
+ doThrow(new RuntimeException("처리 실패"))
+ .when(productMetricsService).incrementLikeCount(any(), anyLong());
+
+ List> records = List.of(
+ new ConsumerRecord<>("like-events", 0, 0L, 0L, TimestampType.CREATE_TIME, 0, 0, "key", invalidEvent, headers1, Optional.empty()),
+ new ConsumerRecord<>("like-events", 0, 1L, 0L, TimestampType.CREATE_TIME, 0, 0, "key", validEvent, headers2, Optional.empty())
+ );
+
+ // act
+ productMetricsConsumer.consumeLikeEvents(records, acknowledgment);
+
+ // assert
+ verify(eventHandledService).isAlreadyHandled(eventId1);
+ verify(eventHandledService).isAlreadyHandled(eventId2);
+ verify(productMetricsService, atLeastOnce()).incrementLikeCount(any(), anyLong());
+ verify(acknowledgment).acknowledge();
+ }
+
+ @DisplayName("개별 이벤트 처리 실패 시에도 acknowledgment를 수행한다.")
+ @Test
+ void acknowledgesEvenWhenIndividualEventFails() {
+ // arrange
+ String eventId = "test-event-id-8";
+ Long productId = 1L;
+ Long userId = 100L;
+
+ LikeEvent.LikeAdded event = new LikeEvent.LikeAdded(userId, productId, LocalDateTime.now());
+
+ Headers headers = new RecordHeaders();
+ headers.add(new RecordHeader("eventId", eventId.getBytes(StandardCharsets.UTF_8)));
+ headers.add(new RecordHeader("version", "8".getBytes(StandardCharsets.UTF_8)));
+
+ // 서비스 호출 시 예외 발생
+ when(eventHandledService.isAlreadyHandled(eventId)).thenReturn(false);
+ doThrow(new RuntimeException("서비스 처리 실패"))
+ .when(productMetricsService).incrementLikeCount(eq(productId), anyLong());
+
+ List> records = List.of(
+ new ConsumerRecord<>("like-events", 0, 0L, 0L, TimestampType.CREATE_TIME, 0, 0, "key", event, headers, Optional.empty())
+ );
+
+ // act
+ productMetricsConsumer.consumeLikeEvents(records, acknowledgment);
+
+ // assert
+ // 개별 이벤트 실패는 내부 catch 블록에서 처리되고 계속 진행되므로 acknowledgment는 호출됨
+ verify(eventHandledService).isAlreadyHandled(eventId);
+ verify(productMetricsService).incrementLikeCount(eq(productId), anyLong());
+ // 예외 발생 시 markAsHandled는 호출되지 않음
+ verify(eventHandledService, never()).markAsHandled(any(), any(), any());
+ verify(acknowledgment).acknowledge();
+ }
+
+ @DisplayName("이미 처리된 이벤트는 스킵한다.")
+ @Test
+ void skipsAlreadyHandledEvent() {
+ // arrange
+ String eventId = "test-event-id";
+ Long productId = 1L;
+ Long userId = 100L;
+ LikeEvent.LikeAdded event = new LikeEvent.LikeAdded(userId, productId, LocalDateTime.now());
+
+ Headers headers = new RecordHeaders();
+ headers.add(new RecordHeader("eventId", eventId.getBytes(StandardCharsets.UTF_8)));
+
+ ConsumerRecord record = new ConsumerRecord<>(
+ "like-events", 0, 0L, 0L, TimestampType.CREATE_TIME, 0, 0, "key", event, headers, Optional.empty()
+ );
+ List> records = List.of(record);
+
+ when(eventHandledService.isAlreadyHandled(eventId)).thenReturn(true);
+
+ // act
+ productMetricsConsumer.consumeLikeEvents(records, acknowledgment);
+
+ // assert
+ verify(eventHandledService).isAlreadyHandled(eventId);
+ verify(productMetricsService, never()).incrementLikeCount(any(), anyLong());
+ verify(eventHandledService, never()).markAsHandled(any(), any(), any());
+ verify(acknowledgment).acknowledge();
+ }
+
+ @DisplayName("eventId가 없는 메시지는 건너뛴다.")
+ @Test
+ void skipsEventWithoutEventId() {
+ // arrange
+ Long productId = 1L;
+ Long userId = 100L;
+ LikeEvent.LikeAdded event = new LikeEvent.LikeAdded(userId, productId, LocalDateTime.now());
+
+ ConsumerRecord record = new ConsumerRecord<>(
+ "like-events", 0, 0L, "key", event
+ );
+ List> records = List.of(record);
+
+ // act
+ productMetricsConsumer.consumeLikeEvents(records, acknowledgment);
+
+ // assert
+ verify(eventHandledService, never()).isAlreadyHandled(any());
+ verify(productMetricsService, never()).incrementLikeCount(any(), anyLong());
+ verify(acknowledgment).acknowledge();
+ }
+
+ @DisplayName("동시성 상황에서 DataIntegrityViolationException이 발생하면 정상 처리로 간주한다.")
+ @Test
+ void handlesDataIntegrityViolationException() {
+ // arrange
+ String eventId = "test-event-id";
+ Long productId = 1L;
+ Long userId = 100L;
+ LikeEvent.LikeAdded event = new LikeEvent.LikeAdded(userId, productId, LocalDateTime.now());
+
+ Headers headers = new RecordHeaders();
+ headers.add(new RecordHeader("eventId", eventId.getBytes(StandardCharsets.UTF_8)));
+ headers.add(new RecordHeader("version", "9".getBytes(StandardCharsets.UTF_8)));
+
+ ConsumerRecord record = new ConsumerRecord<>(
+ "like-events", 0, 0L, 0L, TimestampType.CREATE_TIME, 0, 0, "key", event, headers, Optional.empty()
+ );
+ List> records = List.of(record);
+
+ when(eventHandledService.isAlreadyHandled(eventId)).thenReturn(false);
+ doThrow(new DataIntegrityViolationException("UNIQUE constraint violation"))
+ .when(eventHandledService).markAsHandled(eventId, "LikeAdded", "like-events");
+
+ // act
+ productMetricsConsumer.consumeLikeEvents(records, acknowledgment);
+
+ // assert
+ verify(eventHandledService).isAlreadyHandled(eventId);
+ verify(productMetricsService).incrementLikeCount(eq(productId), anyLong());
+ verify(eventHandledService).markAsHandled(eventId, "LikeAdded", "like-events");
+ verify(acknowledgment).acknowledge();
+ }
+
+ @DisplayName("중복 메시지 재전송 시 한 번만 처리되어 멱등성이 보장된다.")
+ @Test
+ void handlesDuplicateMessagesIdempotently() {
+ // arrange
+ String eventId = "duplicate-event-id";
+ Long productId = 1L;
+ Long userId = 100L;
+ Long eventVersion = 1L;
+ LikeEvent.LikeAdded event = new LikeEvent.LikeAdded(userId, productId, LocalDateTime.now());
+
+ Headers headers = new RecordHeaders();
+ headers.add(new RecordHeader("eventId", eventId.getBytes(StandardCharsets.UTF_8)));
+ headers.add(new RecordHeader("version", String.valueOf(eventVersion).getBytes(StandardCharsets.UTF_8)));
+
+ // 동일한 eventId를 가진 메시지 3개 생성
+ List> records = List.of(
+ new ConsumerRecord<>("like-events", 0, 0L, 0L, TimestampType.CREATE_TIME, 0, 0, "key", event, headers, Optional.empty()),
+ new ConsumerRecord<>("like-events", 0, 1L, 0L, TimestampType.CREATE_TIME, 0, 0, "key", event, headers, Optional.empty()),
+ new ConsumerRecord<>("like-events", 0, 2L, 0L, TimestampType.CREATE_TIME, 0, 0, "key", event, headers, Optional.empty())
+ );
+
+ // 첫 번째 메시지는 처리되지 않았으므로 false, 나머지는 이미 처리되었으므로 true
+ when(eventHandledService.isAlreadyHandled(eventId))
+ .thenReturn(false) // 첫 번째: 처리됨
+ .thenReturn(true) // 두 번째: 이미 처리됨 (스킵)
+ .thenReturn(true); // 세 번째: 이미 처리됨 (스킵)
+
+ // act
+ productMetricsConsumer.consumeLikeEvents(records, acknowledgment);
+
+ // assert
+ // isAlreadyHandled는 3번 호출됨 (각 메시지마다)
+ verify(eventHandledService, times(3)).isAlreadyHandled(eventId);
+
+ // incrementLikeCount는 한 번만 호출되어야 함 (첫 번째 메시지만 처리)
+ verify(productMetricsService, times(1)).incrementLikeCount(eq(productId), eq(eventVersion));
+
+ // markAsHandled는 한 번만 호출되어야 함 (첫 번째 메시지만 처리)
+ verify(eventHandledService, times(1)).markAsHandled(eventId, "LikeAdded", "like-events");
+
+ // acknowledgment는 한 번만 호출되어야 함 (배치 처리 완료)
+ verify(acknowledgment, times(1)).acknowledge();
+ }
+}
diff --git a/modules/kafka/src/main/java/com/loopers/confg/kafka/KafkaConfig.java b/modules/kafka/src/main/java/com/loopers/confg/kafka/KafkaConfig.java
index a73842775..33222efb1 100644
--- a/modules/kafka/src/main/java/com/loopers/confg/kafka/KafkaConfig.java
+++ b/modules/kafka/src/main/java/com/loopers/confg/kafka/KafkaConfig.java
@@ -1,6 +1,7 @@
package com.loopers.confg.kafka;
import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
@@ -8,6 +9,7 @@
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.*;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.support.converter.BatchMessagingMessageConverter;
@@ -30,19 +32,19 @@ public class KafkaConfig {
public static final int MAX_POLL_INTERVAL_MS = 2 * 60 * 1000; // max poll interval = 2m
@Bean
- public ProducerFactory producerFactory(KafkaProperties kafkaProperties) {
+ public ProducerFactory producerFactory(KafkaProperties kafkaProperties) {
Map props = new HashMap<>(kafkaProperties.buildProducerProperties());
return new DefaultKafkaProducerFactory<>(props);
}
@Bean
- public ConsumerFactory consumerFactory(KafkaProperties kafkaProperties) {
+ public ConsumerFactory consumerFactory(KafkaProperties kafkaProperties) {
Map props = new HashMap<>(kafkaProperties.buildConsumerProperties());
return new DefaultKafkaConsumerFactory<>(props);
}
@Bean
- public KafkaTemplate kafkaTemplate(ProducerFactory producerFactory) {
+ public KafkaTemplate kafkaTemplate(ProducerFactory producerFactory) {
return new KafkaTemplate<>(producerFactory);
}
@@ -52,7 +54,7 @@ public ByteArrayJsonMessageConverter jsonMessageConverter(ObjectMapper objectMap
}
@Bean(name = BATCH_LISTENER)
- public ConcurrentKafkaListenerContainerFactory defaultBatchListenerContainerFactory(
+ public ConcurrentKafkaListenerContainerFactory defaultBatchListenerContainerFactory(
KafkaProperties kafkaProperties,
ByteArrayJsonMessageConverter converter
) {
@@ -64,7 +66,7 @@ public ConcurrentKafkaListenerContainerFactory defaultBatchListe
consumerConfig.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, HEARTBEAT_INTERVAL_MS);
consumerConfig.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, MAX_POLL_INTERVAL_MS);
- ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory<>();
+ ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfig));
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL); // 수동 커밋
factory.setBatchMessageConverter(new BatchMessagingMessageConverter(converter));
@@ -72,4 +74,94 @@ public ConcurrentKafkaListenerContainerFactory defaultBatchListe
factory.setBatchListener(true);
return factory;
}
+
+ /**
+ * Like 도메인 이벤트 토픽.
+ *
+ * 파티션 키: productId (상품별 좋아요 수 집계를 위해)
+ *
+ */
+ @Bean
+ public NewTopic likeEventsTopic() {
+ return TopicBuilder.name("like-events")
+ .partitions(3)
+ .replicas(1)
+ .config("min.insync.replicas", "1")
+ .build();
+ }
+
+ /**
+ * Product 도메인 이벤트 토픽.
+ *
+ * 파티션 키: productId (상품별 재고 관리를 위해)
+ *
+ */
+ @Bean
+ public NewTopic productEventsTopic() {
+ return TopicBuilder.name("product-events")
+ .partitions(3)
+ .replicas(1)
+ .config("min.insync.replicas", "1")
+ .build();
+ }
+
+ /**
+ * Order 도메인 이벤트 토픽.
+ *
+ * 파티션 키: orderId (주문별 이벤트 순서 보장을 위해)
+ *
+ */
+ @Bean
+ public NewTopic orderEventsTopic() {
+ return TopicBuilder.name("order-events")
+ .partitions(3)
+ .replicas(1)
+ .config("min.insync.replicas", "1")
+ .build();
+ }
+
+ /**
+ * Payment 도메인 이벤트 토픽.
+ *
+ * 파티션 키: orderId (주문별 결제 처리 순서 보장을 위해)
+ *
+ */
+ @Bean
+ public NewTopic paymentEventsTopic() {
+ return TopicBuilder.name("payment-events")
+ .partitions(3)
+ .replicas(1)
+ .config("min.insync.replicas", "1")
+ .build();
+ }
+
+ /**
+ * Coupon 도메인 이벤트 토픽.
+ *
+ * 파티션 키: orderId (주문별 쿠폰 할인 적용 순서 보장을 위해)
+ *
+ */
+ @Bean
+ public NewTopic couponEventsTopic() {
+ return TopicBuilder.name("coupon-events")
+ .partitions(3)
+ .replicas(1)
+ .config("min.insync.replicas", "1")
+ .build();
+ }
+
+ /**
+ * User 도메인 이벤트 토픽.
+ *
+ * 파티션 키: userId (사용자별 포인트 처리 순서 보장을 위해)
+ *
+ */
+ @Bean
+ public NewTopic userEventsTopic() {
+ return TopicBuilder.name("user-events")
+ .partitions(3)
+ .replicas(1)
+ .config("min.insync.replicas", "1")
+ .build();
+ }
}
diff --git a/modules/kafka/src/main/resources/kafka.yml b/modules/kafka/src/main/resources/kafka.yml
index 9609dbf85..a2a73417b 100644
--- a/modules/kafka/src/main/resources/kafka.yml
+++ b/modules/kafka/src/main/resources/kafka.yml
@@ -15,6 +15,10 @@ spring:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
retries: 3
+ properties:
+ acks: all # 모든 리플리카에 쓰기 확인 (At Least Once 보장)
+ enable.idempotence: true # 중복 방지 (At Least Once 보장)
+ max.in.flight.requests.per.connection: 5 # idempotence=true일 때 필수
consumer:
group-id: loopers-default-consumer
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
@@ -29,11 +33,13 @@ spring.config.activate.on-profile: local, test
spring:
kafka:
- bootstrap-servers: localhost:19092
+ # Testcontainers를 사용하는 경우 BOOTSTRAP_SERVERS가 자동으로 설정됨
+ # 로컬 개발 환경에서는 localhost:19092 사용
+ bootstrap-servers: ${BOOTSTRAP_SERVERS:localhost:19092}
admin:
properties:
- bootstrap.servers: kafka:9092
-
+ bootstrap.servers: ${BOOTSTRAP_SERVERS:localhost:19092}
+ auto-create: true
---
spring.config.activate.on-profile: dev
diff --git a/modules/kafka/src/testFixtures/java/com/loopers/testcontainers/KafkaTestContainersConfig.java b/modules/kafka/src/testFixtures/java/com/loopers/testcontainers/KafkaTestContainersConfig.java
new file mode 100644
index 000000000..4500d3b0b
--- /dev/null
+++ b/modules/kafka/src/testFixtures/java/com/loopers/testcontainers/KafkaTestContainersConfig.java
@@ -0,0 +1,35 @@
+package com.loopers.testcontainers;
+
+import org.springframework.context.annotation.Configuration;
+import org.testcontainers.kafka.ConfluentKafkaContainer;
+
+/**
+ * Kafka Testcontainers 설정.
+ *
+ * 테스트 실행 시 자동으로 Kafka 컨테이너를 시작하고,
+ * Spring Boot의 Kafka 설정에 동적으로 포트를 주입합니다.
+ *
+ *
+ * 동작 방식:
+ * 1. Kafka 컨테이너를 시작
+ * 2. 동적으로 할당된 포트를 System Property로 설정
+ * 3. kafka.yml의 ${BOOTSTRAP_SERVERS}가 이 값을 사용
+ *
+ */
+@Configuration
+public class KafkaTestContainersConfig {
+
+ private static final ConfluentKafkaContainer kafkaContainer;
+
+ static {
+ // Kafka 컨테이너 생성 및 시작
+ // ConfluentKafkaContainer는 confluentinc/cp-kafka 이미지를 사용
+ kafkaContainer = new ConfluentKafkaContainer("confluentinc/cp-kafka:7.5.0");
+ kafkaContainer.start();
+
+ // Spring Boot의 Kafka 설정에 동적으로 포트 주입
+ // kafka.yml의 ${BOOTSTRAP_SERVERS}가 이 값을 사용
+ String bootstrapServers = kafkaContainer.getBootstrapServers();
+ System.setProperty("BOOTSTRAP_SERVERS", bootstrapServers);
+ }
+}
diff --git a/modules/kafka/src/testFixtures/java/com/loopers/utils/KafkaCleanUp.java b/modules/kafka/src/testFixtures/java/com/loopers/utils/KafkaCleanUp.java
new file mode 100644
index 000000000..51207364a
--- /dev/null
+++ b/modules/kafka/src/testFixtures/java/com/loopers/utils/KafkaCleanUp.java
@@ -0,0 +1,194 @@
+package com.loopers.utils;
+
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.DeleteConsumerGroupsResult;
+import org.apache.kafka.clients.admin.DeleteTopicsResult;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.springframework.kafka.config.TopicBuilder;
+import org.springframework.kafka.core.KafkaAdmin;
+import org.springframework.stereotype.Component;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Kafka 테스트 정리 유틸리티.
+ *
+ * 테스트 간 Kafka 메시지 격리를 위해 토픽을 삭제하고 재생성합니다.
+ *
+ *
+ * 사용 방법:
+ *
+ * - 통합 테스트에서 `@BeforeEach` 또는 `@AfterEach`에서 호출하여 테스트 간 격리 보장
+ * - 단위 테스트는 Mock을 사용하므로 불필요
+ *
+ *
+ *
+ * 주의:
+ * 프로덕션 환경에서는 사용하지 마세요. 테스트 환경에서만 사용해야 합니다.
+ *
+ */
+@Component
+public class KafkaCleanUp {
+
+ private static final List TEST_TOPICS = List.of(
+ "like-events",
+ "order-events",
+ "product-events",
+ "payment-events",
+ "coupon-events",
+ "user-events"
+ );
+
+ private final KafkaAdmin kafkaAdmin;
+
+ public KafkaCleanUp(KafkaAdmin kafkaAdmin) {
+ this.kafkaAdmin = kafkaAdmin;
+ }
+
+ /**
+ * 테스트용 토픽의 모든 메시지를 삭제합니다.
+ *
+ * 토픽을 삭제하고 재생성하여 모든 메시지를 제거합니다.
+ *
+ *
+ * 주의: 프로덕션 환경에서는 사용하지 마세요.
+ *
+ */
+ public void deleteAllTestTopics() {
+ try (AdminClient adminClient = createAdminClient()) {
+ // 존재하는 토픽만 삭제
+ Set existingTopics = adminClient.listTopics()
+ .names()
+ .get(5, TimeUnit.SECONDS);
+
+ List topicsToDelete = TEST_TOPICS.stream()
+ .filter(existingTopics::contains)
+ .toList();
+
+ if (topicsToDelete.isEmpty()) {
+ return;
+ }
+
+ // 토픽 삭제 (모든 메시지 제거)
+ DeleteTopicsResult deleteResult = adminClient.deleteTopics(topicsToDelete);
+ deleteResult.all().get(10, TimeUnit.SECONDS);
+
+ // 토픽 삭제 후 재생성 대기 (Kafka가 토픽 삭제를 완료할 때까지)
+ Thread.sleep(1000);
+ } catch (Exception e) {
+ // 토픽이 없거나 이미 삭제된 경우 무시
+ // 테스트 환경에서는 토픽이 없을 수 있음
+ }
+ }
+
+ /**
+ * 테스트용 토픽을 재생성합니다.
+ *
+ * 삭제된 토픽을 원래 설정으로 재생성합니다.
+ *
+ */
+ public void recreateTestTopics() {
+ try (AdminClient adminClient = createAdminClient()) {
+ for (String topicName : TEST_TOPICS) {
+ try {
+ // 토픽이 이미 존재하는지 확인
+ adminClient.describeTopics(Collections.singletonList(topicName))
+ .allTopicNames()
+ .get(2, TimeUnit.SECONDS);
+ // 이미 존재하면 스킵
+ continue;
+ } catch (Exception e) {
+ // 토픽이 없으면 생성
+ }
+
+ // 토픽 생성
+ NewTopic newTopic = TopicBuilder.name(topicName)
+ .partitions(3)
+ .replicas(1)
+ .config("min.insync.replicas", "1")
+ .build();
+
+ adminClient.createTopics(Collections.singletonList(newTopic))
+ .all()
+ .get(5, TimeUnit.SECONDS);
+ }
+ } catch (Exception e) {
+ // 토픽 생성 실패는 무시 (이미 존재할 수 있음)
+ }
+ }
+
+ /**
+ * 테스트용 토픽을 삭제하고 재생성합니다.
+ *
+ * 모든 메시지를 제거하고 깨끗한 상태로 시작합니다.
+ *
+ */
+ public void resetAllTestTopics() {
+ deleteAllTestTopics();
+ recreateTestTopics();
+ }
+
+ /**
+ * 모든 Consumer Group을 삭제하여 offset을 리셋합니다.
+ *
+ * 테스트 간 격리를 위해 사용합니다.
+ *
+ *
+ * 주의: 모든 Consumer Group을 삭제하므로 프로덕션 환경에서는 사용하지 마세요.
+ *
+ */
+ public void resetAllConsumerGroups() {
+ try (AdminClient adminClient = createAdminClient()) {
+ // 모든 Consumer Group 목록 조회
+ Set consumerGroups = adminClient.listConsumerGroups()
+ .all()
+ .get(5, TimeUnit.SECONDS)
+ .stream()
+ .map(group -> group.groupId())
+ .collect(java.util.stream.Collectors.toSet());
+
+ if (consumerGroups.isEmpty()) {
+ return;
+ }
+
+ // Consumer Group 삭제 (offset 리셋)
+ DeleteConsumerGroupsResult deleteResult = adminClient.deleteConsumerGroups(consumerGroups);
+ deleteResult.all().get(5, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ // Consumer Group이 없거나 이미 삭제된 경우 무시
+ // 테스트 환경에서는 Consumer Group이 없을 수 있음
+ }
+ }
+
+ /**
+ * 특정 Consumer Group을 삭제합니다.
+ *
+ * @param groupId 삭제할 Consumer Group ID
+ */
+ public void resetConsumerGroup(String groupId) {
+ try (AdminClient adminClient = createAdminClient()) {
+ DeleteConsumerGroupsResult deleteResult = adminClient.deleteConsumerGroups(
+ Collections.singletonList(groupId)
+ );
+ deleteResult.all().get(5, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ // Consumer Group이 없거나 이미 삭제된 경우 무시
+ }
+ }
+
+ /**
+ * AdminClient를 생성합니다.
+ */
+ private AdminClient createAdminClient() {
+ Properties props = new Properties();
+ Object bootstrapServers = kafkaAdmin.getConfigurationProperties()
+ .getOrDefault(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:19092");
+ props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+ return AdminClient.create(props);
+ }
+}