Streams 치트시트

// 1) 토폴로지를 문자열로 출력해 repartition 노드를 찾습니다
Topology topology = builder.build();
System.out.println(topology.describe());
// "Sink: KSTREAM-FILTER-...-repartition-filter" 처럼
// -repartition 이 들어간 노드가 있으면 리파티션이 생긴 것입니다

# 내부 토픽은 -- 규칙을 따릅니다
# (공식 문서는 이 규칙이 향후 릴리스에서 보장되지 않는다고 명시합니다)
bin/kafka-topics.sh --bootstrap-server localhost:9092 --list \
  | grep -E "^my-streams-app-" | sort

# 출력 예
# my-streams-app-KSTREAM-AGGREGATE-STATE-STORE-0000000003-changelog
# my-streams-app-order-count-repartition

import java.time.Duration;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.SlidingWindows;
import org.apache.kafka.streams.kstream.SessionWindows;
import org.apache.kafka.streams.kstream.JoinWindows;

// Tumbling — 크기 5분. advance 가 size 와 같은 hopping 이라고 볼 수 있습니다.
TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5));

// Hopping — 크기 5분, 1분마다 전진. 한 레코드가 최대 5개 윈도에 들어갑니다.
TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5))
           .advanceBy(Duration.ofMinutes(1));

// Sliding (집계) — 타임스탬프 차이 10분, grace 30분
SlidingWindows.ofTimeDifferenceAndGrace(
    Duration.ofMinutes(10), Duration.ofMinutes(30));

// Sliding (조인) — KStream-KStream 조인은 JoinWindows 를 씁니다
JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(10));

// Session — 비활성 간격 5분
SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(5));

# 1) 양쪽 파티션 수 확인
bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic orders
bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic customers

// KStream 이 적은 쪽이면
KStream<String, Order> repartitioned =
    orders.repartition(Repartitioned.numberOfPartitions(24));

// KTable 이 적은 쪽이면 — toStream → repartition → toTable
KTable<String, Customer> repartitionedTable =
    customers.toStream()
             .repartition(Repartitioned.numberOfPartitions(24))
             .toTable();

// stream-table 조인이라면 KStream 쪽을 리파티션하는 것이 권장됩니다.
// KTable 을 리파티션하면 상태 저장소가 하나 더 생길 수 있습니다.

export BS=localhost:9092

# Streams 그룹 목록
bin/kafka-streams-groups.sh --bootstrap-server $BS --list

# 내부 토픽 포함 전체 확인
bin/kafka-topics.sh --bootstrap-server $BS --list | grep "^my-streams-app-"

# lag 확인 — Streams 도 결국 컨슈머 그룹입니다 (application.id = group.id)
bin/kafka-consumer-groups.sh --bootstrap-server $BS --describe --group my-streams-app

# 앱 완전 초기화:
#   1) 애플리케이션 인스턴스를 모두 내립니다
#   2) 아래 도구로 오프셋과 내부 토픽을 정리합니다
bin/kafka-streams-application-reset.sh --bootstrap-server $BS \
  --application-id my-streams-app \
  --input-topics orders

#   3) 로컬 상태 디렉터리도 직접 지워야 합니다 (도구가 지우지 않습니다)
rm -rf /var/lib/kafka-streams/my-streams-app