Transactional Outbox 2부: Kafka 우선 발행으로 주문 트랜잭션 줄이기

성공 경로에서는 이벤트 행을 쓰지 않고, 실패 경로에서만 outbox 비용을 부담합니다.

2026-06-30 · 트랜잭셔널 아웃박스 시리즈 · 2부

1부는 전형적인 트랜잭셔널 아웃박스에 가까웠습니다. 도메인 행과 outbox 행을 같은 트랜잭션에 넣고, 릴레이가 outbox 이벤트를 Kafka로 전송하는 방식입니다. 안정적이고 설명하기 쉽지만, 요청이 집중되는 쓰기 경로에서는 매번 outbox 행을 하나 더 기록해야 합니다.

2부는 다른 질문에서 시작합니다.

Kafka가 정상일 때도 반드시 outbox 테이블에 먼저 기록해야 할까?

이번 실전 예제의 답은 “항상 그렇지는 않다”입니다. 주문 저장 트랜잭션은 orders만 책임집니다. 커밋한 뒤 이벤트를 Kafka로 바로 보내고, 직접 발행이 실패하거나 제한 시간을 넘기면 event_publications에 NOT_PUBLISHED 행을 남깁니다. Kafka 전면 장애나 브로커 응답 지연이 길어지면 outbox 릴레이 경로로 전환합니다.

성공 경로는 주문 행만 기록합니다. 직접 발행이 실패할 때만 대체 행과 릴레이가 등장합니다.

전형적인 outbox와 무엇이 다른가

전형적인 outbox는 보통 이렇게 설명합니다.

단계	전형적인 트랜잭셔널 아웃박스
1	같은 DB 트랜잭션에서 orders와 outbox 행을 함께 저장
2	릴레이가 처리할 outbox 행을 선점
3	Kafka 발행에 성공하면 행을 PUBLISHED로 표시

이번 예제는 성공 경로가 다릅니다.

단계	Kafka 우선 발행과 영속 대체
1	DB 트랜잭션에서는 orders만 저장
2	트랜잭션 밖에서 Kafka에 직접 발행
3	직접 발행에 성공하면 event_publications 행을 만들지 않음
4	직접 발행이 실패하거나 제한 시간을 넘길 때만 NOT_PUBLISHED 행을 저장
5	outbox 릴레이가 대체 행을 Kafka로 전송

핵심은 트랜잭션 쓰기 부하를 줄이는 대신, 이벤트 직접 발행 실패와 대체 행 저장 사이에 유실 공백이 생길 수 있음을 인정하는 것입니다. 이 공백을 없다고 가정해서는 안 됩니다. 예제는 FALLBACK_STORE_FAILED 상태와 복구 작업자를 두어 공백을 관찰하고 복구합니다.

주문 트랜잭션은 주문 행만 저장한다

TransactionalOrderWriter는 주문 저장의 트랜잭션 경계입니다. 이 클래스는 outbox 행을 기록하지 않습니다.

@Transactional
fun saveOrder(customerId: String, product: String, quantity: Int): OrderRecord {
    validate(customerId, product, quantity)

    val orderId = OrderTable.insertAndGetId {
        it[OrderTable.customerId] = customerId
        it[OrderTable.product] = product
        it[OrderTable.quantity] = quantity
        it[OrderTable.status] = OrderStatus.PENDING
    }

    return getOrder(orderId.value)
}

테스트도 이 의도를 먼저 고정합니다. transactional writer stores only order row는 주문 저장 뒤 orders = 1, event_publications = 0인지 확인합니다. 이 테스트가 없으면 누군가 안전성을 높인다는 이유로 전형적인 outbox 저장을 트랜잭션 안에 다시 넣어도 알아차리기 어렵습니다.

외부에 공개된 유스케이스는 주문을 저장한 뒤 이벤트를 만들어 발행자에게 넘깁니다.

fun placeOrder(request: OrderRequest): OrderResponse {
    val order = transactionalOrderWriter.saveOrder(
        customerId = request.customerId,
        product = request.product,
        quantity = request.quantity,
    )
    val event = OrderPlacedEvent.from(order)
    val publicationStatus = orderEventPublisher.publishDirectOrFallback(event)
    return OrderResponse.from(order, publicationStatus)
}

이 시점에는 주문 트랜잭션이 이미 끝났습니다. 따라서 API 응답에는 PUBLISHED_DIRECT, FALLBACK_STORED, FALLBACK_STORE_FAILED 같은 발행 상태가 함께 포함됩니다.

직접 발행에 성공하면 대체 테이블은 비어 있습니다. 실패할 때만 NOT_PUBLISHED 행이 생성됩니다.

Kafka 직접 발행은 짧고 명확하게 실패해야 한다

Kafka 발행을 무한히 기다려서는 안 됩니다. 예제는 직접 발행의 시도 횟수와 전체 제한 시간을 정합니다.

for (attempt in 1..properties.directPublishAttempts) {
    val remainingMillis = TimeUnit.NANOSECONDS.toMillis(deadline - System.nanoTime())
    if (remainingMillis <= 0) break

    val timeoutMillis = minOf(properties.directPublishTimeout.toMillis(), remainingMillis)
    var future: CompletableFuture<*>? = null
    try {
        future = kafkaTemplate.send(properties.topic, event.eventId, payload)
        future.get(timeoutMillis, TimeUnit.MILLISECONDS)
        outboxMetrics.recordDirectPublish("success")
        return OrderPublicationStatus.PUBLISHED_DIRECT
    } catch (e: TimeoutException) {
        future?.cancel(true)
        lastFailure = e
        outboxMetrics.recordDirectPublish("timeout")
    } catch (e: Exception) {
        lastFailure = e
        outboxMetrics.recordDirectPublish("failure")
    }
}

기본 테스트는 세 번 재시도한 뒤 대체 행을 저장하는 경로를 확인합니다. 이때 lastErrorSummary에는 secret, token, password 같은 문자열을 그대로 남기지 않습니다. 운영자가 outbox 테이블을 직접 조회할 수 있으므로, 실패 원인은 남기되 인증 정보를 노출해서는 안 됩니다.

대체 행은 별도 트랜잭션에서 저장합니다.

@Transactional(propagation = Propagation.REQUIRES_NEW)
fun upsertNotPublished(
    event: OrderPlacedEvent,
    payload: String,
    directAttemptCount: Int,
    errorCode: String,
    errorSummary: String,
) {
    EventPublicationTable.insert {
        it[eventId] = event.eventId
        it[aggregateType] = "Order"
        it[aggregateId] = event.orderId.toString()
        it[eventType] = "OrderPlaced"
        it[EventPublicationTable.payload] = payload
        it[status] = EventPublicationStatus.NOT_PUBLISHED
        it[EventPublicationTable.directAttemptCount] = directAttemptCount
        it[relayRetryCount] = 0
        it[lastErrorCode] = errorCode
        it[lastErrorSummary] = sanitize(errorSummary)
        it[nextAttemptAt] = now
    }
}

이 행이 만들어지면 이후에는 익숙한 outbox 릴레이 문제로 바뀝니다.

릴레이는 선점한 작업자를 확인한다

대체 행을 Kafka로 보내는 릴레이는 NOT_PUBLISHED 또는 FAILED 행을 선점합니다. 중요한 점은 단순 조회가 아니라 조건부 갱신으로 선점한다는 것입니다.

val updated = EventPublicationTable.update({
    (EventPublicationTable.id eq rowId) and eligibleForClaim(now)
}) {
    it[EventPublicationTable.claimedBy] = claimedBy
    it[EventPublicationTable.claimedUntil] = claimUntil
    it[updatedAt] = now
}

발행에 성공한 뒤에도 현재 선점자가 맞을 때만 PUBLISHED로 표시합니다. 두 릴레이가 같은 행을 동시에 잡거나, 선점 기한이 지난 작업자가 뒤늦게 상태를 덮어쓰는 일을 막기 위한 최소 장치입니다. 테스트는 동시 릴레이 호출이 같은 행을 두 번 선점하지 않는지, 만료된 선점이 TTL 뒤 다시 처리 대상이 되는지 확인합니다.

실패가 반복되면 FAILED로 미루다가 재시도 한도를 넘으면 DEAD_LETTER가 됩니다. 여기까지는 전형적인 outbox와 거의 같습니다. 다른 점은 이 테이블의 행이 모든 이벤트가 아니라 직접 발행에 실패했거나 복구한 이벤트라는 것입니다.

가장 위험한 지점은 대체 행 저장 실패다

이 접근의 약점은 명확합니다.

1. orders는 커밋됐다.
2. Kafka 직접 발행은 실패했다.
3. event_publications 대체 행 저장도 실패했다.

이때 이벤트가 유실될 수 있습니다. 예제는 이 상황을 숨기지 않고 FALLBACK_STORE_FAILED로 API 응답과 메트릭에 드러냅니다. 복구 작업자는 오래된 orders 행 가운데 발행 행이 없는 항목을 찾아 결정적으로 계산한 이벤트 ID로 대체 행을 재구성합니다.

fun reconcileOnce(): ReconcileResult {
    val cutoff = LocalDateTime.now(clock).minus(properties.reconcilerGrace)
    val candidates = eventPublicationRepository.findOrdersWithoutPublications()
        .filter { event -> !event.createdAt.isAfter(cutoff) }

    val reconstructed = candidates.count { event ->
        val payload = objectMapper.writeValueAsString(event)
        eventPublicationRepository.upsertReconstructed(event, payload)
    }

    return ReconcileResult(
        scanned = candidates.size,
        reconstructed = reconstructed,
        duplicateRiskDocumented = true,
    )
}

duplicateRiskDocumented = true가 중요합니다. 복구 작업자는 이벤트 유실 가능성을 줄이지만, 분산 시스템에서 “정확히 한 번” 처리를 보장하지는 않습니다. 소비자는 eventId를 기준으로 멱등성을 보장해야 합니다. 1부에서 다룬 멱등성 설계가 여기서 다시 필요합니다.

언제 이 방식을 고려할 만한가

이 방식이 “전형적인 outbox보다 항상 낫다”는 뜻은 아닙니다. 조건이 맞을 때만 의미가 있습니다.

맞는 경우	피해야 하는 경우
Kafka가 대부분 정상이고 성공 경로의 쓰기 증폭을 줄여야 한다	이벤트 유실 가능성을 허용할 수 없다
소비자가 이벤트 ID를 기준으로 멱등성을 보장한다	트랜잭션과 이벤트 영속화를 같은 원자적 경계로 묶어야 한다
대체 저장·복구 작업·메트릭을 운영할 수 있다	복구 작업자가 재구성할 수 없는 이벤트 본문을 만든다
장애 시 outbox 릴레이로 전환하는 점진적 성능 저하가 필요하다	DB outbox 행 자체가 감사 증적의 기준 정보다

실제 서비스에서는 Redis 같은 완충 계층을 한 단계 더 둘 수도 있습니다. 직접 발행 실패를 Redis Stream/List에 먼저 넣고, 작업자가 DB 대체 행 저장이나 Kafka 발행을 이어가는 방식입니다. 다만 이 경우에도 완충 계층의 내구성, 퇴거 정책, 소비자 그룹 복구와 중복 처리를 다시 설계해야 합니다. DB 트랜잭션에서 행 하나를 줄였다는 장점만 보고 전체 복구 경로를 흐리게 만들면 안 됩니다.

운영에서 확인할 신호

예제는 핵심 메트릭만 제공합니다.

메트릭	의미
workshop.outbox.direct.publish.attempts	Kafka 직접 발행의 성공·실패·시간 초과
workshop.outbox.fallback.stored	대체 행 저장의 성공·실패
workshop.outbox.relay.events	릴레이 발행·실패·배달 불가·선점 상실
workshop.outbox.reconciler.events	복구 작업자가 복원한 유실 공백

운영 대시보드에서는 fallback.stored{result=failure}와 reconciler.events{result=reconstructed}를 높은 우선순위의 경보로 두는 편이 좋습니다. 두 신호는 단순한 Kafka 장애를 넘어 이벤트 유실 가능성이 있는 구간에 진입했음을 뜻합니다.

관리용 엔드포인트도 기본값은 비활성입니다. 릴레이와 복구 작업자를 수동 실행하는 엔드포인트는 장애 대응에 유용하지만, 외부에 노출하면 같은 행을 반복 처리하거나 이벤트 본문을 노출할 수 있습니다. 테스트는 관리용 엔드포인트가 기본적으로 꺼져 있는지, 발행 조회 API가 본문이나 원본 예외를 노출하지 않는지도 확인합니다.

정리

Kafka 우선 대체 발행은 트랜잭셔널 아웃박스를 버리는 방식이 아닙니다. 성공 경로에서는 outbox 비용을 부담하지 않고, 실패 경로에서는 다시 outbox 릴레이를 사용합니다. 그 대가로 직접 발행과 대체 저장 사이의 유실 구간, 복구 작업자의 중복 발행 위험, 소비자 멱등성 요구가 생깁니다.

따라서 이 패턴의 핵심은 “더 빠르다”가 아니라 “어디서 비용을 부담할지 명시한다”입니다. 모든 요청에 outbox 행을 기록하는 비용을 핵심 트랜잭션에서 부담할지, Kafka 실패 시에만 대체 저장과 복구 비용을 부담할지 선택해야 합니다. 예제는 그 선택을 코드와 테스트로 분리해 보여주기 위한 작은 실험입니다.

관련 자료

• 예제 모듈
• 주문 트랜잭션 경계
• 직접 발행과 대체 저장
• 대체 발행 저장소
• 통합 흐름 테스트