Kafka Streams와 EIP Aggregator: 모으는 것보다 끝내는 것이 어렵다

EIP Aggregator 패턴은 이름만 보면 단순해 보입니다.

관련 있는 메시지를 모아서 하나의 메시지로 만든다.

Kafka Streams로 구현한다고 하면 더 쉬워 보입니다.

stream
  .groupByKey()
  .aggregate(...)

하지만 실제로 Aggregator를 구현해보면 어려운 지점은 groupByKey()나 aggregate()가 아닙니다.

진짜 어려운 부분은 이것입니다.

무엇을 같은 묶음으로 볼 것인가?
언제 이 묶음이 완료됐다고 판단할 것인가?
늦게 도착한 메시지는 어떻게 처리할 것인가?
완료된 뒤 다시 온 메시지는 버릴 것인가, 보정할 것인가?

Aggregator는 “모으는 패턴”이라기보다 “끝내는 조건을 가진 stateful pattern”에 가깝습니다.

여러 이벤트 도형이 상태 그룹으로 모인 뒤 완료 조건을 통과한 그룹만 오른쪽으로 나가는 2D 리소그래프 스타일 일러스트 — Aggregator의 핵심은 이벤트를 모으는 데서 끝나지 않습니다. 어떤 그룹이 완료됐고, 어떤 그룹은 더 기다려야 하는지 판단하는 조건이 설계의 중심입니다.

EIP Aggregator가 다루는 문제

Enterprise Integration Patterns의 Aggregator는 서로 관련된 여러 메시지를 모아 하나의 결과 메시지를 만드는 패턴입니다.

예를 들어 주문 처리 흐름에서 다음 이벤트들이 각각 다른 시점에 도착한다고 해보겠습니다.

OrderCreated
PaymentApproved
InventoryReserved
ShipmentPrepared

우리는 이 이벤트들을 orderId 기준으로 모아 OrderReadyForFulfillment 같은 하나의 메시지를 만들고 싶을 수 있습니다.

이때 Aggregator에는 최소 네 가지 결정이 필요합니다.

Correlation key: 어떤 메시지들이 같은 그룹인가
Aggregation strategy: 메시지를 어떻게 합칠 것인가
Completion condition: 언제 결과를 내보낼 것인가
Message store: 아직 완료되지 않은 그룹을 어디에 둘 것인가

Kafka Streams는 이 네 가지 중 많은 부분을 자연스럽게 지원합니다.

Kafka record key는 correlation key가 될 수 있습니다.
aggregate, reduce, count 같은 DSL 연산은 aggregation strategy를 표현할 수 있습니다.
state store는 아직 완료되지 않은 그룹을 저장할 수 있습니다.
changelog topic은 state store 복구를 가능하게 합니다.
window와 grace period는 시간 기반 완료와 late event 허용 범위를 표현할 수 있습니다.

그래서 Kafka Streams는 EIP Aggregator를 구현하기에 좋은 도구입니다.

다만 “좋은 도구”라는 말은 “자동으로 쉬워진다”는 뜻이 아닙니다.

DSL로 충분한 경우

가장 단순한 Aggregator는 windowed aggregation입니다.

예를 들어 5분 동안 들어온 주문 item event를 주문별로 모아 count를 내고 싶다면 Kafka Streams DSL로 충분할 수 있습니다.

KStream<String, OrderItemEvent> items = builder.stream("order-items");

items
    .selectKey((ignoredKey, event) -> event.orderId())
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeAndGrace(
        Duration.ofMinutes(5),
        Duration.ofMinutes(1)
    ))
    .aggregate(
        OrderItemAggregate::empty,
        (orderId, event, aggregate) -> aggregate.add(event),
        Materialized.as("order-item-aggregate-store")
    )
    .toStream()
    .to("order-item-aggregates");

이 형태는 Kafka Streams가 잘하는 영역입니다. Kafka Streams DSL 문서도 DSL이 aggregations, joins, windowing 같은 stateful transformation을 표현할 수 있다고 설명합니다.

하지만 위 코드는 아직 EIP Aggregator의 전체 문제를 다루지 않습니다.

이 코드는 “5분 window 단위로 모은다”는 의미에 가깝습니다. “주문 하나에 필요한 모든 이벤트가 모였는가”와는 다른 질문입니다.

예를 들어 주문이 준비됐다고 판단하려면 다음 조건이 필요할 수 있습니다.

OrderCreated가 있어야 한다.
PaymentApproved가 있어야 한다.
InventoryReserved가 있어야 한다.
ShipmentPrepared는 특정 상품군에서만 필요하다.
PaymentRejected가 오면 즉시 실패 결과를 내보낸다.
10분 안에 InventoryReserved가 오지 않으면 partial result를 내보낸다.

이쯤 되면 단순 windowed aggregate가 아니라 completion condition 설계 문제가 됩니다.

모으는 것보다 끝내는 것이 어렵다

Aggregator에서 completion condition은 여러 형태가 될 수 있습니다.

첫째, 개수 기반입니다.

같은 correlation key로 N개 메시지가 모이면 완료

둘째, 이벤트 타입 기반입니다.

필수 이벤트 타입이 모두 도착하면 완료

셋째, 마커 이벤트 기반입니다.

OrderClosed 같은 완료 이벤트가 오면 현재까지 모은 결과를 emit

넷째, timeout 기반입니다.

첫 메시지 도착 후 10분이 지나면 partial result emit

다섯째, 혼합 조건입니다.

필수 이벤트가 모두 오면 즉시 완료
그렇지 않으면 10분 뒤 partial result
실패 이벤트가 오면 즉시 실패

이 조건들이 실제 시스템에서는 더 복잡해집니다. 상품 종류, 고객 유형, 지역, 외부 API 응답 여부, 보정 이벤트 여부에 따라 완료 조건이 달라질 수 있습니다.

그래서 Aggregator는 집계 함수 하나로 끝나지 않습니다.

Aggregator = correlation + state + completion + release + cleanup

이벤트가 correlation key로 묶이고 state store에 저장된 뒤 completion condition에 따라 complete result, partial timeout, late path로 나뉘며 cleanup과 closed marker로 이어지는 Aggregator lifecycle 다이어그램 — Aggregator는 단순히 이벤트를 모으는 함수가 아니라, state에 머무르는 그룹을 언제 release하고 어떻게 cleanup할지 결정하는 lifecycle입니다.

Kafka Streams에서 어떤 API를 쓸지는 이 다섯 가지가 얼마나 단순한지에 따라 달라집니다.

window와 suppress가 맞는 경우

시간 window 안에서만 결과를 만들고, window가 닫힌 뒤 최종 결과만 내보내면 되는 경우에는 DSL이 꽤 깔끔합니다.

Kafka Streams에는 windowed aggregation의 중간 결과를 숨기고 window가 닫혔을 때 final result만 내보내는 suppress()가 있습니다. Kafka Streams DSL의 Window Final Results 문서는 window가 닫힐 때까지 intermediate result를 내보내지 않고 final result를 emit하는 예를 보여줍니다.

개념적으로는 이런 흐름입니다.

items
    .selectKey((ignoredKey, event) -> event.orderId())
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeAndGrace(
        Duration.ofMinutes(5),
        Duration.ofMinutes(1)
    ))
    .aggregate(
        OrderItemAggregate::empty,
        (orderId, event, aggregate) -> aggregate.add(event),
        Materialized.as("order-item-window-store")
    )
    .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()))
    .toStream()
    .to("order-item-final-aggregates");

이 방식은 좋습니다. 다만 의미를 정확히 이해해야 합니다.

suppress()는 “비즈니스 완료 조건이 만족됐다”는 뜻이 아닙니다. “window가 닫혔고, 더 이상 그 window의 결과가 바뀌지 않는다고 Kafka Streams가 판단했다”는 뜻에 가깝습니다.

따라서 다음과 같은 Aggregator에는 잘 맞습니다.

5분 단위 주문 item 요약
1시간 단위 센서 측정값 집계
하루 단위 사용자 행동 count
세션 window 단위 activity summary

하지만 다음과 같은 Aggregator에는 부족할 수 있습니다.

필수 이벤트 타입이 모두 모이면 즉시 emit
마커 이벤트가 오면 emit
실패 이벤트가 오면 실패 결과 emit
timeout이면 partial emit
완료 후 늦게 온 보정 이벤트를 별도 topic으로 emit

이 경우에는 Processor API가 더 자연스럽습니다.

Processor API가 필요한 순간

Kafka Streams Processor API 문서는 custom processor를 정의하고 state store와 연결해 더 자유로운 processing logic을 구성할 수 있다고 설명합니다.

EIP Aggregator를 Processor API로 구현하면 구조는 대략 이렇게 됩니다.

process(record):
  key = correlationKey(record)
  aggregate = store.get(key) or new aggregate
  aggregate.add(record)

  if aggregate.isComplete():
    forward(result)
    store.delete(key)
    closedStore.put(key, completionMarker)
  else:
    store.put(key, aggregate)

punctuate(now):
  for each aggregate in store:
    if aggregate.isExpired(now):
      forward(partialResult)
      store.delete(key)

실제 Java 코드는 더 길지만 핵심은 단순합니다.

public class OrderAggregatorProcessor
    implements Processor<String, OrderEvent, String, OrderAggregateResult> {

  private ProcessorContext<String, OrderAggregateResult> context;
  private KeyValueStore<String, OrderAggregate> store;

  @Override
  public void init(ProcessorContext<String, OrderAggregateResult> context) {
    this.context = context;
    this.store = context.getStateStore("order-aggregate-store");

    context.schedule(
        Duration.ofMinutes(1),
        PunctuationType.WALL_CLOCK_TIME,
        this::emitExpiredAggregates
    );
  }

  @Override
  public void process(Record<String, OrderEvent> record) {
    String orderId = record.value().orderId();
    OrderAggregate aggregate = store.get(orderId);

    if (aggregate == null) {
      aggregate = OrderAggregate.empty(orderId, record.timestamp());
    }

    aggregate = aggregate.add(record.value(), record.timestamp());

    if (aggregate.isComplete()) {
      context.forward(record.withKey(orderId).withValue(aggregate.toResult()));
      store.delete(orderId);
      return;
    }

    store.put(orderId, aggregate);
  }

  private void emitExpiredAggregates(long now) {
    try (KeyValueIterator<String, OrderAggregate> iterator = store.all()) {
      while (iterator.hasNext()) {
        KeyValue<String, OrderAggregate> entry = iterator.next();
        if (entry.value.isExpired(now)) {
          context.forward(new Record<>(
              entry.key,
              entry.value.toPartialResult(),
              now
          ));
          store.delete(entry.key);
        }
      }
    }
  }
}

이 코드는 설명용입니다. 실제 운영 코드에서는 serialization, error handling, iterator 비용, store scan 비용, output key, tombstone, metrics, replay behavior까지 더 신중히 다뤄야 합니다.

그래도 이 예시는 DSL과 Processor API의 차이를 보여줍니다.

DSL은 windowed aggregation을 짧게 표현합니다. Processor API는 completion condition과 release strategy를 코드로 직접 갖습니다.

stream-time과 wall-clock-time을 구분해야 한다

Timeout 기반 Aggregator를 만들 때 특히 조심할 것이 있습니다.

Kafka Streams에는 stream-time과 wall-clock-time 개념이 있습니다. Processor API의 punctuation도 PunctuationType.STREAM_TIME과 PunctuationType.WALL_CLOCK_TIME 중 하나를 선택합니다.

Processor API 문서는 stream-time punctuation은 record processing으로 stream time이 전진해야 호출된다고 설명합니다. 입력이 멈추면 stream-time도 전진하지 않을 수 있습니다.

이 차이는 timeout Aggregator에서 중요합니다.

stream-time timeout:
  이벤트 시간 기준으로 오래된 그룹을 닫는다.
  입력이 멈추면 timeout도 같이 멈출 수 있다.

wall-clock-time timeout:
  실제 시간 기준으로 오래된 그룹을 닫는다.
  입력이 없어도 주기적으로 확인할 수 있다.

비즈니스 의미가 “event time 기준으로 10분 window가 닫혔다”라면 stream-time이 자연스럽습니다. 운영 의미가 “실제 시간으로 10분 동안 더 안 오면 partial result를 내보낸다”라면 wall-clock-time이 더 맞을 수 있습니다.

이 선택을 모호하게 두면, 테스트에서는 잘 되는데 운영에서 timeout이 예상과 다르게 움직일 수 있습니다.

state store는 구현 세부가 아니라 설계의 중심이다

Aggregator는 stateful pattern입니다. 아직 완료되지 않은 그룹을 어딘가에 기억해야 합니다.

Kafka Streams에서 그 역할을 하는 것이 state store입니다. Processor API의 state store 문서는 state store가 최근 record 기억, rolling aggregate 추적, deduplication 등에 쓰일 수 있다고 설명합니다.

운영 관점에서 state store는 단순한 local cache가 아닙니다.

생각해야 할 것이 많습니다.

aggregate 하나의 크기
동시에 열려 있는 correlation key 수
timeout 전까지 보관할 최대 시간
late event를 위해 닫힌 group marker를 보관할 기간
RocksDB local state 크기
changelog topic 크기
rebalance나 장애 후 복구 시간
standby replica 필요 여부

특히 “완료된 뒤 다시 온 메시지”를 어떻게 처리할지 정해야 합니다.

완료 후 늦게 온 이벤트를 그냥 새 aggregate로 시작하면 이상한 결과가 나올 수 있습니다. 이미 OrderReadyForFulfillment를 내보낸 뒤 같은 orderId의 PaymentApproved가 다시 오면, 그 이벤트를 새 주문 흐름처럼 보면 안 됩니다.

그래서 운영 구현에서는 completed marker store를 별도로 둘 때가 있습니다.

open aggregate store:
  아직 완료되지 않은 그룹

closed aggregate marker store:
  이미 완료된 correlation key와 completion timestamp

이 marker는 영원히 보관할 필요는 없습니다. 하지만 late event 허용 범위나 재처리 정책만큼은 유지해야 합니다.

Kafka key와 repartition을 가볍게 보면 안 된다

Aggregator는 같은 correlation key의 메시지를 같은 task에서 봐야 합니다. Kafka Streams에서는 record key와 partitioning이 중요합니다.

입력 topic의 key가 이미 orderId라면 좋습니다. 하지만 그렇지 않다면 selectKey()나 groupBy() 과정에서 repartition이 발생할 수 있습니다.

이 자체가 나쁜 것은 아닙니다. 문제는 이를 인지하지 못하는 것입니다.

입력 key: eventId
correlation key: orderId
Aggregator key: orderId

이 경우 Kafka Streams는 orderId 기준으로 grouping하기 위해 repartition topic을 만들 수 있습니다. 내부 topic이 늘어나고, 데이터가 한 번 더 Kafka를 거치며, topology 이름과 store 이름이 운영에 영향을 줍니다.

따라서 Aggregator를 만들 때는 다음을 확인해야 합니다.

입력 topic key가 correlation key와 같은가?
key 변경이 필요하다면 repartition 비용을 받아들일 수 있는가?
internal topic 이름과 state store 이름이 안정적인가?
partition 수는 expected throughput과 key distribution에 맞는가?
특정 key가 너무 큰 aggregate를 만들어 hotspot이 되지 않는가?

Kafka Streams DSL은 편하지만, 내부 topology와 internal topic을 모른 채 쓰면 운영에서 당황할 수 있습니다.

output은 업데이트인가, 최종 결과인가

Kafka Streams의 aggregate() 결과는 보통 KTable입니다. KTable은 “현재까지의 최신 상태”를 표현합니다. 이를 toStream()으로 내보내면 aggregation이 업데이트될 때마다 downstream에 변경이 흘러갈 수 있습니다.

하지만 EIP Aggregator에서 원하는 output은 보통 하나입니다.

완료된 그룹당 결과 메시지 하나

따라서 output semantics를 명확히 해야 합니다.

중간 상태도 downstream에 보낼 것인가?
최종 결과만 보낼 것인가?
partial result도 정상 output인가?
timeout result와 complete result를 같은 topic에 넣을 것인가?
late correction은 보정 topic으로 분리할 것인가?

이 질문은 기술 문제가 아니라 계약 문제입니다. Downstream consumer가 “최종 결과 하나”를 기대하는데 intermediate update가 여러 번 오면 consumer 쪽이 복잡해집니다.

반대로 downstream이 진행 상황을 보고 싶다면 intermediate update가 유용할 수 있습니다.

중요한 것은 결과 topic의 의미를 이름과 schema로 드러내는 것입니다.

order-aggregate-updates
order-aggregate-finalized
order-aggregate-timeouts
order-aggregate-corrections

이렇게 분리하면 consumer가 훨씬 덜 혼란스럽습니다.

Exactly-once는 completion condition을 대신하지 않는다

Kafka Streams에는 processing guarantee 설정이 있습니다. Exactly-once processing은 state store update와 output topic write를 더 안전하게 묶는 데 도움이 됩니다.

하지만 이것이 Aggregator의 completion condition을 대신하지는 않습니다.

Exactly-once는 “동일한 처리 결과가 중복으로 반영되는 위험”을 줄이는 데 도움이 됩니다. 그러나 다음 질문에는 답하지 않습니다.

어떤 이벤트가 모두 모였는가?
timeout이면 partial로 내보낼 것인가?
late event를 correction으로 처리할 것인가?
이미 완료된 group에 다시 온 event를 어떻게 볼 것인가?

이 질문은 비즈니스 규칙입니다. Kafka Streams의 processing guarantee가 아니라 Aggregator 설계가 답해야 합니다.

운영에서 봐야 할 신호

Kafka Streams Aggregator를 운영할 때는 broker metric만 보면 부족합니다.

봐야 할 신호는 다음과 같습니다.

input topic lag
output topic throughput
state store 크기
RocksDB memory와 disk 사용량
changelog topic throughput과 size
restore 시간
open aggregate count
expired aggregate count
completed aggregate count
late event count
duplicate event count
partial result count
processor error count

특히 open aggregate count와 expired aggregate count는 중요합니다. 이 값이 예상보다 계속 늘면 completion condition이 너무 엄격하거나, 특정 upstream 이벤트가 누락되고 있거나, correlation key가 잘못 잡혔을 수 있습니다.

Konduo처럼 Kafka, JVM, host, container, alert 신호를 함께 보는 도구는 이런 Kafka Streams 애플리케이션을 운영할 때도 도움이 됩니다. Aggregator의 지연은 단순히 consumer lag만의 문제가 아니라 state store 크기, RocksDB I/O, JVM memory, broker changelog traffic, input partition skew가 함께 만든 결과일 수 있기 때문입니다.

테스트는 시간과 재처리를 포함해야 한다

Aggregator 테스트는 happy path만으로 부족합니다.

다음 케이스를 넣어야 합니다.

필수 이벤트가 모두 순서대로 도착한다.
필수 이벤트가 역순으로 도착한다.
동일 이벤트가 중복 도착한다.
필수 이벤트 일부가 오지 않아 timeout된다.
완료 후 late event가 도착한다.
실패 이벤트가 먼저 도착한다.
재시작 후 state store가 changelog에서 복구된다.
reprocessing 시 결과가 중복 emit되지 않는다.

Kafka Streams에는 TopologyTestDriver를 이용한 테스트 흐름이 있습니다. 단순 aggregation 함수 테스트와 topology-level 테스트를 나눠두면 completion condition을 훨씬 명확하게 검증할 수 있습니다.

Aggregator는 “몇 개 모였는가”보다 “이 상태에서 내보내도 되는가”가 더 중요합니다. 테스트도 그 질문을 중심으로 만들어야 합니다.

마무리

Kafka Streams는 EIP Aggregator를 구현하기 좋은 도구입니다.

Kafka의 key 기반 partitioning, Streams의 state store, windowed aggregation, changelog, Processor API는 Aggregator에 필요한 많은 재료를 이미 갖고 있습니다.

하지만 Aggregator를 groupByKey().aggregate()로만 생각하면 중요한 절반을 놓칩니다.

모으는 것은 시작이다.
끝내는 조건이 Aggregator의 본질이다.

단순 시간 window와 final result면 DSL과 suppress()로 충분할 수 있습니다. 하지만 비즈니스 완료 조건, timeout, partial result, late event, duplicate handling, closed group marker가 필요해지면 Processor API와 명시적인 state store 설계가 더 자연스럽습니다.

Kafka Streams로 Aggregator를 만들 때 가장 먼저 정해야 할 것은 API가 아닙니다.

correlation key
completion condition
late event policy
state retention
output contract

이 다섯 가지가 정리되면 구현은 훨씬 덜 흔들립니다. 반대로 이 다섯 가지가 애매하면, 어떤 API를 써도 Aggregator는 운영에서 애매하게 동작합니다.

함께 읽기 좋은 글

이 글의 관점과 이어지는 주제를 더 보려면 아래 글도 함께 읽어볼 수 있습니다.

Kafka는 왜 Outbox와 Inbox 사이에 잘 맞을까 - 비동기 메시징에서 중복과 재처리를 견디는 애플리케이션 계약을 다룹니다.
Kafka가 끼어도 Trace는 끊기지 않아야 한다 - Kafka Streams 전후 흐름을 추적하려면 trace context를 어떻게 이어야 하는지 설명합니다.
Consumer Lag은 건강 점수가 아니다: Kafka Consuming Pressure라는 관점 - Streams 애플리케이션 지연을 lag보다 넓은 처리 압력 관점에서 봅니다.