[volume - 8] Decoupling with Kafka #191

minor7295 · 2025-12-18T16:19:50Z

📌 Summary

API 서비스(commerce-api)와 독립된 집계 서비스(commerce-streamer)를 추가하고, 두 서비스 간 느슨한 연결을 Kafka 기반 이벤트 파이프라인으로 구현했습니다.

주요 구현 내용:

Producer (commerce-api): Transactional Outbox Pattern을 통한 이벤트 발행 (도메인 트랜잭션과 동일 트랜잭션에서 Outbox 저장, 별도 스케줄러가 Kafka로 발행)
Consumer (commerce-streamer): 이벤트 수취 및 상품 메트릭 집계 (좋아요 수, 판매량, 조회 수)
멱등성 보장: event_handled 테이블(UUID 기반 eventId)과 version 필드(aggregateId별 순차적 버전)를 통한 중복 처리 방지
순서 보장: 파티션 키 기반 이벤트 순서 보장 및 offset.reset: latest 설정으로 불필요한 과거 메시지 처리 방지

구현된 이벤트:

like-events: LikeAdded, LikeRemoved → 좋아요 수 집계
order-events: OrderCreated → 판매량 집계
product-events: ProductViewed → 조회 수 집계

💬 Review Points

1. Transactional Outbox Pattern 구현 방식의 적절성

배경 및 문제 상황:
외부 시스템(Kafka)과의 통신이 필요한 상황에서, 도메인 트랜잭션과 Kafka 발행을 동일 트랜잭션으로 묶을 수 없기 때문에 이벤트 유실 가능성이 있었습니다. 예를 들어, 주문 생성 트랜잭션이 성공했지만 Kafka 발행이 실패하면, 집계 서비스는 해당 주문 이벤트를 받지 못하게 됩니다. 반대로 Kafka 발행은 성공했지만 도메인 트랜잭션이 롤백되면, 실제로는 주문이 생성되지 않았는데 집계 서비스는 주문 이벤트를 받게 되는 문제가 발생합니다.

해결 방안:
이러한 문제를 해결하기 위해 Transactional Outbox Pattern을 적용했습니다. 도메인 트랜잭션과 같은 트랜잭션에서 OutboxEvent를 DB에 먼저 저장하고, 별도 스케줄러가 주기적으로 PENDING 상태의 이벤트를 읽어 Kafka로 발행하는 구조입니다. 이렇게 하면 도메인 트랜잭션이 성공하면 Outbox에 이벤트가 저장되고, 트랜잭션이 롤백되면 Outbox 저장도 함께 롤백되어 일관성이 보장됩니다.

구현 세부사항:

ApplicationEvent → OutboxEvent 변환: OutboxBridgeEventListener가 @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)로 설정되어, 도메인 트랜잭션이 커밋된 후에만 Outbox에 저장합니다. 이렇게 하면 도메인 로직이 실패하여 트랜잭션이 롤백되면 Outbox 저장도 롤백되어 불필요한 이벤트가 저장되지 않습니다.
스케줄러 기반 발행: OutboxEventPublisher가 1초마다 실행되어 PENDING 상태의 이벤트를 최대 100개씩 읽어 Kafka로 발행합니다. 발행 성공 시 PUBLISHED 상태로 변경하고, 실패 시 FAILED 상태로 변경하여 다음 스케줄에서 재시도할 수 있도록 했습니다.

관련 코드:

// OutboxBridgeEventListener.java - 도메인 트랜잭션 커밋 후에만 Outbox에 저장
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
public void handleLikeAdded(LikeEvent.LikeAdded event) {
    outboxEventService.saveEvent(/* ... */);
}

// OutboxEventPublisher.java - 1초마다 PENDING 이벤트를 읽어 Kafka로 발행
@Scheduled(fixedDelay = 1000)
public void publishPendingEvents() {
    List<OutboxEvent> pendingEvents = outboxEventRepository.findPendingEvents(BATCH_SIZE);
    for (OutboxEvent event : pendingEvents) {
        publishEvent(event);
        event.markAsPublished();  // PUBLISHED 상태로 변경
    }
}

고민한 점:

@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)를 사용한 이유는, 도메인 트랜잭션이 성공적으로 커밋된 후에만 Outbox에 저장하여 일관성을 보장하기 위함입니다. 만약 AFTER_COMMIT이 아닌 다른 시점에 저장하면, 도메인 로직이 실패하여 롤백되었는데도 Outbox에 이벤트가 저장될 수 있습니다. 하지만 AFTER_COMMIT은 트랜잭션이 완전히 커밋된 후에 실행되므로, Outbox 저장 실패 시 도메인 트랜잭션을 롤백할 수 없다는 단점이 있습니다. 이 부분에 대한 검토가 필요합니다.
스케줄러 주기(1초)와 배치 크기(100)는 현재 트래픽을 고려하여 설정했지만, 실제 운영 환경에서는 이벤트 발생 빈도와 Kafka 처리 속도를 고려하여 조정이 필요할 수 있습니다. 너무 짧은 주기(예: 100ms)는 DB 부하를 증가시킬 수 있고, 너무 긴 주기(예: 10초)는 이벤트 발행 지연이 발생할 수 있습니다.
개별 이벤트 발행 실패 시 FAILED 상태로 변경하고 계속 진행하도록 했는데, 이렇게 하면 일부 이벤트만 실패해도 다음 스케줄에서 재시도할 수 있습니다. 하지만 FAILED 상태의 이벤트를 별도로 모니터링하거나, 재시도 횟수 제한을 두는 등의 추가 로직이 필요할 수 있습니다.

2. 멱등성 처리 전략: event_handled 테이블과 version 필드의 조합

배경 및 문제 상황:
Kafka는 기본적으로 At Least Once 보장을 제공하므로, 네트워크 오류나 Consumer 재시작 등의 상황에서 동일한 메시지가 여러 번 전달될 수 있습니다. 또한 Producer 측에서도 acks=all, enable.idempotence=true 설정으로 At Least Once를 보장하므로, 동일한 이벤트가 중복 발행될 수 있습니다. 이러한 중복 메시지를 그대로 처리하면 좋아요 수나 판매량이 중복 집계되어 잘못된 메트릭이 생성됩니다.

해결 방안:
중복 처리를 방지하기 위해 두 가지 전략을 조합했습니다:

event_handled 테이블: 동일 eventId의 중복 처리 방지 (동일 이벤트의 완전 중복 방지)
version 필드: 오래된 이벤트가 최신 상태를 덮어쓰는 것 방지 (순서가 뒤바뀐 이벤트 처리 방지)

구현 세부사항:

1) event_handled 테이블을 통한 중복 처리 방지:

각 이벤트에 UUID 기반의 고유한 eventId를 부여합니다.
Consumer에서 이벤트를 처리하기 전에 event_handled 테이블에서 해당 eventId가 이미 처리되었는지 확인합니다.
이미 처리된 경우 스킵하고, 처리되지 않은 경우에만 비즈니스 로직을 실행한 후 event_handled 테이블에 기록을 저장합니다.
event_handled 테이블의 event_id 컬럼에 UNIQUE 제약조건을 설정하여, 동시성 상황에서도 중복 처리를 방지합니다. 만약 두 개의 Consumer 인스턴스가 동시에 같은 이벤트를 처리하려고 하면, 하나는 성공하고 다른 하나는 UNIQUE 제약조건 위반 예외가 발생하여 중복 처리가 방지됩니다.

2) version 필드를 통한 오래된 이벤트 처리 방지:

OutboxEvent에 aggregateId별로 순차적으로 증가하는 version 필드를 부여합니다. 예를 들어, productId=1에 대한 첫 번째 이벤트는 version=1, 두 번째 이벤트는 version=2가 됩니다.
이 version은 Kafka 메시지 헤더에 포함되어 Consumer로 전달됩니다.
Consumer에서 이벤트를 처리할 때, ProductMetrics의 현재 version과 이벤트의 version을 비교합니다. 이벤트의 version이 메트릭의 version보다 크면 업데이트하고, 그렇지 않으면 스킵합니다.
이렇게 하면 네트워크 지연이나 파티션 순서 문제로 인해 오래된 이벤트가 나중에 도착하더라도, 이미 더 최신 버전의 메트릭이 존재하면 오래된 이벤트는 무시됩니다.

관련 코드:

// OutboxEventService.java - eventId(UUID)와 version(aggregateId별 순차 증가) 부여
public void saveEvent(...) {
    String eventId = UUID.randomUUID().toString();
    Long nextVersion = outboxEventRepository.findLatestVersionByAggregateId(...) + 1L;
    // OutboxEvent에 eventId와 version 저장
}

// OutboxEventPublisher.java - Kafka 헤더에 eventId와 version 포함
private void publishEvent(OutboxEvent event) {
    messageBuilder
        .setHeader("eventId", event.getEventId())
        .setHeader("version", event.getVersion());
}

// ProductMetricsConsumer.java - 멱등성 체크 및 버전 비교
public void consumeLikeEvents(...) {
    String eventId = extractEventId(record);
    if (eventHandledService.isAlreadyHandled(eventId)) continue;  // 중복 체크
    
    Long eventVersion = extractVersion(record);
    productMetricsService.incrementLikeCount(productId, eventVersion);  // version 비교 포함
    eventHandledService.markAsHandled(eventId, ...);
}

// ProductMetricsService.java - version 비교로 최신 이벤트만 반영
public void incrementLikeCount(Long productId, Long eventVersion) {
    if (!metrics.shouldUpdate(eventVersion)) return;  // 오래된 이벤트 스킵
    metrics.incrementLikeCount();
}

고민한 점:

event_handled 테이블의 UNIQUE 제약조건으로 동시성 상황에서도 중복 처리를 방지했지만, 이 테이블이 계속 증가하는 문제가 있습니다. 시간이 지나면서 이 테이블의 데이터가 무한정 증가하게 되는데, 이는 스토리지 비용과 조회 성능에 영향을 줄 수 있습니다. TTL(Time To Live)을 설정하여 일정 기간이 지난 레코드를 자동으로 삭제하거나, 아카이빙 전략을 수립하여 오래된 데이터를 별도 테이블로 이동시키는 등의 방안이 필요할 수 있습니다.
version 필드는 aggregateId별로 자동 증가하도록 구현했는데, 이 방식의 장점은 간단하고 순차적인 버전 관리가 가능하다는 것입니다. 하지만 updatedAt 기반 방식과 비교했을 때, updatedAt은 시간 기반이므로 네트워크 지연이나 시스템 시간 불일치 문제가 발생할 수 있습니다. 반면 version은 순차적으로 증가하므로 이러한 문제가 없습니다. 다만, aggregateId별로 별도의 버전을 관리해야 하므로 복잡도가 증가합니다. 이 방식이 적절한지, 또는 다른 방식(예: updatedAt 기반, 또는 이벤트 발생 시점의 타임스탬프 기반)이 더 나은지 검토가 필요합니다.
event_handled 테이블과 version 필드를 모두 사용하는 것이 중복일 수 있다는 의문도 있습니다. 하지만 두 가지는 서로 다른 목적을 가지고 있습니다. event_handled는 동일한 이벤트의 완전 중복을 방지하고, version은 순서가 뒤바뀐 이벤트를 처리하지 않도록 합니다. 예를 들어, 네트워크 문제로 인해 version=3 이벤트가 먼저 도착하고 version=2 이벤트가 나중에 도착하는 경우, event_handled로는 중복을 감지할 수 없지만 version으로는 오래된 이벤트임을 감지할 수 있습니다.

3. 파티션 키 기반 순서 보장과 offset.reset: latest 설정의 조합

배경 및 문제 상황:
Kafka는 기본적으로 파티션 내에서만 순서를 보장하고, 서로 다른 파티션 간의 순서는 보장하지 않습니다. 또한 Consumer가 재시작되거나 새로운 Consumer Group이 시작될 때, 과거의 모든 메시지를 다시 처리하게 되면 이미 처리된 오래된 메시지를 중복 처리하거나, 테스트 환경에서 이전 테스트의 메시지가 다음 테스트에 영향을 줄 수 있습니다.

해결 방안:

1) 파티션 키 기반 순서 보장:

같은 도메인에서 발생한 이벤트는 동일한 파티션에서 처리되도록 파티션 키를 설정했습니다.
like-events와 product-events는 productId를 파티션 키로 사용하여, 같은 상품에 대한 이벤트는 항상 같은 파티션에서 순서대로 처리됩니다.
order-events는 orderId를 파티션 키로 사용하여, 같은 주문에 대한 이벤트는 항상 같은 파티션에서 순서대로 처리됩니다.
이렇게 하면 같은 aggregate root에 대한 이벤트는 순서가 보장되어, 예를 들어 LikeAdded → LikeRemoved 순서로 이벤트가 발생했을 때 Consumer도 같은 순서로 처리할 수 있습니다.

2) offset.reset: latest 설정:

Consumer가 새로운 Consumer Group으로 시작할 때(offset이 없을 때), 최신 메시지부터 읽기 시작하도록 offset.reset: latest를 설정했습니다.
이렇게 하면 Consumer가 재시작되거나 새로운 Consumer Group이 시작될 때, 과거의 오래된 메시지를 처리하지 않고 최신 메시지부터 처리할 수 있습니다.
특히 테스트 환경에서는 이전 테스트에서 발행한 메시지가 다음 테스트에 영향을 주지 않도록 하는 데 유용합니다.

다른 옵션과의 비교:

offset.reset: latest 설정은 새로운 Consumer Group이 시작할 때 최신 메시지부터 읽기 시작하므로, earliest를 사용할 경우 발생하는 문제(이전 테스트의 메시지가 다음 테스트에 영향을 주어 테스트 격리 실패)를 방지하고, manual 방식처럼 복잡한 offset 관리 로직 없이도 테스트 격리를 보장할 수 있습니다.
그러나 latest 설정은 새로운 Consumer Group이 시작할 때만 적용되므로, 같은 Consumer Group을 계속 사용하는 경우 이미 커밋된 offset이 있으면 latest는 적용되지 않고 기존 offset부터 계속 읽게 됩니다.
따라서 통합 테스트에서는 KafkaCleanUp.resetAllTestTopics()로 토픽을 삭제하고 재생성하여 Consumer Group을 초기화함으로써, 매 테스트마다 offset.reset: latest 설정이 적용되도록 하고, 이전 테스트의 메시지가 완전히 제거되어 테스트 간 격리를 보장합니다.

구현 세부사항:

파티션 키 설정:

// OutboxBridgeEventListener.java
// productId를 파티션 키로 사용
public void handleLikeAdded(LikeEvent.LikeAdded event) {
    outboxEventService.saveEvent(
        "LikeAdded",
        event.productId().toString(),  // aggregateId
        "Product",
        event,
        "like-events",
        event.productId().toString()  // partitionKey
    );
}

// OutboxEventPublisher.java
// Kafka 메시지에 파티션 키 설정
private void publishEvent(OutboxEvent event) {
    Object payload = objectMapper.readValue(event.getPayload(), Object.class);
    
    var messageBuilder = MessageBuilder
        .withPayload(payload)
        .setHeader(KafkaHeaders.KEY, event.getPartitionKey())  // 파티션 키 설정
        .setHeader("eventId", event.getEventId())
        .setHeader("version", event.getVersion());
    
    kafkaTemplate.send(event.getTopic(), message);
}

offset.reset: latest 설정:

# modules/kafka/src/main/resources/kafka.yml
spring:
  kafka:
    properties:
      auto.offset.reset: latest  # 새로운 Consumer Group 시작 시 최신 메시지부터
    producer:
      properties:
        acks: all                    # 모든 리플리카에 쓰기 확인
        enable.idempotence: true     # 중복 방지
    consumer:
      properties:
        enable-auto-commit: false    # 수동 커밋 사용
    listener:
      ack-mode: manual               # 수동 커밋 모드

테스트 환경에서의 토픽 및 Consumer Group 초기화:

// KafkaCleanUp.java
// 테스트 실행 전에 토픽과 Consumer Group을 초기화하여 offset.reset: latest가 적용되도록 함
public void resetAllTestTopics() {
    deleteAllTestTopics();
    recreateTestTopics();
}

public void resetAllConsumerGroups() {
    try (AdminClient adminClient = createAdminClient()) {
        Set<String> consumerGroups = adminClient.listConsumerGroups()
            .all()
            .get(5, TimeUnit.SECONDS)
            .stream()
            .map(group -> group.groupId())
            .collect(java.util.stream.Collectors.toSet());
        
        if (!consumerGroups.isEmpty()) {
            DeleteConsumerGroupsResult deleteResult = adminClient.deleteConsumerGroups(consumerGroups);
            deleteResult.all().get(5, TimeUnit.SECONDS);
        }
    } catch (Exception e) {
        // Consumer Group이 없거나 이미 삭제된 경우 무시
    }
}

테스트 환경에서의 격리 보장:

offset.reset: latest는 새로운 Consumer Group이 시작할 때만 적용되므로, 이미 offset이 커밋된 Consumer Group에서는 적용되지 않습니다. 따라서 테스트 환경에서는 각 테스트 실행 전에 KafkaCleanUp.resetAllTestTopics()와 resetAllConsumerGroups()를 호출하여 토픽과 Consumer Group을 초기화했습니다.
이렇게 하면 매 테스트마다 offset.reset: latest 설정이 적용되어, 이전 테스트의 메시지가 다음 테스트에 영향을 주지 않습니다.
또한 테스트 프로파일에서 Consumer Group ID를 동적으로 생성(${spring.application.name}-test-${random.uuid})하여, 각 테스트마다 다른 Consumer Group을 사용하도록 했습니다. 이렇게 하면 이전 테스트의 offset이 다음 테스트에 영향을 주지 않습니다.

관련 코드:

// KafkaCleanUp.java - 테스트 격리를 위한 초기화
public void resetAllTestTopics() {
    deleteAllTestTopics();  // 모든 메시지 제거
    recreateTestTopics();   // 깨끗한 상태로 재생성
}

// ProductMetricsConsumerIntegrationTest.java
@BeforeEach
void setUp() {
    kafkaCleanUp.resetAllTestTopics();      // 토픽 초기화
    kafkaCleanUp.resetAllConsumerGroups();  // Consumer Group 초기화
}

고민한 점:

offset.reset: latest는 새로운 Consumer Group이 시작할 때만 적용되므로, 이미 offset이 커밋된 Consumer Group에서는 적용되지 않습니다. 따라서 테스트 환경에서는 각 테스트 실행 전에 KafkaCleanUp.resetAllTestTopics()와 resetAllConsumerGroups()를 호출하여 토픽과 Consumer Group을 초기화했습니다. 이렇게 하면 매 테스트마다 offset.reset: latest 설정이 적용되어, 이전 테스트의 메시지가 다음 테스트에 영향을 주지 않습니다. 하지만 이 방식은 테스트 실행 시간을 증가시킬 수 있고, 프로덕션 환경에서는 사용할 수 없습니다. 다른 테스트 격리 전략(예: 각 테스트마다 고유한 Consumer Group ID 사용, 또는 테스트용 별도 토픽 사용)이 더 나은지 검토가 필요합니다.
파티션 키를 productId 또는 orderId로 설정했는데, 이로 인한 파티션 불균형 문제가 발생할 수 있습니다. 예를 들어, 특정 상품에 대한 이벤트가 매우 많으면 해당 상품의 파티션에만 메시지가 집중되어 다른 파티션은 비어있을 수 있습니다. 하지만 현재 구현에서는 파티션 키를 사용하여 순서를 보장하는 것이 더 중요하다고 판단했습니다. 만약 파티션 불균형이 심각한 문제가 된다면, 파티션 키를 해시 함수로 변환하거나, 복합 키를 사용하는 등의 방안을 고려할 수 있습니다.
파티션 키를 사용하지 않고 랜덤 키를 사용하면 어떤 문제가 발생할까요? 예를 들어, 같은 상품에 대한 LikeAdded와 LikeRemoved 이벤트가 서로 다른 파티션에 발행되면, Consumer가 LikeRemoved를 먼저 처리하고 LikeAdded를 나중에 처리할 수 있습니다. 이 경우 좋아요 수가 음수가 되거나 잘못된 메트릭이 생성될 수 있습니다. 따라서 파티션 키를 사용하여 같은 aggregate root에 대한 이벤트는 항상 같은 파티션에서 순서대로 처리되도록 하는 것이 중요합니다.

구현 세부사항

1. 내부 이벤트와 외부 이벤트의 구분

배경:
기존에는 JVM 내에서 처리 가능한 느슨한 연결을 위해 Spring Application Event를 사용했습니다. 예를 들어, 주문 생성 시 재고 차감이나 포인트 적립 등의 로직은 같은 애플리케이션 내에서 처리되므로 Application Event로 충분했습니다. 하지만 이번에는 외부 시스템(집계 서비스)과의 통신이 필요하므로, JVM을 벗어나 네트워크를 통해 메시지를 전달할 수 있는 Kafka를 사용하도록 구성했습니다.

내부 이벤트 (Application Event):

같은 JVM 내에서 처리되는 이벤트
예: 주문 생성 → 재고 차감, 포인트 적립 등
Spring의 ApplicationEventPublisher를 통해 발행
동기 또는 비동기로 처리 가능
트랜잭션 내에서 처리되므로 일관성 보장이 상대적으로 쉬움

외부 이벤트 (Kafka Event):

다른 서비스(집계 서비스)로 전달되어야 하는 이벤트
예: 주문 생성 → 판매량 집계, 좋아요 추가 → 좋아요 수 집계 등
Kafka를 통해 네트워크로 전달
비동기로 처리되며, 네트워크 오류나 서비스 다운 등의 상황을 고려해야 함
트랜잭션 경계를 넘어서므로 일관성 보장이 복잡함 (Outbox 패턴 필요)

구조:

도메인 로직 (예: 주문 생성)
    ↓
ApplicationEvent 발행 (JVM 내부)
    ↓
OutboxBridgeEventListener (ApplicationEvent → OutboxEvent 변환)
    - @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    - 도메인 트랜잭션과 같은 트랜잭션에서 OutboxEvent 저장
    ↓
OutboxEvent (DB 저장, PENDING 상태)
    ↓
OutboxEventPublisher (스케줄러가 주기적으로 PENDING 이벤트 읽기)
    - @Scheduled(fixedDelay = 1000)
    - Kafka로 발행 후 PUBLISHED 상태로 변경
    ↓
Kafka Topic (like-events, order-events, product-events)
    ↓
ProductMetricsConsumer (집계 서비스, 다른 JVM)
    - 이벤트 수취 및 메트릭 집계
    - event_handled 테이블로 멱등성 보장
    - version 필드로 최신 이벤트만 반영

관련 코드:

// 내부 이벤트: ApplicationEvent (JVM 내부)
applicationEventPublisher.publishEvent(new OrderEvent.OrderCreated(...));

// 외부 이벤트: ApplicationEvent → OutboxEvent → Kafka
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
public void handleOrderCreated(OrderEvent.OrderCreated event) {
    outboxEventService.saveEvent(/* OutboxEvent로 변환하여 DB 저장 */);
}

@Scheduled(fixedDelay = 1000)
public void publishPendingEvents() {
    // PENDING 이벤트를 읽어 Kafka로 발행
}

2. Producer 설정: At Least Once 보장 및 이벤트 생성 실패 처리

배경:
외부 시스템과의 통신이기 때문에 DB에서 데이터를 직접 조회하거나, 기존의 서비스 로직을 사용하여 검증처리하기는 어려운 상황입니다. 따라서 이벤트 publisher 자체의 설정을 통해 메시지 유실을 방지하고, Consumer 측에서 멱등 처리를 통해 중복을 방지하는 구조로 구성했습니다.

설정 내용:

acks=all: Producer가 메시지를 발행할 때, 모든 리플리카에 쓰기가 완료될 때까지 대기합니다. 이렇게 하면 리플리카 중 하나가 실패하더라도 다른 리플리카에 메시지가 저장되어 유실을 방지할 수 있습니다.
enable.idempotence=true: Producer가 동일한 메시지를 여러 번 발행하더라도 Kafka 브로커가 중복을 제거하여 한 번만 저장하도록 합니다. 이를 통해 네트워크 오류로 인한 재시도 시 중복 메시지가 저장되는 것을 방지합니다.
max.in.flight.requests.per.connection=5: idempotence=true일 때 필수 설정입니다. 동시에 전송할 수 있는 미확인 요청의 최대 개수를 제한합니다.

관련 코드:

# modules/kafka/src/main/resources/kafka.yml
spring:
  kafka:
    producer:
      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일 때 필수

고민한 점:

acks=all은 메시지 유실을 방지하지만, 모든 리플리카에 쓰기가 완료될 때까지 대기하므로 지연 시간이 증가할 수 있습니다. 하지만 메시지 유실을 방지하는 것이 더 중요하다고 판단하여 acks=all을 선택했습니다.
enable.idempotence=true는 Producer 측에서 중복을 제거하지만, 네트워크 오류나 Consumer 재시작 등의 상황에서 동일한 메시지가 여러 번 전달될 수 있습니다. 따라서 Consumer 측에서도 멱등 처리가 필요합니다.

3. Consumer 설정: Manual Ack 처리

배경:
Consumer가 이벤트를 처리하는 과정에서 오류가 발생할 수 있습니다. 만약 자동 커밋을 사용하면, 이벤트를 처리하기 전에 offset이 커밋되어 이벤트가 유실될 수 있습니다. 반대로 이벤트 처리 실패 시에도 offset이 커밋되지 않도록 하여 재처리가 가능하도록 해야 합니다.

설정 내용:

enable-auto-commit: false: 자동 커밋을 비활성화하여 수동 커밋을 사용합니다.
ack-mode: manual: 수동 커밋 모드를 사용하여, 이벤트 처리 성공 후에만 Acknowledgment.acknowledge()를 호출하여 offset을 커밋합니다.

처리 흐름:

Consumer가 메시지를 수신합니다.
각 메시지를 처리합니다 (멱등성 체크, 비즈니스 로직 실행, event_handled 테이블에 기록).
모든 메시지 처리 완료 후 acknowledgment.acknowledge()를 호출하여 offset을 커밋합니다.
만약 처리 중 오류가 발생하면 acknowledgment.acknowledge()를 호출하지 않아 offset이 커밋되지 않으므로, Consumer가 재시작되거나 다음 poll 시 동일한 메시지를 다시 받아 재처리할 수 있습니다.

관련 코드:

# modules/kafka/src/main/resources/kafka.yml
spring:
  kafka:
    consumer:
      group-id: loopers-default-consumer
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.ByteArrayDeserializer
      properties:
        enable-auto-commit: false    # 자동 커밋 비활성화
    listener:
      ack-mode: manual                # 수동 커밋 모드

// ProductMetricsConsumer.java - 처리 성공 후에만 수동 커밋
@KafkaListener(topics = "like-events")
public void consumeLikeEvents(..., Acknowledgment acknowledgment) {
    // 이벤트 처리 로직
    acknowledgment.acknowledge();  // 성공 시에만 커밋
}

고민한 점:

Manual Ack를 사용하면 이벤트 처리 성공 후에만 offset이 커밋되므로, 처리 실패 시 재처리가 가능합니다. 하지만 Consumer가 재시작되거나 장애가 발생하면 처리 중이던 메시지들이 다시 처리될 수 있습니다. 따라서 멱등 처리가 필수적입니다.
배치 처리 시 모든 메시지 처리 완료 후 한 번에 커밋하는 방식과, 각 메시지 처리 후 개별적으로 커밋하는 방식 중 선택할 수 있습니다. 현재는 배치 처리 후 한 번에 커밋하는 방식을 사용했는데, 이는 성능상 유리하지만 일부 메시지 처리 실패 시 전체 배치가 재처리됩니다. 개별 커밋 방식은 성능이 떨어지지만 실패한 메시지만 재처리할 수 있습니다.

4. 멱등성 처리: event_handled 테이블

배경:
Kafka의 At Least Once 보장과 Manual Ack 처리로 인해, 동일한 이벤트가 여러 번 전달될 수 있습니다. 또한 Consumer가 여러 인스턴스로 실행되는 경우, 동시에 같은 이벤트를 처리하려고 할 수 있습니다. 이러한 상황에서 중복 처리를 방지하기 위해 event_handled 테이블을 사용합니다.

구현 방식:

각 이벤트에 고유한 UUID 기반 eventId를 부여합니다.
Consumer에서 이벤트를 처리하기 전에 event_handled 테이블에서 해당 eventId가 이미 처리되었는지 확인합니다.
이미 처리된 경우 스킵하고, 처리되지 않은 경우에만 비즈니스 로직을 실행한 후 event_handled 테이블에 기록을 저장합니다.
event_handled 테이블의 event_id 컬럼에 UNIQUE 제약조건을 설정하여, 동시성 상황에서도 중복 처리를 방지합니다.

관련 코드:

// EventHandled.java - eventId에 UNIQUE 제약조건
@Entity
public class EventHandled {
    @Id
    @Column(unique = true)
    private String eventId;  // UNIQUE 제약조건으로 동시성 보장
}

// ProductMetricsConsumer.java - 중복 체크
if (eventHandledService.isAlreadyHandled(eventId)) continue;
productMetricsService.incrementLikeCount(...);
eventHandledService.markAsHandled(eventId, ...);  // UNIQUE 제약조건으로 중복 방지

고민한 점:

event_handled 테이블을 DB로 구현했지만, Redis를 사용하는 것도 고려할 수 있습니다. Redis는 TTL을 쉽게 설정할 수 있고 조회 성능이 빠르지만, 영속성이 보장되지 않습니다. DB는 영속성이 보장되지만 TTL 설정이 복잡하고 조회 성능이 상대적으로 느릴 수 있습니다. 현재는 DB를 선택했는데, 이는 영속성이 중요하고 TTL은 별도 아카이빙 전략으로 해결할 수 있다고 판단했기 때문입니다.
event_handled 테이블이 계속 증가하는 문제가 있습니다. 시간이 지나면서 이 테이블의 데이터가 무한정 증가하게 되는데, 이는 스토리지 비용과 조회 성능에 영향을 줄 수 있습니다. TTL을 설정하여 일정 기간이 지난 레코드를 자동으로 삭제하거나, 아카이빙 전략을 수립하여 오래된 데이터를 별도 테이블로 이동시키는 등의 방안이 필요할 수 있습니다.

5. 버전 기반 최신 이벤트만 반영

배경:
네트워크 지연이나 파티션 순서 문제로 인해, 이벤트가 발생한 순서와 Consumer가 받는 순서가 다를 수 있습니다. 예를 들어, version=3 이벤트가 먼저 도착하고 version=2 이벤트가 나중에 도착하는 경우, version=2 이벤트를 처리하면 이미 version=3으로 업데이트된 메트릭을 덮어쓰게 되어 잘못된 상태가 됩니다.

해결 방안:

OutboxEvent에 aggregateId별로 순차적으로 증가하는 version 필드를 부여합니다.
이 version은 Kafka 메시지 헤더에 포함되어 Consumer로 전달됩니다.
Consumer에서 이벤트를 처리할 때, ProductMetrics의 현재 version과 이벤트의 version을 비교합니다.
이벤트의 version이 메트릭의 version보다 크면 업데이트하고, 그렇지 않으면 스킵합니다.

구현 방식:

OutboxEventService.saveEvent()에서 aggregateId별 최신 버전을 조회한 후 +1하여 새로운 버전을 부여합니다.
ProductMetrics에도 version 필드를 두고, 업데이트할 때마다 증가시킵니다.
ProductMetrics.shouldUpdate() 메서드로 이벤트 버전과 메트릭 버전을 비교하여, 이벤트가 최신인 경우에만 업데이트합니다.

전체 플로우 다이어그램:

[도메인 이벤트 발생]
    ↓
[OutboxBridgeEventListener]
    ↓
[OutboxEventService.saveEvent()]
    ├─ aggregateId별 최신 버전 조회 (DB)
    ├─ 버전 +1 계산
    └─ OutboxEvent에 version 저장 (DB)
    ↓
[OutboxEventPublisher (스케줄러, 1초마다)]
    ├─ PENDING 이벤트 조회
    ├─ Kafka 메시지 헤더에 version 추가
    └─ Kafka로 발행
    ↓
[ProductMetricsConsumer]
    ├─ Kafka 헤더에서 version 추출
    ├─ ProductMetricsService에 version 전달
    └─ version 비교로 최신 이벤트만 반영

핵심 코드 위치:

1) 버전 생성: OutboxEventService.saveEvent() (59-60줄)

// 집계 ID별 최신 버전 조회 후 +1
Long latestVersion = outboxEventRepository.findLatestVersionByAggregateId(aggregateId, aggregateType);
Long nextVersion = latestVersion + 1L;

2) Kafka 헤더에 추가: OutboxEventPublisher.publishEvent() (99-102줄)

// version이 있으면 헤더에 추가
if (event.getVersion() != null) {
    messageBuilder.setHeader("version", event.getVersion());
}

3) Consumer에서 추출: ProductMetricsConsumer.extractVersion() (374-387줄)

// Kafka 헤더에서 version 추출
Header header = record.headers().lastHeader(VERSION_HEADER);
return Long.parseLong(new String(header.value(), StandardCharsets.UTF_8));

4) 버전 비교: ProductMetricsService에서 eventVersion과 metrics.version 비교하여 최신 이벤트만 반영

이렇게 aggregateId별로 순차적인 버전이 생성되어 Kafka 헤더로 전달되고, Consumer에서 최신 이벤트만 반영하는 데 사용됩니다.

관련 코드:

// OutboxEventService.java - aggregateId별 순차적 version 부여
Long nextVersion = findLatestVersionByAggregateId(...) + 1L;

// ProductMetrics.java - version 비교로 최신 이벤트만 반영
public boolean shouldUpdate(Long eventVersion) {
    return eventVersion > this.version;  // 이벤트 버전이 더 크면 업데이트
}

// ProductMetricsService.java
if (!metrics.shouldUpdate(eventVersion)) return;  // 오래된 이벤트 스킵
metrics.incrementLikeCount();

고민한 점:

version 필드는 aggregateId별로 자동 증가하도록 구현했는데, 이 방식의 장점은 간단하고 순차적인 버전 관리가 가능하다는 것입니다. 하지만 updatedAt 기반 방식과 비교했을 때, updatedAt은 시간 기반이므로 네트워크 지연이나 시스템 시간 불일치 문제가 발생할 수 있습니다. 반면 version은 순차적으로 증가하므로 이러한 문제가 없습니다. 다만, aggregateId별로 별도의 버전을 관리해야 하므로 복잡도가 증가합니다.
version 필드가 aggregateId별로 관리되므로, 같은 aggregateId에 대한 이벤트는 순차적으로 처리되어야 합니다. 하지만 파티션 키를 aggregateId로 설정했으므로, 같은 aggregateId에 대한 이벤트는 같은 파티션에서 순서대로 처리되므로 이 문제는 해결됩니다.

6. 파티션 키 설정

배경:
Kafka는 기본적으로 파티션 내에서만 순서를 보장하고, 서로 다른 파티션 간의 순서는 보장하지 않습니다. 따라서 같은 aggregate root에 대한 이벤트는 같은 파티션에서 처리되어야 순서가 보장됩니다.

7. 퍼블리셔와 컨슈머가 동일 이벤트를 판단하는 기준 설정

배경 및 문제 상황:
aggregate root id(예: productId, orderId)만으로는 해당 event가 동일한 항목을 지정하는지 보장하기 어렵습니다. 예를 들어, 같은 productId에 대해 여러 번 좋아요가 추가되면, 각각은 서로 다른 이벤트이지만 productId만으로는 구분할 수 없습니다. 또한 네트워크 오류나 Consumer 재시작으로 인해 동일한 이벤트가 여러 번 전달될 수 있는데, productId만으로는 이것이 중복인지 새로운 이벤트인지 판단할 수 없습니다.

해결 방안:
이러한 문제를 해결하기 위해 두 가지 식별자를 조합하여 사용합니다:

eventId (UUID): 각 이벤트에 고유한 UUID를 부여하여, 동일한 이벤트는 한 번만 처리될 수 있도록 합니다.
version: 같은 aggregate root에 대한 이벤트의 순서를 보장하고, 오래된 이벤트가 최신 상태를 덮어쓰는 것을 방지합니다.

구현 세부사항:

1) eventId를 통한 동일 이벤트 판단:

OutboxEventService.saveEvent()에서 각 이벤트에 UUID 기반의 고유한 eventId를 부여합니다.
이 eventId는 Kafka 메시지 헤더에 포함되어 Consumer로 전달됩니다.
Consumer에서 이벤트를 처리하기 전에 event_handled 테이블에서 해당 eventId가 이미 처리되었는지 확인합니다.
이미 처리된 경우 스킵하고, 처리되지 않은 경우에만 비즈니스 로직을 실행한 후 event_handled 테이블에 기록을 저장합니다.
이렇게 하면 동일한 이벤트가 여러 번 전달되더라도 한 번만 처리됩니다.

2) version을 통한 불필요한 이벤트 처리 방지:

OutboxEventService.saveEvent()에서 aggregateId별로 순차적으로 증가하는 version을 부여합니다.
이 version은 Kafka 메시지 헤더에 포함되어 Consumer로 전달됩니다.
Consumer에서 이벤트를 처리할 때, ProductMetrics의 현재 version과 이벤트의 version을 비교합니다.
이벤트의 version이 메트릭의 version보다 크면 업데이트하고, 그렇지 않으면 스킵합니다.
이렇게 하면 네트워크 지연이나 파티션 순서 문제로 인해 오래된 이벤트가 나중에 도착하더라도, 이미 더 최신 버전의 메트릭이 존재하면 오래된 이벤트는 무시됩니다.

관련 코드:

// OutboxEventService.java - eventId(UUID)와 version(aggregateId별 순차 증가) 부여
String eventId = UUID.randomUUID().toString();
Long nextVersion = findLatestVersionByAggregateId(...) + 1L;

// OutboxEventPublisher.java - Kafka 헤더에 eventId와 version 포함
messageBuilder.setHeader("eventId", event.getEventId())
              .setHeader("version", event.getVersion());

// ProductMetricsConsumer.java - eventId로 중복 체크, version으로 최신 이벤트만 반영
String eventId = extractEventId(record);
if (eventHandledService.isAlreadyHandled(eventId)) continue;
Long eventVersion = extractVersion(record);
productMetricsService.incrementLikeCount(productId, eventVersion);  // version 비교 포함

고민한 점:

eventId와 version을 모두 사용하는 것이 중복일 수 있다는 의문이 있습니다. 하지만 두 가지는 서로 다른 목적을 가지고 있습니다. eventId는 동일한 이벤트의 완전 중복을 방지하고, version은 순서가 뒤바뀐 이벤트를 처리하지 않도록 합니다. 예를 들어, 네트워크 문제로 인해 version=3 이벤트가 먼저 도착하고 version=2 이벤트가 나중에 도착하는 경우, eventId로는 중복을 감지할 수 없지만 version으로는 오래된 이벤트임을 감지할 수 있습니다. 반대로, 동일한 이벤트가 네트워크 오류로 인해 여러 번 전달되는 경우, version으로는 중복을 감지할 수 없지만 eventId로는 중복을 감지할 수 있습니다.
aggregateId만으로는 동일 이벤트를 판단할 수 없는 이유는, 같은 aggregateId에 대해 여러 번 이벤트가 발생할 수 있기 때문입니다. 예를 들어, 같은 상품에 대해 좋아요가 여러 번 추가되면, 각각은 서로 다른 이벤트이지만 productId만으로는 구분할 수 없습니다. 따라서 각 이벤트에 고유한 eventId를 부여하여 구분해야 합니다.
version은 aggregateId별로 관리되므로, 같은 aggregateId에 대한 이벤트는 순차적으로 처리되어야 합니다. 하지만 파티션 키를 aggregateId로 설정했으므로, 같은 aggregateId에 대한 이벤트는 같은 파티션에서 순서대로 처리되므로 이 문제는 해결됩니다.

6. 파티션 키 설정

배경:
Kafka는 기본적으로 파티션 내에서만 순서를 보장하고, 서로 다른 파티션 간의 순서는 보장하지 않습니다. 따라서 같은 aggregate root에 대한 이벤트는 같은 파티션에서 처리되어야 순서가 보장됩니다.

설정 내용:

like-events, product-events: productId를 파티션 키로 사용하여, 같은 상품에 대한 이벤트는 항상 같은 파티션에서 순서대로 처리됩니다.
order-events: orderId를 파티션 키로 사용하여, 같은 주문에 대한 이벤트는 항상 같은 파티션에서 순서대로 처리됩니다.

관련 코드:

// OutboxBridgeEventListener.java - 파티션 키 설정
outboxEventService.saveEvent(..., partitionKey: productId.toString());  // like-events, product-events
outboxEventService.saveEvent(..., partitionKey: orderId.toString());     // order-events

// OutboxEventPublisher.java - Kafka 메시지에 파티션 키 설정
messageBuilder.setHeader(KafkaHeaders.KEY, event.getPartitionKey());

고민한 점:

파티션 키를 productId 또는 orderId로 설정했는데, 이로 인한 파티션 불균형 문제가 발생할 수 있습니다. 예를 들어, 특정 상품에 대한 이벤트가 매우 많으면 해당 상품의 파티션에만 메시지가 집중되어 다른 파티션은 비어있을 수 있습니다. 하지만 현재 구현에서는 파티션 키를 사용하여 순서를 보장하는 것이 더 중요하다고 판단했습니다. 만약 파티션 불균형이 심각한 문제가 된다면, 파티션 키를 해시 함수로 변환하거나, 복합 키를 사용하는 등의 방안을 고려할 수 있습니다.
파티션 키를 사용하지 않고 랜덤 키를 사용하면 어떤 문제가 발생할까요? 예를 들어, 같은 상품에 대한 LikeAdded와 LikeRemoved 이벤트가 서로 다른 파티션에 발행되면, Consumer가 LikeRemoved를 먼저 처리하고 LikeAdded를 나중에 처리할 수 있습니다. 이 경우 좋아요 수가 음수가 되거나 잘못된 메트릭이 생성될 수 있습니다. 따라서 파티션 키를 사용하여 같은 aggregate root에 대한 이벤트는 항상 같은 파티션에서 순서대로 처리되도록 하는 것이 중요합니다.

✅ Checklist

도메인(애플리케이션) 이벤트 설계
- apps/commerce-api/src/main/java/com/loopers/domain/like/LikeEvent.java
- apps/commerce-api/src/main/java/com/loopers/domain/order/OrderEvent.java
- apps/commerce-api/src/main/java/com/loopers/domain/product/ProductEvent.java
- 이벤트 타입: LikeAdded, LikeRemoved, OrderCreated, ProductViewed
Producer 앱에서 도메인 이벤트 발행
- apps/commerce-api/src/main/java/com/loopers/application/outbox/OutboxBridgeEventListener.java
- apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventPublisher.java
- 토픽: like-events, order-events, product-events
PartitionKey 기반의 이벤트 순서 보장
- apps/commerce-api/src/main/java/com/loopers/domain/outbox/OutboxEvent.java (partitionKey 필드)
- apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventPublisher.java (KafkaHeaders.KEY 설정)
- 파티션 키: like-events, product-events → productId, order-events → orderId
At Least Once 보장 (acks=all, idempotence=true)
- modules/kafka/src/main/resources/kafka.yml (19-21줄)
- 설정: acks: all, enable.idempotence: true, max.in.flight.requests.per.connection: 5
Transactional Outbox Pattern 구현
- apps/commerce-api/src/main/java/com/loopers/domain/outbox/OutboxEvent.java
- apps/commerce-api/src/main/java/com/loopers/application/outbox/OutboxEventService.java
- apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventPublisher.java
- apps/commerce-api/src/main/java/com/loopers/application/outbox/OutboxBridgeEventListener.java
메시지 발행 실패 처리
- apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventPublisher.java (63-69줄)
- 개별 이벤트 발행 실패 시 FAILED 상태로 변경하고 다음 스케줄에서 재시도

⚾ Consumer (7/7)

Consumer가 Metrics 집계 처리
- apps/commerce-streamer/src/main/java/com/loopers/interfaces/consumer/ProductMetricsConsumer.java
- apps/commerce-streamer/src/main/java/com/loopers/application/metrics/ProductMetricsService.java
- 집계 대상: 좋아요 수, 판매량, 조회 수
Manual Ack 처리
- modules/kafka/src/main/resources/kafka.yml (27, 29줄)
- apps/commerce-streamer/src/main/java/com/loopers/interfaces/consumer/ProductMetricsConsumer.java (141, 217줄)
- 설정: enable-auto-commit: false, ack-mode: manual
event_handled 테이블 기반 멱등 처리
- apps/commerce-streamer/src/main/java/com/loopers/domain/eventhandled/EventHandled.java
- apps/commerce-streamer/src/main/java/com/loopers/application/eventhandled/EventHandledService.java
- apps/commerce-streamer/src/main/java/com/loopers/interfaces/consumer/ProductMetricsConsumer.java (91, 275줄)
- UNIQUE 제약조건으로 동시성 상황에서도 중복 방지
version 기준 최신 이벤트만 반영
- apps/commerce-api/src/main/java/com/loopers/domain/outbox/OutboxEvent.java (56줄, version 필드)
- apps/commerce-api/src/main/java/com/loopers/application/outbox/OutboxEventService.java (59-60줄, 버전 생성)
- apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventPublisher.java (99-102줄, 헤더에 추가)
- apps/commerce-streamer/src/main/java/com/loopers/interfaces/consumer/ProductMetricsConsumer.java (100, 284줄, 헤더에서 추출)
- apps/commerce-streamer/src/main/java/com/loopers/application/metrics/ProductMetricsService.java (100, 126줄, 버전 비교)
- apps/commerce-streamer/src/main/java/com/loopers/domain/metrics/ProductMetrics.java (shouldUpdate 메서드)
product_metrics 테이블에 upsert
- apps/commerce-streamer/src/main/java/com/loopers/domain/metrics/ProductMetrics.java
- apps/commerce-streamer/src/main/java/com/loopers/application/metrics/ProductMetricsService.java
- apps/commerce-streamer/src/main/java/com/loopers/infrastructure/metrics/ProductMetricsRepositoryImpl.java
중복 메시지 재전송 테스트
- apps/commerce-streamer/src/test/java/com/loopers/interfaces/consumer/ProductMetricsConsumerTest.java (352줄, handlesDuplicateMessagesIdempotently())
- 동일한 eventId를 가진 메시지가 한 번만 처리되는지 검증
재고 소진 시 상품 캐시 갱신
- apps/commerce-api/src/main/java/com/loopers/application/product/ProductEventHandler.java (147-151줄)
- apps/commerce-api/src/main/java/com/loopers/application/product/ProductCacheService.java (evictProductDetailCache 메서드)
- 재고 차감 후 stock == 0 체크하여 캐시 무효화

📎 References

Summary by CodeRabbit

릴리스 노트

새로운 기능
- 상품 조회, 좋아요, 주문 이벤트에 대한 실시간 메트릭 추적 시스템 추가
- 신뢰할 수 있는 이벤트 기반 아키텍처로 데이터 처리 안정성 향상
- 중복 이벤트 처리 방지를 위한 멱등성 보장 메커니즘 구현
테스트
- 이벤트 처리 및 메트릭 추적 관련 종합 테스트 스위트 추가

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

coderabbitai · 2025-12-18T16:19:56Z

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은 transactional outbox 패턴을 구현하여 도메인 이벤트를 데이터베이스에 저장한 후 별도 프로세스에서 Kafka로 발행하는 이벤트 기반 아키텍처를 도입합니다. commerce-api에서 ProductEvent, LikeEvent, OrderEvent를 발행하고, OutboxEventPublisher가 주기적으로 pending 이벤트를 Kafka에 배포하며, commerce-streamer의 ProductMetricsConsumer가 이벤트를 소비하여 상품 메트릭을 업데이트합니다.

Changes

Cohort / File(s)	변경 요약
Kafka 모듈 설정 `modules/kafka/src/main/java/com/loopers/confg/kafka/KafkaConfig.java`, `modules/kafka/src/main/resources/kafka.yml`, `modules/kafka/src/testFixtures/java/com/loopers/testcontainers/KafkaTestContainersConfig.java`, `modules/kafka/src/testFixtures/java/com/loopers/utils/KafkaCleanUp.java`	KafkaConfig에서 제네릭 타입을 Object에서 String, Object로 변경하고 6개 토픽 빈 추가; kafka.yml에 At-Least-Once 발행 설정 및 환경 변수 지원 추가; Kafka 테스트 컨테이너 설정 및 토픽/컨슈머 그룹 관리 유틸리티 추가
Outbox 패턴 도메인 및 서비스 `apps/commerce-api/src/main/java/com/loopers/domain/outbox/OutboxEvent.java`, `apps/commerce-api/src/main/java/com/loopers/domain/outbox/OutboxEventRepository.java`	OutboxEvent JPA 엔티티 및 OutboxEventRepository 인터페이스 추가; 트랜잭셔널 outbox 패턴을 위한 PENDING/PUBLISHED/FAILED 상태 관리
Outbox 애플리케이션 계층 `apps/commerce-api/src/main/java/com/loopers/application/outbox/OutboxEventService.java`, `apps/commerce-api/src/main/java/com/loopers/application/outbox/OutboxBridgeEventListener.java`	OutboxEventService로 이벤트 직렬화 및 저장; OutboxBridgeEventListener에서 LikeEvent, OrderEvent, ProductEvent를 수신하여 Outbox에 저장
Outbox 인프라 및 발행 `apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventJpaRepository.java`, `apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventRepositoryImpl.java`, `apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventPublisher.java`	JPA 리포지토리 및 구현; OutboxEventPublisher는 1초 주기로 pending 이벤트를 배치로 조회하여 Kafka에 발행
Product 이벤트 `apps/commerce-api/src/main/java/com/loopers/domain/product/ProductEvent.java`, `apps/commerce-api/src/main/java/com/loopers/domain/product/ProductEventPublisher.java`, `apps/commerce-api/src/main/java/com/loopers/infrastructure/product/ProductEventPublisherImpl.java`, `apps/commerce-api/src/main/java/com/loopers/application/catalog/CatalogFacade.java`	ProductEvent 및 ProductViewed 레코드 도입; ProductEventPublisher 인터페이스 및 구현; CatalogFacade에서 product 조회 시 ProductViewed 이벤트 발행
의존성 및 설정 `apps/commerce-api/build.gradle.kts`, `apps/commerce-api/src/main/resources/application.yml`	Kafka 모듈 의존성 추가; kafka.yml 임포트 추가
Streamer 도메인 및 서비스 `apps/commerce-streamer/src/main/java/com/loopers/domain/event/LikeEvent.java`, `apps/commerce-streamer/src/main/java/com/loopers/domain/event/OrderEvent.java`, `apps/commerce-streamer/src/main/java/com/loopers/domain/event/ProductEvent.java`, `apps/commerce-streamer/src/main/java/com/loopers/domain/eventhandled/EventHandled.java`, `apps/commerce-streamer/src/main/java/com/loopers/domain/eventhandled/EventHandledRepository.java`, `apps/commerce-streamer/src/main/java/com/loopers/domain/metrics/ProductMetrics.java`, `apps/commerce-streamer/src/main/java/com/loopers/domain/metrics/ProductMetricsRepository.java`	이벤트 DTO 레코드 정의; EventHandled 엔티티 및 리포지토리로 멱등성 관리; ProductMetrics 엔티티로 상품 메트릭 추적
Streamer 애플리케이션 계층 `apps/commerce-streamer/src/main/java/com/loopers/application/eventhandled/EventHandledService.java`, `apps/commerce-streamer/src/main/java/com/loopers/application/metrics/ProductMetricsService.java`	EventHandledService로 이벤트 중복 처리 방지; ProductMetricsService로 like/sales/view count 업데이트 및 버전 관리
Streamer 인프라 및 컨슈머 `apps/commerce-streamer/src/main/java/com/loopers/infrastructure/eventhandled/EventHandledJpaRepository.java`, `apps/commerce-streamer/src/main/java/com/loopers/infrastructure/eventhandled/EventHandledRepositoryImpl.java`, `apps/commerce-streamer/src/main/java/com/loopers/infrastructure/metrics/ProductMetricsJpaRepository.java`, `apps/commerce-streamer/src/main/java/com/loopers/infrastructure/metrics/ProductMetricsRepositoryImpl.java`, `apps/commerce-streamer/src/main/java/com/loopers/interfaces/consumer/ProductMetricsConsumer.java`	JPA 리포지토리 구현; ProductMetricsConsumer는 like-events, order-events, product-events 토픽을 배치로 수신하여 메트릭 업데이트 및 멱등성 확인
테스트 `apps/commerce-api/src/test/java/com/loopers/application/outbox/Test.java`, `apps/commerce-api/src/test/java/com/loopers/domain/outbox/Test.java`, `apps/commerce-api/src/test/java/com/loopers/infrastructure/outbox/Test.java`, `apps/commerce-streamer/src/test/java/com/loopers/application/eventhandled/Test.java`, `apps/commerce-streamer/src/test/java/com/loopers/application/metrics/Test.java`, `apps/commerce-streamer/src/test/java/com/loopers/domain/metrics/Test.java`, `apps/commerce-streamer/src/test/java/com/loopers/interfaces/consumer/*Test.java`	OutboxBridgeEventListener, OutboxEventService, OutboxEvent, OutboxEventPublisher, EventHandledService, ProductMetricsService, ProductMetrics, ProductMetricsConsumer에 대한 단위 및 통합 테스트

Sequence Diagram

sequenceDiagram
    participant CatalogFacade
    participant ProductEventPublisher
    participant OutboxBridgeEventListener
    participant OutboxEventService
    participant OutboxEventRepository
    participant Database
    participant OutboxEventPublisher_Infra
    participant Kafka
    participant ProductMetricsConsumer
    participant ProductMetricsService

    CatalogFacade->>ProductEventPublisher: publish(ProductViewed)
    ProductEventPublisher->>OutboxBridgeEventListener: publishEvent(ProductEvent)
    OutboxBridgeEventListener->>OutboxEventService: saveEvent(eventType, aggregateId, ...)
    OutboxEventService->>OutboxEventRepository: save(OutboxEvent)
    OutboxEventRepository->>Database: INSERT outbox_event (status=PENDING)
    
    rect rgb(200, 220, 240)
        Note over OutboxEventPublisher_Infra: 1초 주기 스케줄
        OutboxEventPublisher_Infra->>OutboxEventRepository: findPendingEvents(100)
        OutboxEventRepository->>Database: SELECT * WHERE status=PENDING
        Database-->>OutboxEventPublisher_Infra: List<OutboxEvent>
    end
    
    loop 각 Pending 이벤트
        OutboxEventPublisher_Infra->>Kafka: send(topic, key, payload, headers)
        alt 발행 성공
            OutboxEventPublisher_Infra->>OutboxEventRepository: save(status=PUBLISHED, publishedAt)
        else 발행 실패
            OutboxEventPublisher_Infra->>OutboxEventRepository: save(status=FAILED)
        end
    end
    
    ProductMetricsConsumer->>Kafka: poll(like-events, order-events, product-events)
    Kafka-->>ProductMetricsConsumer: ConsumerRecord<String, Object>
    
    loop 배치 처리
        rect rgb(240, 220, 200)
            Note over ProductMetricsConsumer: 멱등성 확인
            ProductMetricsConsumer->>ProductMetricsService: eventHandledService.isAlreadyHandled(eventId)
        end
        
        alt 미처리 이벤트
            ProductMetricsConsumer->>ProductMetricsService: incrementViewCount(productId, version)
            ProductMetricsService->>Database: SELECT * FOR UPDATE WHERE productId (PESSIMISTIC_WRITE)
            ProductMetricsService->>Database: UPDATE product_metrics SET viewCount++, version++
            ProductMetricsConsumer->>ProductMetricsService: markAsHandled(eventId)
        else 중복 이벤트
            Note over ProductMetricsConsumer: 스킵 (DataIntegrityViolationException)
        end
    end
    
    ProductMetricsConsumer->>Kafka: acknowledge(batch)

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~60 minutes

추가 주의가 필요한 영역:

OutboxEventPublisher: 배치 처리 중 개별 이벤트 실패 시 격리 처리, Kafka 헤더 구성 및 비동기 전송 로직의 정확성
ProductMetricsConsumer: 배치 멀티 토픽 처리, JSON 역직렬화 폴백, 멱등성을 위한 DataIntegrityViolationException 처리
ProductMetrics 동시성: pessimistic write lock을 사용한 version 기반 업데이트 로직 및 shouldUpdate 검증
Outbox 상태 관리: PENDING → PUBLISHED/FAILED 상태 전이 및 publishedAt 타이밍
이벤트 직렬화/역직렬화: ObjectMapper를 통한 payload 처리 및 에러 케이스

Possibly related PRs

[volume-7] Decoupling with Event #172: 같은 transactional outbox 패턴, OutboxEventService/Publisher 및 도메인 이벤트 타입/핸들러 클래스 구현
[volume-7] Decoupling with Event #186: 동일한 이벤트 기반 분리 작업 (product/order 이벤트, 애플리케이션 이벤트 발행/처리, Kafka/outbox 통합)
[volume-3] 커머스 상품/좋아요/주문 구현 #93: 유사한 product event 타입 및 이벤트 발행 기반 구조 도입

Suggested labels

enhancement

Poem

🐰 토끼가 소식을 전하네,
우편함에 담아 안전하게,
카프카 물줄기 따라 흘러,
메트릭 수집은 우아하게.
멱등성의 춤, 일관성의 노래~

Pre-merge checks and finishing touches

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 50.00% which is insufficient. The required threshold is 80.00%.	You can run `@coderabbitai generate docstrings` to improve docstring coverage.

✅ Passed checks (2 passed)

Check name	Status	Explanation
Description check	✅ Passed	PR 설명이 템플릿의 모든 필수 섹션(📌 Summary, 💬 Review Points, ✅ Checklist)을 포함하고 있으며, 각 섹션이 상세하게 작성되어 있습니다.
Title check	✅ Passed	PR 제목은 주요 변경사항인 Kafka 도입과 마이크로서비스 간 이벤트 기반 통신 아키텍처를 명확하게 요약하고 있습니다.

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

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

minor7295 · 2025-12-18T16:20:18Z

@coderabbitai review

coderabbitai · 2025-12-18T16:20:24Z

✅ Actions performed

Review triggered.

Note: CodeRabbit is an incremental review system and does not re-review already reviewed commits. This command is applicable only when automatic reviews are paused.

coderabbitai

Actionable comments posted: 7

🧹 Nitpick comments (16)

apps/commerce-streamer/src/main/java/com/loopers/domain/event/ProductEvent.java (1)
21-26: DTO 검증 로직 추가를 고려하세요.

commerce-api의 ProductEvent.ProductViewed는 생성자에서 productId != null 검증을 수행하지만, 이 DTO에는 검증 로직이 없습니다. Kafka에서 잘못된 데이터를 수신할 경우 null 값이 전파되어 예상치 못한 동작이 발생할 수 있습니다.
🔎 다음 검증 로직 추가를 고려하세요:
 public record ProductViewed(
 Long productId,
 Long userId,
 LocalDateTime occurredAt
 ) {
+ public ProductViewed {
+ if (productId == null) {
+ throw new IllegalArgumentException("productId는 필수입니다.");
+ }
+ }
 }
참고: Line 10의 주석에서 언급한 대로, 향후 공유 모듈로 분리하면 이러한 중복과 불일치를 해결할 수 있습니다.
apps/commerce-api/src/main/java/com/loopers/application/catalog/CatalogFacade.java (1)

122-124: 캐시 히트 시 트랜잭션 오버헤드를 고려하세요.

캐시에서 상품 정보를 조회하는 경우에도 @Transactional 메서드 내에서 실행되므로 DB 커넥션을 획득하게 됩니다. 캐시 히트율이 높은 경우 불필요한 트랜잭션 오버헤드가 발생할 수 있습니다.

성능이 중요한 경우, 다음과 같은 최적화를 고려할 수 있습니다:

이벤트 발행을 별도의 비동기 메서드로 분리

캐시 히트 경로에서는 @Transactional(propagation = Propagation.NOT_SUPPORTED) 사용

다만, 현재 구조가 outbox 패턴의 일관성을 보장하므로, 실제 성능 문제가 확인되기 전까지는 현재 구현을 유지하는 것이 좋습니다.

apps/commerce-api/src/test/java/com/loopers/infrastructure/outbox/OutboxEventPublisherIntegrationTest.java (1)

37-42: TODO 주석: 통합 테스트 구현이 필요합니다.

통합 테스트의 뼈대는 잘 구성되어 있지만, 실제 검증 로직이 구현되지 않았습니다. Outbox 패턴의 핵심 기능인 이벤트 저장 및 Kafka 발행을 검증하는 테스트를 추가해야 합니다.

테스트 구현을 도와드릴까요? 다음과 같은 시나리오를 포함할 수 있습니다:

OutboxEvent 저장 후 PENDING 상태 확인

OutboxEventPublisher 실행 후 Kafka 메시지 수신 확인

발행 후 PUBLISHED 상태로 변경 확인

또는 이 작업을 추적하기 위한 이슈를 생성해 드릴까요?
apps/commerce-api/src/test/java/com/loopers/domain/outbox/OutboxEventTest.java (1)
51-75: Thread.sleep 사용을 개선할 수 있습니다.

Line 66의 Thread.sleep(1)은 타임스탬프 차이를 보장하기 위해 사용되었지만, 테스트를 느리게 만들고 불안정할 수 있습니다.
🔎 다음 방법들을 고려해보세요:

방법 1: 단순히 publishedAt이 null이 아닌지만 확인
- LocalDateTime beforePublish = outboxEvent.getCreatedAt();
- Thread.sleep(1); // 시간 차이를 보장하기 위한 작은 지연
-
 // act
 outboxEvent.markAsPublished();
 
 // assert
 assertThat(outboxEvent.getStatus()).isEqualTo(OutboxEvent.OutboxStatus.PUBLISHED);
 assertThat(outboxEvent.getPublishedAt()).isNotNull();
- assertThat(outboxEvent.getPublishedAt()).isAfter(beforePublish);
+ assertThat(outboxEvent.getPublishedAt()).isAfterOrEqualTo(outboxEvent.getCreatedAt());
방법 2: Clock을 주입받도록 OutboxEvent 리팩토링 (더 복잡하지만 테스트하기 좋음)
apps/commerce-api/src/test/java/com/loopers/application/outbox/OutboxEventServiceTest.java (1)
132-147: 테스트 정확성 개선 제안: 실제 예외 타입 사용

ObjectMapper.writeValueAsString()은 JsonProcessingException (checked exception)을 발생시킵니다. 실제 동작을 더 정확히 테스트하려면 RuntimeException 대신 JsonProcessingException을 throw하도록 수정하는 것이 좋습니다.
🔎 제안된 수정 사항:
 @DisplayName("JSON 직렬화 실패 시 예외를 발생시킨다.")
 @Test
 void throwsException_whenJsonSerializationFails() throws Exception {
 // arrange
 LikeEvent.LikeAdded event = new LikeEvent.LikeAdded(100L, 1L, LocalDateTime.now());
 when(objectMapper.writeValueAsString(event))
- .thenThrow(new RuntimeException("JSON 직렬화 실패"));
+ .thenThrow(new com.fasterxml.jackson.core.JsonProcessingException("JSON 직렬화 실패") {});

 // act & assert
 assertThatThrownBy(() -> 
 outboxEventService.saveEvent("LikeAdded", "1", "Product", event, "like-events", "1")
 ).isInstanceOf(RuntimeException.class)
 .hasMessageContaining("Outbox 이벤트 저장 실패");

 verify(outboxEventRepository, never()).save(any());
 }
apps/commerce-streamer/src/main/java/com/loopers/domain/eventhandled/EventHandled.java (1)
56-61: 선택적 개선: 생성자 파라미터 null 검증

필수 필드(eventId, eventType, topic)에 대한 null 검증을 생성자에 추가하면 DB 제약조건 위반 전에 빠르게 실패할 수 있습니다. 현재 구현도 DB 레벨에서 보호되므로 선택사항입니다.
🔎 제안된 수정 사항:
 public EventHandled(String eventId, String eventType, String topic) {
+ if (eventId == null || eventId.isBlank()) {
+ throw new IllegalArgumentException("eventId는 필수입니다.");
+ }
+ if (eventType == null || eventType.isBlank()) {
+ throw new IllegalArgumentException("eventType은 필수입니다.");
+ }
+ if (topic == null || topic.isBlank()) {
+ throw new IllegalArgumentException("topic은 필수입니다.");
+ }
 this.eventId = eventId;
 this.eventType = eventType;
 this.topic = topic;
 this.handledAt = LocalDateTime.now();
 }
apps/commerce-api/src/main/java/com/loopers/domain/product/ProductEvent.java (1)
5-14: 문서화 개선 제안: 중복 구조에 대한 주의사항 추가

commerce-streamer의 ProductEvent와 동일한 구조를 가지고 있습니다. LikeEvent에서처럼 향후 공유 모듈로 분리를 고려한다는 주석을 추가하면 일관성이 향상됩니다.
🔎 JavaDoc 보완 예시:
 /**
 * 상품 도메인 이벤트.
 * 
 * 상품 도메인의 중요한 상태 변화를 나타내는 이벤트들입니다.
 * 
+ * 
+ * 주의: 이 클래스는 commerce-streamer의 ProductEvent와 동일한 구조를 가집니다.
+ * 향후 공유 모듈로 분리하는 것을 고려해야 합니다.
+ * 
 *
 * @author Loopers
 * @version 1.0
 */
apps/commerce-api/src/main/java/com/loopers/application/outbox/OutboxEventService.java (1)
76-80: 예외 처리 개선을 고려해 주세요.

Exception을 포괄적으로 catch하는 대신 JsonProcessingException과 같은 특정 예외를 처리하면 디버깅이 더 용이합니다.
🔎 개선 제안:
- } catch (Exception e) {
+ } catch (JsonProcessingException e) {
 log.error("Outbox 이벤트 저장 실패: eventType={}, aggregateId={}", 
 eventType, aggregateId, e);
- throw new RuntimeException("Outbox 이벤트 저장 실패", e);
+ throw new RuntimeException("Outbox 이벤트 JSON 직렬화 실패", e);
 }
apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventJpaRepository.java (1)
21-25: 네이티브 쿼리의 데이터베이스 이식성 고려.

LIMIT 절은 MySQL/PostgreSQL에서 동작하지만, Oracle이나 SQL Server에서는 다른 구문이 필요합니다. 다양한 데이터베이스를 지원해야 한다면 JPQL과 Pageable을 사용하는 것이 좋습니다.

또한, 여러 인스턴스가 스케줄러를 실행하는 경우 중복 발행을 방지하기 위해 비관적 락(FOR UPDATE SKIP LOCKED)을 고려해 주세요.
🔎 개선 제안:
- @Query(value = "SELECT * FROM outbox_event e " +
- "WHERE e.status = 'PENDING' " +
- "ORDER BY e.created_at ASC " +
- "LIMIT :limit", nativeQuery = true)
- List<OutboxEvent> findPendingEvents(@Param("limit") int limit);
+ @Query(value = "SELECT * FROM outbox_event e " +
+ "WHERE e.status = 'PENDING' " +
+ "ORDER BY e.created_at ASC " +
+ "LIMIT :limit " +
+ "FOR UPDATE SKIP LOCKED", nativeQuery = true)
+ List<OutboxEvent> findPendingEvents(@Param("limit") int limit);
apps/commerce-streamer/src/test/java/com/loopers/domain/metrics/ProductMetricsTest.java (1)

40-41: Thread.sleep 대안 고려.

Thread.sleep(1)은 타임스탬프 차이를 보장하기 위한 것이지만, 테스트 안정성을 위해 java.time.Clock을 주입받아 테스트에서 시간을 제어하는 방식을 고려해 볼 수 있습니다.
apps/commerce-api/src/main/java/com/loopers/domain/outbox/OutboxEventRepository.java (1)
37-38: findById 반환 타입에 Optional 사용을 고려해 주세요.

ID로 조회 시 결과가 없을 수 있으므로 Optional<OutboxEvent>를 반환하면 호출자가 null 체크를 강제하게 되어 NPE를 방지할 수 있습니다.
🔎 개선 제안:
- OutboxEvent findById(Long id);
+ Optional<OutboxEvent> findById(Long id);
import 추가:
import java.util.Optional;
apps/commerce-api/src/test/java/com/loopers/infrastructure/outbox/OutboxEventPublisherTest.java (1)
52-52: 배치 크기 상수 추출을 고려해 주세요.

100이 여러 테스트에서 반복 사용되고 있습니다. 상수로 추출하면 실제 Publisher의 배치 크기가 변경될 때 테스트 유지보수가 쉬워집니다.
🔎 개선 제안:
private static final int BATCH_SIZE = 100;

// 사용 예시
when(outboxEventRepository.findPendingEvents(BATCH_SIZE)).thenReturn(pendingEvents);
apps/commerce-api/src/main/java/com/loopers/application/outbox/OutboxBridgeEventListener.java (1)
53-140: 토픽 이름을 상수로 추출하는 것을 고려하세요

"like-events", "order-events", "product-events" 등 토픽 이름이 여러 곳에서 사용됩니다. 오타 방지와 유지보수를 위해 상수나 설정으로 관리하면 좋습니다.
🔎 토픽 상수 추출 예시:
public final class KafkaTopics {
 public static final String LIKE_EVENTS = "like-events";
 public static final String ORDER_EVENTS = "order-events";
 public static final String PRODUCT_EVENTS = "product-events";
 
 private KafkaTopics() {}
}
modules/kafka/src/testFixtures/java/com/loopers/utils/KafkaCleanUp.java (1)
35-51: 테스트 유틸리티가 프로덕션에서 실수로 활성화될 위험이 있습니다.

@Component 어노테이션은 컴포넌트 스캔 시 어떤 컨텍스트에서든 이 빈을 등록할 수 있습니다. testFixtures 소스 세트는 일반적으로 프로덕션에 포함되지 않지만, 명시적으로 테스트 전용임을 표시하는 것이 더 안전합니다.
🔎 테스트 프로파일 제한을 추가하는 것을 고려하세요:
+import org.springframework.context.annotation.Profile;
+
+@Profile("test")
 @Component
 public class KafkaCleanUp {
또는 이 클래스를 @Configuration 클래스에서 명시적으로 @Bean으로 등록하여 제어를 강화할 수 있습니다.
apps/commerce-streamer/src/main/java/com/loopers/interfaces/consumer/ProductMetricsConsumer.java (1)
232-240: 파싱 메서드의 일관성을 개선하세요.

parseLikeEvent는 instanceof 체크 없이 바로 JSON 파싱을 시도하는 반면, parseOrderCreatedEvent와 parseProductViewedEvent는 instanceof 체크를 먼저 수행합니다. 일관성을 위해 parseLikeEvent도 동일한 패턴을 따르는 것이 좋습니다.
🔎 일관된 파싱 패턴 적용:
 private LikeEvent.LikeAdded parseLikeEvent(Object value) {
 try {
+ if (value instanceof LikeEvent.LikeAdded) {
+ return (LikeEvent.LikeAdded) value;
+ }
+ 
 // JSON 문자열인 경우 파싱
 String json = value instanceof String ? (String) value : objectMapper.writeValueAsString(value);
 return objectMapper.readValue(json, LikeEvent.LikeAdded.class);
 } catch (Exception e) {
 throw new RuntimeException("LikeEvent 파싱 실패", e);
 }
 }
modules/kafka/src/main/java/com/loopers/confg/kafka/KafkaConfig.java (1)
78-166: 토픽 설정이 개발/테스트 환경에 적합하게 구성되었습니다.

각 토픽의 파티션 키 전략이 명확히 문서화되어 있으며, 3개 파티션은 적절한 병렬 처리를 제공합니다. 다만 replicas=1 설정은 개발/테스트 환경에만 적합하므로, 프로덕션 환경에서는 최소 3개 이상의 레플리카를 사용하도록 설정을 외부화하는 것을 권장합니다.
🔎 프로덕션 환경을 위한 설정 외부화 제안:

application.yml에 토픽 설정을 추가하여 환경별로 다른 레플리카 수를 사용하도록 할 수 있습니다:
kafka:
 topic:
 partitions: 3
 replicas: 3 # production
 min-insync-replicas: 2
그리고 @Value 또는 @ConfigurationProperties를 통해 이 값들을 주입받아 사용합니다.

📜 Review details

Configuration used: Repository UI

Review profile: CHILL

Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between fb769b2 and 3964c8e.

📒 Files selected for processing (41)

apps/commerce-api/build.gradle.kts (2 hunks)
apps/commerce-api/src/main/java/com/loopers/application/catalog/CatalogFacade.java (4 hunks)
apps/commerce-api/src/main/java/com/loopers/application/outbox/OutboxBridgeEventListener.java (1 hunks)
apps/commerce-api/src/main/java/com/loopers/application/outbox/OutboxEventService.java (1 hunks)
apps/commerce-api/src/main/java/com/loopers/domain/outbox/OutboxEvent.java (1 hunks)
apps/commerce-api/src/main/java/com/loopers/domain/outbox/OutboxEventRepository.java (1 hunks)
apps/commerce-api/src/main/java/com/loopers/domain/product/ProductEvent.java (1 hunks)
apps/commerce-api/src/main/java/com/loopers/domain/product/ProductEventPublisher.java (1 hunks)
apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventJpaRepository.java (1 hunks)
apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventPublisher.java (1 hunks)
apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventRepositoryImpl.java (1 hunks)
apps/commerce-api/src/main/java/com/loopers/infrastructure/product/ProductEventPublisherImpl.java (1 hunks)
apps/commerce-api/src/main/resources/application.yml (1 hunks)
apps/commerce-api/src/test/java/com/loopers/application/outbox/OutboxBridgeEventListenerTest.java (1 hunks)
apps/commerce-api/src/test/java/com/loopers/application/outbox/OutboxEventServiceTest.java (1 hunks)
apps/commerce-api/src/test/java/com/loopers/domain/outbox/OutboxEventTest.java (1 hunks)
apps/commerce-api/src/test/java/com/loopers/infrastructure/outbox/OutboxEventPublisherIntegrationTest.java (1 hunks)
apps/commerce-api/src/test/java/com/loopers/infrastructure/outbox/OutboxEventPublisherTest.java (1 hunks)
apps/commerce-streamer/src/main/java/com/loopers/application/eventhandled/EventHandledService.java (1 hunks)
apps/commerce-streamer/src/main/java/com/loopers/application/metrics/ProductMetricsService.java (1 hunks)
apps/commerce-streamer/src/main/java/com/loopers/domain/event/LikeEvent.java (1 hunks)
apps/commerce-streamer/src/main/java/com/loopers/domain/event/OrderEvent.java (1 hunks)
apps/commerce-streamer/src/main/java/com/loopers/domain/event/ProductEvent.java (1 hunks)
apps/commerce-streamer/src/main/java/com/loopers/domain/eventhandled/EventHandled.java (1 hunks)
apps/commerce-streamer/src/main/java/com/loopers/domain/eventhandled/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/eventhandled/EventHandledJpaRepository.java (1 hunks)
apps/commerce-streamer/src/main/java/com/loopers/infrastructure/eventhandled/EventHandledRepositoryImpl.java (1 hunks)
apps/commerce-streamer/src/main/java/com/loopers/infrastructure/metrics/ProductMetricsJpaRepository.java (1 hunks)
apps/commerce-streamer/src/main/java/com/loopers/infrastructure/metrics/ProductMetricsRepositoryImpl.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/eventhandled/EventHandledServiceTest.java (1 hunks)
apps/commerce-streamer/src/test/java/com/loopers/application/metrics/ProductMetricsServiceTest.java (1 hunks)
apps/commerce-streamer/src/test/java/com/loopers/domain/metrics/ProductMetricsTest.java (1 hunks)
apps/commerce-streamer/src/test/java/com/loopers/interfaces/consumer/ProductMetricsConsumerIntegrationTest.java (1 hunks)
apps/commerce-streamer/src/test/java/com/loopers/interfaces/consumer/ProductMetricsConsumerTest.java (1 hunks)
modules/kafka/src/main/java/com/loopers/confg/kafka/KafkaConfig.java (4 hunks)
modules/kafka/src/main/resources/kafka.yml (2 hunks)
modules/kafka/src/testFixtures/java/com/loopers/testcontainers/KafkaTestContainersConfig.java (1 hunks)
modules/kafka/src/testFixtures/java/com/loopers/utils/KafkaCleanUp.java (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-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-api/src/main/java/com/loopers/application/outbox/OutboxEventService.java
apps/commerce-streamer/src/test/java/com/loopers/interfaces/consumer/ProductMetricsConsumerTest.java
apps/commerce-streamer/src/main/java/com/loopers/application/eventhandled/EventHandledService.java
apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventPublisher.java
apps/commerce-api/src/test/java/com/loopers/infrastructure/outbox/OutboxEventPublisherIntegrationTest.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/domain/outbox/OutboxEventRepository.java

📚 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-streamer/src/test/java/com/loopers/domain/metrics/ProductMetricsTest.java
apps/commerce-streamer/src/main/java/com/loopers/infrastructure/metrics/ProductMetricsJpaRepository.java
apps/commerce-streamer/src/main/java/com/loopers/domain/metrics/ProductMetricsRepository.java
apps/commerce-streamer/src/test/java/com/loopers/application/metrics/ProductMetricsServiceTest.java
apps/commerce-streamer/src/main/java/com/loopers/domain/event/ProductEvent.java
apps/commerce-streamer/src/main/java/com/loopers/domain/metrics/ProductMetrics.java
apps/commerce-streamer/src/main/java/com/loopers/application/metrics/ProductMetricsService.java
apps/commerce-api/src/main/java/com/loopers/domain/product/ProductEvent.java
apps/commerce-api/src/main/java/com/loopers/application/catalog/CatalogFacade.java

📚 Learning: 2025-11-12T13:04:50.782Z

Learnt from: kilian-develop
Repo: Loopers-dev-lab/loopers-spring-java-template PR: 72
File: core/domain/src/main/java/com/loopers/core/domain/productlike/vo/ProductLikeId.java:5-7
Timestamp: 2025-11-12T13:04:50.782Z
Learning: In the com.loopers codebase, domain entity ID value objects (e.g., ProductLikeId, OrderItemId, ProductId, PaymentId, OrderId, BrandId) are system-generated identifiers and do not require pattern validation (regex, length checks). They are implemented as simple records with a String value and an empty() factory method returning null for unsaved entities. This differs from UserIdentifier, which is a user-supplied login ID that requires format validation. Domain IDs should not be flagged for missing validation logic in the create() method.
<!-- [add_learning]
UserIdentifier와 같은 사용자 입력 ID와 ProductLikeId, OrderItemId 등의 도메인 ID는 검증 패턴이 다릅니다. UserIdentifier는 사용자가 입력하는 로그인 ID로서 정규식, 길이 등의 형식 검증이 필요하지만, 도메인 ID는 시스템에서 생성하는 식별자(UUID, DB 생성 ID)이므로 패턴 검증이 불필요합니다. 도메인 ID VO는 단순한 record와 empty() 팩토리 메서드만으로 충분합니다.

Applied to files:

apps/commerce-streamer/src/test/java/com/loopers/domain/metrics/ProductMetricsTest.java
apps/commerce-api/src/main/java/com/loopers/domain/product/ProductEvent.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/java/com/loopers/confg/kafka/KafkaConfig.java
modules/kafka/src/main/resources/kafka.yml

📚 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/application/catalog/CatalogFacade.java

🧬 Code graph analysis (10)

apps/commerce-streamer/src/test/java/com/loopers/interfaces/consumer/ProductMetricsConsumerTest.java (2)

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

LikeEvent (16-37)

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

OrderEvent (17-40)

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

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

LikeEvent (16-37)

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

Slf4j (28-154)

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

apps/commerce-api/src/main/java/com/loopers/application/outbox/OutboxBridgeEventListener.java (1)

Slf4j (39-141)

apps/commerce-api/src/main/java/com/loopers/application/outbox/OutboxEventService.java (1)

Slf4j (23-82)

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

Component (13-39)

apps/commerce-streamer/src/test/java/com/loopers/interfaces/consumer/ProductMetricsConsumerIntegrationTest.java (1)

apps/commerce-api/src/test/java/com/loopers/infrastructure/outbox/OutboxEventPublisherIntegrationTest.java (1)

SpringBootTest (23-43)

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

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

ProductEvent (14-59)

apps/commerce-api/src/test/java/com/loopers/infrastructure/outbox/OutboxEventPublisherIntegrationTest.java (1)

apps/commerce-streamer/src/test/java/com/loopers/interfaces/consumer/ProductMetricsConsumerIntegrationTest.java (1)

SpringBootTest (35-116)

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

apps/commerce-streamer/src/main/java/com/loopers/application/eventhandled/EventHandledService.java (1)

Slf4j (21-64)

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

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

ProductEvent (16-27)

apps/commerce-api/src/main/java/com/loopers/application/catalog/CatalogFacade.java (2)

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

ProductEvent (14-59)

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

Component (26-36)

apps/commerce-api/src/main/java/com/loopers/application/outbox/OutboxBridgeEventListener.java (2)

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

ProductEvent (14-59)

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

ProductEvent (16-27)

🔇 Additional comments (58)

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

23-23: LGTM!

Kafka 설정 파일을 configuration import 목록에 추가한 것은 적절합니다. 이는 PR의 Kafka 통합 목표와 일치합니다.

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

40-57: LGTM!

Upsert 시 동시성 제어를 위한 비관적 락 전략이 잘 문서화되어 있고, 메트릭 집계 시 Lost Update를 방지하기 위한 적절한 접근 방식입니다. Lock 범위를 PK 기반 조회로 최소화한 것도 좋습니다.

Based on learnings, 이 프로젝트는 유즈케이스별로 의도적으로 다른 락 전략을 사용하고 있으며, 메트릭 집계에 비관적 락을 사용하는 것은 합리적인 선택입니다.

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

5-5: LGTM!

Kafka 모듈과 테스트 픽스처 의존성을 추가하여 commerce-api에서 Kafka 기능을 사용할 수 있도록 한 것은 적절합니다. 이는 PR의 이벤트 기반 아키텍처 도입 목표와 일치합니다.

Also applies to: 38-38

apps/commerce-api/src/main/java/com/loopers/application/catalog/CatalogFacade.java (1)

117-125: LGTM!

@Transactional(readOnly = true) 어노테이션 추가는 outbox 패턴을 위한 적절한 선택입니다. 이는 이벤트 발행 시 outbox 엔트리가 트랜잭션 컨텍스트 내에서 저장되도록 보장합니다. 캐시 히트 시에도 이벤트를 발행하여 조회 수 집계를 일관되게 처리하는 것도 좋습니다.

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

1-21: LGTM! 도메인 레이어의 이벤트 발행 인터페이스가 잘 설계되었습니다.

DIP 원칙을 준수하여 도메인 레이어에서 인터페이스를 정의하고 구현을 인프라 레이어에 위임하는 구조가 적절합니다. 문서화도 명확하게 작성되었습니다.

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

1-40: LGTM! 이벤트 처리 기록을 위한 저장소 인터페이스가 잘 정의되었습니다.

이벤트 중복 처리 방지를 위한 메서드들이 적절하게 정의되어 있으며, DIP 원칙을 준수하고 있습니다. 문서화도 명확합니다.

apps/commerce-api/src/test/java/com/loopers/domain/outbox/OutboxEventTest.java (3)

15-49: LGTM! OutboxEvent 생성 및 초기 상태 검증이 잘 구현되었습니다.

필수 필드와 초기 상태(PENDING)를 확인하는 테스트가 명확하고 포괄적입니다.

77-97: LGTM! 실패 상태 전환 테스트가 적절합니다.

이벤트 실패 처리 로직이 올바르게 검증되고 있습니다.

99-123: LGTM! 상태 전환 시나리오 테스트가 잘 구현되었습니다.

발행 완료 후 실패 상태로 변경하는 엣지 케이스를 다루고 있으며, publishedAt이 유지되는지 확인하는 것이 좋습니다.

apps/commerce-streamer/src/test/java/com/loopers/interfaces/consumer/ProductMetricsConsumerTest.java (8)

29-48: LGTM! 테스트 설정이 잘 구성되었습니다.

Mockito를 사용한 의존성 모킹이 적절하게 설정되어 있습니다.

50-108: LGTM! 개별 이벤트 처리 테스트가 포괄적입니다.

LikeAdded와 LikeRemoved 이벤트 처리 로직을 명확하게 검증하고 있으며, eventId 헤더와 version 헤더를 포함한 완전한 시나리오를 다루고 있습니다.

110-149: LGTM! OrderCreated 이벤트의 복잡한 시나리오가 잘 테스트되었습니다.

여러 주문 아이템을 포함한 OrderCreated 이벤트 처리와 각 상품별 판매 수량 증가가 올바르게 검증되고 있습니다.

151-189: LGTM! 배치 처리 테스트가 적절합니다.

여러 이벤트를 배치로 처리하는 시나리오를 검증하고 있으며, acknowledgment가 한 번만 호출되는지 확인하는 것이 좋습니다.

191-263: LGTM! 오류 복원력 테스트가 훌륭합니다.

개별 이벤트 처리 실패 시에도 배치 처리를 계속하고 acknowledgment를 수행하는 로직이 잘 검증되어 있습니다. 이는 Kafka 소비자의 중요한 특성입니다.

Based on learnings, 이 코드베이스에서 Kafka 소비자는 에러 처리와 이벤트 추적을 EventInboxAspect를 통해 서비스 레이어에 위임하는 것으로 알고 있습니다. 이 테스트는 소비자 레벨의 에러 처리를 잘 검증하고 있습니다.

265-292: LGTM! 멱등성 테스트가 적절합니다.

이미 처리된 이벤트를 건너뛰는 로직이 올바르게 검증되고 있으며, 중복 처리 방지 메커니즘이 잘 작동함을 보여줍니다.

294-314: LGTM! eventId 누락 처리가 적절합니다.

eventId가 없는 메시지를 건너뛰는 방어적 프로그래밍이 잘 구현되어 있습니다.

316-346: LGTM! 동시성 예외 처리가 훌륭합니다.

DataIntegrityViolationException이 발생하는 경쟁 조건(race condition) 시나리오를 다루고 있습니다. 이는 여러 소비자 인스턴스가 동일한 이벤트를 거의 동시에 처리하려 할 때 발생할 수 있는 상황을 잘 반영하고 있습니다.

apps/commerce-api/src/test/java/com/loopers/application/outbox/OutboxBridgeEventListenerTest.java (6)

19-29: LGTM! 테스트 설정이 적절합니다.

Mockito를 사용한 의존성 모킹이 올바르게 구성되어 있습니다.

31-73: LGTM! Like 이벤트 처리 테스트가 포괄적입니다.

LikeAdded와 LikeRemoved 이벤트를 Outbox에 저장하는 로직이 명확하게 검증되고 있으며, 올바른 파라미터로 OutboxEventService가 호출되는지 확인하고 있습니다.

75-100: LGTM! OrderCreated 이벤트 처리가 적절합니다.

주문 생성 이벤트를 Outbox에 저장하는 로직이 올바르게 검증되고 있습니다.

102-124: LGTM! ProductViewed 이벤트 처리가 적절합니다.

상품 조회 이벤트를 Outbox에 저장하는 로직이 올바르게 검증되고 있습니다.

126-141: LGTM! 에러 격리 테스트가 훌륭합니다.

Outbox 저장 실패 시에도 예외를 전파하지 않는 것은 transactional outbox 패턴의 중요한 특성입니다. 주 트랜잭션이 성공하더라도 outbox 저장 실패로 인해 전체 작업이 롤백되지 않도록 하는 방어적 설계가 잘 검증되었습니다.

143-162: LGTM! 순차적 이벤트 처리 테스트가 적절합니다.

여러 이벤트를 순차적으로 처리하는 시나리오가 잘 검증되고 있습니다.

apps/commerce-streamer/src/main/java/com/loopers/infrastructure/eventhandled/EventHandledJpaRepository.java (1)

1-31: LGTM! JPA 저장소 인터페이스가 잘 정의되었습니다.

Spring Data JPA의 derived query 메서드를 적절하게 활용하고 있으며, 이벤트 중복 처리 방지를 위한 조회 메서드들이 명확하게 정의되어 있습니다.

apps/commerce-streamer/src/test/java/com/loopers/application/eventhandled/EventHandledServiceTest.java (4)

18-28: LGTM! 테스트 설정이 적절합니다.

Mockito를 사용한 의존성 모킹이 올바르게 구성되어 있습니다.

30-58: LGTM! 이벤트 처리 확인 로직이 잘 테스트되었습니다.

처리되지 않은 이벤트와 이미 처리된 이벤트에 대한 두 가지 경로를 모두 검증하고 있으며, 저장소와의 상호작용이 올바르게 확인되고 있습니다.

60-76: LGTM! 성공적인 이벤트 저장 테스트가 적절합니다.

새로운 이벤트를 정상적으로 저장하는 경로가 잘 검증되고 있습니다.

78-95: LGTM! 중복 이벤트 처리 예외 테스트가 훌륭합니다.

DataIntegrityViolationException을 통해 eventId의 UNIQUE 제약 조건 위반을 검증하고 있습니다. 이는 데이터베이스 수준에서 멱등성을 보장하는 중요한 메커니즘입니다.

apps/commerce-api/src/test/java/com/loopers/application/outbox/OutboxEventServiceTest.java (1)

1-73: LGTM! 테스트가 잘 구성되어 있습니다.

Outbox 이벤트 저장의 핵심 시나리오가 잘 검증되었습니다. ArgumentCaptor를 활용하여 저장된 엔티티의 속성들을 세밀하게 검증하는 방식이 좋습니다.

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

28-61: 잘 설계된 멱등성 보장 엔티티입니다.

eventId를 PK로 사용하여 DB 레벨에서 중복 처리를 방지하는 설계가 적절합니다. handled_at 인덱스는 이벤트 보존 정책이나 정리 작업에 유용합니다.

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

27-58: 잘 구현된 도메인 이벤트입니다.

productId에 대한 null 검증과 두 가지 팩토리 메서드 제공이 좋습니다. userId가 nullable인 것도 비로그인 사용자 케이스를 고려한 적절한 설계입니다.

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

1-37: LGTM! DTO 구조가 적절합니다.

Kafka 이벤트 파싱용 DTO로서 적절한 구조입니다. JavaDoc에서 commerce-api의 LikeEvent와의 중복 및 향후 공유 모듈 분리 고려 사항을 명확히 문서화한 점이 좋습니다.

apps/commerce-streamer/src/main/java/com/loopers/infrastructure/metrics/ProductMetricsRepositoryImpl.java (1)

20-49: LGTM! 표준적인 Repository 구현입니다.

JPA 리포지토리에 위임하는 깔끔한 구현입니다. findByProductIdForUpdate 메서드를 통해 비관적 락을 지원하는 것이 메트릭 업데이트의 동시성 제어에 적합합니다.

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

17-40: LGTM! 구조가 깔끔합니다.

Kafka 이벤트 역직렬화를 위한 record 기반 DTO 구조가 적절합니다. 주석에서 언급된 대로 commerce-api의 OrderEvent와 중복되므로 향후 공유 모듈로 분리하는 것이 좋겠습니다.

한 가지 확인 사항: subtotal은 Integer이고 usedPointAmount는 Long인데, 금액 관련 필드의 타입이 의도적으로 다른 것인지 확인해 주세요.

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

26-36: LGTM! DIP 원칙을 잘 따르고 있습니다.

단일 책임 원칙에 따라 ApplicationEvent만 발행하고, Kafka 전송은 OutboxBridgeEventListener가 처리하는 관심사 분리가 적절합니다.

apps/commerce-streamer/src/test/java/com/loopers/domain/metrics/ProductMetricsTest.java (1)

10-155: LGTM! 도메인 엔티티의 동작을 잘 검증하고 있습니다.

ProductMetrics의 주요 동작(좋아요, 판매량, 조회수 증감, 멱등성, 버전 관리)에 대한 테스트 커버리지가 충분합니다.

apps/commerce-streamer/src/test/java/com/loopers/application/metrics/ProductMetricsServiceTest.java (1)

21-217: LGTM! 서비스 레이어의 동작을 잘 검증하고 있습니다.

버전 기반 업데이트 로직(오래된 이벤트 스킵, 최신 이벤트 적용), 메트릭 생성/업데이트, 유효하지 않은 수량 무시 등 주요 시나리오가 잘 테스트되어 있습니다.

apps/commerce-api/src/test/java/com/loopers/infrastructure/outbox/OutboxEventPublisherTest.java (1)

29-299: LGTM! OutboxEventPublisher의 주요 동작을 잘 검증하고 있습니다.

배치 처리, 개별 실패 시 계속 진행, 상태 전환(PUBLISHED/FAILED), JSON 역직렬화 실패 처리, 파티션 키 검증 등 중요한 시나리오가 모두 테스트되어 있습니다.

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

87-116: publishEvent 메서드 구현 검토

Kafka 헤더에 eventId와 version을 포함시켜 멱등성 처리를 지원하는 설계가 적절합니다. 다만 위에서 언급한 비동기 전송 이슈를 해결하면 전체적으로 안정적인 구현이 됩니다.

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

13-39: JPA 레포지토리 위임 구현 LGTM

도메인 레포지토리 인터페이스를 JPA 레포지토리에 깔끔하게 위임하고 있습니다. findById에서 조회 실패 시 예외를 던지는 것은 도메인 계층에서 Optional 대신 확정된 값을 기대하는 경우에 적합한 설계입니다.

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

44-58: 버전 기반 이벤트 처리 로직 LGTM

shouldUpdate(eventVersion) 체크를 통해 이미 처리된 오래된 이벤트를 스킵하는 멱등성 처리가 잘 구현되어 있습니다. 비관적 락과 함께 사용하여 동시성 제어도 적절합니다. 학습된 내용에 따르면, 이 코드베이스에서는 유즈케이스별로 의도적으로 다른 락 전략을 사용합니다.

apps/commerce-streamer/src/main/java/com/loopers/infrastructure/metrics/ProductMetricsJpaRepository.java (1)

37-39: 비관적 락 쿼리 구현 LGTM

PESSIMISTIC_WRITE 락 모드와 명시적 JPQL 쿼리를 사용한 findByProductIdForUpdate 구현이 적절합니다. 학습된 패턴에 따르면 이 코드베이스에서는 동시성 제어가 필요한 경우 비관적 락을 사용합니다.

apps/commerce-api/src/main/java/com/loopers/application/outbox/OutboxBridgeEventListener.java (1)

51-68: @async와 @TransactionalEventListener 패턴의 예외 처리 고려

@Async와 @TransactionalEventListener(AFTER_COMMIT) 조합은 원본 트랜잭션 커밋 후 비동기로 Outbox 저장을 수행하여 에러 격리를 잘 구현했습니다. 애플리케이션에 이미 @EnableAsync가 구성되어 있고 Spring Boot가 자동으로 TaskExecutor를 제공하므로 추가 설정이 필요하지 않습니다. 다만 void 반환 타입의 메서드에서는 발생한 예외가 호출자에게 전파되지 않으므로, 비동기 작업 중 예외가 발생해도 로깅만 되고 원본 트랜잭션에는 영향을 주지 않습니다.

apps/commerce-streamer/src/main/java/com/loopers/application/eventhandled/EventHandledService.java (1)

24-63: LGTM! 멱등성 보장 패턴이 올바르게 구현되었습니다.

데이터베이스 UNIQUE 제약조건을 활용한 멱등성 보장 구현이 적절합니다. isAlreadyHandled()와 markAsHandled() 조합 사용 시 발생할 수 있는 TOCTOU(Time-of-check to time-of-use) 이슈는 UNIQUE 제약조건이 최종적으로 방지하므로, DataIntegrityViolationException을 재발생시켜 호출자가 동시성 중복을 처리하도록 하는 방식이 올바릅니다.

apps/commerce-streamer/src/main/java/com/loopers/infrastructure/eventhandled/EventHandledRepositoryImpl.java (1)

21-38: LGTM! 표준 리포지토리 어댑터 패턴이 적절하게 구현되었습니다.

JPA 리포지토리로의 단순 위임 구조가 명확하고 깔끔합니다.

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

58-65: LGTM! 생성자 초기화가 적절합니다.

모든 카운터와 버전을 0으로 초기화하고 생성 시점을 기록하는 로직이 명확합니다.

79-85: 좋아요 감소 로직이 안전하게 구현되었습니다.

음수 방지 가드가 적절하며, 조건 불충족 시 조용히 무시하는 방식은 방어적 코딩으로 허용됩니다.

119-126: 버전 기반 이벤트 순서 보장 로직이 올바르게 구현되었습니다.

shouldUpdate 메서드는 null 이벤트 버전을 허용(하위 호환성)하면서 이벤트 버전 비교를 통해 오래된 이벤트가 최신 메트릭을 덮어쓰는 것을 방지합니다. 이 로직은 ProductMetricsService에서 비관적 락(findByProductIdForUpdate)과 함께 사용되어 동시성 제어가 보장됩니다.

apps/commerce-api/src/main/java/com/loopers/domain/outbox/OutboxEvent.java (3)

22-68: LGTM! Outbox 엔티티 구조가 트랜잭셔널 아웃박스 패턴에 적합합니다.

eventId의 UNIQUE 제약조건은 중복 이벤트를 방지하고, (status, created_at) 인덱스는 PENDING 이벤트의 효율적인 폴링을 지원합니다.

69-90: LGTM! 빌더 패턴 생성자가 적절하게 구현되었습니다.

초기 상태를 PENDING으로 설정하고 생성 시점을 기록하는 로직이 명확합니다.

92-105: 상태 전환 메서드가 단순하고 명확합니다.

현재 상태를 검증하지 않고 전환을 허용하는 방식은 아웃박스 패턴에서 재시도 시나리오를 간단하게 처리할 수 있게 합니다. markAsPublished()를 여러 번 호출해도 멱등적이므로 문제없습니다.

modules/kafka/src/testFixtures/java/com/loopers/utils/KafkaCleanUp.java (2)

95-123: LGTM! 테스트 토픽 재생성 로직이 적절합니다.

3개 파티션, 1개 레플리카, min.insync.replicas=1 설정은 테스트 환경에 적합하며, 토픽별 에러 핸들링을 통해 부분 실패를 허용합니다.

187-193: LGTM! AdminClient 생성 로직이 적절합니다.

KafkaAdmin의 설정에서 부트스트랩 서버를 가져오며, localhost:19092 폴백은 기본값으로 합리적입니다. 테스트 환경에서는 KafkaTestContainersConfig가 올바른 동적 포트를 설정하므로 폴백이 사용될 가능성은 낮습니다.

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

164-224: LGTM! 주문 이벤트 처리 로직이 올바르게 구현되었습니다.

OrderCreated 이벤트의 여러 주문 항목을 순회하며 각 상품의 판매량을 증가시키는 로직이 정확하며, 에러 핸들링 패턴도 일관성 있게 적용되었습니다.

256-312: LGTM! 상품 조회 이벤트 처리 로직이 일관성 있게 구현되었습니다.

다른 리스너 메서드들과 동일한 패턴을 따르며, ProductViewed 이벤트 처리가 간단명료합니다.

360-387: LGTM! 헤더 추출 로직이 안전하게 구현되었습니다.

eventId와 version 추출 시 null 처리와 파싱 에러 핸들링이 적절하며, UTF-8 인코딩 사용도 표준을 따릅니다.

129-147: 에러 핸들링 패턴이 적절합니다.

개별 레코드 에러를 로깅 후 계속 진행하고, 배치 레벨 에러는 재발생시켜 커밋을 방지하는 방식이 합리적입니다. DataIntegrityViolationException을 동시성 중복으로 처리하는 것도 올바릅니다. 이 구현은 적절한 At-Least-Once 전달 보장을 제공합니다.

modules/kafka/src/main/java/com/loopers/confg/kafka/KafkaConfig.java (1)

35-76: LGTM! 제네릭 타입 변경으로 타입 안전성이 향상되었습니다.

키 타입을 Object에서 String으로 변경하여 Kafka 메시지의 파티션 키가 항상 문자열임을 명시적으로 표현하며, 실수로 다른 타입의 키를 사용하는 것을 방지합니다.

apps/commerce-api/src/main/java/com/loopers/application/outbox/OutboxEventService.java

apps/commerce-api/src/main/java/com/loopers/infrastructure/outbox/OutboxEventPublisher.java

apps/commerce-streamer/src/main/java/com/loopers/application/metrics/ProductMetricsService.java

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

...mer/src/test/java/com/loopers/interfaces/consumer/ProductMetricsConsumerIntegrationTest.java

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

modules/kafka/src/testFixtures/java/com/loopers/testcontainers/KafkaTestContainersConfig.java

* chore:kafka producer 설정 * chore: kafka 토픽 자동 생성 설정 추가 * feat: kafka event publisher, comsumer 추가 * test: 집계 도메인 단위 테스트 코드 추가 * feat: 집계 도메인 domain 레이어 구현 * feat: 집계 도메인 infra 레이어 구현 * chore: kafka 토픽 자동 생성 설정 추가 * chore: kafka 빌드 의존성 추가 * test: 집계 통합 테스트 추가 * feat: 집계 서비스 로직 구현 * test: kafka consumer 테스트 코드 추가 * feat: kafka comsumer 구현 * outbox 패턴 적용위해 기존 kafka 설정 삭제 * test: outboxevent 단위 테스트 추가 * feat: outbox 도메인 구현 * feat: outbox infrastructure repository구현 * metric 오타 수정 * refactor: consumer 관련 로직들은 commerce-streamer 모듈로 이동 * test: outbox 테스트 코드 추가 * test: outbox 구현 * outbox event listener 구현 * feat: 상품 조회 이벤트 추가 * feat: 상품 조회시 이벤트 발행 * chore: kafka 설정 수정 * fix: outbox 처리되지 않는 오류 수정 * chore: 테스트 코드 실행시 kafka 사용할 수 있도록 test container 설정 추가 * test: offset.reset: latest 설정이 제대로 적용되는지 확인하는 테스트 코드 추가 * test: kafka 파티션 키 설정에 대한 테스트 코드 추가 * chore: commerce-api 테스트 환경에서 카프카 사용하도록 설ㄹ정 * test: event id 기준으로 한 번만 publish, consume하는 것을 검증하는 테스트 코드 추가 * chore: 충돌 발생한 테스트 코드 수정 * feat: event id 기준 1회 처리되도록 로직 구현 * test: 버전 기준으로 최신 이벤트만 처리하도록 테스트 코드 수정 * feat: version 기준으로 최신 이벤트만 처리하도록 함 * test: 중복 메시지 재전송 시 한 번만 처리되는지 검증하는 테스트 코드 추가 * feat: kafka 이벤트 publish 할 때 콜백 사용하여 이벤트 유실 방지 * feat: kafka메시지 헤더에 event type 추가 * feat: 버전 조회와 저장 사이의 경쟁 조건 가능성 해결 * feat: 신규 상품 등록시 event 발행에서 발생하는 경합 문제 수정

looper88888 · 2025-12-20T03:33:03Z

✅ Checklist

🎾 Producer
[ ] 도메인(애플리케이션) 이벤트 설계
[ ] Producer 앱에서 도메인 이벤트 발행 (catalog-events, order-events, 등)
[ ] PartitionKey 기반의 이벤트 순서 보장
[ ] 메세지 발행이 실패했을 경우에 대해 고민해보기

⚾ Consumer
[ ] Consumer 가 Metrics 집계 처리
[ ] event_handled 테이블을 통한 멱등 처리 구현
[ ] 재고 소진 시 상품 캐시 갱신
[ ] 중복 메세지 재전송 테스트 → 최종 결과가 한 번만 반영되는지 확인

@coderabbitai 체크리스트 확인해

coderabbitai · 2025-12-20T03:34:25Z

Tip

For best results, initiate chat on the files or code changes.

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Search for stock depletion related code
echo "=== Searching for StockDepleted events ==="
rg -i "stockdepleted" --type java -C 3

echo ""
echo "=== Searching for cache refresh or invalidation ==="
rg -i "cache.*refresh|cache.*invalidate|evict" --type java -C 3

echo ""
echo "=== Searching for product cache operations ==="
rg -i "product.*cache" --type java -C 3

echo ""
echo "=== Listing all event types in domain ==="
fd -e java -p "Event\.java" apps/

Length of output: 18813

@looper88888, 체크리스트를 검토했습니다:

🎾 Producer 구현 상태

✅ 도메인(애플리케이션) 이벤트 설계

ProductEvent.ProductViewed, LikeEvent.LikeAdded/LikeRemoved, OrderEvent.OrderCreated 등 도메인 이벤트 정의 완료
추가로 CouponEvent, PaymentEvent, PointEvent도 존재

✅ Producer 앱에서 도메인 이벤트 발행

OutboxBridgeEventListener가 도메인 이벤트를 수신하여 OutboxEvent로 변환
OutboxEventPublisher가 주기적으로 pending 이벤트를 Kafka로 발행
like-events, order-events, product-events 토픽으로 발행 구현됨

✅ PartitionKey 기반의 이벤트 순서 보장

OutboxEvent에 partitionKey 필드 존재
Kafka 메시지 헤더에 KafkaHeaders.KEY로 partitionKey 설정됨
동일 aggregateId(예: productId, orderId)에 대한 순서 보장

✅ 메시지 발행 실패 시 처리

OutboxEventPublisher에서 발행 실패 시 OutboxStatus.FAILED로 마킹
실패한 이벤트는 로그 기록 후 계속 처리 (다른 이벤트 처리 차단 안 함)
트랜잭션 경계 내에서 상태 변경 저장

⚾ Consumer 구현 상태

✅ Consumer가 Metrics 집계 처리

ProductMetricsConsumer가 like-events, order-events, product-events 소비
ProductMetricsService로 likeCount, salesCount, viewCount 집계
ProductMetrics 엔티티에 pessimistic lock으로 동시성 제어

✅ event_handled 테이블을 통한 멱등 처리 구현

EventHandled 엔티티와 EventHandledService 구현 완료
eventId 기반 중복 처리 방지 (primary key로 uniqueness 보장)
DataIntegrityViolationException 처리로 동시 중복 이벤트 허용

❌ 재고 소진 시 상품 캐시 갱신

현재 PR에 StockDepletedEvent 또는 재고 소진 관련 Consumer가 없음
ProductCacheService의 evict() 인프라는 존재하지만, 재고 소진 이벤트 구독/처리 로직 누락
🔴 구현 필요: StockDepleted 이벤트 Consumer 추가 및 캐시 무효화 로직 구현

✅ 중복 메시지 재전송 테스트

ProductMetricsConsumerTest에 중복 이벤트 처리 테스트 존재
이미 처리된 eventId는 스킵되는 로직 검증됨
EventHandledService.isAlreadyHandled() 체크로 멱등성 보장

📝 종합 요약

완료: 7개 항목
미완료: 1개 항목 (재고 소진 시 상품 캐시 갱신)

재고 소진 캐시 갱신 기능을 추가하려면:

StockDepletedEvent consumer 구현 필요
해당 consumer에서 productCacheService.evict() 호출하여 상품 캐시 무효화
통합 테스트 추가 권장

🧠 Learnings used

Learnt from: toongri
Repo: Loopers-dev-lab/loopers-spring-kotlin-template PR: 68
File: docs/week8/round8-detailed-design.md:151-178
Timestamp: 2025-12-19T20:59:50.009Z
Learning: In the Loopers-dev-lab/loopers-spring-kotlin-template repository's Kafka event pipeline, only 5 domain events are intentionally published to Kafka via CloudEventEnvelopeFactory: OrderPaidEventV1, LikeCreatedEventV1, LikeCanceledEventV1, ProductViewedEventV1, and StockDepletedEventV1. Other domain events (OrderCreatedEventV1, OrderCanceledEventV1, PaymentCreatedEventV1, PaymentPaidEventV1, PaymentFailedEventV1) are internal-only and intentionally not mapped in resolveMetadata(), which correctly returns null for them to exclude them from Outbox publication.

Learnt from: toongri
Repo: Loopers-dev-lab/loopers-spring-kotlin-template PR: 68
File: apps/commerce-streamer/src/main/kotlin/com/loopers/interfaces/consumer/product/ProductEventConsumer.kt:0-0
Timestamp: 2025-12-19T23:39:16.424Z
Learning: In Loopers-dev-lab/loopers-spring-kotlin-template, toongri prefers pragmatic idempotency handling in Kafka consumers: when idempotency key persistence fails after successful business logic execution, log a warning and continue (accepting low risk of duplicates on retry) rather than rolling back business data. This keeps business logic decoupled from idempotency store (which might be Redis/external system, not RDB), avoids over-engineering for rare edge cases, and prioritizes core business value over auxiliary metadata. Transaction-based atomicity between business data and idempotency keys is considered excessive when duplicate event probability is very low.

coderabbitai bot reviewed Dec 18, 2025

View reviewed changes

minor7295 force-pushed the base-pr-round8 branch from 239a920 to 71bc388 Compare December 18, 2025 17:13

minor7295 changed the title ~~Feature/kafka (#36)~~ [volume - 8] Decoupling with Kafka Dec 19, 2025

minor7295 self-assigned this Dec 19, 2025

minor7295 added the enhancement New feature or request label Dec 19, 2025

minor7295 marked this pull request as ready for review December 19, 2025 03:25

minor7295 force-pushed the base-pr-round8 branch 2 times, most recently from c123b34 to bf7b8e3 Compare December 19, 2025 03:57

minor7295 force-pushed the base-pr-round8 branch from bf7b8e3 to e4403b1 Compare December 19, 2025 13:33

minor7295 merged commit 3af88f9 into Loopers-dev-lab:minor7295 Dec 21, 2025
2 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[volume - 8] Decoupling with Kafka #191

[volume - 8] Decoupling with Kafka #191

Uh oh!

minor7295 commented Dec 18, 2025 •

edited

Loading

Uh oh!

coderabbitai bot commented Dec 18, 2025 •

edited

Loading

Review skipped

Uh oh!

minor7295 commented Dec 18, 2025

Uh oh!

coderabbitai bot commented Dec 18, 2025

Uh oh!

coderabbitai bot left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

looper88888 commented Dec 20, 2025

Uh oh!

coderabbitai bot commented Dec 20, 2025

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

[volume - 8] Decoupling with Kafka #191

[volume - 8] Decoupling with Kafka #191

Uh oh!

Conversation

minor7295 commented Dec 18, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

📌 Summary

💬 Review Points

1. Transactional Outbox Pattern 구현 방식의 적절성

2. 멱등성 처리 전략: event_handled 테이블과 version 필드의 조합

3. 파티션 키 기반 순서 보장과 offset.reset: latest 설정의 조합

구현 세부사항

1. 내부 이벤트와 외부 이벤트의 구분

2. Producer 설정: At Least Once 보장 및 이벤트 생성 실패 처리

3. Consumer 설정: Manual Ack 처리

4. 멱등성 처리: event_handled 테이블

5. 버전 기반 최신 이벤트만 반영

6. 파티션 키 설정

7. 퍼블리셔와 컨슈머가 동일 이벤트를 판단하는 기준 설정

6. 파티션 키 설정

✅ Checklist

⚾ Consumer (7/7)

📎 References

Summary by CodeRabbit

릴리스 노트

Uh oh!

coderabbitai bot commented Dec 18, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Review skipped

Walkthrough

Changes

Sequence Diagram

Estimated code review effort

Possibly related PRs

Suggested labels

Poem

Pre-merge checks and finishing touches

Uh oh!

minor7295 commented Dec 18, 2025

Uh oh!

coderabbitai bot commented Dec 18, 2025

Uh oh!

coderabbitai bot left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

looper88888 commented Dec 20, 2025

Uh oh!

coderabbitai bot commented Dec 20, 2025

🎾 Producer 구현 상태

⚾ Consumer 구현 상태

📝 종합 요약

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

minor7295 commented Dec 18, 2025 •

edited

Loading

coderabbitai bot commented Dec 18, 2025 •

edited

Loading