2.3 Kafka Streams DSL 基础

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

KStream<String, Order> largeOrders = orders
    .filter((key, order) -> order.amount() > 1000);

// map：修改 key 和 value
KStream<String, String> simplified = orders
    .map((key, order) -> KeyValue.pair(order.userId(), order.symbol()));

// mapValues：只修改 value（更高效，不触发 repartition）
KStream<String, Double> amounts = orders
    .mapValues(Order::amount);

// 一条记录展开为多条（如订单含多个商品）
KStream<String, OrderItem> items = orders
    .flatMapValues(order -> order.items());

KTable<String, Order> latestOrder = orders
    .groupByKey()
    .reduce((oldOrder, newOrder) -> newOrder);  // 保留最新订单

KTable<String, Double> totalAmount = orders
    .groupByKey()
    .aggregate(
        () -> 0.0,                                    // 初始值
        (key, order, total) -> total + order.amount(), // 累加
        Materialized.<String, Double, KeyValueStore<Bytes, byte[]>>as("total-amount")
            .withValueSerde(Serdes.Double())
    );

KTable<String, Long> orderCount = orders
    .groupByKey()
    .count(Materialized.as("order-count-store"));

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

KStream<String, EnrichedOrder> enriched = orders
    .leftJoin(users, (order, user) ->
        new EnrichedOrder(order.id(), order.amount(), user.name()));

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

orders.join(payments,
    (order, payment) -> new CompletedOrder(order, payment),
    JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(10))
);

KTable<String, Double> balances = builder.table("balances");
KTable<String, Double> limits = builder.table("limits");

KTable<String, Double> available = balances
    .join(limits, (balance, limit) -> limit - balance);

Map<String, KStream<String, Order>> branches = orders
    .split(Named.as("orders-"))
    .branch((key, order) -> order.amount() > 10000, Branched.as("large"))
    .branch((key, order) -> order.amount() > 1000, Branched.as("medium"))
    .defaultBranch(Branched.as("small"));

branches.get("orders-large").to("large-orders");
branches.get("orders-medium").to("medium-orders");
branches.get("orders-small").to("small-orders");

标记 key 可能变化、后续按 key 操作时需要 repartition 的：
- map
- selectKey
- groupBy（带 key 映射）
- join / 其它按 key 的有状态操作（上游标记未消除时）

不会标记的：
- filter
- mapValues
- flatMapValues

// ❌ 修改 key：map 是 key-changing 操作，后续聚合要 repartition
stream.map((k, v) -> KeyValue.pair(v.userId(), v));

// ❌ selectKey 同样被标记为 key-changing，内部一样会 repartition
stream.selectKey((k, v) -> v.userId());

// ✅ 只改 value、不动 key：mapValues 不触发 repartition
stream.mapValues(this::normalize);

@Test
void testAggregate() {
    StreamsBuilder builder = new StreamsBuilder();
    builder.<String, Integer>stream("input")
        .groupByKey()
        .reduce(Integer::sum, Materialized.as("sum-store"))
        .toStream()
        .to("output");

    Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-app");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");
    // stream()/to()/reduce 都没有显式 Serde，默认值必须与 Integer 对齐
    props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
    props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.Integer().getClass());

    try (TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
        TestInputTopic<String, Integer> input = driver.createInputTopic(
            "input", new StringSerializer(), new IntegerSerializer());
        TestOutputTopic<String, Integer> output = driver.createOutputTopic(
            "output", new StringDeserializer(), new IntegerDeserializer());

        input.pipeInput("alice", 10);
        input.pipeInput("alice", 20);
        input.pipeInput("bob", 5);

        assertEquals(10, output.readValue());
        assertEquals(30, output.readValue());  // 10 + 20
        assertEquals(5, output.readValue());
    }
}