Outbox Relay 최적화, 10건에서 155건으로

Outbox Relay 최적화: 10건/초에서 155건/초까지

TL;DR

Outbox Relay의 초기 구현은 10건/초였다. PROCESSING 상태 추가 + FOR UPDATE SKIP LOCKED + parallelStream으로 155건/초까지 개선했다. 부하 테스트로 병목이 Phase 2(Kafka 동기 발행)임을 확인했고, 5분 PROCESSING 복구 threshold가 충분히 안전한 것을 수치로 증명했다.

---

1. 초기 구현의 한계 — 왜 10건/초인가

[이전 글](커밋 후 발행 vs 발행 후 커밋)에서 Outbox Pattern을 "왜" 선택했는지 다뤘다.

이 글은 그 다음 질문에 답한다. Outbox를 선택했으면, 처리량은 어떻게 끌어올리는가?

초기 구현은 단순했다.

5초 간격 폴링 × 50건 배치 = 최대 10건/초

이것이 전부였다. @Scheduled(fixedDelay = 5000)으로 5초마다 깨어나 PENDING 상태인 이벤트 50건을 조회하고, 순차적으로 Kafka에 발행한 뒤, 상태를 PUBLISHED로 변경했다.

처리량 자체도 문제였지만, 더 근본적인 문제는 멀티 인스턴스에서 발생했다. 두 인스턴스가 동시에 같은 PENDING 이벤트를 조회하면, 같은 이벤트가 Kafka에 두 번 발행된다. Consumer 쪽에서 멱등성으로 방어할 수 있지만, 그건 "문제를 Consumer에게 떠넘기는 것"이다.

"폴링 간격을 줄이면 되지 않나?"라는 생각이 들 수 있다. 5초를 1초로 줄이면 처리량은 50건/초로 오른다. 하지만 중복 발행 문제는 해결되지 않는다. 오히려 폴링이 빨라질수록 두 인스턴스가 동시에 같은 행을 조회할 확률이 높아진다.

처리량과 정합성, 두 문제를 동시에 해결해야 했다.

---

2. PROCESSING 상태 — 상태 머신 설계

첫 번째 결정은 상태를 추가하는 것이었다.

public enum OutboxStatus {
    PENDING,      // 발행 대기
    PROCESSING,   // 발행 중 (멀티 인스턴스 중복 방지)
    PUBLISHED,    // 발행 완료
    FAILED        // 발행 실패
}

초기 구현에는 PENDING, PUBLISHED, FAILED 세 가지만 있었다. 여기에 PROCESSING을 추가했다.

왜 4개 상태가 필요한가? 각 상태의 전이 조건을 보면 명확하다.

핵심은 Relay를 두 단계로 분리한 것이다.

• Phase 1: PENDING → PROCESSING 전환. DB 상태만 변경하는 작업이라 빠르다. 이 단계에서 행 수준 락(row-level lock)을 잡고, 상태를 변경한 뒤, 즉시 커밋하고 락을 해제한다.
• Phase 2: PROCESSING → PUBLISHED / FAILED. Kafka에 실제로 메시지를 보내는 작업이라 느리다. 하지만 이 시점에서는 이미 행이 PROCESSING이므로, 다른 인스턴스가 같은 행을 가져갈 수 없다.

Phase 1에서 빠르게 락을 해제해야 다른 인스턴스가 다음 배치를 잡을 수 있다. 만약 Phase 1과 Phase 2를 하나의 트랜잭션에서 처리하면, Kafka 발행이 끝날 때까지 DB 락을 잡고 있게 된다. Kafka 응답이 2초 걸리면, 그 2초 동안 다른 인스턴스는 새로운 PENDING 행을 잡을 수 없다.

// Phase 1: PENDING → PROCESSING (빠름, 락 최소 보유)
@Scheduled(fixedDelay = 1000)
@Transactional
public void markPendingAsProcessing() {
    long startTime = System.currentTimeMillis();

    List<OutboxEventEntity> pendingEvents =
        outboxEventJpaRepository.findPendingEventsForUpdate(BATCH_SIZE);

    if (pendingEvents.isEmpty()) {
        return;
    }

    for (OutboxEventEntity event : pendingEvents) {
        event.markProcessing();
    }
    outboxEventJpaRepository.saveAll(pendingEvents);

    long duration = System.currentTimeMillis() - startTime;
    metrics.recordPhase1Duration(duration);
}

Phase 1은 @Transactional로 묶여 있다. 메서드가 끝나면 커밋되고, 락이 해제된다. 이 과정에서 소요되는 시간은 뒤의 부하 테스트에서 측정한다.

---

3. FOR UPDATE SKIP LOCKED — 경합 없는 행 분배

Phase 1에서 findPendingEventsForUpdate가 호출하는 쿼리를 보자.

@Query(value = """
    SELECT * FROM outbox_event
    WHERE status = 'PENDING'
    ORDER BY created_at ASC
    LIMIT :limit
    FOR UPDATE SKIP LOCKED
    """, nativeQuery = true)
List<OutboxEventEntity> findPendingEventsForUpdate(@Param("limit") int limit);

 

FOR UPDATE는 SELECT한 행에 배타적 행 수준 락(exclusive row-level lock)을 건다. 다른 트랜잭션이 같은 행을 SELECT ... FOR UPDATE하면 대기한다. 앞선 트랜잭션이 커밋하거나 롤백해야 락이 풀린다.

FOR UPDATE SKIP LOCKED는 동작이 다르다. 이미 잠긴 행을 만나면 대기하지 않고 건너뛴다. 이것이 멀티 인스턴스 환경에서의 핵심이다.

 

인스턴스 A: SELECT ... FOR UPDATE SKIP LOCKED LIMIT 500
  → row 1~500 락 획득

인스턴스 B: SELECT ... FOR UPDATE SKIP LOCKED LIMIT 500 (동시)
  → row 1~500은 이미 잠김 → SKIP
  → row 501~1000 락 획득

 

일반 FOR UPDATE였다면 인스턴스 B는 인스턴스 A가 커밋할 때까지 대기한다. 결과적으로 두 인스턴스가 직렬화되어 멀티 인스턴스의 의미가 없어진다. SKIP LOCKED는 각 인스턴스가 겹치지 않는 행을 동시에 가져가므로, 인스턴스 수에 비례하여 처리량이 증가한다.

Native Query로 전환한 이유

JPQL은 DB 벤더 독립적인 쿼리 언어다. 하지만 SKIP LOCKED는 SQL 표준이 아닌 DB 벤더 확장 기능이다. Hibernate가 @Lock(LockModeType.PESSIMISTIC_WRITE)와 @QueryHints를 통해 일부 지원하지만, SKIP LOCKED는 JPQL 레벨에서 정상 동작하지 않는 경우가 있다. Hibernate의 락 힌트 처리가 DB 방언(dialect)에 따라 달라지기 때문이다.

확실하게 FOR UPDATE SKIP LOCKED가 실행되도록 Native Query를 선택했다. 이 쿼리가 실제로 중복을 방지하는지는 통합 테스트에서 검증했다.

@Test
@DisplayName("멀티 스레드 환경에서 FOR UPDATE SKIP LOCKED가 중복 조회를 방지한다")
void for_update_skip_locked_prevents_duplicate_acquisition() throws InterruptedException {
    // given: 100개의 PENDING 이벤트 생성
    for (int i = 1; i <= 100; i++) {
        outboxEventService.save("ORDER", (long) i, "OrderConfirmedEvent",
                Map.of("orderId", i), "order-events-v1", String.valueOf(i));
    }

    // when: 4개의 스레드가 동시에 PENDING → PROCESSING 전환 시도
    int threadCount = 4;
    int batchSize = 50;
    ExecutorService executor = Executors.newFixedThreadPool(threadCount);
    CountDownLatch latch = new CountDownLatch(threadCount);
    Map<Integer, List<Long>> acquiredEventsByThread = new ConcurrentHashMap<>();

    for (int threadId = 0; threadId < threadCount; threadId++) {
        int finalThreadId = threadId;
        executor.submit(() -> {
            try {
                List<Long> acquired = transactionTemplate.execute(status -> {
                    List<OutboxEventEntity> events =
                        outboxEventJpaRepository.findPendingEventsForUpdate(batchSize);
                    events.forEach(OutboxEventEntity::markProcessing);
                    outboxEventJpaRepository.saveAll(events);
                    return events.stream().map(OutboxEventEntity::getId).toList();
                });
                acquiredEventsByThread.put(finalThreadId, acquired != null ? acquired : List.of());
            } finally {
                latch.countDown();
            }
        });
    }

    latch.await(10, TimeUnit.SECONDS);

    // then: 모든 이벤트가 정확히 1번씩만 조회됨
    List<Long> allAcquiredIds = acquiredEventsByThread.values().stream()
            .flatMap(List::stream)
            .toList();

    assertThat(allAcquiredIds).hasSize(100);          // 100개 모두 조회됨
    assertThat(allAcquiredIds).doesNotHaveDuplicates(); // 중복 없음
}

 

4개 스레드가 동시에 50건씩 조회해도, 100건이 정확히 한 번씩만 분배된다. SKIP LOCKED 덕분이다.

SKIP LOCKED는 MySQL 8.0+, PostgreSQL 9.5+에서 지원된다. Oracle은 SKIP LOCKED라는 같은 구문을 훨씬 오래전부터 지원해왔다. 이 프로젝트는 MySQL 8.0을 사용한다.

---

4. parallelStream으로 Partition Key별 병렬 발행

Phase 2는 Kafka에 실제로 메시지를 보내는 단계다. 여기서 고려해야 할 것은 순서 보장이다.

Kafka는 Partition 단위로 순서를 보장한다. 같은 Partition Key를 가진 메시지는 같은 Partition에 들어가고, Consumer는 Partition 내에서 순서대로 읽는다. 따라서:

• 같은 Partition Key → 같은 Partition → 순서 보장 필요 → 순차 발행
• 다른 Partition Key → 다른 Partition → 순서 무관 → 병렬 발행

이 규칙을 코드로 표현하면, Partition Key별로 그룹핑한 뒤, 그룹 간에는 병렬, 그룹 내에서는 순차 처리하면 된다.

private void executePhase2() {
    long startTime = System.currentTimeMillis();

    List<OutboxEventEntity> processingEvents =
        outboxEventJpaRepository.findProcessingEvents(BATCH_SIZE);

    if (processingEvents.isEmpty()) {
        return;
    }

    // Partition Key별 그룹핑 (LinkedHashMap으로 삽입 순서 유지)
    var groupedByPartitionKey = processingEvents.stream()
            .collect(Collectors.groupingBy(
                    OutboxEventEntity::getPartitionKey,
                    LinkedHashMap::new,
                    Collectors.toList()
            ));

    // Partition Key별 병렬 발행 (다른 Key는 병렬, 같은 Key는 순차)
    groupedByPartitionKey.values().parallelStream().forEach(events -> {
        for (OutboxEventEntity event : events) {
            try {
                publishToKafka(event);
            } catch (Exception e) {
                event.markFailed("Unexpected error: " + e.getMessage());
                metrics.recordPublishFailure();
            }
        }
    });

    // 상태 업데이트 일괄 커밋
    transactionTemplate.executeWithoutResult(status -> {
        outboxEventJpaRepository.saveAll(processingEvents);
    });

    long duration = System.currentTimeMillis() - startTime;
    metrics.recordPhase2Duration(duration);
}

 

LinkedHashMap을 사용한 이유는 삽입 순서를 유지하기 위해서다. HashMap은 순서를 보장하지 않으므로, 같은 Partition Key 내에서 이벤트 순서가 뒤섞일 수 있다. LinkedHashMap은 put 순서를 유지하므로, findProcessingEvents가 ORDER BY created_at ASC로 반환한 순서가 그룹 내에서도 보존된다.

parallelStream은 ForkJoinPool.commonPool()을 사용한다. 이 풀의 크기는 Runtime.getRuntime().availableProcessors() - 1이다. CPU 바운드 작업을 위해 설계된 풀에 I/O 바운드 작업(Kafka 네트워크 전송)을 실행하는 것이다. 최적은 아니지만, Partition Key가 10개인 환경에서 충분한 병렬도를 제공했다. 이 한계는 8장에서 다시 다룬다.

 

Kafka 발행: 동기 .get()

각 이벤트의 Kafka 발행은 동기로 처리한다.

private void publishToKafka(OutboxEventEntity event) {
    try {
        var producerRecord = new ProducerRecord<Object, Object>(
                event.getTopic(), null, event.getPartitionKey(), event.getPayload());
        producerRecord.headers()
                .add("X-Event-Type", event.getEventType().getBytes(UTF_8))
                .add("X-Aggregate-Type", event.getAggregateType().getBytes(UTF_8))
                .add("X-Outbox-Id", String.valueOf(event.getId()).getBytes(UTF_8));

        SendResult<Object, Object> result = kafkaTemplate
                .send(producerRecord)
                .get(10, TimeUnit.SECONDS);  // 동기 대기

        event.markPublished();
        metrics.recordPublishSuccess();

    } catch (ExecutionException e) {
        event.markFailed(e.getCause().getMessage());
        metrics.recordPublishFailure();
    } catch (TimeoutException e) {
        event.markFailed("Kafka send timeout (10s)");
        metrics.recordPublishFailure();
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        event.markFailed("Interrupted");
        metrics.recordPublishFailure();
    }
}

 

.get(10, SECONDS)로 Kafka 응답을 동기 대기한다. 왜 비동기 콜백(whenComplete)이 아닌 동기 방식을 선택했는지는 7장에서 설명한다.

Kafka 헤더에 X-Event-Type, X-Aggregate-Type, X-Outbox-Id를 넣는 이유는, Consumer가 payload를 역직렬화하기 전에 헤더만으로 이벤트 유형을 판별하고 라우팅할 수 있게 하기 위해서다.

---

5. 부하 테스트 결과 — 이론 vs 실측

최적화 후의 이론값을 먼저 계산해보자.

1초 폴링 × 500건 배치 = 500건/초 (이론 최대)

 

실제로는 얼마나 나올까? Testcontainers 기반 E2E 테스트로 측정했다.

시나리오 1: 1,000건 기준선

10개 Partition Key에 100건씩, 총 1,000건을 생성한 뒤 Phase 1과 Phase 2를 순차 실행했다.

 

단계	1차 배치 (500건)	2차 배치 (500건)	합계
Phase 1	62~165ms	40~120ms	102~285ms
Phase 2	1.8~4.6초	1.5~3.8초	3.3~8.4초

전체 처리량: 84.5건/초

Phase 1은 배치당 수십~수백 밀리초다. DB 상태 변경만 하기 때문이다. Phase 2는 배치당 수 초가 걸린다. 500건을 Kafka에 보내는 시간이 지배적이다.

시나리오 2: 5,000건 burst

한꺼번에 5,000건을 넣고 Phase 1 + Phase 2를 반복했다.

지표	측정값
Phase 1 평균	~40ms/배치
Phase 2 평균	~2.4초/배치
전체 소화 시간	~32.3초
사이클 수	10 (500건 × 10)
실제 처리량	154.6건/초

이론값 500건/초 vs 실측 154.6건/초. 3.2배 차이.

이 차이는 어디서 오는가? Phase 2의 .get(10, SECONDS) 동기 발행이다. 500건을 parallelStream으로 보내더라도, commonPool의 스레드 수(CPU 코어 - 1)가 병렬도를 제한한다. 10개 Partition Key를 동시에 처리하고 싶지만, 테스트 환경에서 commonPool 크기가 7~11 정도이므로, 모든 그룹이 완전히 병렬로 실행되지는 않는다.

Phase 1은 배치당 40ms. 500건의 UPDATE 문을 실행하는 데 충분히 빠르다. DB는 병목이 아니다.

---

6. Kafka 지연 시뮬레이션 — PROCESSING 복구 안전성

Phase 2에서 Kafka 발행이 지연되면 어떤 일이 벌어지는가? PROCESSING 상태로 전환된 이벤트가 오래 머물면, 복구 로직이 이를 다시 PENDING으로 되돌린다.

@Scheduled(fixedDelay = 300000) // 5분
@Transactional
public void recoverStalledProcessingEvents() {
    ZonedDateTime threshold = ZonedDateTime.now().minusMinutes(5);
    List<OutboxEventEntity> stalledEvents =
        outboxEventJpaRepository.findStalledProcessingEvents(threshold);

    if (stalledEvents.isEmpty()) {
        return;
    }

    stalledEvents.forEach(OutboxEventEntity::markRetry);
    outboxEventJpaRepository.saveAll(stalledEvents);
}

 

5분 threshold는 충분히 안전한가? Kafka가 얼마나 느려져야 5분에 도달하는지 시뮬레이션했다.

KafkaTemplate의 send()에 인위적인 지연을 주입하여, 건당 0ms / 50ms / 200ms 지연 상황에서 500건 배치의 Phase 2 소요 시간을 측정했다.

 

건당 지연	Phase 2 소요 (500건)	5분(300초) 대비
0ms	5.0초	1.7%
50ms	6.9초	2.3%
200ms	14.7초	4.9%

 

건당 200ms 지연이 발생해도 Phase 2는 14.7초면 끝난다. 5분의 4.9%에 불과하다. parallelStream으로 10개 Partition Key가 병렬 처리되기 때문에, 건당 지연이 있어도 전체 소요 시간은 비례적으로 증가하지 않는다.

5분에 도달하려면 건당 약 6초의 지연이 필요하다. 이 정도면 Kafka 브로커가 거의 응답 불능 상태다. 정상적인 Kafka 지연(수십 ms)과는 차원이 다른 수준이다.

결론: PROCESSING 복구 5분 threshold는 충분히 안전하다.

만약 Kafka가 정말 죽었다면? .get(10, SECONDS) 타임아웃이 먼저 발동하여 FAILED로 전환된다. 5분 복구 대상은 "Kafka는 살아 있는데 애플리케이션이 크래시한 경우"에 해당한다.

---

7. 동기 vs 비동기 — 왜 느린 쪽을 선택했는가

"동기 .get()이 병목이라면, 비동기 콜백으로 바꾸면 되지 않나?"

이 프로젝트와 별도로 운영하는 kafka-pipeline-lab에서 비동기 콜백 방식을 실험했다. 결과는 버그였다.

// kafka-pipeline-lab에서 발견한 문제
kafkaTemplate.send(record).whenComplete((result, ex) -> {
    if (ex == null) {
        outboxEvent.markPublished();
        repository.save(outboxEvent);  // 이 save()가 영속되지 않는다
    }
});

 

whenComplete()는 Kafka Producer의 I/O 스레드(sender thread)에서 실행된다. 이 스레드는 Spring의 트랜잭션 컨텍스트 밖에 있다. @Transactional이 설정한 ThreadLocal 기반의 EntityManager에 접근할 수 없다.

결과적으로 repository.save(outboxEvent)가 호출되어도, 영속성 컨텍스트가 없으므로 DB에 반영되지 않는다. 이벤트는 Kafka에 발행되었지만, outbox_event 테이블의 상태는 PROCESSING 그대로다. 다음 복구 사이클에서 PENDING으로 되돌아가고, 다시 발행된다. 무한 재발행 루프다.

이 문제를 해결하려면 콜백 내에서 새로운 트랜잭션을 열거나(TransactionTemplate), 비동기 결과를 수집하여 메인 스레드에서 일괄 커밋해야 한다. 두 방법 모두 복잡도가 크게 증가한다.

동기 .get()은 느리지만 명확하다.

1. kafkaTemplate.send().get(10, SECONDS)  → Kafka 응답 대기
2. event.markPublished()                   → 메모리 상태 변경
3. 루프 끝난 후 saveAll()                  → 일괄 DB 커밋

 

모든 작업이 같은 스레드에서 순차적으로 일어난다. 트랜잭션 경계가 명확하고, 어디서 실패했는지 추적이 쉽다.

Outbox는 속도보다 정확성이 목적이다. 이벤트를 한 번만, 확실하게 발행하기 위한 패턴이다. 처리량이 정말 병목이 되면, Relay 자체를 폴링에서 CDC(Change Data Capture)로 전환하는 것이 근본적인 해법이다.

---

8. 다음 단계 — 처리량 향상 선택지

155건/초가 현재 프로젝트에서 충분한가? 트래픽 규모에 따라 다르다. 만약 부족하다면, 세 가지 선택지가 있다.

전용 ExecutorService

현재 parallelStream은 ForkJoinPool.commonPool()을 사용한다. 이 풀은 JVM 전체에서 공유되며, CPU 바운드 작업에 최적화되어 있다. Kafka 발행처럼 I/O 대기가 긴 작업에는 스레드 수가 부족하다.

전용 ExecutorService를 만들어 I/O 대기에 맞는 스레드 수(예: 20~50)를 설정하면, 병렬도를 높여 처리량을 개선할 수 있다. 트레이드오프는 스레드 풀 관리 복잡도가 증가하고, 과도한 스레드 수는 Kafka 브로커에 커넥션 부담을 줄 수 있다는 것이다.

배치 크기 증가

현재 500건 배치를 2,000건으로 늘리면, 폴링 1회당 처리량이 4배가 된다. 하지만 Phase 1에서 FOR UPDATE SKIP LOCKED로 잡는 행 수가 늘어나면, 락 보유 시간도 길어진다. 2,000건의 UPDATE가 40ms에서 끝나면 괜찮지만, 160ms가 걸린다면 다른 인스턴스의 대기 시간이 늘어난다.

또한 Phase 2에서 2,000건을 한 번에 처리하면, Kafka 발행 실패 시 재처리 범위도 커진다. 적절한 배치 크기는 Phase 1 소요 시간과 Phase 2 실패 빈도를 함께 고려해야 한다.

CDC (Debezium)

가장 근본적인 해법이다. Debezium 같은 CDC 도구가 MySQL의 binlog를 실시간으로 읽어 Kafka에 전달한다. outbox_event 테이블에 INSERT가 일어나면, binlog에 기록되고, Debezium이 이를 감지하여 Kafka로 전송한다.

기존: App → INSERT → Relay(폴링) → Kafka
CDC:  App → INSERT → binlog → Debezium → Kafka

 

폴링 자체가 사라진다. PROCESSING 상태도, FOR UPDATE SKIP LOCKED도, @Scheduled도 필요 없다. Relay 서비스 전체가 불필요해진다.

트레이드오프는 인프라 복잡도다. Debezium Connector를 Kafka Connect 클러스터에서 운영해야 하고, binlog 접근 권한 설정, 스키마 변경 시 호환성 관리, Connector 장애 시 복구 등 운영 부담이 늘어난다. 폴링 기반 Relay가 애플리케이션 코드만으로 완결되는 것과 대비된다.

현재 프로젝트에서는 폴링 기반 Relay의 155건/초가 트래픽 대비 충분하므로, CDC 전환은 처리량이 실제로 병목이 될 때 검토할 계획이다.

---

정리

항목	초기 구현	최적화 후
폴링 간격	5초	1초
배치 크기	50건	500건
이론 처리량	10건/초	500건/초
실측 처리량	-	155건/초
중복 방지	없음	FOR UPDATE SKIP LOCKED
상태	3개 (PENDING/PUBLISHED/FAILED)	4개 (+PROCESSING)
Kafka 발행	순차	Partition Key별 병렬
복구	없음	5분 threshold + Graceful Shutdown

처리량 15배 향상(10 → 155건/초)의 핵심은 세 가지였다.

1. PROCESSING 상태 분리: Phase 1(DB, 빠름)과 Phase 2(Kafka, 느림)를 나눠 락 보유 시간을 최소화
2. FOR UPDATE SKIP LOCKED: 멀티 인스턴스가 서로 다른 행을 가져가 경합 없이 처리량 확장
3. parallelStream: Partition Key별 병렬 발행으로 Phase 2 소요 시간 단축

병목은 Phase 2의 Kafka 동기 발행이다. 그리고 이 병목은 의도적인 선택이다. Outbox Pattern의 본질은 "안전한 발행"이다. 처리량이 한계에 도달하면, 폴링 Relay를 최적화하는 것이 아니라, CDC로 전환하는 것이 올바른 방향이다.

---

References

• Chris Richardson, Microservices Patterns — Transactional Outbox Pattern
• Martin Kleppmann, Designing Data-Intensive Applications — 이중 쓰기(dual write) 문제와 CDC
• MySQL 8.0 Reference Manual — SELECT ... FOR UPDATE SKIP LOCKED
• Debezium Documentation — Outbox Event Router