feat: Kafka Producer Outbox Pattern 적용 (#99)#100
Conversation
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
Outbox 패턴을 도입하여 Kafka 이벤트 발행의 신뢰성을 크게 향상시킨 점은 매우 긍정적입니다. 비즈니스 트랜잭션과 이벤트 저장을 원자적으로 처리하고, 실패 시 재시도 메커니즘을 도입한 것은 시스템 안정성에 큰 기여를 할 것입니다.하지만 현재 Outbox 패턴 구현에서 몇 가지 개선할 점이 있습니다.1. 책임 분리 및 중복 로직: 이벤트 핸들러/서비스에서 Outbox에 이벤트를 저장하는 로직과, Kafka Producer에서 직접 Kafka로 이벤트를 발행하고 Outbox 상태를 업데이트하는 로직이 혼재되어 있습니다. 이는 Outbox 패턴의 핵심 원칙인 책임 분리를 저해하고, 중복된 처리 및 잠재적인 상태 불일치를 야기할 수 있습니다.2. 스케줄러의 블로킹 호출: OutboxRetryScheduler에서 Kafka 발행 시 get() 메서드를 사용하여 블로킹 호출을 하고 있습니다. 이는 처리할 이벤트가 많거나 Kafka 응답이 느릴 경우 스케줄러의 성능 저하를 초래할 수 있습니다.3. SENT 상태의 모호성: 최대 재시도 횟수를 초과한 이벤트도 SENT 상태로 변경됩니다. 이는 SENT 상태가 "성공적으로 발행됨"이 아닌 "더 이상 재시도하지 않음"을 의미하게 되어 상태의 의미가 모호해질 수 있습니다. 특히 알림 이벤트의 경우, 재시도 실패 시 영구적인 유실로 이어질 수 있습니다.이러한 문제점들을 개선하여 Outbox 패턴의 장점을 최대한 활용하고 시스템의 견고성을 더욱 높일 수 있을 것입니다.
| private final ObjectMapper objectMapper; | ||
|
|
||
| @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) | ||
| @Transactional(propagation = Propagation.REQUIRES_NEW) |
There was a problem hiding this comment.
TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)는 부모 트랜잭션이 커밋된 후에 실행됩니다. 이 시점에는 부모 트랜잭션이 존재하지 않으므로, REQUIRES_NEW는 새로운 트랜잭션을 시작합니다. 하지만 outboxEventService.save 메서드는 Propagation.MANDATORY로 설정되어 있어, 현재 트랜잭션이 없으면 예외가 발생합니다. AFTER_COMMIT 페이즈에서 MANDATORY 전파 속성을 가진 메서드를 호출하는 것은 논리적으로 맞지 않습니다. 제안: outboxEventService.save는 비즈니스 로직 트랜잭션 내에서 호출되어야 Outbox 패턴의 원자성을 보장할 수 있습니다. 따라서 이 핸들러에서 outboxEventService.save를 호출하는 대신, 비즈니스 로직을 수행하는 서비스 계층에서 outboxEventService.save를 호출하도록 리팩토링하는 것을 고려해 주세요. 만약 이 핸들러에서 Outbox 저장이 필요하다면, outboxEventService.save의 트랜잭션 전파 속성을 REQUIRES_NEW로 변경하거나, 이 핸들러의 @Transactional을 제거하고 outboxEventService.save를 호출하는 별도의 @Transactional(REQUIRES_NEW) 서비스 메서드를 호출해야 합니다.
| User user = userRepository.findById(userId) | ||
| .orElseThrow(() -> new CustomException(ErrorCode.USER_NOT_FOUND)); | ||
| StudyPost post = studyPostRepository.saveAndFlush(request.toEntity(user)); | ||
| saveOutboxEvent(post.getId(), PostSyncOperationType.UPSERT); |
There was a problem hiding this comment.
saveOutboxEvent를 통해 Outbox에 저장했다면, PostSyncEvent를 직접 발행하여 PostSyncKafkaProducer가 Kafka로 보내는 것은 Outbox 패턴의 목적과 충돌합니다. PostSyncKafkaProducer가 PostSyncEvent를 수신하여 Kafka로 보내는 대신, OutboxRetryScheduler가 Outbox 테이블에서 이벤트를 읽어 Kafka로 보내야 합니다. 제안: eventPublisher.publishEvent(new PostSyncEvent(...)) 호출을 제거하고, OutboxRetryScheduler가 POST_SYNC_TOPIC의 PENDING 이벤트를 처리하도록 해야 합니다.
| log.info("Outbox 재발행 성공 - id: {}, topic: {}", outbox.getId(), outbox.getTopic()); | ||
| } catch (Exception e) { | ||
| log.warn("Outbox 재발행 실패 - id: {}, topic: {}", outbox.getId(), outbox.getTopic(), e); | ||
| handleRetryFailure(outbox, e); |
There was a problem hiding this comment.
get() 메서드는 Kafka 발행이 완료될 때까지 현재 스케줄러 스레드를 블로킹합니다. 이는 처리해야 할 이벤트가 많거나 Kafka 브로커의 응답이 느릴 경우 스케줄러의 지연을 유발하고, 다음 스케줄링 주기에 영향을 줄 수 있습니다. 제안: CompletableFuture의 whenComplete 또는 thenRun 등을 사용하여 비동기적으로 처리하거나, 별도의 TaskExecutor를 사용하여 Kafka 발행 작업을 위임하는 것이 좋습니다. 예를 들어, kafkaTemplate.send(...).whenComplete((result, ex) -> { /* 처리 로직 */ }); 와 같이 변경하여 블로킹을 피할 수 있습니다.
| log.error("Outbox 최대 재시도 초과 - id: {}, topic: {}", outbox.getId(), outbox.getTopic()); | ||
| if (outbox.getTopic().equals(KafkaConstants.POST_SYNC_TOPIC)) { | ||
| PostSyncEvent event = objectMapper.readValue(outbox.getPayload(), PostSyncEvent.class); | ||
| failedPostSyncRepository.save(FailedPostSync.from(event, e.getMessage())); |
There was a problem hiding this comment.
최대 재시도 횟수(MAX_OUTBOX_RETRY)를 초과한 이벤트도 SENT 상태로 변경됩니다. NOTIFICATION_TOPIC의 경우, 이 시점에서 알림이 영구적으로 유실됩니다. SENT 상태는 "성공적으로 발행됨"을 의미하는 것이 아니라 "더 이상 재시도하지 않음"을 의미하게 되어 상태의 의미가 모호해집니다. 제안: FAILED_PERMANENTLY와 같은 새로운 OutboxEventStatus를 도입하여, 최대 재시도 횟수를 초과한 이벤트를 명확히 구분하는 것이 좋습니다. 또한, NOTIFICATION_TOPIC과 같이 FailedPostSync와 같은 별도 저장소가 없는 경우, 영구적인 실패 시 알림 유실에 대한 정책(예: 관리자 알림, 별도 로깅)을 고려해야 합니다.
|
Warning Gemini encountered an error creating the review. You can try again by commenting |
|
/gemini review |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request implements the Transactional Outbox pattern to ensure reliable message delivery to Kafka by decoupling database transactions from message publishing. It introduces a new OutboxEvent entity, a background retry scheduler, and integrates these components into the post and notification domains. The review feedback highlights several critical and high-severity issues: the use of non-unique message keys to mark events as sent which could lead to data loss, unhandled checked exceptions in JSON processing that may cause compilation or runtime failures, and the lack of timeouts on blocking Kafka calls within the scheduler. Additionally, improvements regarding query pagination and service return types were suggested to enhance scalability and traceability.
| @Query("UPDATE OutboxEvent o " + | ||
| "SET o.status = 'SENT', o.sentAt = NOW() " + | ||
| "WHERE o.messageKey = :messageKey AND o.topic = :topic AND o.status = 'PENDING'") | ||
| int markSentByMessageKeyAndTopic(@Param("messageKey") String messageKey, @Param("topic") String topic); |
There was a problem hiding this comment.
The markSentByMessageKeyAndTopic method uses messageKey and topic to identify records. However, messageKey (e.g., postId or receiverId) is not unique per event. If multiple events for the same key are in a PENDING state, calling this method will mark all of them as SENT even if only one was actually processed. This can lead to event loss. It is highly recommended to use the unique id of the OutboxEvent for marking status updates.
@Modifying
@Query("UPDATE OutboxEvent o " +
"SET o.status = 'SENT', o.sentAt = :sentAt " +
"WHERE o.id = :id AND o.status = 'PENDING'")
int markSentById(@Param("id") Long id, @Param("sentAt") java.time.LocalDateTime sentAt);| @Transactional(propagation = Propagation.MANDATORY) | ||
| public void save(String topic, String messageKey, String payload) { | ||
| outboxEventRepository.save(OutboxEvent.pending(topic, messageKey, payload)); | ||
| } |
There was a problem hiding this comment.
The save method should return the saved OutboxEvent or its id so that the caller (e.g., StudyPostService or NotificationHandler) can pass it to the Kafka producer for precise status tracking.
| @Transactional(propagation = Propagation.MANDATORY) | |
| public void save(String topic, String messageKey, String payload) { | |
| outboxEventRepository.save(OutboxEvent.pending(topic, messageKey, payload)); | |
| } | |
| @Transactional(propagation = Propagation.MANDATORY) | |
| public OutboxEvent save(String topic, String messageKey, String payload) { | |
| return outboxEventRepository.save(OutboxEvent.pending(topic, messageKey, payload)); | |
| } |
| if (outbox.getRetryCount() >= MAX_OUTBOX_RETRY) { | ||
| log.error("Outbox 최대 재시도 초과 - id: {}, topic: {}", outbox.getId(), outbox.getTopic()); | ||
| if (outbox.getTopic().equals(KafkaConstants.POST_SYNC_TOPIC)) { | ||
| PostSyncEvent event = objectMapper.readValue(outbox.getPayload(), PostSyncEvent.class); |
There was a problem hiding this comment.
objectMapper.readValue throws a checked exception (JsonProcessingException or IOException). This should be handled or wrapped in a RuntimeException to prevent the scheduler from crashing on corrupted payloads.
| @CircuitBreaker(name = "kafka", fallbackMethod = "sendFallback") | ||
| public void send(UUID receiverId, NotificationType type, String message, UUID targetId) { | ||
| NotificationEvent event = new NotificationEvent(receiverId, type, message, targetId, 0); | ||
| String payload = objectMapper.writeValueAsString(event); |
There was a problem hiding this comment.
objectMapper.writeValueAsString throws a checked exception. Since this method does not declare throws, this code will likely fail to compile unless the exception is handled.
|
|
||
| private void retry(OutboxEvent outbox) { | ||
| try { | ||
| kafkaTemplate.send(outbox.getTopic(), outbox.getMessageKey(), outbox.getPayload()).get(); |
There was a problem hiding this comment.
Using get() on a CompletableFuture without a timeout can block the scheduler thread indefinitely if Kafka is unresponsive. It's safer to specify a timeout.
| kafkaTemplate.send(outbox.getTopic(), outbox.getMessageKey(), outbox.getPayload()).get(); | |
| kafkaTemplate.send(outbox.getTopic(), outbox.getMessageKey(), outbox.getPayload()).get(10, java.util.concurrent.TimeUnit.SECONDS); |
| List<OutboxEvent> pendingEvents = outboxEventRepository | ||
| .findAllByStatusAndCreatedAtBefore(OutboxEventStatus.PENDING, LocalDateTime.now().minusSeconds(30)); |
There was a problem hiding this comment.
Fetching all pending events without a limit can lead to memory issues if the backlog is large. Consider adding a limit to the query (e.g., using Pageable).
There was a problem hiding this comment.
이번 PR은 Kafka 메시지 발행에 Outbox 패턴을 도입하여 이벤트 발행의 신뢰성을 크게 향상시켰습니다. Kafka 브로커의 일시적인 장애에도 이벤트가 유실되지 않고 재시도될 수 있도록 견고한 아키텍처를 구축한 점은 매우 긍정적입니다. 하지만 몇 가지 치명적인 버그와 개선 사항이 발견되어 코드 리뷰를 요청합니다. 특히 OutboxEvent의 상태를 업데이트하는 로직에서 데이터 일관성 문제가 발생할 수 있으며, PostSyncEvent의 경우 Outbox 이벤트가 중복으로 생성될 수 있는 문제가 있습니다.
| String message = event.postTitle() + " 스터디 지원이 승인되었습니다."; | ||
| NotificationEvent notificationEvent = new NotificationEvent( | ||
| event.applicantId(), NotificationType.APPLY_APPROVED, message, event.postId(), 0); | ||
| String payload = objectMapper.writeValueAsString(notificationEvent); |
There was a problem hiding this comment.
현재 ApplyApprovedHandler는 outboxEventService.saveWithNewTx를 호출하여 Outbox에 이벤트를 저장한 후, kafkaProducer.send를 직접 호출하고 있습니다. NotificationKafkaProducer 내부에서도 Kafka 발행 성공 시 outboxEventService.markSent를 호출하고 있습니다. 이는 다음과 같은 문제를 야기할 수 있습니다:
- 이벤트 중복 발행 가능성:
outboxEventService.saveWithNewTx로 Outbox에 저장된 이벤트와kafkaProducer.send로 즉시 발행되는 이벤트가 별개로 처리될 수 있습니다.kafkaProducer.send가 실패하면 Outbox 스케줄러가 재시도하겠지만,kafkaProducer.send가 성공하더라도 Outbox에 저장된 이벤트는markSent가 호출되지 않으면PENDING상태로 남아 스케줄러에 의해 다시 발행될 수 있습니다. markSent로직의 문제:NotificationKafkaProducer의markSent는messageKey와topic으로 Outbox 이벤트를 찾아서SENT로 변경합니다. 만약 동일한applicantId에 대한 알림 이벤트가 짧은 시간 내에 여러 번 발생하여 Outbox에 여러 개의PENDING이벤트가 쌓여있다면, 하나의 이벤트가 성공적으로 발행되었을 때markSent가 모든PENDING이벤트를SENT로 변경할 수 있습니다. 이는 치명적인 데이터 일관성 버그입니다.
개선 방안:
NotificationKafkaProducer의send메서드가OutboxEvent객체를 인자로 받아 해당id로markSent를 호출하도록 변경해야 합니다.ApplyApprovedHandler(및 다른 알림 핸들러)에서는outboxEventService.saveWithNewTx를 호출한 후, 반환된OutboxEvent의id를kafkaProducer.send에 전달하여 특정 Outbox 이벤트를SENT로 마킹하도록 해야 합니다.- 또는,
kafkaProducer.send내부에서OutboxEvent를 생성하고 발행하는 로직을 통합하여, 핸들러는kafkaProducer.send만 호출하도록 단순화하는 것이 좋습니다.
| public void save(String topic, String messageKey, String payload) { | ||
| outboxEventRepository.save(OutboxEvent.pending(topic, messageKey, payload)); | ||
| } | ||
|
|
There was a problem hiding this comment.
markSent 메서드는 messageKey와 topic을 사용하여 Outbox 이벤트를 찾습니다. 위에서 언급했듯이, 동일한 messageKey와 topic을 가진 여러 PENDING 이벤트가 존재할 경우, 이 메서드는 모든 해당 이벤트를 SENT로 변경할 수 있는 치명적인 버그를 가지고 있습니다. markSent는 특정 OutboxEvent의 id를 기반으로 동작해야 합니다.
개선 방안:
markSent(Long outboxEventId)형태로 변경하고, 호출하는 쪽에서outboxEventId를 전달하도록 수정해야 합니다.
|
|
||
| for (OutboxEvent outbox : pendingEvents) { | ||
| retry(outbox); | ||
| } |
There was a problem hiding this comment.
outboxEventService.markSent(outbox.getMessageKey(), outbox.getTopic()); 이 라인 또한 messageKey와 topic을 사용하여 Outbox 이벤트를 마킹하므로, OutboxEventService.markSent의 버그와 동일한 문제를 가지고 있습니다. outbox.getId()를 사용하여 특정 이벤트를 마킹하도록 수정해야 합니다.
|
|
||
| List<OutboxEvent> findAllByStatusAndCreatedAtBefore(OutboxEventStatus status, LocalDateTime createdAt); | ||
|
|
||
| @Modifying |
There was a problem hiding this comment.
markSentByMessageKeyAndTopic 쿼리는 messageKey와 topic을 기준으로 PENDING 상태의 모든 이벤트를 SENT로 변경합니다. 이는 동일한 messageKey와 topic을 가진 여러 PENDING 이벤트가 있을 경우, 하나의 성공적인 발행으로 인해 모든 관련 이벤트가 SENT로 처리되는 치명적인 데이터 일관성 버그를 유발합니다.
개선 방안:
- 이 쿼리를
UPDATE OutboxEvent o SET o.status = 'SENT', o.sentAt = NOW() WHERE o.id = :id AND o.status = 'PENDING'형태로 변경하고,id를 파라미터로 받도록 수정해야 합니다.
There was a problem hiding this comment.
이번 PR은 Kafka 메시지 발행에 Outbox 패턴을 도입하여 시스템의 신뢰성과 데이터 일관성을 크게 향상시켰습니다. 기존의 CircuitBreaker를 제거하고 Outbox를 통해 메시지 발행 실패 시 재시도 메커니즘을 구현한 점이 핵심입니다. 전반적으로 Outbox 패턴의 적용이 매우 잘 이루어졌으며, 트랜잭션 전파(Propagation) 설정도 각 상황에 맞게 적절히 사용되었습니다. 이는 분산 시스템에서 이벤트 기반 아키텍처의 견고성을 높이는 중요한 개선 사항입니다. 몇 가지 코드 품질 및 유지보수성 측면에서 제안할 사항이 있습니다.
| private final UserRepository userRepository; | ||
| private final ApplicationEventPublisher eventPublisher; | ||
| private final OutboxEventService outboxEventService; | ||
| private final ObjectMapper objectMapper; |
There was a problem hiding this comment.
saveOutboxEvent 헬퍼 메서드를 통해 Outbox 이벤트 저장 로직을 캡슐화한 것은 좋은 리팩토링입니다. 이를 통해 createPost, updatePost, deletePost, closePost 메서드의 가독성이 향상되었습니다.
| private final UserRepository userRepository; | ||
| private final ApplicationEventPublisher eventPublisher; | ||
| private final OutboxEventService outboxEventService; | ||
| private final ObjectMapper objectMapper; |
There was a problem hiding this comment.
outboxEventService.save는 Propagation.MANDATORY로 설정되어 있어, StudyPostService의 @Transactional 트랜잭션 내에서 Outbox 이벤트가 저장되도록 보장합니다. 이는 비즈니스 로직과 Outbox 이벤트 저장이 하나의 트랜잭션으로 묶여 원자성을 보장하는 올바른 접근 방식입니다.
|
|
||
| @Service | ||
| @RequiredArgsConstructor | ||
| public class OutboxEventService { |
There was a problem hiding this comment.
@Transactional(propagation = Propagation.MANDATORY)는 현재 진행 중인 트랜잭션이 없으면 예외를 발생시킵니다. 이는 Outbox 이벤트 저장이 항상 비즈니스 로직의 트랜잭션의 일부로 실행되도록 강제하여 데이터 일관성을 보장하는 좋은 방법입니다.
| private final OutboxEventRepository outboxEventRepository; | ||
|
|
||
| @Transactional(propagation = Propagation.MANDATORY) | ||
| public Long save(String topic, String messageKey, String payload) { |
There was a problem hiding this comment.
@Transactional(propagation = Propagation.REQUIRES_NEW)는 새로운 트랜잭션을 시작하여 Outbox 이벤트를 저장합니다. 이는 TransactionalEventListener(AFTER_COMMIT)와 같이 메인 트랜잭션 커밋 이후에 실행되거나, 스케줄러처럼 독립적인 트랜잭션이 필요한 경우에 유용합니다.
| @Component | ||
| @RequiredArgsConstructor | ||
| public class OutboxRetryScheduler { | ||
|
|
There was a problem hiding this comment.
@Scheduled(fixedDelay = 30_000)는 Outbox 이벤트 재시도 주기를 30초로 설정합니다. 이는 너무 짧지도 길지도 않은 적절한 간격으로 보이지만, 시스템의 부하와 이벤트의 중요도에 따라 조정될 수 있습니다. createdAtBefore(LocalDateTime.now().minusSeconds(30))와 함께 사용되어, 방금 생성되거나 실패한 이벤트가 즉시 재시도되는 것을 방지합니다.
|
|
||
| private static final int MAX_OUTBOX_RETRY = 3; | ||
|
|
||
| private final OutboxEventRepository outboxEventRepository; |
There was a problem hiding this comment.
findAllByStatusAndCreatedAtBefore 쿼리는 status와 createdAt 필드에 인덱스가 필수적입니다. 그렇지 않으면 Outbox 이벤트가 많아질 경우 성능 병목이 발생할 수 있습니다. 인덱스 추가를 고려해 주세요.
| kafkaTemplate.send(outbox.getTopic(), outbox.getMessageKey(), outbox.getPayload()) | ||
| .whenComplete((result, ex) -> { | ||
| if (ex == null) { | ||
| outboxEventService.markSent(outbox.getId()); |
There was a problem hiding this comment.
MAX_OUTBOX_RETRY 상수를 사용하여 최대 재시도 횟수를 관리하는 것은 좋은 관행입니다. 이를 통해 재시도 정책을 중앙에서 쉽게 변경할 수 있습니다.
| outboxEventService.markSent(outbox.getId()); | ||
| log.info("Outbox 재발행 성공 - id: {}, topic: {}", outbox.getId(), outbox.getTopic()); | ||
| } else { | ||
| log.warn("Outbox 재발행 실패 - id: {}, topic: {}", outbox.getId(), outbox.getTopic(), ex); |
There was a problem hiding this comment.
POST_SYNC_TOPIC에 대한 이벤트가 최대 재시도 횟수를 초과했을 때 FailedPostSync 엔티티로 저장하는 것은 매우 중요한 로직입니다. 이는 ES 동기화와 같은 중요한 이벤트의 영구적인 실패를 기록하고, 수동 개입 또는 별도의 복구 프로세스를 위한 기반을 마련합니다.
| @Schema(description = "Kafka 메시지 키") | ||
| @Column(nullable = false) | ||
| private String messageKey; | ||
|
|
There was a problem hiding this comment.
payload 필드의 columnDefinition = "TEXT"는 가변적인 길이의 JSON 데이터를 저장하기에 적합합니다. 다만, 매우 큰 페이로드가 자주 발생한다면 DB 성능에 영향을 줄 수 있으므로, 해당 필드에 대한 접근 패턴을 모니터링하는 것이 좋습니다.
| public interface OutboxEventRepository extends JpaRepository<OutboxEvent, Long> { | ||
|
|
||
| List<OutboxEvent> findAllByStatusAndCreatedAtBefore(OutboxEventStatus status, LocalDateTime createdAt); | ||
|
|
There was a problem hiding this comment.
markSentById 쿼리에서 o.status = 'PENDING' 조건을 추가한 것은 낙관적 잠금(Optimistic Locking)과 유사한 효과를 줍니다. 이미 다른 스레드나 프로세스에 의해 상태가 변경된 경우 불필요한 업데이트를 방지하고 데이터 일관성을 높이는 좋은 방법입니다.
1 participant
관련 이슈
closes #99
작업 내용
테스트
참고 사항