Round8

[sylee6529]

📌 Summary

Kafka 기반 이벤트 파이프라인을 구축하여 At Least Once Producer와 At Most Once Consumer 패턴을 구현했습니다.

[commerce-api (Producer)]
     ↓ Spring Events
[Transactional Outbox Pattern]
     ↓ Polling (5초 주기)
[Kafka Cluster] (6개 토픽, Partition Key 기반)
     ↓ Manual Ack
[commerce-streamer (Consumer)]
     ↓ Batch Processing
[ProductMetrics 집계 + 멱등성 보장]

1. Producer (commerce-api) - At Least Once 보장

• Transactional Outbox Pattern
  • OutboxEvent 엔티티: DB 기반 이벤트 큐 (PENDING → PUBLISHED/FAILED)
  • KafkaOutboxEventListener: @TransactionalEventListener(BEFORE_COMMIT)로 원자성 보장
  • OutboxEventPoller: 5초마다 PENDING 이벤트를 Kafka로 발행
  • OutboxEventCleaner: 7일 지난 PUBLISHED, 30일 지난 FAILED 이벤트 자동 삭제
• Kafka 설정
  • acks=all: 모든 in-sync replica 확인
  • idempotence=true: 중복 발행 방지
  • max.in.flight.requests.per.connection=5: 순서 보장

2. Consumer (commerce-streamer) - At Most Once 보장

• 멱등성 보장
  • event_handled 테이블: eventId(PK) 기반 중복 처리 방지
  • 타임스탬프 기반 Out-of-Order 처리: ProductMetrics.lastUpdated 비교
• ProductMetrics 집계
  • 좋아요 수 (likeCount)
  • 조회 수 (viewCount)
  • 판매 수량 (salesCount)
  • 판매 금액 (salesAmount)
  • 낙관적 락 (@Version) 적용
• 에러 처리 (DLQ 패턴)
  • DlqMessage 테이블: 복구 불가능한 에러 저장
  • RetryTracker: 메시지별 재시도 횟수 추적 (최대 3회)
  • 복구 불가능한 에러 (JSON 파싱 실패 등) → 즉시 DLQ 전송
  • 일시적 에러 → 3회 재시도 후 DLQ 전송
• Consumer 설정
  • Manual Ack: 모든 메시지 처리 성공 후에만 offset commit
  • Batch Listener: 한 번에 여러 메시지 처리 (성능 최적화)

3. 이벤트 발행 지점

• ProductFacade.getProductDetail(): 상품 조회 → ProductViewedEvent
• LikeFacade.like(): 좋아요 → ProductLikedEvent
• LikeFacade.unlike(): 좋아요 취소 → ProductUnlikedEvent
• OrderPlacementService: 주문 완료 → OrderCompletedEvent

---

💬 Review Points

1. “왜 이벤트 핸들링 테이블과 로그 테이블을 분리하는가?"에 대해 제가 이해한 내용은 다음과 같습니다.

로그 테이블은 감사(Audit) 및 트래킹하기 위해 사용합니다. 언제 어떤 이벤트가 발행되었는지 추적하고, 이벤트 발행 이력을 통해 문제 분석할 수 있습니다.
이벤트 핸들링 테이블은 멱등성 보장 목적으로 사용합니다. Consumer가 같은 이벤트를 중복 처리하지 않도록 방지합니다. 단순히 처리했는지 여부만 체크하는 용도로써 사용된다고 이해했습니다.
부가적인 이유로는 책임 분리가 되기도 하고, 각각의 생명주기가 다를 수도 있을 것 같습니다.

2. 여러 Customer 인스턴스가 서로 다른 토픽에서 같은 값을 업데이트하는 케이스가 만약 있다면, 낙관적 락을 적용하기도 하나요?
3. Outbox/EventHandled 테이블은 장기적으로는 삭제전략이 필요할 것 같습니다. 주로 어떤 방식으로 관리(보관, 삭제 주기 등)하시나요?

---

✅ Checklist

Must-Have

• Kafka 설정 및 의존성 추가
• Event Pipeline 구축 (Producer → Kafka → Consumer)
• At Least Once Producer (Transactional Outbox Pattern)
• At Most Once Consumer (event_handled 멱등성)

Producer

• 도메인(애플리케이션) 이벤트 설계 (6개 이벤트)
• Producer 앱에서 도메인 이벤트 발행 (6개 토픽)
• PartitionKey 기반의 이벤트 순서 보장 (productId/orderNo)
• 메세지 발행 실패 시 재시도 메커니즘 (OutboxEventPoller, 최대 3회)

Consumer

• Consumer가 Metrics 집계 처리 (좋아요/조회수/판매량)
• event_handled 테이블을 통한 멱등 처리 구현
• 재고 소진 시 상품 캐시 갱신 (CacheInvalidationService)
• 중복 메세지 재전송 테스트 → 멱등성 검증 완료

Nice-to-Have

• 상품별 유저 이벤트 집계 테이블 (product_metrics)
• 메세지 배치 처리 (Batch Listener)
• DLQ 구현 (Dead Letter Queue)

테스트

• 멱등성 테스트
• DLQ Publisher 테스트

Summary by CodeRabbit

릴리스 노트

• 새로운 기능

  • 제품 조회 추적 기능 추가
  • 주문 완료 시 상품 상세 정보 포함
  • 제품 좋아요, 조회, 판매량 메트릭 수집 및 집계
  • 이벤트 중복 처리 방지를 위한 멱등성 보장

• 인프라 개선

  • Kafka 기반 이벤트 스트리밍 통합
  • 실패한 메시지 처리를 위한 Dead Letter Queue 추가
  • 재시도 로직 및 오류 추적 개선
  • 재고 소진 시 캐시 자동 무효화

_{✏️ Tip: You can customize this high-level summary in your review settings.}

[coderabbitai]

Important

Review skipped

Auto incremental reviews are disabled on this repository.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Walkthrough

이 PR은 Kafka 기반의 이벤트 기반 아키텍처를 도입하여 상거래 시스템에 비동기 이벤트 처리를 구현합니다. Outbox 패턴, 이벤트 라우팅, 메시지 큐잉, 메트릭 수집, DLQ 처리 기능을 추가하며 상품 조회, 주문 완료 등 주요 비즈니스 이벤트를 Kafka로 발행합니다.

Changes

Cohort / File(s)	Summary
Kafka 기반 구조 apps/commerce-api/src/main/java/com/loopers/infrastructure/kafka/KafkaTopicRouter.java, KafkaEventEnvelope.java	이벤트 타입을 Kafka 토픽명으로 매핑하는 라우터와 메타데이터 포함 이벤트 엔벨로프 레코드 추가
Outbox 패턴 구현 apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEvent.java, OutboxEventWriter.java, OutboxEventRepository.java, OutboxEventPoller.java, OutboxEventCleaner.java	트랜잭셔널 아웃박스 엔티티, 레포지토리, 폴러(5초 간격), 재시도 로직(30초 간격), 정기적 정리 작업 구현
이벤트 정의 apps/commerce-api/src/main/java/com/loopers/application/event/product/ProductViewedEvent.java, apps/commerce-api/src/main/java/com/loopers/application/event/order/OrderCompletedEvent.java (+ commerce-streamer 동일 파일)	상품 조회, 주문 완료 등록, 상품 좋아요/싫어요 이벤트 레코드 추가
이벤트 리스너 apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/KafkaOutboxEventListener.java, apps/commerce-api/src/main/java/com/loopers/application/event/listener/OrderStatusEventListener.java	도메인 이벤트를 아웃박스에 기록하는 리스너 및 주문 완료 이벤트에 상품 항목 정보 추가
메트릭 집계 서비스 apps/commerce-streamer/src/main/java/com/loopers/application/metrics/MetricsAggregationService.java, apps/commerce-streamer/src/main/java/com/loopers/domain/metrics/ProductMetrics.java, ProductMetricsRepository.java	이벤트 핸들링 기반 상품 메트릭(좋아요, 조회, 판매) 업데이트 및 낙관적 잠금 적용
DLQ 및 재시도 관리 apps/commerce-streamer/src/main/java/com/loopers/domain/dlq/DlqMessage.java, DlqMessageRepository.java, apps/commerce-streamer/src/main/java/com/loopers/infrastructure/kafka/DlqPublisher.java, RetryTracker.java	실패한 메시지용 Dead Letter Queue 엔티티 및 재시도 추적 메커니즘
Kafka 컨슈머 apps/commerce-streamer/src/main/java/com/loopers/interfaces/consumer/ProductMetricsConsumer.java	배치 기반 Kafka 컨슈머로 메트릭 집계 서비스 위임 및 DLQ/재시도 처리
상품 및 주문 서비스 업데이트 apps/commerce-api/src/main/java/com/loopers/application/product/ProductFacade.java, apps/commerce-api/src/main/java/com/loopers/domain/product/repository/ProductRepository.java, ProductRepositoryImpl.java, apps/commerce-api/src/main/java/com/loopers/domain/order/service/OrderPlacementService.java	ProductFacade에 상품 조회 이벤트 발행 및 브랜드ID 조회 추가, 주문 시 재고 소진 캐시 무효화 추가
캐시 관리 apps/commerce-api/src/main/java/com/loopers/infrastructure/cache/CacheInvalidationService.java	재고 소진 시 캐시 무효화 메서드 추가
설정 및 의존성 apps/commerce-api/build.gradle.kts, apps/commerce-api/src/main/resources/application.yml, modules/kafka/src/main/resources/kafka.yml	Kafka 모듈 의존성 추가, Kafka 설정 파일 임포트, Producer 설정(멱등성, acks=all) 적용
테스트 및 구성 apps/commerce-streamer/src/test/java/com/loopers/config/TestConfig.java, apps/commerce-streamer/src/test/resources/application-test.yml, apps/commerce-api/src/test/java/com/loopers/infrastructure/kafka/KafkaConnectionTest.java, apps/commerce-streamer/src/test/java/com/loopers/application/metrics/MetricsAggregationServiceIdempotencyTest.java, DlqPublisherTest.java, RetryTrackerTest.java	Kafka 자동 구성 비활성화, 테스트용 MySQL 컨테이너 설정, Kafka 연결 및 메트릭 멱등성 테스트 추가

Sequence Diagram

sequenceDiagram
    participant App as Application<br/>(ProductFacade)
    participant Outbox as Outbox<br/>Repository
    participant Poller as Outbox<br/>Poller
    participant Kafka as Kafka<br/>Broker
    participant Consumer as Kafka<br/>Consumer
    participant Metrics as Metrics<br/>Service
    participant DLQ as DLQ<br/>Storage

    App->>Outbox: 1. Event Handler<br/>(ProductViewed)
    Note over App,Outbox: BEFORE_COMMIT<br/>(same transaction)
    Outbox->>Outbox: 2. Create OutboxEvent<br/>(status=PENDING)

    rect rgb(200, 220, 255)
    Note over Poller,Kafka: Outbox Poller (5s interval)
    Poller->>Outbox: 3. Poll PENDING<br/>events
    Poller->>Kafka: 4. Publish to Topic<br/>(KafkaEventEnvelope)
    Kafka-->>Poller: 5. Send Success
    Poller->>Outbox: 6. Mark PUBLISHED
    end

    rect rgb(200, 255, 220)
    Note over Consumer,Metrics: Event Processing
    Kafka->>Consumer: 7. Consume batch<br/>events
    Consumer->>Metrics: 8. Delegate to<br/>handler
    Metrics->>Metrics: 9. Check idempotency<br/>(EventHandled)
    Metrics->>Metrics: 10. Update metrics<br/>(likeCount, viewCount)
    Metrics->>Metrics: 11. Record as handled
    Consumer->>Consumer: 12. Batch commit
    end

    rect rgb(255, 220, 200)
    Note over Consumer,DLQ: Error Handling
    Consumer--xMetrics: Error (e.g.,<br/>JsonProcessingException)
    Consumer->>DLQ: 13. Determine DLQ<br/>vs Retry
    alt DLQ Candidate
        Consumer->>DLQ: 14. Publish to DLQ
        DLQ->>DLQ: 15. Store DlqMessage
    else Retryable
        Consumer->>Consumer: 16. Track retry<br/>(RetryTracker)
        Note over Consumer: Retry up to<br/>3 attempts
    end
    end

    rect rgb(220, 200, 255)
    Note over Poller: Cleanup Tasks
    Poller->>Outbox: 17. Clean old<br/>PUBLISHED (7 days)
    Poller->>Outbox: 18. Clean old FAILED<br/>(30 days, retries>3)
    Poller->>Outbox: 19. Log statistics<br/>(hourly)
    end

Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~60 minutes

주의가 필요한 영역:

• OutboxEventPoller.java: 폴링/재시도 로직 및 동시성 제어 검토 필요 (KafkaTemplate 동기 전송, 트랜잭션 경계)
• MetricsAggregationService.java: 멱등성 처리(EventHandled 체크), 이벤트 순서 검증(isEventOutdated) 로직 확인
• ProductMetricsConsumer.java: 배치 처리 중 부분 실패 케이스, DLQ 라우팅 조건, 재시도 카운팅 정확성
• ProductFacade.java: 새로운 의존성 추가(ProductRepository, ApplicationEventPublisher)로 인한 constructor 변경 영향도 및 이벤트 발행 타이밍
• OrderStatusEventListener.java: OrderCompletedEvent 페이로드 변경(items 추가)에 따른 하위 시스템 호환성
• Outbox 패턴 구현: 트랜잭션 경계(BEFORE_COMMIT), 멱등성, 중복 처리 로직 전반

Possibly related PRs

• [volume-7] Decoupling with Event #172: Outbox 패턴, Kafka 통합, 이벤트 엔벨로프, 컨슈머 및 DLQ 구현 등 이벤트 드리븐 아키텍처 관련 변경사항이 동일함
• [volume-7] Decoupling with Event #183: ProductViewedEvent, OrderCompletedEvent 포함 아이템 상세정보 등 동일한 도메인 이벤트 타입 수정
• [volume-6] 외부 시스템 장애 및 지연 대응  #154: ProductFacade 클래스 수정(이벤트 발행 및 브랜드 조회), 메시징/인프라 의존성 변경이 겹침

Suggested labels

enhancement, event-driven-architecture, kafka-integration, outbox-pattern

Poem

🐰 이벤트가 흐르는 강이 되어,
Outbox 패턴이 주문을 지켜내고,
Kafka 메시지 버스를 타 멀리 날아,
메트릭 수집은 꾸준히 쌓여가고,
DLQ는 실패의 이야기를 품네! 🚀

Pre-merge checks and finishing touches

❌ Failed checks (2 warnings)

Check name	Status	Explanation	Resolution
Title check	⚠️ Warning	제목이 너무 모호하고 구체적이지 않습니다. 'Round8'은 변경사항의 주요 내용을 전혀 설명하지 않으며, 팀원이 커밋 히스토리를 훑어볼 때 이 PR의 목적을 알 수 없습니다.	PR 제목을 'Kafka 기반 이벤트 파이프라인 구축: Transactional Outbox 및 멱등성 Consumer 구현' 같이 변경하여 주요 변경사항을 명확히 반영해주세요.
Docstring Coverage	⚠️ Warning	Docstring coverage is 63.21% which is insufficient. The required threshold is 80.00%.	You can run @coderabbitai generate docstrings to improve docstring coverage.

✅ Passed checks (1 passed)

Check name	Status	Explanation
Description check	✅ Passed	PR 설명이 템플릿 구조를 모두 충족하며, 상세한 아키텍처 다이어그램, 구현 내용, 리뷰 포인트, 체크리스트가 완벽하게 작성되어 있습니다.

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 13

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

modules/kafka/src/main/resources/kafka.yml (1)

26-27: Consumer 속성도 브래킷 표기법이 필요합니다.

Producer 속성과 동일하게, consumer properties 맵 내의 enable-auto-commit도 브래킷 표기법을 사용해야 합니다.

🔎 제안된 수정

      properties:
-       enable-auto-commit: false
+       "[enable.auto.commit]": false

🧹 Nitpick comments (32)

apps/commerce-streamer/src/main/java/com/loopers/infrastructure/kafka/event/KafkaEventEnvelope.java (1)

12-41: 이벤트 Envelope 설계가 적절합니다.

불변 레코드 구조와 ZonedDateTime 사용은 분산 시스템의 이벤트 데이터에 적합합니다. 몇 가지 개선 사항을 제안합니다:

1. Compact constructor에서 유효성 검증 추가 고려: eventId, eventType 등 필수 필드에 대한 null/빈 문자열 체크를 추가하면 런타임 오류를 조기에 발견할 수 있습니다.

2. partitionKey 문서 명확화: 현재 주석에는 "디버깅 및 로깅용"으로만 설명되어 있으나, PR 설명에 따르면 productId/orderNo를 기준으로 파티션 키 순서 보장이 이루어집니다. 실제로 Kafka 파티셔닝에도 사용된다면 문서를 업데이트하는 것이 좋습니다.

🔎 선택적 개선안: Compact constructor 예시

 public record KafkaEventEnvelope<T>(
     String eventId,
     String eventType,
     String partitionKey,
     T payload,
     ZonedDateTime occurredAt
-) {
+) {
+    public KafkaEventEnvelope {
+        if (eventId == null || eventId.isBlank()) {
+            throw new IllegalArgumentException("eventId must not be null or blank");
+        }
+        if (eventType == null || eventType.isBlank()) {
+            throw new IllegalArgumentException("eventType must not be null or blank");
+        }
+        if (occurredAt == null) {
+            throw new IllegalArgumentException("occurredAt must not be null");
+        }
+    }
 }

apps/commerce-streamer/src/test/resources/application-test.yml (1)

15-18: Kafka 로깅 설정 정리 고려 (선택사항)

Kafka가 비활성화되어 있는데 org.springframework.kafka를 DEBUG 레벨로 설정한 것은 다소 불일치해 보입니다. 특정 테스트에서 Kafka를 활성화하는 경우가 있다면 유용할 수 있지만, 그렇지 않다면 이 로깅 설정을 제거하는 것을 고려해보세요.

apps/commerce-streamer/src/main/java/com/loopers/confg/kafka/KafkaConfig.java (1)

22-34: 배치 처리 성능 최적화를 위해 설정 추가를 고려하세요.

현재 구성은 MANUAL ack와 배치 리스너를 올바르게 설정했습니다. Spring Kafka 2.3 이상에서는 enable.auto.commit이 자동으로 false로 설정되므로 별도 설정이 필요 없습니다.

프로덕션 환경의 성능 최적화를 위해 다음 속성 추가를 권장합니다:

• max.poll.records: 배치당 최대 레코드 수 제한 (메모리 관리, 기본값 500)
• concurrency: 동시 처리 스레드 수 (처리량 향상, 파티션 수와 일치 권장)
• max.poll.interval.ms: 배치 처리 시간 초과 방지 (기본값 300000ms)

예시:

factory.setConcurrency(3);  // 환경에 맞게 조정

또는 application.yml에서:

spring:
  kafka:
    consumer:
      max-poll-records: 500
      properties:
        "[max.poll.interval.ms]": 300000
    listener:
      concurrency: 3

apps/commerce-api/src/test/java/com/loopers/infrastructure/kafka/KafkaConnectionTest.java (3)

50-53: 테스트에서 System.out.println 대신 로깅을 사용하세요.

테스트 출력은 System.out.println 대신 적절한 로깅 프레임워크(예: SLF4J)를 사용하거나 제거하는 것이 좋습니다. 테스트 로그는 테스트 프레임워크의 리포팅 메커니즘을 통해 관리되어야 합니다.

🔎 개선 제안

-        System.out.println("✅ Kafka Producer 연결 성공!");
-        System.out.println("Topic: " + result.getRecordMetadata().topic());
-        System.out.println("Partition: " + result.getRecordMetadata().partition());
-        System.out.println("Offset: " + result.getRecordMetadata().offset());
+        // 테스트 성공 시 출력은 assertion만으로 충분하며, 필요시 로깅 프레임워크 사용

---

45-48: 메시지 키와 값에 대한 검증을 추가하세요.

현재 테스트는 토픽 이름만 확인하고 있습니다. 메시지의 키와 값이 올바르게 전송되었는지 검증하면 테스트가 더 견고해집니다.

🔎 개선 제안

         // then
         SendResult<String, Object> result = future.get(10, TimeUnit.SECONDS);
         assertThat(result).isNotNull();
         assertThat(result.getRecordMetadata().topic()).isEqualTo(topic);
+        assertThat(result.getProducerRecord().key()).isEqualTo(key);
+        assertThat(result.getProducerRecord().value()).isEqualTo(message);

---

67-70: 테스트에서 System.out.println 대신 로깅을 사용하세요.

테스트 출력은 System.out.println 대신 적절한 로깅 프레임워크를 사용하거나 제거하는 것이 좋습니다.

🔎 개선 제안

-        System.out.println("✅ Producer 설정 확인:");
-        System.out.println("acks: " + configs.get("acks"));
-        System.out.println("enable.idempotence: " + configs.get("enable.idempotence"));
-        System.out.println("retries: " + configs.get("retries"));
+        // 설정 확인은 assertion만으로 충분하며, 필요시 로깅 프레임워크 사용

apps/commerce-api/src/main/java/com/loopers/domain/order/service/OrderPlacementService.java (2)

14-14: 아키텍처 개선: 도메인 이벤트를 통한 캐시 무효화 분리 고려.

현재 도메인 서비스(OrderPlacementService)가 인프라스트럭처 레이어(CacheInvalidationService)에 직접 의존하고 있습니다.

이 PR에서 이미 이벤트 기반 아키텍처를 구현하고 있으므로, 재고 소진 시 StockDepletedEvent를 발행하고 별도의 이벤트 리스너에서 캐시 무효화를 처리하는 방식으로 분리할 수 있습니다. 이렇게 하면 레이어 간 의존성이 깔끔해집니다.

Also applies to: 33-33

---

90-95: 재고 확인과 캐시 무효화 사이의 경쟁 조건.

decreaseStock과 getStockQuantity 호출 사이에 다른 트랜잭션이 재고를 변경할 수 있습니다. 다만 캐시 무효화는 eventually consistent 특성이므로, 최악의 경우 중복 무효화가 발생하거나 드물게 무효화가 누락될 수 있지만 기능상 문제는 없습니다.

더 정확한 처리를 원한다면 decreaseStock이 갱신된 재고 수량을 반환하도록 수정하거나, 단일 쿼리로 재고 감소와 잔여 수량 확인을 함께 수행하는 것을 고려해 보세요.

apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEvent.java (2)

76-88: 선택사항: 중복 초기화 제거 가능

Line 86에서 retryCount = 0을 명시적으로 설정하고 있지만, Line 65에서 이미 필드 선언 시 기본값 0으로 초기화되어 있습니다. 명시적 설정을 제거하면 코드가 더 간결해집니다.

🔎 제안된 리팩토링

 public static OutboxEvent create(
         String partitionKey,
         String eventType,
         String payload
 ) {
     OutboxEvent event = new OutboxEvent();
     event.partitionKey = partitionKey;
     event.eventType = eventType;
     event.payload = payload;
     event.status = OutboxStatus.PENDING;
-    event.retryCount = 0;
     return event;
 }

---

101-105: PUBLISHED → FAILED 전환 시 publishedAt 처리 고려

현재 markAsFailed()는 상태를 FAILED로 변경하지만, 이전에 PUBLISHED 상태였다면 publishedAt 필드가 남아있게 됩니다. FAILED 상태에서는 publishedAt이 null이어야 의미상 더 명확할 수 있습니다. 재시도 후 실패하는 시나리오에서 혼란을 방지하려면 publishedAt을 null로 재설정하는 것을 고려하세요.

🔎 제안된 수정

 public void markAsFailed(String errorMessage) {
     this.status = OutboxStatus.FAILED;
     this.retryCount++;
     this.errorMessage = errorMessage;
+    this.publishedAt = null;
 }

apps/commerce-streamer/src/main/java/com/loopers/infrastructure/kafka/RetryTracker.java (1)

65-78: 주석과 구현 불일치 및 메모리 누수 위험

Line 72의 주석에서 "실제 운영에서는 타임스탬프 기반으로 오래된 항목 삭제"라고 명시되어 있지만, 실제 구현은 단순히 크기가 10,000을 초과할 때 전체를 클리어합니다. 이는 다음과 같은 문제를 야기할 수 있습니다:

1. 메모리 누수: clearRetryCount()가 호출되지 않는 경우(예: 컨슈머 비정상 종료) 항목이 무한정 누적됩니다.
2. 급격한 메모리 해제: 10,000개 도달 시 전체 삭제는 진행 중인 재시도 상태를 모두 잃게 만듭니다.
3. 임계값 근거 부족: 10,000이라는 숫자의 선택 근거가 불명확합니다.

타임스탬프 기반 TTL 방식(예: 1시간 이상 경과한 항목만 삭제)으로 구현하거나, 주기적인 cleanup을 스케줄링하는 것을 권장합니다.

🔎 타임스탬프 기반 정리 예시

-private final ConcurrentHashMap<String, Integer> retryCountMap = new ConcurrentHashMap<>();
+private final ConcurrentHashMap<String, RetryEntry> retryCountMap = new ConcurrentHashMap<>();
+
+private static class RetryEntry {
+    int count;
+    long timestamp;
+    RetryEntry(int count) {
+        this.count = count;
+        this.timestamp = System.currentTimeMillis();
+    }
+}

 public boolean canRetry(String topic, int partition, long offset) {
     String key = buildKey(topic, partition, offset);
-    int currentCount = retryCountMap.compute(key, (k, v) -> v == null ? 1 : v + 1);
+    RetryEntry entry = retryCountMap.compute(key, (k, v) -> 
+        v == null ? new RetryEntry(1) : new RetryEntry(v.count + 1)
+    );
+    int currentCount = entry.count;
     
     log.debug("[RetryTracker] Retry count for {}: {}/{}", key, currentCount, MAX_RETRY_COUNT);
     
     return currentCount <= MAX_RETRY_COUNT;
 }

 public void cleanupOldEntries() {
-    int sizeBefore = retryCountMap.size();
-    if (sizeBefore > 10000) {
-        log.warn("[RetryTracker] Clearing all retry counts (size: {})", sizeBefore);
-        retryCountMap.clear();
-    }
+    long now = System.currentTimeMillis();
+    long oneHourAgo = now - 3600_000;
+    retryCountMap.entrySet().removeIf(entry -> entry.getValue().timestamp < oneHourAgo);
+    log.debug("[RetryTracker] Cleaned up old entries, current size: {}", retryCountMap.size());
 }

그리고 스케줄링 추가:

@Scheduled(fixedRate = 300000) // 5분마다
public void scheduledCleanup() {
    cleanupOldEntries();
}

apps/commerce-streamer/src/main/java/com/loopers/domain/metrics/ProductMetrics.java (1)

71-80: 초기 생성 시 lastUpdated 설정의 필요성 재고

create() 메서드에서 lastUpdated를 ZonedDateTime.now()로 초기화하고 있습니다. 이는 생성 직후 발생한 과거 이벤트(예: 생성 시점 이전의 타임스탬프를 가진 이벤트)가 isEventOutdated() 체크에 걸려 무시될 수 있습니다.

초기 생성 시에는 lastUpdated를 null로 두어 첫 번째 이벤트가 어떤 타임스탬프를 가지든 수용되도록 하는 것이 더 안전할 수 있습니다.

🔎 제안된 수정

 public static ProductMetrics create(Long productId) {
     ProductMetrics metrics = new ProductMetrics();
     metrics.productId = productId;
     metrics.likeCount = 0;
     metrics.viewCount = 0;
     metrics.salesCount = 0;
     metrics.salesAmount = 0;
-    metrics.lastUpdated = ZonedDateTime.now();
+    metrics.lastUpdated = null; // 첫 이벤트의 타임스탬프를 수용
     return metrics;
 }

apps/commerce-streamer/src/main/java/com/loopers/application/event/product/ProductViewedEvent.java (1)

5-11: 이벤트 정의 중복을 고려해보세요.

이 이벤트가 commerce-api 모듈에도 동일하게 정의되어 있습니다. 이벤트 스키마가 producer와 consumer 간에 일치해야 하므로 공유 모듈로 추출하는 것을 고려해보세요. 이렇게 하면 스키마 변경 시 일관성을 보장할 수 있습니다.

apps/commerce-streamer/src/main/java/com/loopers/interfaces/consumer/ProductMetricsConsumer.java (1)

247-261: DLQ 전송 후 중복된 clearRetryCount 호출을 정리하세요.

Lines 251과 260에서 retryTracker.clearRetryCount()가 중복 호출되고 있습니다. 두 경우 모두 DLQ로 메시지를 전송한 후 재시도 카운터를 클리어하는데, 이를 한 번만 호출하도록 리팩토링할 수 있습니다.

🔎 제안하는 개선

 private void handleFailedRecord(
     ConsumerRecord<String, String> record,
     Exception exception,
     List<ConsumerRecord<String, String>> failedRecords
 ) {
+    boolean shouldPublishToDlq = false;
+    int retryCount = retryTracker.getRetryCount(record.topic(), record.partition(), record.offset());
+    
     if (dlqPublisher.shouldSendToDlq(exception)) {
         // 복구 불가능한 에러 → DLQ로 전송
-        int retryCount = retryTracker.getRetryCount(record.topic(), record.partition(), record.offset());
-        dlqPublisher.publishToDlq(record, exception, retryCount);
-        retryTracker.clearRetryCount(record.topic(), record.partition(), record.offset());
+        shouldPublishToDlq = true;
     } else {
         // 일시적 에러 → 재시도 가능 여부 확인
         if (retryTracker.canRetry(record.topic(), record.partition(), record.offset())) {
             failedRecords.add(record);
         } else {
             // 최대 재시도 횟수 초과 → DLQ로 전송
-            int retryCount = retryTracker.getRetryCount(record.topic(), record.partition(), record.offset());
-            dlqPublisher.publishToDlq(record, exception, retryCount);
-            retryTracker.clearRetryCount(record.topic(), record.partition(), record.offset());
+            shouldPublishToDlq = true;
         }
     }
+    
+    if (shouldPublishToDlq) {
+        dlqPublisher.publishToDlq(record, exception, retryCount);
+        retryTracker.clearRetryCount(record.topic(), record.partition(), record.offset());
+    }
 }

apps/commerce-api/src/main/java/com/loopers/infrastructure/kafka/KafkaTopicRouter.java (1)

22-38: 이벤트 타입을 상수 또는 Enum으로 관리하는 것을 권장합니다.

KafkaOutboxEventListener에서도 동일한 문자열 리터럴("ORDER_PLACED", "PRODUCT_LIKED" 등)을 사용하고 있습니다. 오타로 인한 런타임 에러를 방지하기 위해 상수 또는 Enum으로 관리하는 것이 좋습니다.

🔎 상수 클래스 예시

public final class EventTypes {
    public static final String ORDER_PLACED = "ORDER_PLACED";
    public static final String ORDER_COMPLETED = "ORDER_COMPLETED";
    public static final String PAYMENT_COMPLETED = "PAYMENT_COMPLETED";
    public static final String PRODUCT_LIKED = "PRODUCT_LIKED";
    public static final String PRODUCT_UNLIKED = "PRODUCT_UNLIKED";
    public static final String PRODUCT_VIEWED = "PRODUCT_VIEWED";
    
    private EventTypes() {}
}

apps/commerce-streamer/src/main/java/com/loopers/infrastructure/kafka/DlqPublisher.java (1)

69-87: NullPointerException을 DLQ로 즉시 전송하는 것이 적절한지 검토하세요.

NullPointerException은 일시적인 데이터 문제나 경합 상태에서도 발생할 수 있습니다. 다른 RuntimeException처럼 재시도 후 DLQ로 보내는 것이 더 안전할 수 있습니다. 또한, Java 16+ 패턴 매칭을 사용하면 코드가 더 간결해집니다.

🔎 패턴 매칭 사용 예시

 public boolean shouldSendToDlq(Exception exception) {
-    // 역직렬화 에러 → DLQ (재시도 불가능)
-    if (exception instanceof com.fasterxml.jackson.core.JsonProcessingException) {
-        return true;
-    }
-
-    // IllegalArgumentException → DLQ (데이터 오류)
-    if (exception instanceof IllegalArgumentException) {
-        return true;
-    }
-
-    // NullPointerException → DLQ (데이터 오류)
-    if (exception instanceof NullPointerException) {
-        return true;
-    }
-
-    // 그 외의 에러는 일시적 장애일 수 있으므로 재시도
-    return false;
+    return switch (exception) {
+        case com.fasterxml.jackson.core.JsonProcessingException _ -> true;  // 역직렬화 에러
+        case IllegalArgumentException _ -> true;  // 데이터 오류
+        // NullPointerException은 재시도 후 DLQ로 보내는 것을 고려
+        default -> false;
+    };
 }

apps/commerce-streamer/src/test/java/com/loopers/infrastructure/kafka/DlqPublisherTest.java (2)

86-96: 테스트 격리성 향상을 위해 findAll().get(0) 대신 더 명확한 조회 방법을 사용하세요.

@Transactional 어노테이션으로 롤백되지만, 다른 테스트에서 남은 데이터가 있을 경우 get(0)이 예상치 못한 결과를 반환할 수 있습니다. 테스트 시작 시 데이터를 정리하거나, 저장된 레코드를 직접 반환받는 방식을 고려하세요.

🔎 개선된 테스트 패턴

     @Test
     @DisplayName("실패한 메시지를 DLQ 테이블에 저장")
     void shouldSaveFailedMessageToDlq() {
         // given
+        dlqMessageRepository.deleteAll();  // 테스트 격리 보장
         ConsumerRecord<String, String> record = new ConsumerRecord<>(
             "loopers.commerce.product-liked-v1",
             0,
             100L,
             "product-123",
             "{\"invalid\":\"json}"
         );
         // ...

---

56-67: 테스트 커버리지 향상을 위해 추가 테스트 케이스를 고려하세요.

NullPointerException에 대한 테스트와 DLQ 저장 실패 시 예외가 발생하지 않는지 확인하는 테스트도 추가하면 좋습니다.

추가 테스트 케이스를 생성해 드릴까요?

apps/commerce-streamer/src/main/java/com/loopers/domain/event/EventHandledRepository.java (1)

11-14: existsById 재선언은 불필요합니다.

JpaRepository가 이미 existsById(ID id) 메서드를 제공하므로 재선언할 필요가 없습니다.

🔎 제거 제안

 public interface EventHandledRepository extends JpaRepository<EventHandled, String> {
 
-    /**
-     * 이벤트 처리 여부 확인 (멱등성 체크)
-     */
-    boolean existsById(String eventId);
-
     /**
      * 특정 기간 이전의 처리 완료 이벤트 삭제 (정리용)

apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/KafkaOutboxEventListener.java (1)

34-38: 이벤트 타입 문자열을 KafkaTopicRouter와 공유하는 것을 권장합니다.

"ORDER_PLACED", "PRODUCT_LIKED" 등의 문자열이 KafkaTopicRouter와 이 클래스에서 중복 사용되고 있습니다. 오타로 인한 런타임 에러를 방지하기 위해 공유 상수를 사용하세요.

Also applies to: 48-52, 62-66, 76-80, 90-94, 104-108

apps/commerce-streamer/src/main/java/com/loopers/domain/dlq/DlqMessageRepository.java (1)

17-22: 대량의 미해결 DLQ 메시지가 있을 경우를 대비해 페이지네이션을 고려하세요.

findByResolvedFalseOrderByFailedAtDesc()와 findByOriginalTopicOrderByFailedAtDesc()가 제한 없이 모든 결과를 반환합니다. 운영 환경에서 DLQ 메시지가 많이 쌓일 경우 메모리 문제가 발생할 수 있습니다.

🔎 페이지네이션 적용 예시

List<DlqMessage> findByResolvedFalseOrderByFailedAtDesc(Pageable pageable);

apps/commerce-streamer/src/main/java/com/loopers/domain/event/EventHandled.java (1)

53-60: 팩토리 메서드에서 null 값 검증을 추가하는 것을 고려하세요.

eventId, eventType, partitionKey가 null일 경우 DB 저장 시점에서야 에러가 발생합니다. 팩토리 메서드에서 조기에 검증하면 디버깅이 용이해집니다.

🔎 검증 추가 예시

 public static EventHandled create(String eventId, String eventType, String partitionKey) {
+    Objects.requireNonNull(eventId, "eventId must not be null");
+    Objects.requireNonNull(eventType, "eventType must not be null");
+    Objects.requireNonNull(partitionKey, "partitionKey must not be null");
+    
     EventHandled handled = new EventHandled();
     handled.eventId = eventId;
     handled.eventType = eventType;
     handled.partitionKey = partitionKey;
     handled.handledAt = ZonedDateTime.now();
     return handled;
 }

apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventPoller.java (2)

39-73: 단일 트랜잭션에서 대량의 이벤트 처리 시 DB 커넥션 점유 문제를 고려하세요.

현재 구조에서는 100개의 이벤트가 하나의 트랜잭션 내에서 순차적으로 처리됩니다. Kafka 전송 지연이 발생하면 DB 커넥션이 오래 점유될 수 있습니다. 각 이벤트를 개별 트랜잭션으로 처리하거나, 커넥션 풀 설정을 검토하는 것이 좋습니다.

🔎 개선 방안

이벤트별 개별 트랜잭션을 사용하면 한 이벤트의 실패가 다른 이벤트에 영향을 주지 않으며, DB 커넥션 점유 시간도 단축됩니다:

@Scheduled(fixedDelay = 5000, initialDelay = 5000)
public void pollAndPublish() {
    List<OutboxEvent> pendingEvents = outboxRepository.findByStatusOrderByCreatedAtAsc(
        OutboxEvent.OutboxStatus.PENDING,
        BATCH_SIZE
    );
    
    for (OutboxEvent event : pendingEvents) {
        processEventInTransaction(event);  // 별도 @Transactional 메서드로 분리
    }
}

@Transactional(propagation = Propagation.REQUIRES_NEW)
public void processEventInTransaction(OutboxEvent event) {
    // 개별 이벤트 처리
}

---

27-28: MAX_RETRY_COUNT를 OutboxEventCleaner와 공유하는 것을 권장합니다.

OutboxEventCleaner에서도 동일하게 3을 사용하고 있습니다. 값이 불일치하면 FAILED 이벤트가 의도치 않게 삭제되거나 재시도될 수 있습니다.

apps/commerce-streamer/src/main/java/com/loopers/application/metrics/MetricsAggregationService.java (1)

194-196: 타임존 문자열을 상수로 추출하는 것을 권장합니다.

"Asia/Seoul" 문자열이 하드코딩되어 있습니다. 여러 곳에서 사용될 수 있으므로 상수나 설정으로 관리하면 유지보수성이 향상됩니다.

제안된 리팩토링

+    private static final ZoneId ZONE_ID_SEOUL = ZoneId.of("Asia/Seoul");
+
     /**
      * LocalDateTime을 ZonedDateTime으로 변환 (Asia/Seoul)
      */
     private ZonedDateTime toZonedDateTime(java.time.LocalDateTime localDateTime) {
-        return localDateTime.atZone(ZoneId.of("Asia/Seoul"));
+        return localDateTime.atZone(ZONE_ID_SEOUL);
     }

apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventCleaner.java (3)

31-31: 타임존을 명시적으로 지정하는 것을 권장합니다.

ZonedDateTime.now()는 시스템 기본 타임존을 사용합니다. 서버 환경에 따라 다른 결과가 나올 수 있으므로, 명시적으로 타임존을 지정하면 더 예측 가능합니다.

제안된 수정

-        ZonedDateTime cutoffDate = ZonedDateTime.now().minusDays(7);
+        ZonedDateTime cutoffDate = ZonedDateTime.now(ZoneId.of("Asia/Seoul")).minusDays(7);

---

61-65: 매직 넘버를 상수로 추출하는 것을 권장합니다.

재시도 횟수 3이 하드코딩되어 있습니다. OutboxEventPoller에서도 동일한 값을 사용할 가능성이 높으므로, 공통 상수로 관리하면 일관성을 유지하기 좋습니다.

제안된 수정

+    private static final int MAX_RETRY_COUNT = 3;
+    private static final int PUBLISHED_RETENTION_DAYS = 7;
+    private static final int FAILED_RETENTION_DAYS = 30;
+
     @Scheduled(cron = "0 0 4 * * SUN")
     @Transactional
     public void cleanOldFailedEvents() {
-        ZonedDateTime cutoffDate = ZonedDateTime.now().minusDays(30);
+        ZonedDateTime cutoffDate = ZonedDateTime.now().minusDays(FAILED_RETENTION_DAYS);
         // ...
             int deleted = outboxRepository.deleteFailedEventsOlderThan(
                 OutboxEvent.OutboxStatus.FAILED,
                 cutoffDate,
-                3  // 최대 재시도 횟수 초과한 이벤트만 삭제
+                MAX_RETRY_COUNT
             );

---

91-99: 경고 임계값도 상수로 관리하면 좋습니다.

PENDING 1000개, FAILED 100개 임계값이 하드코딩되어 있습니다. 운영 환경에서 조정이 필요할 수 있으므로 상수 또는 설정 프로퍼티로 관리하는 것을 권장합니다.

apps/commerce-streamer/src/main/java/com/loopers/domain/dlq/DlqMessage.java (2)

132-148: 스택 트레이스에서 cause chain이 누락됩니다.

현재 구현은 최상위 예외의 스택 트레이스만 캡처합니다. 래핑된 예외(wrapped exception)의 경우 근본 원인을 파악하기 어려울 수 있습니다. 디버깅에 필요하다면 cause chain 포함을 고려해보세요.

제안된 개선

 private static String getStackTraceAsString(Exception exception) {
     StringBuilder sb = new StringBuilder();
     sb.append(exception.toString()).append("\n");

     StackTraceElement[] stackTrace = exception.getStackTrace();
     int maxLines = Math.min(stackTrace.length, 10);

     for (int i = 0; i < maxLines; i++) {
         sb.append("\tat ").append(stackTrace[i]).append("\n");
     }

     if (stackTrace.length > maxLines) {
         sb.append("\t... ").append(stackTrace.length - maxLines).append(" more\n");
     }

+    // Capture cause chain
+    Throwable cause = exception.getCause();
+    if (cause != null) {
+        sb.append("Caused by: ").append(cause.toString()).append("\n");
+        StackTraceElement[] causeTrace = cause.getStackTrace();
+        int causeMaxLines = Math.min(causeTrace.length, 5);
+        for (int i = 0; i < causeMaxLines; i++) {
+            sb.append("\tat ").append(causeTrace[i]).append("\n");
+        }
+    }

     return sb.toString();
 }

---

125-127: 재처리 완료 시각 추가를 고려해보세요.

markAsResolved() 메서드에서 resolved 플래그만 설정하고 있습니다. 재처리 완료 시각(resolvedAt)을 기록하면 운영 모니터링과 감사(audit) 시 유용할 수 있습니다.

제안된 개선

+    @Column(name = "resolved_at")
+    private ZonedDateTime resolvedAt;
+
     public void markAsResolved() {
         this.resolved = true;
+        this.resolvedAt = ZonedDateTime.now();
     }

apps/commerce-api/src/main/java/com/loopers/application/product/ProductFacade.java (1)

103-107: 중복 DB 조회 발생 - 성능 최적화 필요

productReadService.getProductDetail() 내부에서 이미 Product 엔티티를 조회하고 있는데, 여기서 productRepository.findById()로 동일한 Product를 다시 조회하고 있습니다. 상품 상세 조회마다 불필요한 DB 호출이 발생합니다.

또한, CoreException과 ErrorType은 fully qualified name 대신 import 문을 사용하는 것이 일관성 있습니다.

🔎 개선 제안

ProductDetailInfo에 brandId 필드를 추가하거나, productReadService.getProductDetail()이 brandId도 함께 반환하도록 수정하면 중복 조회를 제거할 수 있습니다.

+import com.loopers.support.error.CoreException;
+import com.loopers.support.error.ErrorType;

또는 캐시된 정보에서 brandId를 얻을 수 있도록 DTO를 확장하는 방안을 고려해주세요.

apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventRepository.java (1)

50-55: @Modifying 옵션 추가 고려

벌크 삭제 후 영속성 컨텍스트에 삭제된 엔티티가 남아있을 수 있습니다. 동일 트랜잭션 내에서 후속 조회가 있다면 stale 데이터를 반환할 수 있습니다.

🔎 개선 제안

-    @Modifying
+    @Modifying(clearAutomatically = true)
     @Query("DELETE FROM OutboxEvent o WHERE o.status = :status AND o.publishedAt < :publishedBefore")

Line 64의 deleteFailedEventsOlderThan에도 동일하게 적용 권장합니다.

📜 Review details

Configuration used: Repository UI

Review profile: CHILL

Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between 2bdad02 and 3174797.

📒 Files selected for processing (42)

• apps/commerce-api/build.gradle.kts (1 hunks)
• apps/commerce-api/src/main/java/com/loopers/application/event/listener/OrderStatusEventListener.java (1 hunks)
• apps/commerce-api/src/main/java/com/loopers/application/event/order/OrderCompletedEvent.java (1 hunks)
• apps/commerce-api/src/main/java/com/loopers/application/event/product/ProductViewedEvent.java (1 hunks)
• apps/commerce-api/src/main/java/com/loopers/application/product/ProductFacade.java (4 hunks)
• apps/commerce-api/src/main/java/com/loopers/domain/order/service/OrderPlacementService.java (3 hunks)
• apps/commerce-api/src/main/java/com/loopers/domain/product/repository/ProductRepository.java (1 hunks)
• apps/commerce-api/src/main/java/com/loopers/infrastructure/cache/CacheInvalidationService.java (1 hunks)
• apps/commerce-api/src/main/java/com/loopers/infrastructure/kafka/KafkaTopicRouter.java (1 hunks)
• apps/commerce-api/src/main/java/com/loopers/infrastructure/kafka/event/KafkaEventEnvelope.java (1 hunks)
• apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/KafkaOutboxEventListener.java (1 hunks)
• apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEvent.java (1 hunks)
• apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventCleaner.java (1 hunks)
• apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventPoller.java (1 hunks)
• apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventRepository.java (1 hunks)
• apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventWriter.java (1 hunks)
• apps/commerce-api/src/main/java/com/loopers/infrastructure/product/ProductRepositoryImpl.java (1 hunks)
• apps/commerce-api/src/main/resources/application.yml (1 hunks)
• apps/commerce-api/src/test/java/com/loopers/domain/product/InMemoryProductRepository.java (1 hunks)
• apps/commerce-api/src/test/java/com/loopers/infrastructure/kafka/KafkaConnectionTest.java (1 hunks)
• apps/commerce-streamer/src/main/java/com/loopers/application/event/like/ProductLikedEvent.java (1 hunks)
• apps/commerce-streamer/src/main/java/com/loopers/application/event/like/ProductUnlikedEvent.java (1 hunks)
• apps/commerce-streamer/src/main/java/com/loopers/application/event/order/OrderCompletedEvent.java (1 hunks)
• apps/commerce-streamer/src/main/java/com/loopers/application/event/product/ProductViewedEvent.java (1 hunks)
• apps/commerce-streamer/src/main/java/com/loopers/application/metrics/MetricsAggregationService.java (1 hunks)
• apps/commerce-streamer/src/main/java/com/loopers/confg/kafka/KafkaConfig.java (1 hunks)
• apps/commerce-streamer/src/main/java/com/loopers/domain/dlq/DlqMessage.java (1 hunks)
• apps/commerce-streamer/src/main/java/com/loopers/domain/dlq/DlqMessageRepository.java (1 hunks)
• apps/commerce-streamer/src/main/java/com/loopers/domain/event/EventHandled.java (1 hunks)
• apps/commerce-streamer/src/main/java/com/loopers/domain/event/EventHandledRepository.java (1 hunks)
• apps/commerce-streamer/src/main/java/com/loopers/domain/metrics/ProductMetrics.java (1 hunks)
• apps/commerce-streamer/src/main/java/com/loopers/domain/metrics/ProductMetricsRepository.java (1 hunks)
• apps/commerce-streamer/src/main/java/com/loopers/infrastructure/kafka/DlqPublisher.java (1 hunks)
• apps/commerce-streamer/src/main/java/com/loopers/infrastructure/kafka/RetryTracker.java (1 hunks)
• apps/commerce-streamer/src/main/java/com/loopers/infrastructure/kafka/event/KafkaEventEnvelope.java (1 hunks)
• apps/commerce-streamer/src/main/java/com/loopers/interfaces/consumer/ProductMetricsConsumer.java (1 hunks)
• apps/commerce-streamer/src/test/java/com/loopers/application/metrics/MetricsAggregationServiceIdempotencyTest.java (1 hunks)
• apps/commerce-streamer/src/test/java/com/loopers/config/TestConfig.java (1 hunks)
• apps/commerce-streamer/src/test/java/com/loopers/infrastructure/kafka/DlqPublisherTest.java (1 hunks)
• apps/commerce-streamer/src/test/java/com/loopers/infrastructure/kafka/RetryTrackerTest.java (1 hunks)
• apps/commerce-streamer/src/test/resources/application-test.yml (1 hunks)
• modules/kafka/src/main/resources/kafka.yml (1 hunks)

🧰 Additional context used

🧠 Learnings (8)

📓 Common learnings

Learnt from: kilian-develop
Repo: Loopers-dev-lab/loopers-spring-java-template PR: 190
File: apps/commerce-streamer/src/main/java/com/loopers/applications/streamer/consumer/product/IncreaseProductViewKafkaConsumer.java:25-35
Timestamp: 2025-12-18T13:24:54.339Z
Learning: In this codebase, Kafka consumers delegate error handling and event tracking to the service layer via EventInboxAspect. Service methods annotated with InboxEvent are intercepted by the aspect, which handles failures by saving failed EventInbox entries and logging errors. This centralized approach avoids duplicating error handling logic across multiple consumers.

📚 Learning: 2025-11-27T09:09:24.961Z

Learnt from: sky980221
Repo: Loopers-dev-lab/loopers-spring-java-template PR: 121
File: apps/commerce-api/src/main/java/com/loopers/infrastructure/product/ProductJpaRepository.java:22-24
Timestamp: 2025-11-27T09:09:24.961Z
Learning: Product 엔티티 (apps/commerce-api/src/main/java/com/loopers/domain/product/Product.java)는 유즈케이스별로 의도적으로 다른 락 전략을 사용한다: 좋아요 기능에는 비관적 락(findByIdForUpdate)을, 재고 차감에는 낙관적 락(Version + 재시도)을 사용한다.

Applied to files:

• apps/commerce-api/src/main/java/com/loopers/domain/product/repository/ProductRepository.java
• apps/commerce-api/src/main/java/com/loopers/infrastructure/cache/CacheInvalidationService.java
• apps/commerce-streamer/src/main/java/com/loopers/application/event/product/ProductViewedEvent.java
• apps/commerce-api/src/main/java/com/loopers/application/event/product/ProductViewedEvent.java
• apps/commerce-streamer/src/main/java/com/loopers/application/event/like/ProductUnlikedEvent.java
• apps/commerce-streamer/src/main/java/com/loopers/application/event/like/ProductLikedEvent.java
• apps/commerce-api/src/main/java/com/loopers/application/product/ProductFacade.java
• apps/commerce-streamer/src/main/java/com/loopers/domain/metrics/ProductMetricsRepository.java
• apps/commerce-api/src/test/java/com/loopers/domain/product/InMemoryProductRepository.java
• apps/commerce-streamer/src/main/java/com/loopers/domain/metrics/ProductMetrics.java
• apps/commerce-api/src/main/java/com/loopers/infrastructure/product/ProductRepositoryImpl.java

📚 Learning: 2025-12-02T08:12:06.383Z

Learnt from: junoade
Repo: Loopers-dev-lab/loopers-spring-java-template PR: 0
File: :0-0
Timestamp: 2025-12-02T08:12:06.383Z
Learning: ProductQueryService에서 상품 목록 조회 시 Redis 캐시를 적용했으며, 캐시 키는 brandId, sortType, pageNumber, pageSize의 조합으로 구성되고 TTL은 5분으로 설정되어 있다.

Applied to files:

• apps/commerce-api/src/main/java/com/loopers/infrastructure/cache/CacheInvalidationService.java
• apps/commerce-api/src/main/java/com/loopers/application/product/ProductFacade.java

📚 Learning: 2025-12-19T03:23:51.674Z

Learnt from: minor7295
Repo: Loopers-dev-lab/loopers-spring-java-template PR: 191
File: modules/kafka/src/testFixtures/java/com/loopers/testcontainers/KafkaTestContainersConfig.java:24-34
Timestamp: 2025-12-19T03:23:51.674Z
Learning: In the loopers-spring-java-template project, test container configurations (MySQL, Redis, Kafka) use static initialization blocks for consistency across the template. This pattern should be maintained for uniformity, even if alternative lifecycle management approaches exist.

Applied to files:

• apps/commerce-streamer/src/test/java/com/loopers/config/TestConfig.java
• apps/commerce-streamer/src/test/resources/application-test.yml
• apps/commerce-api/src/test/java/com/loopers/infrastructure/kafka/KafkaConnectionTest.java

📚 Learning: 2025-12-18T13:24:54.339Z

Learnt from: kilian-develop
Repo: Loopers-dev-lab/loopers-spring-java-template PR: 190
File: apps/commerce-streamer/src/main/java/com/loopers/applications/streamer/consumer/product/IncreaseProductViewKafkaConsumer.java:25-35
Timestamp: 2025-12-18T13:24:54.339Z
Learning: In this codebase, Kafka consumers delegate error handling and event tracking to the service layer via EventInboxAspect. Service methods annotated with InboxEvent are intercepted by the aspect, which handles failures by saving failed EventInbox entries and logging errors. This centralized approach avoids duplicating error handling logic across multiple consumers.

Applied to files:

• apps/commerce-streamer/src/main/java/com/loopers/infrastructure/kafka/DlqPublisher.java
• apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/KafkaOutboxEventListener.java
• apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventPoller.java
• apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventWriter.java
• apps/commerce-streamer/src/main/java/com/loopers/infrastructure/kafka/event/KafkaEventEnvelope.java
• apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEvent.java

📚 Learning: 2025-12-18T13:24:51.650Z

Learnt from: kilian-develop
Repo: Loopers-dev-lab/loopers-spring-java-template PR: 190
File: apps/commerce-streamer/src/main/java/com/loopers/applications/streamer/consumer/product/IncreaseProductViewKafkaConsumer.java:25-35
Timestamp: 2025-12-18T13:24:51.650Z
Learning: Adopt centralized error handling for Kafka consumers by using the EventInboxAspect to intercept methods annotated with InboxEvent. Ensure that service methods annotated with InboxEvent save failed EventInbox entries and log errors, avoiding duplicating error handling logic in individual consumers. Apply this pattern broadly to similar consumer/service layers within the commerce-streamer module.

Applied to files:

• apps/commerce-streamer/src/main/java/com/loopers/interfaces/consumer/ProductMetricsConsumer.java

📚 Learning: 2025-11-09T10:41:39.297Z

Learnt from: ghojeong
Repo: Loopers-dev-lab/loopers-spring-kotlin-template PR: 25
File: apps/commerce-api/src/main/kotlin/com/loopers/domain/product/ProductRepository.kt:1-12
Timestamp: 2025-11-09T10:41:39.297Z
Learning: In this codebase, domain repository interfaces are allowed to use Spring Data's org.springframework.data.domain.Page and org.springframework.data.domain.Pageable types. This is an accepted architectural decision and should not be flagged as a DIP violation.

Applied to files:

• apps/commerce-api/src/main/java/com/loopers/application/product/ProductFacade.java
• apps/commerce-streamer/src/main/java/com/loopers/domain/metrics/ProductMetricsRepository.java
• apps/commerce-streamer/src/main/java/com/loopers/domain/dlq/DlqMessageRepository.java

📚 Learning: 2025-12-18T01:01:55.894Z

Learnt from: jikimee64
Repo: Loopers-dev-lab/loopers-spring-kotlin-template PR: 65
File: modules/kafka/src/main/resources/kafka.yml:25-32
Timestamp: 2025-12-18T01:01:55.894Z
Learning: In Spring Boot Kafka configuration YAML, properties inside spring.kafka.producer.properties and spring.kafka.consumer.properties maps must use exact Kafka client property names with dot notation, and must be quoted with bracket notation like "[enable.idempotence]": true and "[enable.auto.commit]": false to prevent YAML from parsing dots as nested keys. Spring Boot's relaxed binding only applies to top-level Spring Kafka properties, not to the properties map.

Applied to files:

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

🧬 Code graph analysis (6)

apps/commerce-streamer/src/main/java/com/loopers/infrastructure/kafka/DlqPublisher.java (2)

apps/commerce-streamer/src/main/java/com/loopers/infrastructure/kafka/RetryTracker.java (1)

• Slf4j (13-86)

apps/pg-simulator/src/main/kotlin/com/loopers/interfaces/api/ApiControllerAdvice.kt (1)

• log (22-119)

apps/commerce-streamer/src/main/java/com/loopers/domain/dlq/DlqMessage.java (3)

apps/commerce-streamer/src/main/java/com/loopers/domain/event/EventHandled.java (1)

• Entity (15-61)

apps/commerce-streamer/src/main/java/com/loopers/domain/metrics/ProductMetrics.java (1)

• Entity (15-160)

apps/pg-simulator/src/main/kotlin/com/loopers/support/error/CoreException.kt (1)

• errorType (3-6)

apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/KafkaOutboxEventListener.java (3)

apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventCleaner.java (1)

• Slf4j (16-105)

apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventPoller.java (1)

• Slf4j (22-144)

apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventWriter.java (1)

• Slf4j (14-50)

apps/commerce-streamer/src/main/java/com/loopers/infrastructure/kafka/RetryTracker.java (1)

apps/commerce-streamer/src/main/java/com/loopers/infrastructure/kafka/DlqPublisher.java (1)

• Slf4j (16-88)

apps/commerce-api/src/main/java/com/loopers/application/product/ProductFacade.java (1)

apps/pg-simulator/src/main/kotlin/com/loopers/interfaces/api/ApiResponse.kt (1)

• result (7-19)

apps/commerce-streamer/src/main/java/com/loopers/domain/event/EventHandled.java (3)

apps/commerce-streamer/src/main/java/com/loopers/domain/dlq/DlqMessage.java (1)

• Entity (15-149)

apps/commerce-streamer/src/main/java/com/loopers/domain/metrics/ProductMetrics.java (1)

• Entity (15-160)

apps/pg-simulator/src/main/kotlin/com/loopers/domain/payment/Payment.kt (1)

• name (14-87)

🔇 Additional comments (33)

apps/commerce-streamer/src/test/resources/application-test.yml (2)

2-4: LGTM! 표준 Testcontainers 설정

Testcontainers를 사용한 MySQL 설정이 올바르게 구성되어 있습니다. TC_DAEMON=true 파라미터는 테스트 간 컨테이너 재사용을 허용하여 성능을 개선합니다.

---

5-11: LGTM! 테스트에 적합한 JPA 설정

ddl-auto: create-drop는 테스트 격리를 보장하고, SQL 로깅 설정은 디버깅에 유용합니다.

apps/commerce-streamer/src/test/java/com/loopers/config/TestConfig.java (1)

8-17: Kafka 자동 설정 제외 시 빈 의존성 확인 필요

Kafka 자동 설정을 제외하는 접근 방식은 올바르지만, 프로덕션 Kafka 컨슈머가 KafkaTemplate, ConsumerFactory 등의 Kafka 빈을 직접 주입받고 있다면 테스트 컨텍스트 로딩이 실패할 수 있습니다. 다음을 확인해주세요:

1. Kafka 관련 빈을 직접 @Autowired하는 컨슈머나 서비스 클래스의 존재 여부
2. 실제 테스트 실행 시 빈 초기화 오류 발생 여부

이런 문제가 있다면, 해당 Kafka 의존 빈들을 별도 설정으로 분리하거나 테스트 환경에서만 필요한 특정 패키지만 스캔하도록 수정해주세요.

apps/commerce-api/build.gradle.kts (1)

5-5: LGTM!

Kafka 모듈 의존성 추가가 PR 목표인 Kafka 기반 이벤트 파이프라인 구현과 잘 맞습니다.

apps/commerce-api/src/main/resources/application.yml (1)

24-24: LGTM!

kafka.yml 설정 import 추가가 Kafka 모듈 의존성과 일관성 있게 구성되었습니다.

apps/commerce-api/src/main/java/com/loopers/infrastructure/cache/CacheInvalidationService.java (1)

38-46: LGTM!

재고 소진 시 캐시 무효화 메서드가 기존 invalidateOnLikeChange, invalidateOnProductUpdate 패턴과 일관성 있게 구현되었습니다.

apps/commerce-api/src/main/java/com/loopers/application/event/order/OrderCompletedEvent.java (1)

11-18: LGTM!

주문 완료 이벤트에 주문 항목 정보(OrderItemInfo)를 추가하여 downstream consumer에서 상품별 메트릭(salesCount, salesAmount) 집계가 가능해졌습니다. 중첩 record 패턴으로 관련 타입을 함께 유지하는 것이 좋습니다.

apps/commerce-streamer/src/test/java/com/loopers/infrastructure/kafka/RetryTrackerTest.java (1)

1-115: LGTM!

테스트 커버리지가 우수하며 RetryTracker의 핵심 동작을 모두 검증하고 있습니다:

• 재시도 허용 로직 (최대 3회)
• 메시지별 독립적인 추적
• 성공 시 카운터 초기화 및 재시작
• 존재하지 않는 메시지의 기본값

모든 테스트 케이스가 명확하고 예상 동작을 정확히 검증하고 있습니다.

apps/commerce-api/src/main/java/com/loopers/domain/product/repository/ProductRepository.java (1)

29-29: LGTM!

재고 수량 조회를 위한 읽기 전용 메서드 추가가 적절합니다. 기존 패턴과 일관성이 있으며, 도메인 레이어에 상태 변경 없이 재고 정보를 조회할 수 있는 기능을 제공합니다.

apps/commerce-api/src/main/java/com/loopers/infrastructure/product/ProductRepositoryImpl.java (1)

141-149: LGTM!

QueryDSL을 사용한 효율적인 부분 필드 조회 구현입니다. 필요한 stock.quantity만 선택하여 불필요한 데이터 로딩을 방지하고 있으며, null 체크를 통해 안전한 기본값(0)을 반환합니다. 읽기 전용 작업으로 부작용이 없습니다.

apps/commerce-api/src/test/java/com/loopers/domain/product/InMemoryProductRepository.java (1)

98-105: LGTM!

테스트용 인메모리 구현이 프로덕션 코드와 동일한 계약을 충실히 이행하고 있습니다. 존재하지 않는 상품에 대해 0을 반환하는 기본 동작이 ProductRepositoryImpl과 일관성 있게 구현되어 있습니다.

apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEvent.java (1)

93-96: 타임존 명시적 지정 필요

BaseEntity와 OutboxEvent 모두 ZonedDateTime.now()를 사용하여 타임존 처리가 일관되어 있습니다. 다만, 시스템 기본 타임존에 의존하고 있어 분산 환경에서 예측 불가능한 문제가 발생할 수 있습니다. 명시적으로 UTC 타임존을 지정하는 것이 권장됩니다: ZonedDateTime.now(ZoneId.of("UTC")) 또는 ZonedDateTime.now(ZoneOffset.UTC)

Likely an incorrect or invalid review comment.

apps/commerce-streamer/src/main/java/com/loopers/domain/metrics/ProductMetricsRepository.java (2)

25-38: JPQL LIMIT 구문의 JPA 표준 호환성 확인 필요

Lines 25, 31, 37에서 JPQL 쿼리에 LIMIT :limit 구문을 사용하고 있습니다. 이는 Hibernate에서는 지원되지만 JPA 표준 구문은 아닙니다.

JPA 2.0 이전에는 setMaxResults()를 사용해야 하며, JPA 3.1부터는 표준 LIMIT이 추가되었습니다. 현재 사용 중인 JPA 버전과 JPA 프로바이더를 확인하여 호환성을 검증해야 합니다.

#!/bin/bash
# JPA 및 Hibernate 버전 확인
rg "jakarta.persistence|org.hibernate" apps/commerce-streamer/build.gradle apps/commerce-streamer/pom.xml -A 2 -B 2

# Spring Data JPA 버전 확인
rg "spring-boot-starter-data-jpa" apps/commerce-streamer/build.gradle apps/commerce-streamer/pom.xml -A 1

표준 호환성을 위해 네이티브 쿼리로 변경하거나, Pageable을 사용하는 방식도 고려할 수 있습니다.

🔎 표준 호환 대안

-@Query("SELECT pm FROM ProductMetrics pm ORDER BY pm.likeCount DESC LIMIT :limit")
-List<ProductMetrics> findTopByLikeCount(@Param("limit") int limit);
+// 옵션 1: Pageable 사용 (표준 방식)
+@Query("SELECT pm FROM ProductMetrics pm ORDER BY pm.likeCount DESC")
+List<ProductMetrics> findTopByLikeCount(Pageable pageable);
+
+// 사용: repository.findTopByLikeCount(PageRequest.of(0, limit))

---

18-20: findByIdWithLock() 메서드가 정의되었으나 실제로 사용되지 않음 - 디자인 일관성 확인 필요

ProductMetrics 엔티티는 @Version 필드로 낙관적 락을 활성화하고 있지만, ProductMetricsRepository의 findByIdWithLock() 메서드는 PESSIMISTIC_WRITE로 정의되어 있습니다. 그러나 MetricsAggregationService에서는 findById()를 사용하고 있으어 findByIdWithLock()은 실제로 호출되지 않고 있습니다.

현재 구현은 다음과 같이 동작합니다:

• findById()로 엔티티 조회
• 엔티티 메서드 호출 시 timestamp 기반 이벤트 순서 검증
• save() 시 @Version 필드 자동 관리 (JPA 암묵적 처리)
• EventHandledRepository를 통한 멱등성 보장

비관적 락 메서드인 findByIdWithLock()은 정의만 되어 있고 사용되지 않으므로, 다음 중 하나를 진행하세요:

• 불필요한 메서드라면 제거
• 의도된 설계라면 실제 호출 지점에서 사용하도록 변경 및 에러 처리 추가

apps/commerce-api/src/main/java/com/loopers/application/event/product/ProductViewedEvent.java (1)

5-11: LGTM!

이벤트 레코드 구조가 명확하고 적절합니다. 비로그인 사용자 추적을 위한 nullable memberId 설계도 합리적입니다.

apps/commerce-streamer/src/main/java/com/loopers/application/event/like/ProductLikedEvent.java (1)

5-11: LGTM!

이벤트 구조가 명확하고 다른 이벤트 레코드들과 일관성이 있습니다.

apps/commerce-api/src/main/java/com/loopers/application/event/listener/OrderStatusEventListener.java (1)

53-59: LGTM!

주문 아이템을 이벤트 페이로드로 변환하는 로직이 명확하고 정확합니다. 함수형 스타일로 간결하게 구현되었습니다.

apps/commerce-streamer/src/main/java/com/loopers/application/event/order/OrderCompletedEvent.java (1)

7-19: LGTM!

이벤트 구조가 명확하고 중첩된 OrderItemInfo 레코드로 타입 안정성을 확보했습니다. 금액 필드에 BigDecimal 사용도 적절합니다.

apps/commerce-api/src/main/java/com/loopers/infrastructure/kafka/event/KafkaEventEnvelope.java (1)

12-42: LGTM!

Kafka 이벤트 엔벨로프 설계가 우수합니다. 메타데이터와 페이로드를 명확히 분리했고, 멱등성 체크와 순서 보장을 위한 필수 정보를 모두 포함하고 있습니다. ZonedDateTime 사용도 분산 시스템에 적합합니다.

apps/commerce-streamer/src/main/java/com/loopers/application/event/like/ProductUnlikedEvent.java (1)

5-11: LGTM!

이벤트 구조가 명확하고 ProductLikedEvent와 일관성이 있습니다.

apps/commerce-streamer/src/main/java/com/loopers/interfaces/consumer/ProductMetricsConsumer.java (1)

241-263: 중앙 집중식 에러 핸들링 패턴 적용을 검토하세요.

학습된 패턴에 따르면, 이 코드베이스에서는 Kafka 컨슈머의 에러 핸들링을 EventInboxAspect를 통해 서비스 레이어에 위임하는 중앙 집중식 접근 방식을 사용합니다. 그러나 이 컨슈머는 RetryTracker와 DlqPublisher를 직접 사용하는 커스텀 에러 핸들링을 구현하고 있습니다.

현재 구현이 의도적으로 다른 패턴을 사용하는 것인지 확인이 필요합니다. 만약 일관성을 위해 중앙 집중식 패턴을 적용해야 한다면, 서비스 메서드에 @InboxEvent 어노테이션을 사용하고 EventInboxAspect가 에러 핸들링과 이벤트 추적을 담당하도록 리팩토링하는 것을 고려하세요.

Based on learnings from PR #190.

apps/commerce-streamer/src/main/java/com/loopers/infrastructure/kafka/DlqPublisher.java (1)

30-61: LGTM!

DLQ 저장 실패 시 예외를 던지지 않는 방어적 처리가 잘 구현되어 있습니다. 무한 루프 방지를 위한 적절한 패턴입니다.

apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/KafkaOutboxEventListener.java (1)

30-109: 구조가 잘 설계되었습니다.

BEFORE_COMMIT 단계를 사용하여 원본 트랜잭션과 Outbox 저장의 원자성을 보장하고 있습니다. 각 이벤트 핸들러가 일관된 패턴을 따르고 있어 유지보수성이 좋습니다.

apps/commerce-streamer/src/main/java/com/loopers/domain/event/EventHandled.java (1)

15-61: LGTM!

멱등성 보장을 위한 이벤트 처리 이력 엔티티가 잘 설계되었습니다. eventId를 PK로 사용하여 중복 이벤트 처리를 효과적으로 방지할 수 있습니다. 인덱스 설정도 쿼리 패턴에 적합합니다.

apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventWriter.java (1)

29-49: LGTM! Outbox 이벤트 저장 로직이 명확하게 구현되었습니다.

JSON 직렬화 실패 시 IllegalStateException으로 래핑하여 호출 측에서 적절히 처리할 수 있도록 했고, 로깅도 적절히 구성되어 있습니다.

apps/commerce-streamer/src/main/java/com/loopers/application/metrics/MetricsAggregationService.java (2)

36-69: 멱등성 처리 로직이 잘 구현되었습니다.

중복 이벤트 체크, out-of-order 이벤트 처리, 그리고 처리 완료 기록이 체계적으로 구현되어 있습니다. out-of-order 이벤트도 eventHandled에 기록하여 재처리를 방지하는 점이 좋습니다.

---

112-151: 주문 완료 이벤트 처리 로직 확인 요청

주문 아이템이 많을 경우 각 아이템별로 metricsRepository.save() 호출이 발생합니다. 단일 트랜잭션 내에서 처리되므로 일관성은 보장되지만, 대량 주문 시 성능 영향을 모니터링하시기 바랍니다.

또한, 모든 아이템이 out-of-order로 무시되어도 이벤트가 처리 완료로 기록되는데, 이는 의도된 동작인지 확인이 필요합니다.

apps/commerce-streamer/src/test/java/com/loopers/application/metrics/MetricsAggregationServiceIdempotencyTest.java (2)

55-81: 좋은 테스트 구조입니다.

given-when-then 패턴을 따르고, 중복 이벤트 처리 시 상태 변화가 없음을 명확하게 검증하고 있습니다. eventHandledRepository.existsById() 확인도 포함되어 있어 멱등성 메커니즘을 완전히 테스트합니다.

---

152-180: 서로 다른 eventId 테스트가 누적 처리를 잘 검증합니다.

독립적인 이벤트가 각각 처리되어 좋아요 수가 누적되는 것을 확인하는 테스트입니다. 멱등성 로직이 eventId 기반임을 명확히 보여줍니다.

apps/commerce-streamer/src/main/java/com/loopers/domain/dlq/DlqMessage.java (1)

15-20: 인덱스 설계가 적절합니다.

original_topic, failed_at, error_type 컬럼에 대한 인덱스가 일반적인 조회 패턴(토픽별 조회, 시간순 조회, 에러 유형별 조회)을 잘 지원합니다.

apps/commerce-api/src/main/java/com/loopers/application/product/ProductFacade.java (1)

32-33: 의존성 추가 LGTM

ProductRepository와 ApplicationEventPublisher 추가는 이벤트 발행 기능에 필요한 적절한 의존성입니다.

apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventRepository.java (2)

37-42: 디버깅용 쿼리 메서드 LGTM

Spring Data JPA 쿼리 메서드 네이밍 컨벤션을 잘 따르고 있습니다. 디버깅 용도로 적절합니다.

---

17-21: JPQL LIMIT 구문 - Hibernate 6 사용 시 동작하지만 표준 JPA 이식성 문제

LIMIT :limit 구문은 Hibernate의 HQL 확장으로 JPQL에서는 직접 limit 키워드를 포함하지 않는 것이 표준입니다. 이 프로젝트는 Spring Boot 3.4.4를 사용하므로 Hibernate 6이 적용되어 있어 런타임에 이 쿼리는 정상 작동합니다. 그러나 JPQL과 Criteria API는 LIMIT과 OFFSET 절을 쿼리의 일부로 지원하지 않지만 이를 설정하는 API를 제공하며, Hibernate가 생성된 SQL 문에 필요한 절을 추가합니다.

표준 JPA 이식성과 가독성을 위해 Pageable을 사용하거나 Query.setMaxResults()로 변경하는 것이 권장됩니다.

추가로 검토할 사항:

• Line 28: 열거형 상수 대신 하드코딩된 'FAILED' 문자열 사용 - @Param("status") OutboxEvent.OutboxStatus status로 매개변수화 권장
• Lines 50, 64: @Modifying 쿼리에 clearAutomatically = true 옵션 추가 권장

[coderabbitai]

⚠️ Potential issue | 🟡 Minor

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Description: isValidTopic 메서드 사용처 확인
rg -n "isValidTopic" --type=java

Repository: Loopers-dev-lab/loopers-spring-java-template

Length of output: 229

---

isValidTopic() 메서드가 사용되지 않고 있습니다.

현재 코드베이스에서 이 메서드의 호출 지점을 찾을 수 없습니다. 사용되지 않는 메서드이므로 제거하거나, 향후 사용 계획이 있다면 TODO 주석을 추가하세요.

🤖 Prompt for AI Agents

In
apps/commerce-api/src/main/java/com/loopers/infrastructure/kafka/KafkaTopicRouter.java
around lines 43-47, the isValidTopic(String) method is unused; either remove the
method entirely (delete its declaration and any now-unused imports or
references, then run mvn/test to ensure no compile breakage) or, if you intend
to keep it for future use, add a clear TODO comment above the method explaining
expected future usage and add @SuppressWarnings("unused") to avoid linter noise.

[coderabbitai]

⚠️ Potential issue | 🟠 Major

InterruptedException 처리 시 인터럽트 상태를 복원해야 합니다.

InterruptedException을 catch한 후 인터럽트 상태를 복원하지 않으면, 스레드 종료 시그널이 손실되어 graceful shutdown이 제대로 동작하지 않을 수 있습니다.

🔎 수정 제안

         } catch (InterruptedException | ExecutionException e) {
+            if (e instanceof InterruptedException) {
+                Thread.currentThread().interrupt();
+            }
             throw new RuntimeException("Kafka send failed", e);
         }

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change

	} catch (InterruptedException | ExecutionException e) {
	throw new RuntimeException("Kafka send failed", e);
	}
	} catch (InterruptedException | ExecutionException e) {
	if (e instanceof InterruptedException) {
	Thread.currentThread().interrupt();
	}
	throw new RuntimeException("Kafka send failed", e);
	}

🤖 Prompt for AI Agents

In
apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventPoller.java
around lines 140 to 142, the current catch lumps InterruptedException and
ExecutionException together which loses the thread interrupt status; separate
the two exceptions, and in the InterruptedException handler call
Thread.currentThread().interrupt() to restore the interrupt flag and then
rethrow or wrap the exception (e.g., throw new RuntimeException("Kafka send
interrupted", e)); keep a separate ExecutionException handler that throws the
existing RuntimeException("Kafka send failed", e).

[coderabbitai]

⚠️ Potential issue | 🟡 Minor

하드코딩된 enum 값 - 파라미터 바인딩 사용 권장

'FAILED' 문자열이 직접 쿼리에 하드코딩되어 있습니다. 다른 메서드들(findByStatusOrderByCreatedAtAsc)은 :status 파라미터를 사용하는데, 이 메서드만 일관성이 없습니다. enum의 @Enumerated 전략이 변경되거나 값이 변경되면 쿼리가 실패할 수 있습니다.

🔎 개선 제안

-    @Query("SELECT o FROM OutboxEvent o WHERE o.status = 'FAILED' AND o.retryCount < :maxRetryCount ORDER BY o.createdAt ASC LIMIT :limit")
+    @Query("SELECT o FROM OutboxEvent o WHERE o.status = :status AND o.retryCount < :maxRetryCount ORDER BY o.createdAt ASC LIMIT :limit")
     List<OutboxEvent> findRetryableEvents(
+        @Param("status") OutboxEvent.OutboxStatus status,
         @Param("maxRetryCount") int maxRetryCount,
         @Param("limit") int limit
     );

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change

	@Query("SELECT o FROM OutboxEvent o WHERE o.status = 'FAILED' AND o.retryCount < :maxRetryCount ORDER BY o.createdAt ASC LIMIT :limit")
	List<OutboxEvent> findRetryableEvents(
	@Param("maxRetryCount") int maxRetryCount,
	@Param("limit") int limit
	);
	@Query("SELECT o FROM OutboxEvent o WHERE o.status = :status AND o.retryCount < :maxRetryCount ORDER BY o.createdAt ASC LIMIT :limit")
	List<OutboxEvent> findRetryableEvents(
	@Param("status") OutboxEvent.OutboxStatus status,
	@Param("maxRetryCount") int maxRetryCount,
	@Param("limit") int limit
	);

🤖 Prompt for AI Agents

In
apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventRepository.java
around lines 28-32, the JPQL query hardcodes the status string 'FAILED'; replace
that literal with a bound parameter to match other methods and respect the
entity's @Enumerated mapping. Update the @Query to use :status, add an
@Param("status") OutboxStatus status parameter to the method signature, and
update any callers to pass the appropriate enum value so the enum mapping is
used instead of a literal string.

[coderabbitai]

⚠️ Potential issue | 🔴 Critical

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Find KafkaTestContainersConfig
fd -type f -name "*KafkaTestContainersConfig*"

# Search for its usage pattern in tests
rg -n --type=java "@Import.*KafkaTestContainersConfig|@SpringBootTest" -A 5 | head -100

# Check the file structure to understand import paths
git ls-files | grep -i "kafka.*test" | head -20

Repository: Loopers-dev-lab/loopers-spring-java-template

Length of output: 10869

---

🏁 Script executed:

#!/bin/bash
# Look for testFixtures or test container configs
find . -type d -name "testFixtures" -o -name "test-fixtures"

# Search for any KafkaTestContainers related files
rg -l "KafkaTestContainers" --type java

Repository: Loopers-dev-lab/loopers-spring-java-template

Length of output: 151

---

🏁 Script executed:

#!/bin/bash
# Check if the actual test file exists and examine its content
find . -path "*commerce-api*" -name "KafkaConnectionTest.java" -type f

# Also check what's in the kafka test related files
rg -n "class.*Test" --type java | grep -i kafka

Repository: Loopers-dev-lab/loopers-spring-java-template

Length of output: 540

---

🏁 Script executed:

#!/bin/bash
# Search for KafkaTestContainersConfig in the entire repository
rg -l "KafkaTestContainersConfig" --type java

# Check if modules/kafka directory exists
find . -type d -name "kafka" | grep -E "modules|testFixtures"

# Search for any file with TestContainers in kafka context
rg -l "TestContainers" --type java | grep -i kafka

Repository: Loopers-dev-lab/loopers-spring-java-template

Length of output: 157

---

🏁 Script executed:

#!/bin/bash
# Read the full KafkaConnectionTest.java file to see its complete implementation
cat -n apps/commerce-api/src/test/java/com/loopers/infrastructure/kafka/KafkaConnectionTest.java

# Also check DlqPublisherTest to see the pattern used there
cat -n apps/commerce-streamer/src/test/java/com/loopers/infrastructure/kafka/DlqPublisherTest.java

Repository: Loopers-dev-lab/loopers-spring-java-template

Length of output: 8184

---

🏁 Script executed:

#!/bin/bash
# Check if modules/kafka exists and what's inside
ls -la modules/ | grep -i kafka

# If it exists, check testFixtures
ls -la modules/kafka/src/ 2>/dev/null || echo "modules/kafka not found"

Repository: Loopers-dev-lab/loopers-spring-java-template

Length of output: 301

---

🏁 Script executed:

#!/bin/bash
# Explicitly search for any file with KafkaTestContainersConfig in its name or content
find . -name "*KafkaTestContainersConfig*" -type f

# Search for any testcontainers related Kafka config
rg "KafkaTestContainersConfig" 

# Check what testFixtures modules actually have
find ./modules -type d -name "testFixtures"

Repository: Loopers-dev-lab/loopers-spring-java-template

Length of output: 151

---

🏁 Script executed:

#!/bin/bash
# Check the full structure of modules/kafka
find ./modules/kafka -type f -name "*.java" | head -20

# List all testFixtures directories and their content
find ./modules -type d -name "testFixtures" -exec find {} -name "*.java" \;

Repository: Loopers-dev-lab/loopers-spring-java-template

Length of output: 497

---

테스트 컨테이너 설정이 누락되었습니다. 하드코딩된 부트스트랩 서버 주소를 제거하세요.

테스트가 localhost:19092에 의존하고 있어 Kafka 인스턴스가 실행 중이지 않으면 실패합니다. 프로젝트의 MySQL, Redis 테스트처럼 테스트 컨테이너를 사용하여 관리되는 Kafka 인스턴스를 설정해야 합니다.

• modules/kafka/src/testFixtures에 KafkaTestContainersConfig를 생성하고 정적 초기화 블록으로 Kafka 컨테이너를 구성하세요.
• @Import(KafkaTestContainersConfig.class) 어노테이션을 추가하세요.
• 하드코딩된 bootstrap-servers 속성을 제거하고, Producer 설정은 KafkaConfig 또는 application-test.yml에서 관리하세요.

또한 라인 50-53, 67-70의 System.out.println 호출을 @Slf4j 로거로 변경하세요.

🤖 Prompt for AI Agents

In
apps/commerce-api/src/test/java/com/loopers/infrastructure/kafka/KafkaConnectionTest.java
around lines 16-28, the test currently hardcodes
spring.kafka.bootstrap-servers=localhost:19092 and lacks Testcontainers setup;
remove the hardcoded bootstrap-servers property, create a
modules/kafka/src/testFixtures/KafkaTestContainersConfig that starts a
KafkaContainer in a static initializer and exposes its bootstrapServers,
annotate the test with @Import(KafkaTestContainersConfig.class) so the test uses
the container-managed address, move producer configuration out of the test
properties into KafkaConfig or application-test.yml for the test profile, and
replace the System.out.println calls at lines ~50-53 and ~67-70 with logging via
an @Slf4j logger.

[coderabbitai]

⚠️ Potential issue | 🟠 Major

서로 다른 이벤트 타입 간 타임스탬프 충돌 가능성

현재 모든 메트릭 업데이트(좋아요, 조회, 판매)가 단일 lastUpdated 필드를 공유하고 있습니다. 이로 인해 다음과 같은 시나리오에서 데이터 손실이 발생할 수 있습니다:

시나리오 예시:

1. T1 시점: ProductViewedEvent 발생 → viewCount 증가, lastUpdated = T1
2. T0 시점: OrderCompletedEvent가 네트워크 지연으로 늦게 도착 (T0 < T1)
3. T0 이벤트는 isEventOutdated() 체크에서 false를 반환하여 salesCount 업데이트가 무시됨

문제점:
서로 독립적인 메트릭(조회수 vs 판매량)인데도 불구하고, 한 메트릭의 최신 이벤트가 다른 메트릭의 과거 이벤트를 차단합니다.

권장 사항:
메트릭 타입별로 별도의 타임스탬프를 관리하는 것을 고려하세요:

• lastLikeUpdated, lastViewUpdated, lastSalesUpdated

이렇게 하면 각 메트릭이 독립적으로 시간순 정렬을 보장하면서도 서로 영향을 주지 않습니다.

🔎 메트릭별 타임스탬프 분리 예시

-@Column(name = "last_updated", nullable = false)
-private ZonedDateTime lastUpdated;
+@Column(name = "last_like_updated")
+private ZonedDateTime lastLikeUpdated;
+
+@Column(name = "last_view_updated")
+private ZonedDateTime lastViewUpdated;
+
+@Column(name = "last_sales_updated")
+private ZonedDateTime lastSalesUpdated;

 public boolean incrementLikeCount(ZonedDateTime eventOccurredAt) {
-    if (isEventOutdated(eventOccurredAt)) {
+    if (isEventOutdated(eventOccurredAt, this.lastLikeUpdated)) {
         return false;
     }
     this.likeCount++;
-    this.lastUpdated = eventOccurredAt;
+    this.lastLikeUpdated = eventOccurredAt;
     return true;
 }

-private boolean isEventOutdated(ZonedDateTime eventOccurredAt) {
-    return this.lastUpdated != null && eventOccurredAt.isBefore(this.lastUpdated);
+private boolean isEventOutdated(ZonedDateTime eventOccurredAt, ZonedDateTime lastUpdatedForMetric) {
+    return lastUpdatedForMetric != null && eventOccurredAt.isBefore(lastUpdatedForMetric);
 }

Committable suggestion skipped: line range outside the PR's diff.

🤖 Prompt for AI Agents

In
apps/commerce-streamer/src/main/java/com/loopers/domain/metrics/ProductMetrics.java
around lines 56-60, the single lastUpdated field causes events for one metric to
block older-but-valid events for another; add separate timestamp fields (e.g.,
lastLikeUpdated, lastViewUpdated, lastSalesUpdated) annotated with @Column
(nullable = false or true as appropriate), persist them in the entity, and
update all places that call isEventOutdated() or set lastUpdated to instead
compare/update the metric-specific timestamp so likes, views, and sales are
ordered and validated independently.

[coderabbitai]

🛠️ Refactor suggestion | 🟠 Major

중복 코드를 제거하여 유지보수성을 개선하세요.

네 개의 컨슈머 메서드가 거의 동일한 구조를 가지고 있습니다. 이는 코드 중복으로 인해 유지보수가 어렵고 버그 발생 가능성을 높입니다. 공통 로직을 제네릭 헬퍼 메서드로 추출하는 것을 권장합니다.

🔎 제안하는 리팩토링

private <T> void processEvents(
    List<ConsumerRecord<String, String>> records,
    Acknowledgment ack,
    TypeReference<KafkaEventEnvelope<T>> typeRef,
    BiConsumer<String, T> handler,
    String eventName
) {
    log.info("[Consumer] Received {} {} events", records.size(), eventName);
    
    List<ConsumerRecord<String, String>> failedRecords = new ArrayList<>();
    
    for (ConsumerRecord<String, String> record : records) {
        try {
            KafkaEventEnvelope<T> envelope = objectMapper.readValue(
                record.value(), typeRef
            );
            
            handler.accept(envelope.eventId(), envelope.payload());
            retryTracker.clearRetryCount(record.topic(), record.partition(), record.offset());
            
        } catch (Exception e) {
            log.error("[Consumer] Failed to process {} event - offset: {}, key: {}",
                eventName, record.offset(), record.key(), e);
            handleFailedRecord(record, e, failedRecords);
        }
    }
    
    if (!failedRecords.isEmpty()) {
        log.warn("[Consumer] {} records failed, will retry", failedRecords.size());
        throw new RuntimeException(
            String.format("Failed to process %d records", failedRecords.size())
        );
    }
    
    ack.acknowledge();
    log.debug("[Consumer] Acknowledged {} {} events", records.size(), eventName);
}

@KafkaListener(
    topics = {"loopers.commerce.product-liked-v1"},
    groupId = "metrics-aggregator",
    containerFactory = KafkaConfig.BATCH_LISTENER
)
public void consumeProductLiked(
    List<ConsumerRecord<String, String>> records,
    Acknowledgment ack
) {
    processEvents(
        records, ack,
        new TypeReference<KafkaEventEnvelope<ProductLikedEvent>>() {},
        aggregationService::handleProductLiked,
        "product-liked"
    );
}

🤖 Prompt for AI Agents

In
apps/commerce-streamer/src/main/java/com/loopers/interfaces/consumer/ProductMetricsConsumer.java
around lines 43-236, four consumer methods duplicate the same batching,
deserialization, retry-tracking, error-collection and ack logic; extract that
common flow into a single private generic helper (e.g. private <T> void
processEvents(List<ConsumerRecord<String,String>> records, Acknowledgment ack,
TypeReference<KafkaEventEnvelope<T>> typeRef, BiConsumer<String,T> handler,
String eventName)) that: logs receipt, iterates records, deserializes with the
provided TypeReference, calls the handler (use method references to
aggregationService), clears retryTracker on success, uses handleFailedRecord on
catch, throws RuntimeException when failedRecords non-empty, then acknowledges
and logs; replace each consumeX method body with a single call to this helper
passing the appropriate TypeReference, handler and eventName and keep method
visibility and exception handling consistent.

[coderabbitai]

⚠️ Potential issue | 🟡 Minor

🧩 Analysis chain

🏁 Script executed:

# Find TestConfig.java and examine its content
fd -e java "TestConfig.java" -x cat {}

Repository: Loopers-dev-lab/loopers-spring-java-template

Length of output: 612

---

🏁 Script executed:

# Find and examine the test file to verify @Import annotation
fd -e java "MetricsAggregationServiceIdempotencyTest.java" -x cat {}

Repository: Loopers-dev-lab/loopers-spring-java-template

Length of output: 8994

---

TestConfig 임포트를 제거하거나 @Import 어노테이션에 추가하세요.

TestConfig가 임포트되어 있지만 @Import 어노테이션에서 사용되지 않고 있습니다. @DataJpaTest는 JPA 저장소를 자동으로 발견하므로, TestConfig가 필요하지 않다면 임포트를 제거하세요. 또는 테스트에서 TestConfig의 설정이 필요하다면 @Import({MetricsAggregationService.class, TestConfig.class})로 추가하세요.

🤖 Prompt for AI Agents

In
apps/commerce-streamer/src/test/java/com/loopers/application/metrics/MetricsAggregationServiceIdempotencyTest.java
around line 7, the TestConfig import is present but not used in @Import; either
remove the unused import if TestConfig isn't needed or add it to the test's
@Import annotation (e.g., @Import({MetricsAggregationService.class,
TestConfig.class})) so the configuration is actually applied, and then run the
build to ensure there are no unused-import warnings.

[coderabbitai]

⚠️ Potential issue | 🟡 Minor

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Verify spring.kafka.enabled property usage and TestConfig setup

# 1. Check for spring.kafka.enabled property usage
echo "=== Searching for spring.kafka.enabled property references ==="
rg -n "spring\.kafka\.enabled|kafka\.enabled" --type yaml --type java --type kotlin -C 2

# 2. Check for @ConditionalOnProperty with kafka.enabled
echo -e "\n=== Searching for ConditionalOnProperty kafka patterns ==="
rg -n "@ConditionalOnProperty.*kafka|EnableConfigurationProperties.*Kafka" \
  --type java --type kotlin -C 3

# 3. Find and examine TestConfig
echo -e "\n=== Searching for TestConfig file ==="
fd -n "TestConfig" --type f

# 4. Check for KafkaAutoConfiguration exclusion
echo -e "\n=== Searching for KafkaAutoConfiguration references ==="
rg -n "KafkaAutoConfiguration|kafka.*AutoConfiguration" \
  --type java --type kotlin -C 3

# 5. Look for Kafka configuration beans
echo -e "\n=== Searching for Kafka bean definitions ==="
rg -n "@Bean.*kafka|KafkaTemplate|KafkaConsumer|KafkaProducer" \
  --type java --type kotlin -C 2

Repository: Loopers-dev-lab/loopers-spring-java-template

Length of output: 8613

---

spring.kafka.enabled: false 속성 제거 필요

spring.kafka.enabled는 Spring Boot의 표준 속성이 아니며, 코드베이스에서 구현되지도 않았습니다. TestConfig에서 이미 @EnableAutoConfiguration(exclude = {KafkaAutoConfiguration.class})를 통해 Kafka 자동 구성을 제외하고 있으므로 이 YAML 속성은 불필요합니다. 제거하세요.

🤖 Prompt for AI Agents

In apps/commerce-streamer/src/test/resources/application-test.yml around lines
12 to 13, remove the kafka.enabled: false entry because Spring Boot does not
recognize spring.kafka.enabled and Kafka is already excluded via
@EnableAutoConfiguration(exclude = {KafkaAutoConfiguration.class}) in
TestConfig; simply delete this YAML property so the test config remains clean
and avoid relying on a non-existent property.

[coderabbitai]

⚠️ Potential issue | 🟠 Major

Kafka 속성에 브래킷 표기법이 필요합니다.

spring.kafka.producer.properties 맵 내의 속성은 YAML이 점(.)을 중첩 키로 파싱하지 않도록 브래킷 표기법과 따옴표를 사용해야 합니다. 현재 설정은 의도한 대로 작동하지 않을 수 있습니다.

학습된 내용에 따르면, Spring Boot의 relaxed binding은 최상위 Spring Kafka 속성에만 적용되고, properties 맵에는 적용되지 않습니다.

🔎 제안된 수정

      properties:
-       enable.idempotence: true  # 멱등성 프로듀서: 중복 발행 방지
-       max.in.flight.requests.per.connection: 5  # 순서 보장
+       "[enable.idempotence]": true  # 멱등성 프로듀서: 중복 발행 방지
+       "[max.in.flight.requests.per.connection]": 5  # 순서 보장

🤖 Prompt for AI Agents

In modules/kafka/src/main/resources/kafka.yml around lines 19 to 21, the
properties under spring.kafka.producer.properties use dotted keys which YAML
will parse as nested keys; change them to bracket notation with quotes so the
dots are preserved as literal keys (for example use "['enable.idempotence']" and
"['max.in.flight.requests.per.connection']" as map keys) to ensure Spring
receives the exact property names; update the YAML keys accordingly and keep
values unchanged.