Kafka와 Outbox pattern: 데이터베이스 쓰기 시 이벤트 손실 방지

1. 개요

서비스는 보통 두 가지 작업을 연이어 해야 합니다: 데이터베이스에 데이터를 저장하고 다른 서비스들을 위해 Kafka로 이벤트를 발행하는 것입니다. 이 두 시스템은 같은 트랜잭션을 공유하지 않습니다. 한 단계는 성공하고 다른 단계가 실패하면 시스템은 일관성을 잃게 됩니다. 결과적으로 주문은 저장되었지만 OrderPlaced 이벤트를 아무도 받지 못했습니다.

이 글은 dual write 문제를 설명하고 Java 21, Spring Boot 3.x, Spring Kafka를 사용하여 Outbox pattern을 구현하는 방법을 다룹니다. 이 글을 읽고 나면, outbox 테이블에서 이벤트를 Kafka로 밀어올리는 message relay, 테스트, 그리고 idempotent consumer를 갖추게 됩니다.

2. Dual write 문제

다음의 “소박한” 방식을 살펴보세요:

@Transactional
public Order placeOrder(String customerId, BigDecimal amount) {
    Order order = orderRepository.save(new Order(customerId, amount));
    kafkaTemplate.send("orders", order.getId().toString(), toJson(order));
    return order;
}

이 코드에는 최소한 두 가지 오류 시나리오가 있습니다:

• Kafka 전송 성공, 트랜잭션 롤백. 예를 들어 commit 중 오류 또는 제약 조건 위반. ‘유령’ 이벤트가 존재하지 않는 주문에 대해 나갔습니다.
• 트랜잭션 commit 성공, Kafka 전송 실패. Broker에 접근할 수 없거나 commit 직후 애플리케이션이 충돌할 수 있습니다. 주문은 데이터베이스에 있지만 이벤트는 사라졌습니다.

두 명령의 순서를 바꿔도 근본 원인을 해결하지 못합니다. 두 개의 다른 시스템에 대한 두 작업은 분산 트랜잭션이 없으면 원자성을 가질 수 없습니다. 대부분의 팀도 복잡하고 느리기 때문에 XA/2PC를 사용하고 싶어 하지 않습니다.

3. Outbox pattern의 개념

Outbox pattern은 ’데이터베이스 쓰기 + Kafka 전송’을 ’데이터베이스 쓰기 + 데이터베이스 쓰기’로 변환합니다:

1. 같은 트랜잭션 내에서 서비스는 Order를 저장하고 outbox_event 테이블에 행을 삽입합니다.
2. message relay라는 별도의 프로세스가 outbox 테이블에서 아직 보내지 않은 행들을 읽고, Kafka로 보낸 다음 발행됨으로 표시합니다.

1단계는 한 개의 데이터베이스만 건드리므로 완전히 원자적입니다: 주문이 있으면 반드시 이벤트가 있습니다. 2단계는 몇 번이고 재시도할 수 있습니다. 그 대신, at-least-once 의미론을 얻게 됩니다: 이벤트가 중복으로 전송될 수 있지만 절대 손실되지 않습니다.

4. Spring Boot로 구현하기

4.1. Dependency

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-jpa</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.postgresql</groupId>
        <artifactId>postgresql</artifactId>
        <scope>runtime</scope>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>com.h2database</groupId>
        <artifactId>h2</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

다음은 안전한 producer 설정입니다:

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.acks=all
spring.kafka.producer.properties.enable.idempotence=true

acks=all은 모든 replica가 동기화되고 쓰기를 완료한 후 broker가 확인하도록 합니다. enable.idempotence=true는 network 계층에서 재시도할 때 producer가 중복을 자동으로 생성하지 않도록 도와줍니다.

4.2. Outbox 테이블 및 엔티티

CREATE TABLE outbox_event (
    id             UUID PRIMARY KEY,
    aggregate_type VARCHAR(100) NOT NULL,
    aggregate_id   VARCHAR(100) NOT NULL,
    event_type     VARCHAR(100) NOT NULL,
    payload        TEXT         NOT NULL,
    created_at     TIMESTAMP    NOT NULL,
    published_at   TIMESTAMP
);
CREATE INDEX idx_outbox_unpublished ON outbox_event (created_at) WHERE published_at IS NULL;

부분 인덱스는 아직 보내지 않은 행만 포함합니다. 따라서 relay의 쿼리는 테이블에 수백만 개의 이전 행이 있더라도 항상 빠릅니다.

@Entity
@Table(name = "outbox_event")
public class OutboxEvent {

    @Id
    private UUID id;
    private String aggregateType;
    private String aggregateId;
    private String eventType;
    @Column(columnDefinition = "text")
    private String payload;
    private Instant createdAt;
    private Instant publishedAt;

    protected OutboxEvent() {
    }

    public OutboxEvent(String aggregateType, String aggregateId, String eventType, String payload) {
        this.id = UUID.randomUUID();
        this.aggregateType = aggregateType;
        this.aggregateId = aggregateId;
        this.eventType = eventType;
        this.payload = payload;
        this.createdAt = Instant.now();
    }

    public void markPublished(Instant at) {
        this.publishedAt = at;
    }

    public UUID getId() { return id; }
    public String getAggregateId() { return aggregateId; }
    public String getEventType() { return eventType; }
    public String getPayload() { return payload; }
    public Instant getPublishedAt() { return publishedAt; }
}

id는 event id로도 역할합니다. Consumer는 이를 사용하여 중복을 필터링합니다 (섹션 6 참고). aggregateId는 Kafka 메시지의 key가 됩니다.

4.3. 같은 트랜잭션에서 Order와 이벤트 작성

public record OrderPlacedEvent(Long orderId, String customerId, BigDecimal amount) {
}

@Service
public class OrderService {

    private final OrderRepository orderRepository;
    private final OutboxEventRepository outboxRepository;
    private final ObjectMapper objectMapper;

    public OrderService(OrderRepository orderRepository,
                        OutboxEventRepository outboxRepository,
                        ObjectMapper objectMapper) {
        this.orderRepository = orderRepository;
        this.outboxRepository = outboxRepository;
        this.objectMapper = objectMapper;
    }

    @Transactional
    public Order placeOrder(String customerId, BigDecimal amount) {
        Order order = orderRepository.save(new Order(customerId, amount));

        var event = new OrderPlacedEvent(order.getId(), customerId, amount);
        outboxRepository.save(new OutboxEvent(
                "Order", order.getId().toString(), "OrderPlaced", toJson(event)));

        return order;
    }

    private String toJson(Object value) {
        try {
            return objectMapper.writeValueAsString(value);
        } catch (JsonProcessingException e) {
            throw new IllegalStateException("이벤트를 직렬화할 수 없습니다", e);
        }
    }
}

@Transactional이 핵심입니다. 두 save 명령은 함께 commit되거나 함께 rollback됩니다. OrderService는 Kafka에 대해 더 이상 알 필요가 없으므로 트랜잭션 내에 네트워크 호출이 없습니다.

4.4. 이벤트를 Kafka로 밀어올리는 Message Relay

먼저 행 잠금 쿼리가 있는 repository를 작성합니다:

public interface OutboxEventRepository extends JpaRepository<OutboxEvent, UUID> {

    @Query(value = """
            SELECT * FROM outbox_event
            WHERE published_at IS NULL
            ORDER BY created_at
            LIMIT :batchSize
            FOR UPDATE SKIP LOCKED
            """, nativeQuery = true)
    List<OutboxEvent> findBatchToPublish(@Param("batchSize") int batchSize);
}

FOR UPDATE SKIP LOCKED (PostgreSQL, MySQL 8+)를 사용하면 애플리케이션의 여러 인스턴스가 동시에 실행될 수 있습니다. 각 인스턴스는 자신의 배치를 잠금하고 다른 인스턴스가 처리 중인 행을 건너뜁니다. 따라서 두 인스턴스가 동시에 같은 이벤트를 전송하지 않습니다.

다음은 relay입니다:

@Component
public class OutboxRelay {

    private static final Logger log = LoggerFactory.getLogger(OutboxRelay.class);

    private final OutboxEventRepository outboxRepository;
    private final KafkaTemplate<String, String> kafkaTemplate;

    public OutboxRelay(OutboxEventRepository outboxRepository,
                       KafkaTemplate<String, String> kafkaTemplate) {
        this.outboxRepository = outboxRepository;
        this.kafkaTemplate = kafkaTemplate;
    }

    @Scheduled(fixedDelay = 1000)
    @Transactional
    public void publishPending() {
        for (OutboxEvent event : outboxRepository.findBatchToPublish(100)) {
            try {
                kafkaTemplate.send(toRecord(event)).get(5, TimeUnit.SECONDS);
                event.markPublished(Instant.now());
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return;
            } catch (Exception e) {
                log.warn("이벤트 {} 전송 실패, 나중에 재시도하겠습니다", event.getId(), e);
                break; // 순서를 유지하기 위해 중단, 다음 실행에서 재시도
            }
        }
    }

    private ProducerRecord<String, String> toRecord(OutboxEvent event) {
        var record = new ProducerRecord<>("orders", event.getAggregateId(), event.getPayload());
        record.headers()
              .add("eventId", event.getId().toString().getBytes(StandardCharsets.UTF_8))
              .add("eventType", event.getEventType().getBytes(StandardCharsets.UTF_8));
        return record;
    }
}

각 부분을 순서대로 살펴보겠습니다:

• @Scheduled(fixedDelay = 1000)은 이전 실행이 끝난 후 1초 후 relay를 실행합니다. 구성 클래스에 @EnableScheduling을 추가하는 것을 기억하세요.
• .get(5, TimeUnit.SECONDS)은 broker 확인을 기다립니다. Relay는 Kafka가 메시지를 받았을 때만 publishedAt으로 표시합니다.
• markPublished은 관리 중인 엔티티를 수정합니다. 트랜잭션 commit 시 JPA가 자동으로 UPDATE 명령을 생성하므로 save를 호출할 필요가 없습니다.
• 오류가 발생하면 relay는 그 행을 건너뛰지 않고 break합니다. 같은 주문의 이벤트들이 순서가 뒤바뀌지 않습니다.
• Key는 aggregateId이므로 하나의 Order의 모든 이벤트는 같은 파티션으로 가서 Kafka에서 순서가 유지됩니다.

broker가 다운되었을 때의 로그는 다음과 같습니다:

WARN  OutboxRelay : 이벤트 3f6c1e0a-... 전송 실패, 나중에 재시도하겠습니다
org.apache.kafka.common.errors.TimeoutException: Topic orders not present in metadata after 5000 ms.

broker가 다시 온라인 상태가 되면 다음 실행에서 정확히 그 이벤트부터 계속 전송합니다.

참고: Kafka가 메시지를 받은 후 하지만 트랜잭션 commit 전에 애플리케이션이 충돌하면 outbox 행은 여전히 ‘미발행’ 상태입니다. 다음 실행에서 다시 전송합니다. 이것이 바로 at-least-once이며, consumer는 반드시 중복을 처리할 수 있어야 합니다.

5. 테스트

첫 번째 테스트는 각 주문이 정확히 하나의 outbox 행을 생성하는지 확인합니다:

@DataJpaTest
@Import({OrderService.class, JacksonAutoConfiguration.class})
class OrderServiceTest {

    @Autowired
    OrderService orderService;

    @Autowired
    OutboxEventRepository outboxRepository;

    @Test
    void placeOrder_shouldWriteOutboxEventInSameTransaction() {
        Order order = orderService.placeOrder("customer-42", new BigDecimal("150000"));

        List<OutboxEvent> events = outboxRepository.findAll();

        assertThat(events).hasSize(1);
        OutboxEvent event = events.getFirst();
        assertThat(event.getAggregateId()).isEqualTo(order.getId().toString());
        assertThat(event.getEventType()).isEqualTo("OrderPlaced");
        assertThat(event.getPayload()).contains("\"customerId\":\"customer-42\"");
        assertThat(event.getPublishedAt()).isNull();
    }
}

Repository의 native query는 시작 시 검사되지 않습니다. 따라서 H2는 SKIP LOCKED를 완전히 지원하지 않아도 이 테스트를 실행할 수 있습니다. Relay를 실제 쿼리로 테스트하려면 Testcontainers를 통해 PostgreSQL을 사용해야 합니다.

두 번째 테스트는 Kafka가 실패할 때 relay가 발행됨으로 표시하지 않음을 보장합니다:

class OutboxRelayTest {

    OutboxEventRepository repository = mock(OutboxEventRepository.class);
    KafkaTemplate<String, String> kafkaTemplate = mock(KafkaTemplate.class);
    OutboxRelay relay = new OutboxRelay(repository, kafkaTemplate);

    @Test
    void shouldKeepEventUnpublished_whenKafkaFails() {
        var event = new OutboxEvent("Order", "1", "OrderPlaced", "{}");
        when(repository.findBatchToPublish(100)).thenReturn(List.of(event));
        when(kafkaTemplate.send(any(ProducerRecord.class)))
                .thenReturn(CompletableFuture.failedFuture(new RuntimeException("broker down")));

        relay.publishPending();

        assertThat(event.getPublishedAt()).isNull();
    }

    @Test
    void shouldMarkPublished_whenKafkaAcknowledges() {
        var event = new OutboxEvent("Order", "1", "OrderPlaced", "{}");
        when(repository.findBatchToPublish(100)).thenReturn(List.of(event));
        when(kafkaTemplate.send(any(ProducerRecord.class)))
                .thenReturn(CompletableFuture.completedFuture(null));

        relay.publishPending();

        assertThat(event.getPublishedAt()).isNotNull();
    }
}

이 두 테스트는 핵심 동작을 고정시킵니다: broker가 확인할 때만 발행됨으로 표시합니다.

6. Consumer 측: 멱등성 있는 처리

중복을 받을 수 있으므로 consumer는 처리한 eventId를 기억해야 합니다. 이 기억은 비즈니스 로직과 같은 트랜잭션에 있어야 합니다:

@Component
public class InventoryListener {

    private final ProcessedEventRepository processedEvents;
    private final InventoryService inventoryService;

    public InventoryListener(ProcessedEventRepository processedEvents,
                             InventoryService inventoryService) {
        this.processedEvents = processedEvents;
        this.inventoryService = inventoryService;
    }

    @KafkaListener(topics = "orders", groupId = "inventory-service")
    @Transactional
    public void onOrderPlaced(@Header("eventId") String eventId, String payload) {
        if (processedEvents.existsById(eventId)) {
            return; // 이미 처리됨, 중복 건너뛰기
        }
        inventoryService.reserveStock(payload);
        processedEvents.save(new ProcessedEvent(eventId));
    }
}

processed_event 테이블은 event_id를 기본 키로 설정해야 합니다. 두 스레드가 동시에 중복을 처리하려고 하면 제약 조건이 하나를 차단합니다.

7. 흔한 오류와 언제 사용하지 말 것인가

• OrderService에서 Kafka를 호출하기 “빠르게”하기. 이렇게 하면 dual write로 돌아갑니다. 모든 이벤트는 outbox 테이블을 통해야 합니다.
• Outbox 테이블 정리 잊음. 테이블이 매우 빠르게 증가합니다. 며칠보다 오래된 published_at 행을 삭제하는 작업이 있어야 합니다.
• 오류가 있는 행을 건너뛰고 계속 진행하기. 이것은 이벤트 순서를 망칩니다. payload가 영구적으로 손상되었다면 조사를 위해 FAILED 상태로 이동하고, 무한정 재시도하지 마세요.
• 너무 잦은 폴링. 많은 인스턴스로 100ms 주기는 데이터베이스에 상당한 부하를 만듭니다. 일반적으로 0.5–2초 주기면 충분합니다.
• 매우 낮은 지연 시간이나 매우 높은 처리량이 필요합니다. 이 경우 폴링을 CDC (Change Data Capture)로 바꿔야 합니다. 예를 들어 Debezium은 WAL/binlog를 읽은 다음 outbox 행을 직접 Kafka로 밀어올립니다. Outbox 테이블은 그대로 유지되고 relay만 변경됩니다.

참고: 이벤트가 중요하지 않은 경우 Outbox pattern을 사용하지 마세요. 예를 들어 손실되어도 되는 메트릭이나 사용자 행동 로그입니다. 이 경우 Kafka로 직접 전송하는 것이 훨씬 간단합니다. 반대로, Kafka transactions (transactional.id)는 Outbox를 대체할 수 없습니다: 이들은 Kafka 토픽 간의 원자성만 보장하며 데이터베이스를 포함하지 않습니다.

8. 결론

데이터베이스와 Kafka 간의 dual write는 항상 이벤트를 손실하거나 생성할 수 있는 틈이 있습니다. Outbox pattern은 세 가지 단계로 그 틈을 봉합합니다. 첫째, 비즈니스 데이터와 같은 트랜잭션에서 이벤트를 outbox 테이블에 작성합니다. 둘째, 별도의 relay를 사용하여 이벤트를 Kafka로 밀어올리고 broker가 확인할 때만 발행됨으로 표시합니다. 셋째, consumer에서 멱등성 있는 처리를 통해 at-least-once 의미론의 중복을 흡수합니다.

이제 모든 조각이 있습니다: 부분 인덱스가 있는 스키마, Kafka에 의존하지 않는 OrderService, 여러 인스턴스를 실행할 수 있는 FOR UPDATE SKIP LOCKED를 사용하는 relay, 핵심 동작에 대한 테스트, 그리고 중복을 방지하는 consumer입니다. 시스템이 커지면 비즈니스 코드를 수정하지 않고 relay 폴링을 CDC로 바꿀 수 있습니다.