Kafka Streams와 ksqlDB

클러스터가 아니라 라이브러리입니다

스트림 처리 프레임워크는 보통 자체 클러스터와 리소스 매니저를 요구합니다. Kafka Streams는 그 방향을 택하지 않았습니다. 공식 문서의 표현대로 "어떤 Java 애플리케이션에도 쉽게 임베드할 수 있는 단순하고 가벼운 클라이언트 라이브러리"이고, 내부 메시징 계층으로 Apache Kafka 자체 외에 어떤 외부 시스템에도 의존하지 않습니다. 문서는 "Kafka Streams는 리소스 매니저가 아니며, 스트림 처리 애플리케이션이 도는 곳이면 어디서든 돈다"고 명시합니다.

토폴로지 — source · processor · sink

Streams 애플리케이션은 프로세서 토폴로지(processor topology)로 계산 로직을 정의합니다. 토폴로지는 스트림 프로세서(노드)를 스트림(엣지)으로 연결한 그래프이며, 두 개의 특별한 프로세서가 있습니다.

• source processor — 상위 프로세서가 없는 노드. 하나 이상의 Kafka 토픽에서 레코드를 읽어 하위로 전달합니다.
• sink processor — 하위 프로세서가 없는 노드. 받은 레코드를 지정한 Kafka 토픽으로 보냅니다.

토폴로지를 정의하는 방법은 두 가지입니다 — Kafka Streams DSL(map, filter, join, 집계 등을 바로 제공)과 더 저수준인 Processor API(커스텀 프로세서를 직접 정의하고 상태 저장소에 접근). 토폴로지는 논리적 추상일 뿐이고, 런타임에는 병렬 처리를 위해 인스턴스화되고 복제됩니다.

import java.util.Arrays;
import java.util.Locale;
import java.util.Properties;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Produced;

public final class WordCount {

    public static Topology buildTopology() {
        StreamsBuilder builder = new StreamsBuilder();

        // source processor: 입력 토픽
        KStream<String, String> lines = builder.stream("text-lines");

        KTable<String, Long> counts = lines
            // 값만 바꿉니다 → 리파티션을 유발하지 않습니다
            .flatMapValues(line -> Arrays.asList(line.toLowerCase(Locale.ROOT).split("\\W+")))
            // 키를 단어로 바꿉니다 → 여기서 리파티션이 필요해집니다
            .groupBy((key, word) -> word)
            // stateful: 상태 저장소와 changelog 토픽이 생깁니다
            .count(Materialized.as("counts-store"));

        // sink processor: 출력 토픽
        counts.toStream().to("word-counts", Produced.with(Serdes.String(), Serdes.Long()));

        return builder.build();
    }

    public static void main(String[] args) {
        Properties props = new Properties();
        // application.id 는 컨슈머 group.id, client.id 접두어, changelog 토픽 접두어로 함께 쓰입니다
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-1:9092,kafka-2:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        // changelog · 리파티션 토픽의 복제 계수. 기본값 -1 은 브로커 기본값을 따릅니다
        props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 3);

        KafkaStreams streams = new KafkaStreams(buildTopology(), props);

        // 예외로 스트림 스레드가 죽었을 때의 정책을 반드시 정하세요.
        // 정하지 않으면 스레드가 하나씩 사라져 처리 용량이 조용히 줄어듭니다.
        streams.setUncaughtExceptionHandler(
            exception -> StreamThreadExceptionResponse.REPLACE_THREAD);

        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
        streams.start();
    }
}

KStream · KTable · GlobalKTable

같은 토픽을 읽어도 무엇으로 해석하느냐에 따라 결과가 달라집니다. 공식 문서가 드는 예시가 가장 명확합니다 — ("alice", 1) 다음에 ("alice", 3)이 들어올 때 사용자별 값을 합산하면,

• KStream이면 4입니다. 두 번째 레코드는 앞 레코드를 대체하지 않는 별개의 사실이기 때문입니다.
• KTable이면 3입니다. 두 번째 레코드가 앞 값의 갱신으로 해석되기 때문입니다.

stateless와 stateful, 그리고 리파티션

DSL 연산을 분류하는 축은 두 개이고 서로 독립입니다 — 상태 저장소를 만드는가와 리파티션을 유발하는가입니다. map은 stateless이지만 리파티션을 유발하고, count는 stateful이지만 그 자체가 리파티션을 유발하지는 않습니다. 이 두 축을 섞으면 문제를 틀립니다.

상태 저장소와 changelog

stateful 연산을 쓰면 DSL이 상태 저장소(state store)를 자동으로 만들고 관리합니다. 기본 구현은 디스크 기반 키-값 저장소(RocksDB)이고, 인메모리 해시맵도 선택할 수 있습니다. 내고장성은 changelog 토픽이 담당합니다.

• 각 상태 저장소마다 복제되는 changelog Kafka 토픽이 유지되며 모든 상태 갱신이 기록됩니다.
• changelog도 파티션되어 있어 각 저장소 인스턴스(=태스크)마다 전용 changelog 파티션을 가집니다.
• changelog 토픽에는 로그 컴팩션이 켜져 있어 옛 데이터가 안전하게 정리됩니다 — 무한히 커지지 않습니다.
• 태스크가 다른 머신에서 재시작되면 changelog를 재생(replay)해 장애 이전 상태로 복원한 뒤 처리를 이어갑니다.

문제는 복원 시간입니다. 태스크 재초기화 비용은 대부분 changelog 재생 시간이 차지합니다. 이를 줄이는 장치가 standby replica입니다. num.standby.replicas를 올리면 상태의 완전 복제본을 다른 인스턴스에 미리 유지하고, 태스크가 이동할 때 이미 최신 상태를 가진 인스턴스로 배정합니다. 2.6부터는 완전히 따라잡은 로컬 복사본을 가진 인스턴스가 존재한다면 반드시 그 인스턴스에 배정됩니다.

시간 — event time · processing time · ingestion time

Streams는 TimestampExtractor 인터페이스로 모든 레코드에 타임스탬프를 부여합니다. 이 값이 진행 상황을 나타내는 stream time이고, 실제 실행 시각인 wall-clock time과 구분됩니다. stream time은 새 레코드가 프로세서에 도착할 때만 전진합니다 — 데이터가 멈추면 시간도 멈춥니다. 윈도우가 닫히지 않는 현상의 원인이 대부분 이것입니다.

출력 레코드의 타임스탬프는 어떻게 정해지는가

윈도우 4종

윈도우는 stateful 연산을 위해 같은 키를 가진 레코드를 시간 구간으로 다시 묶는 장치입니다. 공식 문서가 강조하는 전제: 윈도우는 레코드 키별로 따로 추적됩니다.

grace period — 늦게 온 데이터를 언제까지 받아 줄 것인가

grace period는 특정 윈도우에 대해 순서가 어긋난(out-of-order) 레코드를 얼마나 기다릴지를 정합니다. 공식 문서의 판정 규칙은 정확히 이렇습니다 — 레코드의 타임스탬프가 어떤 윈도우에 속하지만 현재 stream time이 그 윈도우의 끝 + grace period보다 크면, 그 레코드는 폐기되고 윈도우에 반영되지 않습니다.

윈도우 상태 저장소의 보관 기간(retention)은 grace period와 별개입니다. Streams는 윈도우를 최소 이 기간 동안 유지하며, 기본값은 1일이고 Materialized#withRetention()으로 바꿉니다.

윈도우 계산은 새 데이터가 올 때마다 결과를 갱신해 내려보냅니다. 최종 결과만 필요하면(알림 발송, 갱신을 지원하지 않는 시스템으로 전달 등) suppress로 윈도우가 닫힐 때까지 억제해야 합니다.

조인 매트릭스와 co-partitioning

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

공식 문서는 "조인 시 입력 데이터가 co-partition되어 있어야 하며, 그것을 보장하는 것은 사용자의 책임이다"라고 명시합니다. 요건은 다음과 같습니다.

파티션 수가 다를 때

공식 문서가 권하는 절차는 이렇습니다.

1. 양쪽 중 파티션 수가 적은 쪽을 찾습니다(SMALLER). bin/kafka-topics --describe로 확인합니다.
2. 애플리케이션 안에서 SMALLER를 repartition으로 재분배합니다. LARGER와 같은 파티셔너를 써야 합니다.
   • KStream이면 KStream#repartition(Repartitioned.numberOfPartitions(...))
   • KTable이면 KTable#toStream#repartition(Repartitioned.numberOfPartitions(...).toTable())
3. LARGER와 새 스트림/테이블을 조인합니다.

방향에 대한 문서의 권고도 함께 기억할 만합니다 — 병목을 피하려면 적은 쪽을 많은 쪽에 맞추는 것이 권장됩니다. stream-table 조인에서는 KStream을 재분배하는 쪽이 좋습니다. KTable을 재분배하면 상태 저장소가 하나 더 생길 수 있기 때문입니다. table-table 조인이면 더 작은 KTable을 재분배하는 편이 낫습니다.

// orders: 12 파티션, payments: 6 파티션 → 그대로 조인하면 TopologyException
KStream<String, Order>   orders   = builder.stream("orders");
KStream<String, Payment> payments = builder.stream("payments");

// 적은 쪽(payments)을 12로 맞춥니다
KStream<String, Payment> payments12 =
    payments.repartition(Repartitioned.numberOfPartitions(12));

// KStream-KStream 조인은 반드시 윈도우가 필요합니다
KStream<String, Settlement> settled = orders.join(
    payments12,
    (order, payment) -> Settlement.of(order, payment),
    JoinWindows.ofTimeDifferenceAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(1))
);

병렬성 — 태스크와 스레드

Streams의 병렬성 모델은 Kafka 파티션에 그대로 얹혀 있습니다.

• 각 스트림 파티션은 완전히 순서가 정해진 레코드 시퀀스이며 Kafka 토픽 파티션에 대응합니다.
• Streams는 입력 스트림 파티션에 기반해 고정된 수의 태스크를 만들고, 각 태스크에 파티션 목록을 배정합니다.
• 파티션-태스크 배정은 절대 변하지 않습니다. 태스크가 곧 고정된 병렬 단위입니다.
• 따라서 최대 병렬성은 입력 토픽의 최대 파티션 수입니다. 입력 토픽이 5파티션이면 인스턴스를 최대 5개까지 의미 있게 돌릴 수 있습니다.
• 파티션 수보다 많은 인스턴스는 기동은 되지만 유휴 상태로 남습니다. 대신 바쁜 인스턴스가 죽으면 그중 하나가 일을 이어받습니다 — 무의미한 것은 아니고 hot standby가 됩니다.

exactly_once_v2

Streams의 exactly-once가 보장하는 범위는 정확히 이렇습니다 — 소스 Kafka 토픽에서 읽은 어떤 레코드든, 그 처리 결과가 출력 Kafka 토픽과 stateful 연산의 상태 저장소에 정확히 한 번 반영됩니다. 입력 토픽 오프셋 커밋, 상태 저장소 갱신, 출력 토픽 쓰기가 원자적으로 완료됩니다.

인터랙티브 쿼리

Streams는 상태 저장소를 읽기 전용으로 직접 조회하는 것을 허용합니다. 저장소를 만든 애플리케이션 외부의 메서드·스레드·프로세스·애플리케이션에서도 접근할 수 있으며, 모든 저장소는 이름을 가지고 인터랙티브 쿼리는 읽기 연산만 노출합니다.

이 기능의 가치는 출력 토픽을 거치지 않고 최신 집계 결과를 바로 서빙할 수 있다는 점입니다. 집계 결과를 다시 외부 DB에 넣고 그것을 조회하는 단계가 사라집니다. 다만 상태는 인스턴스에 분산되어 있으므로, 찾는 키가 다른 인스턴스에 있으면 그 인스턴스로 요청을 라우팅해야 합니다. 그 메타데이터를 Streams가 제공합니다.

Streams Rebalance Protocol (KIP-1071)

KIP-848이 일반 컨슈머의 리밸런스 조정을 클라이언트에서 브로커로 옮겼듯이, KIP-1071은 같은 모델을 Kafka Streams 워크로드로 확장합니다. 리밸런스 이벤트마다 클라이언트가 배정을 계산하는 대신, 브로커에서 배정이 지속적으로 계산됩니다. 애플리케이션은 컨슈머 그룹이 아니라 streams group으로 등록됩니다.

streams group에는 컨슈머 그룹에 없는 NOT_READY 상태가 있습니다. 그룹 코디네이터가 토폴로지에 필요한 소스·내부 토픽이 없거나 설정이 맞지 않는 것을 감지하면 이 상태로 들어가고, 모든 멤버는 빈 배정을 받습니다. 코디네이터가 하는 검사에는 co-partition 그룹이 실제로 co-partition되어 있는지 확인하는 단계가 포함됩니다 — 앞에서 본 요건이 브로커 쪽에서도 검증되기 시작한 것입니다.

테스팅 — 브로커 없이 토폴로지를 검증합니다

CCDAK에는 Application Testing 도메인이 별도로 있고(8%), 그 안에서 가장 자주 다뤄지는 것이 TopologyTestDriver입니다. kafka-streams-test-utils 아티팩트를 테스트 의존성으로 추가하면 씁니다.

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams-test-utils</artifactId>
    <version>4.3.0</version>
    <scope>test</scope>
</dependency>

TopologyTestDriver는 라이브러리 런타임을 시뮬레이션합니다 — 입력 토픽에서 레코드를 계속 가져와 토폴로지를 순회하며 처리하는 동작을 대신하고, 결과 레코드를 캡처하며 내장 상태 저장소를 조회할 수 있게 해 줍니다. 브로커가 필요하지 않습니다.

import java.time.Duration;
import java.util.Properties;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.state.KeyValueStore;
import org.junit.jupiter.api.*;
import static org.junit.jupiter.api.Assertions.*;

class WordCountTest {

    private TopologyTestDriver testDriver;
    private TestInputTopic<String, String> input;
    private TestOutputTopic<String, Long> output;

    @BeforeEach
    void setUp() {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-test");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092"); // 실제로 접속하지 않습니다
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        testDriver = new TopologyTestDriver(WordCount.buildTopology(), props);

        input  = testDriver.createInputTopic("text-lines",
                     Serdes.String().serializer(), Serdes.String().serializer());
        output = testDriver.createOutputTopic("word-counts",
                     Serdes.String().deserializer(), Serdes.Long().deserializer());
    }

    @AfterEach
    void tearDown() {
        // 반드시 닫아야 자원이 정리됩니다
        testDriver.close();
    }

    @Test
    void countsWordsPerKey() {
        input.pipeInput("k", "kafka streams kafka");

        assertEquals(new KeyValue<>("kafka", 1L), output.readKeyValue());
        assertEquals(new KeyValue<>("streams", 1L), output.readKeyValue());
        assertEquals(new KeyValue<>("kafka", 2L), output.readKeyValue());
        assertTrue(output.isEmpty());
    }

    @Test
    void stateStoreHoldsLatestCount() {
        input.pipeInput("k", "kafka kafka kafka");

        KeyValueStore<String, Long> store = testDriver.getKeyValueStore("counts-store");
        assertEquals(3L, store.get("kafka"));
    }

    @Test
    void wallClockPunctuationCanBeTriggered() {
        // wall-clock time 은 드라이버가 모킹하므로 직접 전진시킵니다
        testDriver.advanceWallClockTime(Duration.ofSeconds(20));
    }
}

Processor API로 프로세서를 직접 작성했다면 MockProcessorContext로 단위 테스트합니다. 프로세서는 결과를 반환하지 않고 컨텍스트로 forward하므로, forward된 데이터를 캡처하는 목 컨텍스트가 필요합니다. context.forwarded()로 검증하고, context.committed()로 커밋 호출 여부를 확인합니다. 다만 목 컨텍스트는 punctuator를 자동 실행하지 않습니다 — 캡처만 하므로 직접 호출해야 하고, 자동 발동까지 테스트하려면 TopologyTestDriver를 쓰라고 문서가 권합니다. 또한 목 컨텍스트는 changelog를 관리하지 않으므로 상태 저장소는 인메모리 + logging disabled로 등록합니다.

어떤 상황에 무엇을 쓰는가

ksqlDB — Streams 위의 SQL 계층

ksqlDB의 가치는 Java 애플리케이션을 배포하지 않고 SQL 문장으로 스트림 처리를 정의한다는 데 있습니다. Streams의 KStream이 STREAM, KTable이 TABLE에 대응합니다.

-- 기존 토픽을 STREAM 으로 선언 (KStream 에 대응: 각 레코드가 독립 사건)
CREATE STREAM orders (
  order_id   BIGINT KEY,
  user_id    VARCHAR,
  amount     DOUBLE
) WITH (
  KAFKA_TOPIC  = 'orders',
  VALUE_FORMAT = 'AVRO'
);

-- 기존 토픽을 TABLE 로 선언 (KTable 에 대응: 키별 최신 상태)
CREATE TABLE users (
  user_id    VARCHAR PRIMARY KEY,
  country    VARCHAR,
  tier       VARCHAR
) WITH (
  KAFKA_TOPIC  = 'users',
  VALUE_FORMAT = 'AVRO'
);

-- 지속 질의(persistent query): 결과가 새 토픽으로 계속 흘러갑니다
CREATE STREAM high_value_orders WITH (
  KAFKA_TOPIC  = 'high_value_orders',
  VALUE_FORMAT = 'AVRO'
) AS
  SELECT o.order_id, o.amount, u.country, u.tier
  FROM orders o
  JOIN users u ON o.user_id = u.user_id
  WHERE o.amount > 1000
  EMIT CHANGES;

무엇을 모니터링해야 하는가

흔한 오해

시험 포인트 정리

확인 문제

네 가지 문항 유형(단일 선택 · 복수 선택 · 연결형 · 순서 배열)이 섞여 있습니다. 리파티션 유발 여부와 co-partitioning 요건은 반드시 맞히고 넘어가세요.

이어서 볼 곳

• 11장 · 운영 기초 보안·모니터링·성능 튜닝과 Share Groups. CCAAK 중심
• 예제 9 · Streams 실시간 집계 윈도우 집계와 상태 저장소를 실제로 돌려 봅니다. 실무 예제
• Streams 치트시트 DSL 연산자 표(리파티션 유발 여부), 윈도우·조인 매트릭스, 설정 15개. 빠른참조
• CCDAK · Kafka Streams 도메인 12% 도메인의 압축 정리와 자주 나오는 함정. 자격증
• CCDAK · Application Testing 도메인 8% 도메인. TopologyTestDriver와 목 객체 정리. 자격증
• 9장 · Kafka Connect SMT의 한계와 Streams로 넘어가야 하는 경계. CCDAK Kafka Connect

공식 문서 출처

이 장의 설정 기본값·API·의미론은 모두 아래에서 확인했습니다 (Apache Kafka 4.3 문서 기준).

• Streams Core Concepts — 토폴로지·source/sink processor, event/processing/ingestion time, stream time, 출력 타임스탬프 규칙, stream-table 이원성, 윈도우와 grace period, exactly-once v2, 순서 어긋남 처리
• Streams Architecture — 파티션·태스크 관계, 최대 병렬성, StreamsPartitionAssignor 교체 불가, changelog와 로그 컴팩션, standby replica, 랙 인식 배정
• Streams DSL — KStream/KTable/GlobalKTable 정의와 ("alice",1)→("alice",3) 예시, stateless/stateful 연산 표와 리파티션 표시 규칙, 조인 매트릭스, co-partitioning 요구사항, 윈도우 4종
• Streams Configs — num.stream.threads 1, num.standby.replicas 0, replication.factor -1, commit.interval.ms 30000(EOS 시 100), statestore.cache.max.bytes, max.task.idle.ms, topology.optimization
• Testing a Streams Application — kafka-streams-test-utils, TopologyTestDriver, TestInputTopic/TestOutputTopic, advanceWallClockTime, MockProcessorContext의 한계
• Streams Rebalance Protocol — KIP-1071, 지원/미지원 목록, 그룹 레벨 설정, NOT_READY 상태, 오프라인 마이그레이션 절차
• Interactive Queries — 읽기 전용 상태 저장소 조회
• Upgrading Apache Kafka — 4.1 Streams 리밸런스 프로토콜 Early Access, 4.2 production-ready, KAFKA-20254와 4.2.1 수정
• Kafka Streams Monitoring — failed-stream-threads, blocked-time-ns-total, record-e2e-latency-*의 recording level
• Managing Consumer Groups — kafka-consumer-groups.sh --describe 출력, kafka-groups.sh
• kafka-streams-test-utils (Apache Kafka 4.3 소스) — TopologyTestDriver·TestInputTopic·MockProcessorContext의 실제 패키지 경로