주문은 저장됐는데 Kafka에는 안 갔다? - Dual Write에서 멱등 Consumer까지

Intro

tl;dr - DB 커밋과 Kafka 발행은 원자적이지 않다. Outbox로 유실을 막으면 중복이 오고, 중복을 막으면 동시성(Lost Update)이 터진다. 세 가지 문제를 하나씩 해결한 과정을 정리해보자.

 

kafka!

---

본론

1. 이벤트를 발행하는 가장 단순한 방법

커머스 시스템에서 주문이 생성되면 여러 곳에서 이 사실을 알아야 합니다. 상품별 판매량 집계, 조회수 카운팅, 좋아요 메트릭 갱신 - 이런 작업들을 주문 서비스가 직접 하기엔 책임이 너무 커지니까, 이벤트를 발행하고 별도 Consumer가 처리하는 구조를 택했습니다.

 

가장 먼저 떠오르는 흐름은...

@Transactional
public Order createOrder(...) {
    Order order = orderRepository.save(...);
    kafkaTemplate.send("order-events", toJson(order));  // ← 여기가 문제
    return order;
}

로컬에서 100번 돌려도 안 깨집니다. 테스트도 통과하고 지적받을 게 없어 보입니다. 하지만 프로덕션 환경에서는 이 코드가 세 가지 방식으로 깨질 수 있습니다.

 

 

---

2. 메시지가 사라지는 세 가지 순간

시나리오 A - 트랜잭션 안에서 발행 + Kafka 장애

App ─ save() ─▶ DB (O)
App ─ send() ─▶ Kafka (X) (브로커 다운)
App ─ rollback ─▶ DB ↩
→ Kafka 장애가 주문을 죽였다

kafkaTemplate.send()가 트랜잭션 안에서 실행되면, Kafka 예외가 트랜잭션을 롤백시킵니다. 주문 저장까지 날아가는 거죠. Kafka가 죽었다고 주문이 안 되면 안 됩니다.

 

시나리오 B - 트랜잭션 밖에서 발행 + 앱 크래시

App ─ save() ─▶ DB (O) (커밋 완료)
App ─ (OOM Kill)
→ 주문은 DB에 있지만 이벤트는 영원히 소실

"그럼 커밋 후에 보내면 되지 않나?" - 커밋과 send() 사이에 앱이 죽으면 이벤트는 영원히 사라집니다.

 

시나리오 C - 트랜잭션 밖에서 발행 + Kafka 장애

App ─ save() ─▶ DB (O) (커밋 완료)
App ─ send() ─▶ Kafka (X)
→ 재시도하면 중복, 포기하면 유실

DB와 Kafka는 서로 다른 시스템입니다. 두 곳에 쓰는 행위(Dual Write)는 구조적으로 원자적일 수 없습니다.

 

 

---

3. Zero-Latency Outbox: 유실도 막고, 지연도 없게

해결 아이디어는 직관적입니다. 두 시스템에 쓰지 말고, 하나의 시스템(DB)에만 쓰자.

주문을 저장할 때 같은 트랜잭션에 Outbox 레코드도 함께 저장합니다. 같이 커밋되거나 같이 롤백되니까 원자성이 보장됩니다. Kafka 발행은 커밋 후에 비동기로 처리하고, 실패하면 스케줄러가 재시도합니다.

 

이것을 3개 Layer로 나눴습니다.

Layer 1 - BEFORE_COMMIT: 메인 TX에 편승

private void saveInCurrentTxAndPublish(String aggregateType, String aggregateId, String eventType,
                                       String topic, String partitionKey, String payload) {
    Outbox outbox = Outbox.create(aggregateType, aggregateId, eventType, topic, partitionKey, payload);
    outboxJpaRepository.save(outbox);
    log.info("Outbox 저장: {} {}={}", eventType, aggregateType.toLowerCase() + "Id", aggregateId);

    TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
        @Override
        public void afterCommit() {
            asyncOutboxPublisher.publishAsync(outbox.getId(), topic, partitionKey, payload);
        }
    });
}

@TransactionalEventListener(phase = BEFORE_COMMIT)으로 메인 트랜잭션의 커밋 직전에 Outbox를 저장합니다. 주문 INSERT와 Outbox INSERT가 같은 TX — 같이 커밋되거나 같이 롤백됩니다. Dual Write 문제는 여기서 끝입니다.

실제 사용처를 보면 이벤트 타입에 따라 @TransactionalEventListener과 @EventListener를 구분합니다.

 

@TransactionalEventListener(phase = TransactionPhase.BEFORE_COMMIT)
public void handleOrderCreated(OrderCreatedEvent event) {
    saveAndPublish("Order", String.valueOf(event.orderId()), "OrderCreated",
            "order-events", String.valueOf(event.orderId()), event);
}

@EventListener
public void handlePaymentCompleted(PaymentCompletedEvent event) {
    saveAndPublish("Payment", String.valueOf(event.paymentId()), "PaymentCompleted",
            "order-events", String.valueOf(event.orderId()), event);
}

OrderCreatedEvent는 주문 저장 트랜잭션 안에서 발행되니까 BEFORE_COMMIT으로 편승합니다. 반면 PaymentCompletedEvent는 PG 콜백 처리 후 트랜잭션 밖에서 발행되므로 @EventListener를 씁니다.

TX가 없는 이벤트는 saveAndPublish() 내부에서 자동으로 분기합니다.

 

private void saveAndPublish(String aggregateType, String aggregateId, String eventType,
                            String topic, String partitionKey, Object event) {
    String payload = toPayload(eventType, event);

    if (TransactionSynchronizationManager.isSynchronizationActive()) {
        saveInCurrentTxAndPublish(aggregateType, aggregateId, eventType, topic, partitionKey, payload);
    } else {
        saveInNewTxAndPublish(aggregateType, aggregateId, eventType, topic, partitionKey, payload);
    }
}

isSynchronizationActive()가 false면 TransactionTemplate으로 자체 TX를 열어서 Outbox를 저장합니다.

 

Layer 2 - afterCommit + @Async: 즉시 발행으로 지연 제로

@Async("asyncExecutor")
public void publishAsync(Long outboxId, String topic, String partitionKey, String payload) {
    try {
        kafkaTemplate.send(topic, partitionKey, payload).get(5, TimeUnit.SECONDS);
        transactionTemplate.executeWithoutResult(status ->
                outboxJpaRepository.findById(outboxId).ifPresent(Outbox::markPublished)
        );
        log.info("Outbox 즉시 발행 성공: id={}, topic={}", outboxId, topic);
    } catch (Exception e) {
        log.warn("Outbox 즉시 발행 실패, 스케줄러가 재시도: outboxId={}", outboxId, e);
    }
}

TX 커밋 후 afterCommit 콜백에서 @Async로 Kafka 발행을 시도합니다. HTTP 응답 스레드를 블로킹하지 않으면서, 대부분의 메시지를 TX 커밋 직후 즉시 발행합니다.

 

성공하면 markPublished(), 실패하면 로그만 남기고 끝. 여기서 실패해도 데이터는 이미 Outbox 테이블에 있으니까요.

 

이게 왜 "Zero-Latency"인가?

Outbox 패턴의 전통적 약점은 스케줄러 폴링 주기만큼 지연이 발생한다는 것입니다. 5초 주기면 최악의 경우 5초 늦게 발행됩니다. afterCommit + @Async로 대부분의 메시지는 커밋 직후 밀리초 단위로 발행되고, 스케줄러는 순수한 폴백으로만 동작합니다.

 

Layer 3 - OutboxRelayScheduler: 최후의 안전망

@Scheduled(fixedDelay = 5000)
public void relay() {
    List<Outbox> failed = transactionTemplate.execute(status ->
            outboxJpaRepository.findUnpublishedBefore(
                    ZonedDateTime.now().minusSeconds(5), PageRequest.of(0, 100))
    );
    if (failed == null || failed.isEmpty()) return;

    for (Outbox outbox : failed) {
        try {
            kafkaTemplate.send(outbox.getTopic(), outbox.getPartitionKey(), outbox.getPayload())
                    .get(5, TimeUnit.SECONDS);
            transactionTemplate.executeWithoutResult(status ->
                    outboxJpaRepository.findById(outbox.getId())
                            .ifPresent(Outbox::markPublished)
            );
        } catch (Exception e) {
            log.error("Outbox fallback relay 실패: outboxId={}", outbox.getId(), e);
        }
    }
    log.info("Outbox fallback relay 완료: {}건 재시도 처리", failed.size());
}

5초마다 미발행 건을 100건씩 조회해서 재시도합니다. findUnpublishedBefore(now - 5초)로 즉시 발행에 5초 유예를 줘서, Layer 2와 충돌하지 않습니다.

 

네트워크 순단, 앱 재시작, @Async 스레드풀 고갈 - Layer 2가 실패하는 모든 케이스를 이 스케줄러가 커버합니다.

정리하면 이런 구조입니다.

[주문 저장 TX]
  ├─ orderRepository.save(order)
  ├─ outboxJpaRepository.save(outbox)     ← Layer 1: 같은 TX
  └─ COMMIT
       └─ afterCommit callback
            └─ @Async publishAsync()      ← Layer 2: 즉시 발행
                  ├─ 성공 → markPublished()
                  └─ 실패 → 로그만
                        └─ @Scheduled relay()  ← Layer 3: 5초 후 재시도

 

 

---

4. 수신 보장 1: INSERT IGNORE로 멱등성 확보

발행은 보장됐습니다. 그런데 문제가 하나 더 있습니다.

Layer 2(즉시 발행)가 성공했는데 markPublished()가 실패하면? Layer 3(스케줄러)가 같은 메시지를 다시 보냅니다. Outbox 패턴은 구조적으로 At-Least-Once - 같은 메시지가 두 번 올 수 있습니다.

 

그러면 Consumer에서 이미 처리한 이벤트인지 확인하면 되지 않나?

 

Check-Then-Act의 동시성 취약점

// 이렇게 하면 안 된다..
if (eventHandledRepository.existsByEventId(eventId)) {
    return; // 이미 처리됨
}
// ↑ 여기서 Thread B도 "없다"고 판단
eventHandledRepository.save(new EventHandled(eventId));
// → 두 스레드 모두 처리 실행

SELECT와 INSERT 사이에 다른 스레드가 끼어들 수 있습니다 (TOCTOU — Time Of Check, Time Of Use). 배치 리스너가 최대 3000건을 병렬로 처리하는 환경에서 이건 이론적 가능성이 아니라 현실적인 위험입니다.

 

DB 레벨 원자적 연산 — INSERT IGNORE

@Modifying
@Query(value = "INSERT IGNORE INTO event_handled (event_id, occurred_at, handled_at) " +
        "VALUES (:eventId, :occurredAt, NOW())", nativeQuery = true)
int insertIgnore(@Param("eventId") String eventId, @Param("occurredAt") ZonedDateTime occurredAt);

반환값이 1이면 새로 삽입(처리 진행), 0이면 이미 존재(처리 건너뜀). UNIQUE 제약조건이 DB 레벨에서 중복을 차단하니까, Application 코드의 Race Condition과 무관하게 동작합니다.

 

실제 사용은 이렇습니다.

@Transactional
public void handleLikeToggled(String eventId, Long productId, boolean liked, ZonedDateTime occurredAt) {
    if (eventHandledRepository.insertIgnore(eventId, occurredAt) == 0) {
        log.info("이미 처리된 이벤트: eventId={}", eventId);
        return;
    }
    // ... 메트릭 갱신
}

 

eventId는 어떻게 만드나

String eventId = record.topic() + ":" + record.partition() + ":" + record.offset();

topic:partition:offset 조합입니다. Kafka가 보장하는 유일성을 그대로 활용했습니다. UUID를 별도로 생성할 필요도, Producer 쪽에서 키를 심을 필요도 없습니다.

 

 

---

5. 수신 보장 2: JPA Dirty Checking의 Lost Update 극복

중복 처리는 막았습니다. 하지만 메트릭 카운터 값이 실제와 맞지 않습니다. 좋아요가 100번 발생했는데 likeCount가 97입니다.

 

JPA Dirty Checking이 일으키는 Lost Update

JPA 엔티티에 incrementLikeCount() 같은 도메인 메서드를 만들어서 카운터를 변경하면 이런 일이 벌어집니다.

// JPA 엔티티 메서드로 카운터를 변경하면?
ProductMetrics metrics = repository.findById(productId).get();
metrics.incrementLikeCount();  // this.likeCount++ (메모리에서 연산)
// → flush 시점에 UPDATE SET like_count = 97 (스냅샷 기반 덮어쓰기)

Thread A: read(likeCount=96) → likeCount++ → flush(SET likeCount=97)
Thread B: read(likeCount=96) → likeCount++ → flush(SET likeCount=97)
→ 2번 증가했는데 결과는 97. 1번 유실.

JPA Dirty Checking은 스냅샷 비교 → 절대값 덮어쓰기 방식입니다. 동시에 같은 row를 읽으면 마지막 flush가 이전 결과를 덮어씁니다. ProductMetrics 엔티티에 incrementLikeCount() 메서드가 있지만, 동시성 환경에서는 쓸 수 없는 이유입니다.

 

해결 1: 원자적 네이티브 쿼리

@Modifying
@Query("UPDATE ProductMetrics m SET m.likeCount = m.likeCount + 1, m.updatedAt = :occurredAt " +
        "WHERE m.productId = :productId AND m.updatedAt < :occurredAt")
int incrementLikeCount(@Param("productId") Long productId, @Param("occurredAt") ZonedDateTime occurredAt);

likeCount = likeCount + 1 - DB가 현재 값을 기준으로 연산합니다. Read-Modify-Write가 아니라 Modify-In-Place. 두 스레드가 동시에 실행해도 각각 +1이 정확히 반영됩니다.

 

해결 2: Stale Event 방어 - WHERE updatedAt < :occurredAt

이벤트 A (09:00:01 발생) → 네트워크 지연으로 09:00:05에 도착
이벤트 B (09:00:03 발생) → 정상 도착, 09:00:03에 처리

처리 순서: B(09:00:03) → A(09:00:01)
WHERE updatedAt(09:00:03) < occurredAt(09:00:01) → FALSE → UPDATE 무시
→ 시간이 역행하는 이벤트를 자연스럽게 필터링

반환값이 0이면 오래된 이벤트로 무시합니다. 로그를 남기되 예외는 던지지 않습니다.

 

int affected = liked
        ? productMetricsRepository.incrementLikeCount(productId, occurredAt)
        : productMetricsRepository.decrementLikeCount(productId, occurredAt);

if (affected == 0) {
    log.info("오래된 이벤트 무시: eventId={}, productId={}", eventId, productId);
}

 

해결 3: 음수 방어 - CASE WHEN

@Modifying
@Query("UPDATE ProductMetrics m SET m.likeCount = CASE WHEN m.likeCount > 0 THEN m.likeCount - 1 ELSE 0 END, " +
        "m.updatedAt = :occurredAt " +
        "WHERE m.productId = :productId AND m.updatedAt < :occurredAt")
int decrementLikeCount(@Param("productId") Long productId, @Param("occurredAt") ZonedDateTime occurredAt);

좋아요 취소(decrement)가 좋아요(increment)보다 먼저 도착하는 이벤트 순서 역전 케이스입니다. CASE WHEN likeCount > 0으로 음수를 DB 레벨에서 차단합니다.

 

결국 ProductMetricsJpaRepository에는 이런 쿼리들이 들어갔습니다.

 

메서드	핵심	방어
incrementLikeCount	likeCount + 1	Stale Event
decrementLikeCount	CASE WHEN > 0 THEN -1 ELSE 0	Stale Event + 음수
incrementViewCount	viewCount + 1	Stale Event
incrementSalesCount	salesCount + 1	Stale Event
decrementSalesCount	CASE WHEN > 0 THEN -1 ELSE 0	Stale Event + 음수

 

 

---

Conclusion

도메인 메서드는 남겨두기

ProductMetrics 엔티티에 incrementLikeCount() 같은 도메인 메서드가 있습니다. 실제로는 네이티브 쿼리만 사용하지만, 도메인 모델의 "의도 문서화" 역할로 남겨뒀습니다. "이 엔티티는 좋아요 카운트를 증가시킬 수 있다"는 비즈니스 규칙을 코드로 표현하는 거죠. 이게 올바른 판단인지 아직 확신이 없습니다.

 

Outbox 테이블은 계속 커짐..

7일 지난 건을 OutboxCleanupScheduler로 삭제하지만, 트래픽이 폭증하면 7일치가 수백만 건이 될 수 있습니다. 파티셔닝이 필요할 수도 있습니다.

 

모든 이벤트를 Outbox에 태울 필요는 없었다

선착순 쿠폰은 Outbox를 거치지 않고 KafkaCouponIssueMessagePublisher로 발행합니다. 쿠폰 발급 요청 자체가 비동기이고, Redis로 상태를 추적하니까 유실되어도 재요청이 가능하기 때문입니다. 모든 것을 하나의 패턴에 끼워 맞추려는 유혹은 항상 경계해야 합니다.