Kafka Streams Processor API — DSL 밑의 저수준 세계
Kafka Streams GlobalKTable 및 Processor API 공식 문서 종합 (웹 리서치, 2026) — Apache Kafka 공식 문서(streams/developer-guide/dsl-api, streams/developer-guide/processor-api) Apache Kafka 공식문서 Streams Processor API
이 모듈을 다 읽으면
- Topology를 addSource/addProcessor/addSink로 직접 구성하는 저수준 방식을 설명할 수 있다
- Processor 인터페이스와 ProcessorContext#forward()가 DSL 연산자 뒤에서 하는 일을 설명할 수 있다
- 언제 DSL 대신(또는 함께) Processor API를 쓰는지 설명할 수 있다
kafka-learn-37의 KStream/KTable DSL과 kafka-learn-55의 punctuate()가 실제로는 이 Processor API 위에 얹힌 것임을 전제로, Topology를 직접 조립하는 저수준 API의 구조와 DSL과의 관계를 정리한다.
Topology를 직접 조립한다 — addSource, addProcessor, addSink
DSL(KStream/KTable의 filter, map, join 같은 메서드 체이닝)이 내부적으로 만들어주는 Topology를, Processor API에서는 개발자가 직접 세 가지 노드로 조립한다. addSource는 Kafka 토픽을 입력으로 받는 소스 노드를 등록하고(키/값 디시리얼라이저를 지정할 수 있다), addProcessor는 실제 처리 로직을 담은 프로세서 노드를 하나 이상의 부모 노드에 연결해 등록하며, addSink는 그 결과를 Kafka 토픽으로 내보내는 싱크 노드를 등록한다. 즉 DSL이 '무엇을 할지'만 선언하면 되던 것을, Processor API에서는 '노드들을 어떻게 연결할지'까지 명시적으로 그려야 한다.
핵심 포인트
- addSource(토픽→소스노드)/addProcessor(처리로직 노드, 부모 노드 1개 이상과 연결)/addSink(처리결과→토픽) 세 가지로 Topology를 직접 조립
- DSL은 '무엇을 할지'만 선언하면 되지만 Processor API는 노드 간 연결 구조까지 명시적으로 구성해야 함
Processor 인터페이스와 context.forward() — 레코드를 다음 노드로 밀어보내기
커스텀 프로세서는 Processor 인터페이스를 구현해 만든다. init() 메서드는 태스크 구성 단계에서 호출되며 ProcessorContext 인스턴스를 넘겨받는데, 이 컨텍스트로 펑추에이션 함수를 예약(schedule())하거나, 새 레코드를 하위(downstream) 프로세서로 전달(forward())하거나, 커밋을 요청할 수 있다. process() 메서드는 받은 레코드마다 호출되는 실제 처리 로직이다 — DSL의 map()이나 filter() 같은 연산자들도 결국 내부적으로는 process() 안에서 조건에 맞으면 context.forward()를 호출해 다음 노드로 레코드를 넘기고, 안 맞으면 아무것도 안 하는 식으로 구현돼 있다.
핵심 포인트
- init(ProcessorContext)에서 schedule()(펑추에이션 예약)·forward()(하위 전달)·commit 요청 등 컨텍스트 기능에 접근
- process()가 레코드마다 호출되는 실제 로직 — DSL의 map/filter 등도 내부적으로는 process() 안에서 context.forward() 호출 여부로 구현된 것
State Store 연결과 DSL과의 공존
프로세서에 상태 저장소를 붙이는 방법은 두 가지다 — 기존 방식은 별도로 만든 StoreBuilder(예: countStoreBuilder)를 Topology.connectProcessorAndStateStore()로 특정 프로세서 노드에 연결하는 것이고, 더 캡슐화된 방식은 ProcessorSupplier가 ConnectedStoreProvider#stores()를 구현해 '이 프로세서가 필요로 하는 스토어'를 스스로 제공하게 하는 것이다. 그리고 Processor API는 DSL과 배타적이지 않다 — KStream.process()나 transform() 같은 메서드로 DSL 체인 중간에 커스텀 Processor를 끼워 넣을 수 있어, DSL의 편의성과 Processor API의 유연성(복잡한 상태 관리, DSL에 없는 커스텀 로직)을 함께 쓸 수 있다.
핵심 포인트
- state store 연결: StoreBuilder+connectProcessorAndStateStore() 전통 방식, 또는 ProcessorSupplier가 ConnectedStoreProvider#stores()로 스스로 제공하는 캡슐화 방식
- KStream.process()/transform()으로 DSL 체인 중간에 커스텀 Processor를 끼워 넣어 DSL 편의성+Processor API 유연성을 함께 사용 가능