Kafka를 통한 이벤트 소싱: 상태 대신 이벤트 저장하기

1. 개요

대부분의 애플리케이션은 현재 상태만 저장합니다. 주문이 취소될 때, orders 테이블의 행이 CANCELLED로 변경됩니다. 그 전에 주문에 어떤 항목들이 있었는지, 결제했는지 여부는 더 이상 알 수 없습니다.

Event sourcing은 그 반대입니다: 우리는 발생한 이벤트 시퀀스만 저장하고, 상태는 이 시퀀스에서 계산한 결과입니다. 이 기사에서는 Spring Boot 3.x를 사용하여 aggregate Order에 대한 event log로 Kafka를 사용할 것입니다. replay를 통해 상태를 재구성하고, projection으로 read model을 구축하고, 이 방식의 실제 제한사항을 이해할 것입니다.

2. 상태 저장 vs 이벤트 저장

기존 방식에서는 주문이 하나의 레코드입니다:

orders: id=A1 | status=CANCELLED | total=250000

Event sourcing에서는 같은 주문이 추가되기만 하고 수정되거나 삭제되지 않는 이벤트 시퀀스입니다:

OrderCreated   (A1, customer=C7)
ItemAdded      (A1, sku=COFFEE, price=150000)
ItemAdded      (A1, sku=MUG,    price=100000)
OrderCancelled (A1, reason="고객 변심")

상태 CANCELLED, total=250000은 이벤트를 순서대로 “fold”하여 계산됩니다. 이러한 저장 방식은 세 가지 이점을 제공합니다:

• 자연스러운 감사(Audit): 완전한 이력이 바로 원본 데이터입니다.
• 새로운 read model은 migration이 필요 없습니다: 보고서를 추가하려면 로그를 새로운 projection으로 replay하면 됩니다.
• 시간 여행 디버깅: 언제든지 특정 시점의 상태를 재구성할 수 있습니다.

Kafka는 이 로그 역할에 적합합니다. Topic은 각 partition 내에서 순서가 있는 append-only 시퀀스입니다. orderId를 key로 사용하면 동일한 주문의 모든 이벤트가 같은 partition에 배치되고 올바른 순서를 유지합니다.

3. 프로젝트 설정

3.1. 의존성

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-json</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

Spring Boot가 spring-kafka의 버전을 관리하므로 <version>을 선언할 필요가 없습니다. spring-boot-starter-json은 Jackson과 Instant를 직렬화하기 위한 jsr310 모듈을 포함합니다.

3.2. 설정

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all
      properties:
        enable.idempotence: true
        spring.json.add.type.headers: false
    consumer:
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.value.default.type: com.nonla.orders.OrderEvent
        spring.json.trusted.packages: com.nonla.orders
        spring.json.use.type.headers: false

주목할 점:

• acks: all과 enable.idempotence: true는 이벤트가 지속적으로 기록되고 producer 재시도가 중복 레코드를 생성하지 않도록 보장합니다.
• Spring의 type header를 비활성화했습니다.
• 이벤트 유형은 type 속성을 통해 JSON 내에 직접 포함되므로 다른 언어로 작성된 consumer도 읽을 수 있습니다.
• auto-offset-reset: earliest는 새로운 consumer group이 로그의 처음부터 다시 읽을 수 있게 합니다.
• 이것은 replay를 위한 필수 조건입니다.

3.3. 이벤트를 영구적으로 유지하는 Topic

@Configuration
public class KafkaTopicConfig {

    @Bean
    NewTopic orderEvents() {
        return TopicBuilder.name("order-events")
                .partitions(3)
                .replicas(1)
                .config(TopicConfig.RETENTION_MS_CONFIG, "-1")
                .config(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE)
                .build();
    }
}

기본적으로 Kafka는 7일 후 데이터를 삭제합니다. event store의 경우 이는 원본 데이터 손실을 의미합니다. retention.ms=-1은 시간 기반 삭제를 비활성화합니다.

주의: 이벤트 topic에는 cleanup.policy=compact를 사용하지 마세요. Compaction은 각 key의 마지막 레코드만 유지하므로 주문의 전체 이력을 삭제합니다. Compaction은 snapshot topic에만 적합합니다.

4. 이벤트 정의

이벤트는 불변 데이터로, 발생한 일을 설명하며 과거형으로 이름이 지어집니다. sealed interface와 Java 21의 record를 결합하면 이에 매우 적합합니다:

package com.nonla.orders;

@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
@JsonSubTypes({
        @JsonSubTypes.Type(value = OrderEvent.Created.class, name = "OrderCreated"),
        @JsonSubTypes.Type(value = OrderEvent.ItemAdded.class, name = "ItemAdded"),
        @JsonSubTypes.Type(value = OrderEvent.Paid.class, name = "OrderPaid"),
        @JsonSubTypes.Type(value = OrderEvent.Cancelled.class, name = "OrderCancelled")
})
public sealed interface OrderEvent {

    String orderId();
    Instant occurredAt();

    record Created(String orderId, String customerId, Instant occurredAt) implements OrderEvent {}
    record ItemAdded(String orderId, String sku, BigDecimal price, Instant occurredAt) implements OrderEvent {}
    record Paid(String orderId, BigDecimal amount, Instant occurredAt) implements OrderEvent {}
    record Cancelled(String orderId, String reason, Instant occurredAt) implements OrderEvent {}
}

ItemAdded를 전송할 때 Kafka의 payload는 다음과 같은 형태입니다:

{"type":"ItemAdded","orderId":"A1","sku":"COFFEE","price":150000,"occurredAt":"2026-10-11T03:15:00Z"}

@JsonSubTypes의 이름은 장기 계약입니다. Java 클래스 이름은 변경할 수 있지만 "ItemAdded" 문자열은 변경하면 안 됩니다. 그렇지 않으면 이전 이벤트를 다시 읽을 수 없습니다.

5. Aggregate: 이벤트에서 상태 재구성

Order에는 두 가지 유형의 메서드가 있습니다:

• Command (addItem, pay, cancel): 비즈니스 규칙을 확인한 후 이벤트를 반환합니다. Command는 상태를 직접 수정하지 않습니다.
• apply: 이벤트를 받아 상태를 업데이트합니다. 이 함수는 이벤트가 이미 발생했으므로 비즈니스 오류를 throw할 수 없습니다.

public final class Order {

    public enum Status { NEW, PAID, CANCELLED }

    private String id;
    private String customerId;
    private Status status;
    private BigDecimal total = BigDecimal.ZERO;
    private final List<String> items = new ArrayList<>();
    private long version;

    public static Order replay(List<OrderEvent> history) {
        Order order = new Order();
        history.forEach(order::apply);
        return order;
    }

    public void apply(OrderEvent event) {
        switch (event) {
            case OrderEvent.Created e -> { id = e.orderId(); customerId = e.customerId(); status = Status.NEW; }
            case OrderEvent.ItemAdded e -> { items.add(e.sku()); total = total.add(e.price()); }
            case OrderEvent.Paid e -> status = Status.PAID;
            case OrderEvent.Cancelled e -> status = Status.CANCELLED;
        }
        version++;
    }

    public OrderEvent addItem(String sku, BigDecimal price) {
        requireStatus(Status.NEW, "새 주문에만 항목을 추가할 수 있습니다");
        return new OrderEvent.ItemAdded(id, sku, price, Instant.now());
    }

    public OrderEvent pay() {
        requireStatus(Status.NEW, "주문이 결제 대기 상태가 아닙니다");
        if (total.signum() == 0) {
            throw new IllegalStateException("빈 주문을 결제할 수 없습니다");
        }
        return new OrderEvent.Paid(id, total, Instant.now());
    }

    public OrderEvent cancel(String reason) {
        if (status == Status.CANCELLED) {
            throw new IllegalStateException("주문이 이미 취소되었습니다");
        }
        return new OrderEvent.Cancelled(id, reason, Instant.now());
    }

    private void requireStatus(Status expected, String message) {
        if (status != expected) throw new IllegalStateException(message);
    }

    public String id() { return id; }
    public Status status() { return status; }
    public BigDecimal total() { return total; }
    public List<String> items() { return List.copyOf(items); }
    public long version() { return version; }
}

OrderEvent가 sealed이므로 apply 내의 switch가 컴파일러에 의해 완전히 검증됩니다. 새 이벤트 유형을 추가했지만 처리를 잊으면 코드가 컴파일되지 않습니다. 이것은 정말 유용한 안전 장치입니다.

5.1. Aggregate 테스트

Aggregate는 Kafka에 의존하지 않으므로 테스트가 매우 빠릅니다:

class OrderTest {

    private final Instant now = Instant.parse("2026-10-11T03:00:00Z");

    @Test
    void replayRebuildsStateFromHistory() {
        Order order = Order.replay(List.of(
                new OrderEvent.Created("A1", "C7", now),
                new OrderEvent.ItemAdded("A1", "COFFEE", new BigDecimal("150000"), now),
                new OrderEvent.ItemAdded("A1", "MUG", new BigDecimal("100000"), now),
                new OrderEvent.Cancelled("A1", "고객 변심", now)));

        assertThat(order.status()).isEqualTo(Order.Status.CANCELLED);
        assertThat(order.total()).isEqualByComparingTo("250000");
        assertThat(order.items()).containsExactly("COFFEE", "MUG");
        assertThat(order.version()).isEqualTo(4);
    }

    @Test
    void cannotPayCancelledOrder() {
        Order order = Order.replay(List.of(
                new OrderEvent.Created("A1", "C7", now),
                new OrderEvent.Cancelled("A1", "품절", now)));

        assertThatThrownBy(order::pay)
                .isInstanceOf(IllegalStateException.class)
                .hasMessageContaining("결제 대기");
    }
}

첫 번째 테스트는 4개의 이벤트가 일관된 상태로 fold되며 version이 적용된 이벤트의 수와 같음을 보여줍니다. 두 번째 테스트는 이력에서 재구성된 상태를 기반으로 비즈니스 규칙이 검증됨을 보여줍니다.

6. Kafka에 이벤트 기록

6.1. Event store

command를 처리하려면 한 주문의 이력이 필요합니다. Kafka는 “key A1인 모든 레코드를 주세요”라는 쿼리를 지원하지 않으므로 메모리에 복사본을 유지합니다. 이 복사본은 애플리케이션이 시작될 때마다 전체 topic을 다시 읽어 로드됩니다:

@Component
public class OrderEventStore {

    private static final String TOPIC = "order-events";

    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;
    private final Map<String, List<OrderEvent>> streams = new ConcurrentHashMap<>();

    public OrderEventStore(KafkaTemplate<String, OrderEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void append(OrderEvent event) {
        try {
            kafkaTemplate.send(TOPIC, event.orderId(), event).get(10, TimeUnit.SECONDS);
        } catch (Exception e) {
            throw new IllegalStateException("이벤트를 기록할 수 없습니다 " + event, e);
        }
    }

    public List<OrderEvent> load(String orderId) {
        return List.copyOf(streams.getOrDefault(orderId, List.of()));
    }

    @KafkaListener(topics = TOPIC, groupId = "order-store-#{T(java.util.UUID).randomUUID()}")
    void on(OrderEvent event) {
        streams.computeIfAbsent(event.orderId(), id -> new CopyOnWriteArrayList<>()).add(event);
    }
}

각 부분 설명:

• send(TOPIC, event.orderId(), event)는 orderId를 key로 사용합니다. 이렇게 하면 한 주문의 모든 이벤트가 같은 partition에 있고 올바른 순서를 유지합니다.
• .get(...)은 broker의 확인을 기다립니다. Command는 이벤트가 로그에 있을 때만 성공으로 간주됩니다.
• Listener는 임의의 groupId를 사용합니다. 매번 시작할 때 새로운 group이 생성되고 earliest와 결합하면 전체 이력이 map으로 다시 읽힙니다.

6.2. Command service

@Service
public class OrderCommandService {

    private final OrderEventStore store;

    public OrderCommandService(OrderEventStore store) {
        this.store = store;
    }

    public String createOrder(String customerId) {
        String orderId = UUID.randomUUID().toString();
        store.append(new OrderEvent.Created(orderId, customerId, Instant.now()));
        return orderId;
    }

    public void addItem(String orderId, String sku, BigDecimal price) {
        Order order = loadExisting(orderId);
        store.append(order.addItem(sku, price));
    }

    public void pay(String orderId) {
        store.append(loadExisting(orderId).pay());
    }

    private Order loadExisting(String orderId) {
        List<OrderEvent> history = store.load(orderId);
        if (history.isEmpty()) {
            throw new NoSuchElementException("주문을 찾을 수 없습니다 " + orderId);
        }
        return Order.replay(history);
    }
}

처리 흐름은 항상 동일합니다: 이력 로드 → replay → command 호출 → 이벤트 append. UPDATE 명령이 없습니다.

주의: 메모리의 map은 listener를 통해 비동기로 업데이트됩니다. createOrder 직후 addItem을 호출하면 map이 아직 Created를 받지 못했을 수 있습니다. 또한 Kafka에는 “현재 버전이 4인 경우에만 쓰기”라는 optimistic concurrency가 없습니다. 같은 주문을 처리하는 두 인스턴스는 서로 모순되는 두 개의 이벤트를 기록할 수 있습니다. 프로덕션에서는 orderId의 모든 command를 단일 consumer로 라우팅해야 합니다 (command topic의 key는 orderId임), 또는 Kafka Streams와 state store를 사용하세요. 다른 옵션은 전문 event store 또는 데이터베이스를 주 쓰기 소스로 사용한 다음 Kafka에 발행하는 것입니다.

7. Projection: read model 구축

Event log는 쓰기에 좋지만 “고객 C7의 결제된 주문 나열”과 같은 쿼리에는 편리하지 않습니다. Projection이 이를 해결합니다. 이는 이벤트를 읽고 읽기에 최적화된 모델을 유지하는 consumer입니다. 이것이 바로 CQRS의 “Q” 부분입니다.

public record OrderSummary(String orderId, String customerId, String status, BigDecimal total) {}

@Component
public class OrderSummaryProjection {

    private final Map<String, OrderSummary> summaries = new ConcurrentHashMap<>();

    @KafkaListener(topics = "order-events", groupId = "order-summary-projection")
    void on(OrderEvent event) {
        summaries.compute(event.orderId(), (id, current) -> switch (event) {
            case OrderEvent.Created e -> new OrderSummary(id, e.customerId(), "NEW", BigDecimal.ZERO);
            case OrderEvent.ItemAdded e -> new OrderSummary(id, current.customerId(), current.status(),
                    current.total().add(e.price()));
            case OrderEvent.Paid e -> new OrderSummary(id, current.customerId(), "PAID", current.total());
            case OrderEvent.Cancelled e -> new OrderSummary(id, current.customerId(), "CANCELLED", current.total());
        });
    }

    public List<OrderSummary> paidOrdersOf(String customerId) {
        return summaries.values().stream()
                .filter(s -> s.customerId().equals(customerId) && s.status().equals("PAID"))
                .toList();
    }
}

이 예제는 간결함을 위해 ConcurrentHashMap을 사용합니다. 실제로는 PostgreSQL, Elasticsearch 또는 Redis에 쓸 것입니다. 중요한 점은 projection의 groupId가 고정되어 있어서 Kafka가 offset을 기억하고 각 재시작 후 새 이벤트만 처리한다는 것입니다.

Projection은 idempotent해야 합니다. Kafka는 at-least-once를 보장하므로 rebalance 후 이벤트가 다시 전달될 수 있습니다. 데이터베이스의 경우 partition별로 마지막으로 처리된 offset을 저장하고 더 이전 레코드는 무시할 수 있습니다.

7.1. Replay로 read model 재구성

projection에 total을 잘못 계산하는 버그가 있다고 가정해봅시다. 코드를 수정하고 read model 데이터를 삭제한 후 애플리케이션을 중지하고 group의 offset을 처음으로 재설정합니다:

kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group order-summary-projection \
  --topic order-events \
  --reset-offsets --to-earliest --execute

GROUP                     TOPIC         PARTITION  NEW-OFFSET
order-summary-projection  order-events  0          0
order-summary-projection  order-events  1          0
order-summary-projection  order-events  2          0

재시작 시 projection은 전체 이력을 읽고 올바른 read model을 생성합니다. 완전히 새로운 보고서를 추가하려면 새로운 groupId로 새 listener를 작성하기만 하면 됩니다. 그 보고서는 처음날부터 자동으로 모든 데이터를 가집니다. 이것이 상태 대신 이벤트를 저장하는 가장 큰 이점입니다.

8. 일반적인 실수와 사용하지 말아야 할 때

이벤트 schema 변경. 이전 이벤트는 영구적으로 존재합니다. 기본값이 있는 새 필드만 추가하고 이전 필드를 삭제하거나 의미를 변경하지 마세요. 큰 변경이 필요한 경우 새 이벤트 유형 (ItemAddedV2)을 만들거나 읽을 때 이전 이벤트를 변환하는 “upcaster”를 작성하세요. 대규모 시스템의 경우 Avro 또는 Protobuf와 함께 Schema Registry를 사용하는 것이 좋습니다.

이벤트에 상태 포함. OrderUpdated { 전체 주문 }과 같은 이벤트는 Event sourcing을 가장한 CRUD일 뿐입니다. 이벤트는 ItemAdded 또는 OrderCancelled와 같이 비즈니스 의도를 설명해야 합니다.

Replay가 너무 오래 걸림. 수천 개의 이벤트가 있는 aggregate는 로드 속도가 느립니다. 해결책은 snapshot입니다: 상태를 정기적으로 version과 함께 compacted topic에 저장하고, 로드할 때 snapshot을 읽은 후 그 이후의 이벤트만 replay합니다.

apply 내에서 side effect 호출. apply가 이메일을 보내면 매번 replay할 때마다 이메일이 다시 전송됩니다. apply는 메모리의 상태만 변경해야 합니다.

개인정보 및 삭제 권한. 불변 로그는 사용자 데이터 삭제 요청과 충돌합니다. 일반적으로 사용되는 기술은 crypto-shredding입니다: 각 고객을 위한 개인 키로 개인 데이터를 암호화하고 삭제가 필요할 때 키를 파괴합니다.

다음 경우에는 event sourcing을 사용하면 안 됩니다:

• 도메인이 주로 단순 CRUD이고 아무도 이력을 필요로 하지 않을 때.
• 팀이 쓰기 측과 read model 간의 eventual consistency에 익숙하지 않을 때.
• 여러 aggregate에 대한 쓰기 직후 일관된 쿼리가 필요할 때.

주의: Kafka는 훌륭한 event log이지만 완전한 event store는 아닙니다. 두 가지가 부족합니다: aggregate별로 빠르게 읽기와 version에 따른 조건부 쓰기입니다. 많은 시스템은 database 또는 전용 event store를 source of truth로 사용한 다음 Kafka (일반적으로 outbox pattern 통해)를 사용하여 projection 및 다른 service로 이벤트를 배포합니다.

9. 결론

이 기사에서는 주문을 덮어쓰는 단일 레코드 대신 Kafka에서 불변 이벤트 시퀀스로 저장했습니다. sealed interface와 record로 이벤트를 모델링했고, Order.replay로 상태를 재구성했으며, append 전에 비즈니스 규칙을 검증했습니다. projection으로 read model을 구축했고 offset을 재설정하여 이를 재구성했습니다.

기억해야 할 세 가지 점:

• orderId를 key로 사용하여 각 주문의 이벤트 순서를 유지합니다.
• 시간 기반 retention을 비활성화하고 이벤트 topic에 compaction을 사용하지 마세요.
• Kafka를 유일한 source of truth로 보기 전에 concurrency의 제한을 명확히 인식하세요.

올바르게 적용하면 event sourcing은 완전한 이력과 migration 없이 새로운 read model을 만드는 능력을 제공합니다.