round8: 카프카 모듈 적용, 도메인 관련 코드 core로 분리 #207
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,95 @@ | ||
| package com.loopers.infrastructure.event.outbox; | ||
|
|
||
| import com.loopers.domain.event.outbox.OutboxEvent; | ||
| import com.loopers.domain.event.outbox.OutboxEventRepository; | ||
| import com.loopers.domain.event.outbox.OutboxStatus; | ||
| import lombok.RequiredArgsConstructor; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.slf4j.Marker; | ||
| import org.slf4j.MarkerFactory; | ||
| import org.springframework.kafka.core.KafkaTemplate; | ||
| import org.springframework.scheduling.annotation.Scheduled; | ||
| import org.springframework.stereotype.Component; | ||
| import org.springframework.transaction.annotation.Transactional; | ||
|
|
||
| import java.util.List; | ||
|
|
||
| @Slf4j | ||
| @Component | ||
| @RequiredArgsConstructor | ||
| public class OutboxPublisher { | ||
| // fatal marker | ||
| private static Marker FATAL = MarkerFactory.getMarker("FATAL"); | ||
| private final OutboxEventRepository outboxEventRepository; | ||
| private final KafkaTemplate<Object, Object> kafkaTemplate; // KafkaConfig에서 제공하는 Bean 사용 | ||
|
|
||
| private static final int BATCH_SIZE = 100; | ||
| private static final int MAX_RETRY_COUNT = 5; | ||
|
|
||
| /** | ||
| * 주기적으로 PENDING 상태의 Outbox 이벤트를 Kafka로 발행 | ||
| * 5초마다 실행 (필요에 따라 조정) | ||
| * | ||
| * ⚠️ 토픽별로 Relay를 분리하는 것을 권장합니다: | ||
| * - order-events: 1초 주기 (빠른 전달 필요) | ||
| * - catalog-events: 5초 주기 (느슨해도 괜찮음) | ||
| */ | ||
| @Scheduled(fixedDelay = 5000) | ||
| @Transactional | ||
| public void publishPendingEvents() { | ||
| List<OutboxEvent> pendingEvents = outboxEventRepository.findByStatusOrderByCreatedAtAsc(OutboxStatus.PENDING); | ||
|
|
||
| if (pendingEvents.isEmpty()) { | ||
| return; | ||
| } | ||
|
|
||
| log.info("Found {} pending events to publish", pendingEvents.size()); | ||
|
|
||
| int processedCount = 0; | ||
| for (OutboxEvent event : pendingEvents) { | ||
| if (processedCount >= BATCH_SIZE) { | ||
| break; | ||
| } | ||
|
|
||
| String topic = determineTopic(event.getAggregateType()); | ||
| String key = event.getAggregateId(); | ||
|
|
||
| // Kafka로 메시지 발행 | ||
| kafkaTemplate.send(topic, key, event.getPayload()).whenComplete((result, exception) -> { | ||
| if (exception == null) { | ||
| event.markAsPublished(); | ||
| outboxEventRepository.save(event); | ||
| log.debug("Published event: eventId={}, topic={}, key={}", event.getId(), topic, key); | ||
| } else { | ||
| handlePublishFailure(event, exception); | ||
| } | ||
| }); | ||
|
|
||
| processedCount++; | ||
| } | ||
| } | ||
|
Comment on lines
+37
to
+70
There was a problem hiding this comment. 동시성 제어 부재로 인한 중복 발행 가능성 여러 애플리케이션 인스턴스가 동시에 실행될 때, 동일한 PENDING 이벤트를 여러 인스턴스가 동시에 조회하고 발행할 수 있습니다.
해결 방법:
🔎 제안하는 수정안 @Scheduled(fixedDelay = 5000)
@Transactional
public void publishPendingEvents() {
- List<OutboxEvent> pendingEvents = outboxEventRepository.findByStatusOrderByCreatedAtAsc(OutboxStatus.PENDING);
+ // 비관적 락을 사용하여 동시성 제어
+ List<OutboxEvent> pendingEvents = outboxEventRepository.findPendingEventsForUpdate(OutboxStatus.PENDING);또는 Kafka Producer에 멱등성 설정 추가: spring:
kafka:
producer:
enable-idempotence: true
acks: all
🧩 Analysis chain🏁 Script executed: find . -type f -name "*OutboxEventRepository*" -o -name "*OutboxEvent*" | grep -E "\.(java|kt)$" | head -20Repository: Loopers-dev-lab/loopers-spring-java-template Length of output: 567 🏁 Script executed: cat -n ./apps/commerce-core/src/main/java/com/loopers/domain/event/outbox/OutboxEventRepository.javaRepository: Loopers-dev-lab/loopers-spring-java-template Length of output: 477 🏁 Script executed: cat -n ./apps/commerce-core/src/main/java/com/loopers/infrastructure/event/outbox/OutboxEventRepositoryImpl.javaRepository: Loopers-dev-lab/loopers-spring-java-template Length of output: 1269 🏁 Script executed: cat -n ./apps/commerce-core/src/main/java/com/loopers/infrastructure/event/outbox/OutboxEventJpaRepository.javaRepository: Loopers-dev-lab/loopers-spring-java-template Length of output: 1310 비동기 Kafka 콜백에서의 트랜잭션 경계 문제
문제:
해결 방법:
|
||
|
|
||
| private String determineTopic(String aggregateType) { | ||
| return switch (aggregateType) { | ||
| case "PRODUCT" -> "catalog-events"; | ||
| case "ORDER" -> "order-events"; | ||
| default -> throw new IllegalArgumentException("Unknown aggregate type: " + aggregateType); | ||
| }; | ||
| } | ||
|
|
||
| private void handlePublishFailure(OutboxEvent event, Throwable exception) { | ||
| log.error("Failed to publish event: eventId={}, retryCount={}", | ||
| event.getId(), event.getRetryCount(), exception); | ||
|
|
||
| if (event.getRetryCount() >= MAX_RETRY_COUNT) { | ||
| event.markAsFailed(exception.getMessage()); | ||
| outboxEventRepository.save(event); | ||
| log.error(FATAL, "Event exceeded max retry count, marked as FAILED: eventId={}", event.getId()); | ||
| // TODO: DLQ로 전송하거나 알림 발송 | ||
| } else { | ||
| event.incrementRetry(); | ||
| outboxEventRepository.save(event); | ||
| } | ||
| } | ||
| } | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,25 +1,78 @@ | ||
| package com.loopers.infrastructure.like; | ||
|
|
||
| import com.loopers.domain.event.outbox.OutboxEventService; | ||
| import com.loopers.domain.like.product.LikeProductEvent; | ||
| import com.loopers.domain.like.product.LikeProductEventPublisher; | ||
| import lombok.RequiredArgsConstructor; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.springframework.stereotype.Component; | ||
|
|
||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
| import java.util.UUID; | ||
|
|
||
| /** | ||
| * 좋아요 이벤트를 Kafka로 발행하기 위한 Publisher | ||
| * | ||
| * 변경 사항: | ||
| * - 기존: ApplicationEventPublisher를 통한 동기식 이벤트 발행 | ||
| * - 변경: Outbox 패턴을 통한 비동기 Kafka 이벤트 발행 | ||
| * | ||
| * 처리 플로우: | ||
| * 1. LikeProductEvent를 받아서 Outbox 테이블에 저장 (같은 트랜잭션) | ||
| * 2. OutboxPublisher가 주기적으로 Kafka로 발행 | ||
| * 3. Consumer가 Kafka에서 수신하여 Metrics 집계 | ||
| */ | ||
| @Component | ||
| @RequiredArgsConstructor | ||
| @Slf4j | ||
| public class LikeProductEventPublisherImpl implements LikeProductEventPublisher { | ||
| private final OutboxEventService outboxEventService; | ||
|
|
||
| @Override | ||
| public void publishLikeEvent(LikeProductEvent event) { | ||
| if (event == null) { | ||
| log.warn("LikeProductEvent가 null입니다. 이벤트 발행을 건너뜁니다."); | ||
| return; | ||
| } | ||
|
|
||
| try { | ||
| // 이벤트 페이로드 생성 | ||
| Map<String, Object> eventPayload = createEventPayload(event); | ||
|
|
||
| // 이벤트 타입 결정 | ||
| String eventType = event.getLiked() ? "ProductLiked" : "ProductUnliked"; | ||
|
|
||
| // Outbox 테이블에 저장 (도메인 트랜잭션 내에서 호출되어야 함) | ||
| outboxEventService.saveEvent( | ||
| "PRODUCT", | ||
| event.getProductId().toString(), | ||
| eventType, | ||
| eventPayload | ||
| ); | ||
|
|
||
| log.info("LikeProductEvent saved to outbox: productId={}, liked={}, eventType={}", | ||
| event.getProductId(), event.getLiked(), eventType); | ||
| } catch (Exception e) { | ||
| log.error("Failed to save LikeProductEvent to outbox: productId={}, liked={}", | ||
| event.getProductId(), event.getLiked(), e); | ||
| throw e; | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * LikeProductEvent를 Kafka 이벤트 페이로드로 변환 | ||
| */ | ||
| private Map<String, Object> createEventPayload(LikeProductEvent event) { | ||
| Map<String, Object> payload = new HashMap<>(); | ||
| payload.put("eventId", UUID.randomUUID().toString()); | ||
| payload.put("eventType", event.getLiked() ? "ProductLiked" : "ProductUnliked"); | ||
| payload.put("aggregateId", event.getProductId().toString()); | ||
| payload.put("productId", event.getProductId()); | ||
| payload.put("userId", event.getUserId()); | ||
| payload.put("brandId", event.getBrandId()); | ||
| payload.put("liked", event.getLiked()); | ||
| payload.put("createdAt", java.time.Instant.now().toString()); | ||
| return payload; | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🛠️ Refactor suggestion | 🟠 Major
중복된 변환 로직을 헬퍼 메서드로 추출하세요.
PgV1Dto.PgTransactionResponse를 도메인 VO로 변환하는 동일한 로직이 두 곳(retryPaymentRequest와 syncOrderStatusFromPgResponse)에 중복되어 있습니다. DRY 원칙을 위반하며, 향후 변환 로직 변경 시 두 곳을 모두 수정해야 하는 유지보수 리스크가 있습니다.🔎 변환 로직을 헬퍼 메서드로 추출하는 제안
Also applies to: 148-156