Kafka Streams 실시간 집계

clicks-aggregator/
├── pom.xml
└── src/main/java/com/example/streams/
    ├── ClickCountTopology.java   # 토폴로지 정의 (테스트 가능하게 분리)
    ├── ClickTimestampExtractor.java # 이벤트 시각 추출
    └── Main.java                 # 설정 + 기동 + 종료

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>com.example</groupId>
  <artifactId>clicks-aggregator</artifactId>
  <version>1.0.0</version>

  <properties>
    <maven.compiler.release>17</maven.compiler.release>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <kafka.version>4.3.1</kafka.version>
    <slf4j.version>2.0.17</slf4j.version>
  </properties>

  <dependencies>
    <!-- kafka-streams 는 kafka-clients 를 전이 의존성으로 가져옵니다. -->
    <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>kafka-streams</artifactId>
      <version>${kafka.version}</version>
    </dependency>
    <dependency>
      <groupId>org.slf4j</groupId>
      <artifactId>slf4j-simple</artifactId>
      <version>${slf4j.version}</version>
    </dependency>
    <!-- TopologyTestDriver: 브로커 없이 토폴로지를 단위 테스트합니다. -->
    <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>kafka-streams-test-utils</artifactId>
      <version>${kafka.version}</version>
      <scope>test</scope>
    </dependency>
  </dependencies>

  <build>
    <plugins>
      <plugin>
        <groupId>org.codehaus.mojo</groupId>
        <artifactId>exec-maven-plugin</artifactId>
        <version>3.5.0</version>
        <configuration>
          <mainClass>com.example.streams.Main</mainClass>
        </configuration>
      </plugin>
    </plugins>
  </build>
</project>

package com.example.streams;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.streams.processor.TimestampExtractor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * 이벤트 페이로드의 clickedAt 필드를 이벤트 시각으로 씁니다.
 *
 * 기본 추출기는 레코드의 Kafka 타임스탬프를 쓰는데,
 * 그것은 "프로듀서가 발행한 시각"(CreateTime) 또는
 * "브로커가 기록한 시각"(LogAppendTime)입니다.
 * 모바일 오프라인 큐처럼 발행이 늦어지는 경우
 * 실제 클릭 시각과 크게 벌어지므로 페이로드에서 직접 꺼냅니다.
 */
public class ClickTimestampExtractor implements TimestampExtractor {

    private static final Logger log = LoggerFactory.getLogger(ClickTimestampExtractor.class);
    private static final Pattern CLICKED_AT = Pattern.compile("\"clickedAt\"\\s*:\\s*(\\d+)");

    @Override
    public long extract(ConsumerRecord<Object, Object> record, long partitionTime) {
        Object value = record.value();
        if (value instanceof String json) {
            Matcher m = CLICKED_AT.matcher(json);
            if (m.find()) {
                return Long.parseLong(m.group(1));
            }
        }

        // 추출 실패 시 음수를 반환하면 Streams 가 그 레코드를 버립니다.
        // 여기서는 파티션의 현재 시각(스트림 시각)으로 대체해 유실을 막습니다.
        // 어느 쪽이 맞는지는 도메인 판단입니다 — 조용히 버리면 원인을 못 찾습니다.
        log.warn("clickedAt 추출 실패 — partitionTime 으로 대체합니다. topic={} offset={}",
                record.topic(), record.offset());
        return partitionTime >= 0 ? partitionTime : record.timestamp();
    }
}

package com.example.streams;

import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.Grouped;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Produced;
import org.apache.kafka.streams.kstream.Suppressed;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.state.Stores;
import org.apache.kafka.streams.state.WindowBytesStoreSupplier;

import java.time.Duration;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * 상품별 1분 클릭 수를 집계합니다.
 *
 * 토폴로지를 Main 에서 분리해 두면 TopologyTestDriver 로
 * 브로커 없이 단위 테스트할 수 있습니다.
 */
public final class ClickCountTopology {

    public static final String STORE_NAME = "clicks-per-minute-store";

    private static final Pattern PRODUCT_ID = Pattern.compile("\"productId\"\\s*:\\s*\"([^\"]+)\"");

    /** 윈도우 크기 — 1분 tumbling */
    private static final Duration WINDOW_SIZE = Duration.ofMinutes(1);
    /**
     * grace period — 윈도우가 끝난 뒤에도 이만큼은 늦은 이벤트를 받아 줍니다.
     * 시나리오의 "최대 3분 늦게 오는 모바일 이벤트" 를 근거로 3분을 줍니다.
     * 이 값을 늘리면 정확도가 오르지만 결과 확정이 그만큼 늦어지고
     * 상태 저장소가 커집니다. 정확도와 지연의 트레이드오프입니다.
     */
    private static final Duration GRACE = Duration.ofMinutes(3);

    private ClickCountTopology() {
    }

    public static Topology build(String inputTopic, String outputTopic) {
        StreamsBuilder builder = new StreamsBuilder();

        Serde<String> stringSerde = Serdes.String();
        Serde<Long> longSerde = Serdes.Long();

        KStream<String, String> clicks = builder.stream(
                inputTopic,
                Consumed.with(stringSerde, stringSerde)
                        // 이벤트 시각을 페이로드에서 꺼냅니다.
                        .withTimestampExtractor(new ClickTimestampExtractor()));

        // 상태 저장소를 명시적으로 만듭니다.
        // 윈도우 저장소의 보관 기간은 최소한 (윈도우 크기 + grace) 이상이어야 합니다.
        // 짧으면 grace 안에 도착한 이벤트를 받을 저장소가 이미 없어집니다.
        WindowBytesStoreSupplier storeSupplier = Stores.persistentWindowStore(
                STORE_NAME,
                WINDOW_SIZE.plus(GRACE).plus(Duration.ofMinutes(1)),  // retention
                WINDOW_SIZE,                                          // windowSize
                false);                                               // retainDuplicates

        clicks
            // 키를 productId 로 다시 잡습니다.
            // selectKey 는 "리파티션을 유발" 합니다 — 키가 바뀌면 파티션이 바뀌므로
            // Streams 가 내부 리파티션 토픽을 자동으로 만듭니다.
            // (mapValues 처럼 값만 바꾸는 연산은 리파티션을 유발하지 않습니다)
            .selectKey((key, value) -> extractProductId(value))
            // 키가 null 인 레코드는 집계할 수 없으므로 걸러 냅니다.
            // 걸러 내지 않으면 NullPointerException 으로 스트림 스레드가 죽습니다.
            .filter((productId, value) -> productId != null)
            .groupByKey(Grouped.with("clicks-by-product", stringSerde, stringSerde))
            // ofSizeAndGrace 가 4.x 의 표준 API 입니다.
            // (구버전의 TimeWindows.of(...) + until(...) 조합은 제거되었습니다)
            .windowedBy(TimeWindows.ofSizeAndGrace(WINDOW_SIZE, GRACE))
            .count(Materialized.<String, Long>as(storeSupplier)
                    .withKeySerde(stringSerde)
                    .withValueSerde(longSerde))
            // suppress: 윈도우가 닫힐 때까지 중간 결과를 내보내지 않습니다.
            // 이것이 없으면 이벤트가 들어올 때마다 갱신된 카운트가 계속 발행되어
            // 하류 대시보드가 1, 2, 3, 4 ... 를 모두 보게 됩니다.
            // untilWindowCloses 는 StrictBufferConfig 만 받습니다 —
            // 버퍼가 넘칠 때 결과를 조기 방출하면 "윈도우당 한 번" 보장이 깨지기 때문입니다.
            .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()))
            .toStream()
            // Windowed<String> 키를 "productId@윈도우시작시각" 문자열로 펼칩니다.
            // 하류가 일반 컨슈머라면 윈도우 Serde 를 알 필요가 없어집니다.
            .map((Windowed<String> windowedKey, Long count) -> KeyValue.pair(
                    windowedKey.key(),
                    "{\"productId\":\"%s\",\"windowStart\":%d,\"windowEnd\":%d,\"clicks\":%d}"
                            .formatted(windowedKey.key(),
                                    windowedKey.window().start(),
                                    windowedKey.window().end(),
                                    count)))
            .to(outputTopic, Produced.with(stringSerde, stringSerde));

        return builder.build();
    }

    private static String extractProductId(String json) {
        if (json == null) {
            return null;
        }
        Matcher m = PRODUCT_ID.matcher(json);
        return m.find() ? m.group(1) : null;
    }
}

package com.example.streams;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;

public final class Main {

    private static final Logger log = LoggerFactory.getLogger(Main.class);

    private static final String BOOTSTRAP =
            "localhost:29092,localhost:39092,localhost:49092";
    private static final String INPUT = "clickstream";
    private static final String OUTPUT = "clicks-per-minute";

    public static void main(String[] args) {
        boolean eos = args.length > 0 && "eos".equals(args[0]);

        Properties p = new Properties();

        // application.id 는 이 Streams 애플리케이션의 정체성입니다.
        //   - 컨슈머 그룹 ID 가 됩니다.
        //   - 내부 토픽(changelog, repartition) 이름의 접두어가 됩니다.
        //   - 로컬 상태 디렉터리 경로에 들어갑니다.
        // 바꾸면 완전히 새 애플리케이션이 되어 상태를 처음부터 다시 만듭니다.
        p.put(StreamsConfig.APPLICATION_ID_CONFIG, "clicks-aggregator");
        p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP);

        p.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        p.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        // 한 인스턴스가 돌릴 스트림 스레드 수. 기본값은 1 입니다.
        // 태스크 수(=입력 파티션 수 6)를 넘기면 남는 스레드는 유휴 상태가 됩니다.
        p.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 3);

        // 내부 토픽(changelog, repartition)의 복제 계수.
        // 기본값은 -1 로 "브로커 기본값을 따름" 을 의미하며,
        // 브로커 default.replication.factor 가 1 이면 상태가 단일 사본이 됩니다.
        // 반드시 명시하세요.
        p.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 3);

        // 로컬 상태 디렉터리. 컨테이너에서는 반드시 영속 볼륨을 마운트해야 합니다.
        // 휘발성이면 재시작마다 changelog 전체를 재생해 복구가 오래 걸립니다.
        p.put(StreamsConfig.STATE_DIR_CONFIG, "/tmp/kafka-streams");

        // standby replica: 다른 인스턴스가 상태 사본을 미리 따라 만들어 둡니다.
        // 기본값 0. 1 이상으로 두면 인스턴스 장애 시 복구가 훨씬 빠릅니다.
        // 대가는 changelog 재생 트래픽과 디스크입니다.
        p.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);

        if (eos) {
            // processing.guarantee 의 기본값은 at_least_once 이고,
            // 유효값은 at_least_once / exactly_once_v2 두 개뿐입니다
            // (exactly_once 와 exactly_once_beta 는 4.0 에서 제거되었습니다).
            //
            // 켜면 commit.interval.ms 기본값이 30000 -> 100 으로 바뀝니다.
            // 커밋이 잦아져 처리량이 떨어지고 브로커 요청이 늘어나는 대신,
            // 입력 오프셋과 출력·상태 갱신이 원자적으로 커밋됩니다.
            p.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
            log.info("processing.guarantee = exactly_once_v2");
        } else {
            p.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.AT_LEAST_ONCE);
            // at_least_once 의 기본 커밋 주기는 30000 입니다.
            // 대시보드 지연을 줄이려고 짧게 둡니다.
            p.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 5_000);
        }

        Topology topology = ClickCountTopology.build(INPUT, OUTPUT);
        // 토폴로지를 로그로 남기면 리파티션 토픽이 어디 생기는지 눈으로 확인할 수 있습니다.
        log.info("토폴로지:\n{}", topology.describe());

        KafkaStreams streams = new KafkaStreams(topology, p);

        // 스트림 스레드에서 잡히지 않은 예외가 났을 때의 동작을 정합니다.
        // 기본 동작은 스레드 종료이며, 모든 스레드가 죽으면 애플리케이션이 멈춥니다.
        streams.setUncaughtExceptionHandler((Throwable e) -> {
            log.error("스트림 스레드에서 처리되지 않은 예외", e);
            // REPLACE_THREAD        : 그 스레드만 교체하고 계속 처리합니다.
            // SHUTDOWN_CLIENT       : 이 인스턴스만 종료합니다.
            // SHUTDOWN_APPLICATION  : 같은 application.id 의 모든 인스턴스를 종료합니다.
            //
            // 데이터 오류로 인한 예외를 REPLACE_THREAD 로 두면
            // 같은 레코드에서 무한히 실패하는 루프가 됩니다.
            // 원인을 구분할 수 없다면 SHUTDOWN_CLIENT 가 안전합니다.
            return StreamsUncaughtExceptionHandler
                    .StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
        });

        streams.setStateListener((newState, oldState) ->
                log.info("상태 전이 {} → {}", oldState, newState));

        CountDownLatch latch = new CountDownLatch(1);
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            log.info("SIGTERM 수신 — Streams 를 정상 종료합니다");
            // close(Duration) 은 진행 중인 처리를 마치고 오프셋을 커밋한 뒤 닫습니다.
            // 타임아웃 없이 kill 되면 changelog 재생이 필요한 상태로 남습니다.
            streams.close(Duration.ofSeconds(60));
            latch.countDown();
        }, "streams-shutdown-hook"));

        try {
            streams.start();
            latch.await();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } catch (RuntimeException e) {
            log.error("Streams 기동 실패", e);
            System.exit(1);
        }
        log.info("정상 종료");
    }
}

# 0. 예제 1의 클러스터와 토픽
cd kafka-lab && docker compose ps
./kcli kafka-topics.sh --describe --topic clickstream | head -2

# 1. 빌드 후 기동 (at_least_once)
cd ../clicks-aggregator
mvn -q clean package
mvn -q exec:java

# 2. 다른 터미널에서 클릭 이벤트를 넣습니다.
#    clickedAt 을 "지금" 으로 넣어야 윈도우가 정상 동작합니다.
cd kafka-lab
NOW=$(date +%s000)
for i in $(seq 1 30); do
  PID="P-$((i % 3 + 1))"
  echo "$PID:{\"productId\":\"$PID\",\"userId\":\"U-$i\",\"clickedAt\":$NOW}"
done | ./kcli kafka-console-producer.sh --topic clickstream \
        --property parse.key=true --property key.separator=:

# 3. 결과를 봅니다. 윈도우가 닫히고 grace(3분)가 지난 뒤에 나옵니다.
./kcli kafka-console-consumer.sh --topic clicks-per-minute --from-beginning \
  --property print.key=true

./kcli kafka-topics.sh --list | grep clicks-aggregator

clicks-aggregator-clicks-by-product-repartition
clicks-aggregator-clicks-per-minute-store-changelog

./kcli kafka-topics.sh --describe \
  --topic clicks-aggregator-clicks-per-minute-store-changelog

Topic: clicks-aggregator-clicks-per-minute-store-changelog  PartitionCount: 6  ReplicationFactor: 3
	Configs: cleanup.policy=compact,delete,min.insync.replicas=2,retention.ms=...

P-1	{"productId":"P-1","windowStart":1785024000000,"windowEnd":1785024060000,"clicks":10}
P-2	{"productId":"P-2","windowStart":1785024000000,"windowEnd":1785024060000,"clicks":10}
P-3	{"productId":"P-3","windowStart":1785024000000,"windowEnd":1785024060000,"clicks":10}

LATE=$(( $(date +%s) * 1000 - 120000 ))
echo "P-1:{\"productId\":\"P-1\",\"userId\":\"U-LATE\",\"clickedAt\":$LATE}" \
  | ./kcli kafka-console-producer.sh --topic clickstream \
      --property parse.key=true --property key.separator=:

# grace 를 넘는 5분 전 이벤트 — 버려집니다.
TOO_LATE=$(( $(date +%s) * 1000 - 300000 ))
echo "P-1:{\"productId\":\"P-1\",\"userId\":\"U-TOOLATE\",\"clickedAt\":$TOO_LATE}" \
  | ./kcli kafka-console-producer.sh --topic clickstream \
      --property parse.key=true --property key.separator=:

kafka.streams:type=stream-task-metrics,thread-id=*,task-id=*
  Attribute: dropped-records-total
  Attribute: dropped-records-rate

# 애플리케이션을 정상 종료한 뒤 로컬 상태를 지웁니다.
pkill -f 'com.example.streams.Main'
rm -rf /tmp/kafka-streams/clicks-aggregator

# 재시작하면 changelog 를 재생해 상태를 복원합니다.
cd ../clicks-aggregator && mvn -q exec:java

INFO com.example.streams.Main - 상태 전이 CREATED → REBALANCING
INFO o.a.k.s.p.internals.StoreChangelogReader - stream-thread [...] Restoration in progress for 6 partitions.
INFO o.a.k.s.p.internals.StoreChangelogReader - stream-thread [...] Finished restoring changelog clicks-aggregator-clicks-per-minute-store-changelog-0
INFO com.example.streams.Main - 상태 전이 REBALANCING → RUNNING

mvn -q exec:java -Dexec.args="eos"

INFO o.a.k.streams.StreamsConfig - StreamsConfig values:
	commit.interval.ms = 100
	processing.guarantee = exactly_once_v2
	...
INFO o.a.k.c.p.internals.TransactionManager - [Producer clientId=clicks-aggregator-...] ProducerId set to 7000 with epoch 0