Kafka Streams

이 도메인이 묻는 것

참고 챕터는 10장 Kafka Streams와 ksqlDB입니다. Streams는 표준 Kafka 프로듀서·컨슈머 위에 올라간 라이브러리이므로, Application Development 도메인의 지식이 그대로 적용됩니다. Streams 애플리케이션의 병렬성은 결국 입력 토픽의 파티션 수이고, 상태 저장소는 결국 Kafka의 컴팩션 토픽(changelog)입니다. 이 두 연결을 잡고 있으면 문제 대부분이 풀립니다.

핵심 개념 (1) KStream · KTable · GlobalKTable

핵심 개념 (2) stateless · stateful · 리파티션

이 절의 표가 이 도메인 최다 출제 지점입니다. 연산을 세 축으로 분류해야 합니다 — 상태를 갖는가, 리파티션을 유발하는가, 그리고 키를 바꾸는가.

상태 저장소와 changelog

• stateful 연산은 로컬 상태 저장소(기본 RocksDB)를 만듭니다.
• 같은 내용이 changelog 토픽({application.id}-{store}-changelog)에 기록됩니다. 이 토픽은 cleanup.policy=compact입니다.
• 인스턴스가 죽으면 다른 인스턴스가 changelog를 재생해 상태를 복구합니다. 상태가 크면 복구가 오래 걸리므로 num.standby.replicas(기본 0)를 올려 따뜻한 예비 사본을 유지합니다.
• 리파티션 토픽은 {application.id}-{name}-repartition이며 cleanup.policy=delete입니다 (중간 데이터이므로).

핵심 개념 (3) 조인 매트릭스와 co-partitioning

co-partitioning 요건 — 공식 요건은 2개입니다

공식 문서(streams/developer-guide/dsl-api)가 co-partitioning 요구사항으로 열거하는 것은 두 개입니다. 키가 같아야 한다는 것은 요건이 아니라 equi-join 자체의 전제로 별도 서술됩니다.

• 전제 — 양쪽 입력의 키가 같은 의미여야 합니다 (equi-join이 키 기준으로 수행되므로 leftKey == rightKey).
• 요건 ① — 양쪽 입력 토픽의 파티션 수가 같아야 합니다. Streams가 런타임에 검증하는 유일한 항목입니다.
• 요건 ② — 양쪽에 쓰는 모든 애플리케이션이 같은 파티셔닝 전략을 써야 합니다 (같은 partitioner.class 또는 같은 StreamPartitioner). 검증되지 않으므로 어긋나면 예외 없이 조인이 조용히 누락됩니다.

KStream<String, Order>   orders   = builder.stream("orders");
KStream<String, Payment> payments = builder.stream("payments");

// 5분 윈도우 안에서 같은 키를 만나면 조인한다
KStream<String, Settled> settled = orders.join(
    payments,
    (order, payment) -> new Settled(order, payment),   // ValueJoiner
    JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)),
    Joined.with(Serdes.String(), orderSerde, paymentSerde));

핵심 개념 (4) 윈도우 4종

시간 개념과 grace period

• event time — 레코드에 기록된 타임스탬프. Streams의 기본 기준입니다.
• ingestion time — 브로커가 로그에 쓴 시각 (토픽 message.timestamp.type=LogAppendTime).
• processing time — 처리하는 순간의 벽시계 시각.

타임스탬프 추출기의 기본값은 FailOnInvalidTimestamp이며, 음수(무효) 타임스탬프를 만나면 예외를 던집니다. 레거시 데이터를 다룰 때는 LogAndSkipOnInvalidTimestamp나 WallclockTimestampExtractor로 바꾸는 선택지를 씁니다.

grace period는 윈도우가 닫힌 뒤에도 늦게 온 레코드를 받아 주는 유예 기간입니다. …WithNoGrace(…)를 쓰면 유예가 0이므로 지연 도착 레코드가 버려집니다. grace를 크게 주면 정확도가 오르지만 결과 확정이 늦어지고 상태가 커집니다. 중간 결과를 억제하고 최종 결과만 내보내려면 suppress()를 씁니다.

핵심 개념 (5) 병렬성과 exactly-once

• 태스크 수 = 입력 토픽의 파티션 수입니다 (sub-topology별로 계산). 파티션이 6개면 태스크는 6개이고, 그 이상으로는 병렬 처리가 늘지 않습니다.
• num.stream.threads(기본 1)는 한 인스턴스가 돌리는 스레드 수입니다. 태스크는 스레드에 배분됩니다.
• 인스턴스를 늘리면 태스크가 인스턴스 사이로 재배분됩니다. 인스턴스 × 스레드 > 태스크 수가 되면 남는 스레드는 유휴 상태입니다.
• Streams는 내부적으로 컨슈머 그룹을 쓰며, 그룹 ID는 application.id입니다. 그래서 application.id를 바꾸면 처음부터 다시 읽습니다.

인터랙티브 쿼리와 ksqlDB

상태 저장소는 KafkaStreams.store(...)로 직접 조회할 수 있습니다. 인스턴스가 여러 대면 키가 어느 인스턴스에 있는지 메타데이터로 찾아 그 인스턴스에 요청을 전달해야 합니다 (내장 RPC는 없고 애플리케이션이 구현합니다).

ksqlDB는 Kafka Streams 위에 올라간 SQL 계층이며, Apache Kafka 본체가 아니라 Confluent 컴포넌트입니다. CREATE STREAM / CREATE TABLE이 각각 KStream / KTable에 대응합니다. Java 코드 없이 스트림 처리를 하고 싶을 때 쓰고, 복잡한 로직·커스텀 상태가 필요하면 Streams로 갑니다.

반드시 외워야 할 설정값

자주 나오는 함정

함정 1 — map vs mapValues

함정 2 — KStream vs KTable

함정 3 — KTable vs GlobalKTable

함정 4 — 윈도우 4종

함정 5 — co-partitioning vs repartition()

함정 6 — SMT vs Streams (재확인)

Connect 도메인의 같은 함정이 Streams 쪽에서도 나옵니다. 집계·조인·윈도우는 SMT로 불가능하고, 필드 하나 바꾸는 일에 Streams 애플리케이션을 띄우는 것은 과잉입니다. 경계는 "여러 레코드의 관계가 필요한가"입니다.

코드 읽기 문제 대비

스니펫 1 — 리파티션 토픽이 몇 개 생기는가

builder.stream("clicks")                       // key = sessionId
       .filter((k, v) -> v.getStatus() == 200)
       .mapValues(v -> v.getUrl())              // (A)
       .selectKey((k, url) -> url)              // (B)
       .groupByKey()
       .count()
       .toStream()
       .to("url-counts");

스니펫 2 — 이 조인은 동작하는가

KStream<String, Order>    orders    = builder.stream("orders");
KTable<String, Customer>  customers = builder.table("customers");

KStream<String, Enriched> enriched = orders.join(
    customers,
    (order, customer) -> new Enriched(order, customer));

스니펫 3 — 이 설정으로 EOS가 동작하는가

application.id=fraud-detector
bootstrap.servers=localhost:9092
processing.guarantee=exactly_once_v2
default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
default.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde

이어서 볼 곳

• 10장 · Kafka Streams와 ksqlDB 토폴로지, 상태 저장소, 윈도우, 조인의 전체 설명. 기본개념
• 예제 9 · Streams 실시간 집계 동작하는 윈도우 집계 코드와 테스트. 실무 예제
• Application Testing TopologyTestDriver로 토폴로지를 브로커 없이 검증합니다. CCDAK 8%
• 케이스 8 · 상태 토픽이 사라짐 changelog 토픽의 cleanup.policy를 잘못 둔 결과. 실수 케이스
• Streams 치트시트 DSL 연산 표, 윈도우 비교, 조인 매트릭스. 빠른참조
• 함정 사전 40쌍 Streams 구역 5쌍을 포함한 전체 비교표. CCDAK 도구

도메인 미니 퀴즈

공식 문서 출처

• Streams Core Concepts — 스트림·테이블 이원성, 태스크와 스레드, 시간 개념
• Streams DSL — 연산 분류, 리파티션, 조인 매트릭스, co-partitioning 요구사항과 TopologyException, 윈도우 4종 API
• Streams Configs — num.stream.threads, num.standby.replicas, state.dir, processing.guarantee 기본값
• Memory Management — statestore.cache.max.bytes
• Interactive Queries — 상태 저장소 직접 조회
• Streams Configs (표) — 전체 설정 기본값