原理：状态、再平衡、窗口、Join、EOS 怎么工作

2.1状态存储：RocksDB 是缓存，changelog 是真相

有状态算子把状态写进本地 RocksDB，同时每次 put 追加进一条 compacted changelog topic——故障转移靠回放这条日志重建。

一个有状态算子（aggregate / count / reduce / 窗口 / join）需要一块跨记录存活的内存。Streams 默认给它一个嵌入式 RocksDB 实例：一个 LSM-tree 的本地键值库，落盘到 task 的状态目录，所以进程重启后目录还在、状态不丢。RocksDB 前面再挂一个小的写回缓存（record cache），把同一 key 的连续更新合并后再下沉，减少写放大。但本地盘会随实例一起消失——容器被调度走、磁盘坏掉、task 被再平衡搬到别的实例，本地 RocksDB 就没了。所以每次 put 还会同步追加一条记录到一个内部 changelog topic（命名 <application.id>-<store-name>-changelog，配置为 log-compacted——每个 key 只保留最新值 + tombstone）。这条日志才是状态的真相源：RocksDB 丢了可以从零回放 changelog 重建，changelog 丢了 RocksDB 就成了孤儿。这正是流表对偶落到工程上的样子——table 是 changelog 的物化视图，所以 table 永远可以从日志重新长出来。

故障转移的代价由状态大小决定。一个多 GB 的状态从零回放 changelog 会阻塞启动数分钟——这是 Streams 运维里最常被点名的痛点（04 章专门讲它）。缓解手段是 standby 副本（num.standby.replicas）：另一个实例后台尾随同一条 changelog，把状态"保温"在本地 RocksDB 里；owner 挂掉时再平衡把 task 优先分给持有 standby 的实例，它只需追上最后一小段日志，恢复从分钟降到秒。Kafka 4.3 的 KIP-1035 进一步让 StateStore 自己管理 changelog 的已恢复 offset，重启后能跳过已重放的部分，恢复更快。

// 一个有状态 count：状态进本地 RocksDB，并由内部 changelog 撑腰
KStream<String, String> lines = builder.stream("text-lines");
KTable<String, Long> counts = lines
    .flatMapValues(v -> Arrays.asList(v.toLowerCase().split("\\W+")))
    .groupBy((key, word) -> word)            // 改 key → 触发重分区（见 §2.3）
    .count(Materialized.<String, Long>as(    // 物化为一个具名状态存储
        Stores.persistentKeyValueStore("word-counts")));
// 这块状态对应的内部 topic：
//   <application.id>-word-counts-changelog   （compacted，真相源）

带来的代价

① 恢复阻塞：多 GB 状态从零回放 changelog 会阻塞启动数分钟，无 standby 时每次再平衡都会触发。② 写量翻倍：每次 put 既写本地 RocksDB 又写一条 changelog 记录，broker 端多承一份写入和存储。③ 堆外内存：RocksDB 的 block cache 与 memtable 分配在 JVM 堆之外，-Xmx 管不住——容器内存按"堆 + RocksDB 缓存 + 余量"算，否则被 OOMKilled（04 章资源陷阱）。④ 版本化状态存储（KIP-889, 3.5 GA）给每个 key 存多版本 (value, timestamp) 以支持 get(key, asOf) 和正确的乱序处理，代价是更高的存储与写放大。

2.2再平衡与缩放：消费组那套，外加搬状态的代价

缩放就是消费组再平衡——task 在实例和线程间重新分配，但有状态 task 被搬走时要在新落点重建状态。

Streams 不自造调度器。每个 JVM 实例起 num.stream.threads 个 StreamThread，每个线程内嵌一个普通 consumer，全部以 application.id 作为 group.id 加入同一个消费组。task = 子拓扑输入 topic 的每个分区一个，所以 task 总数被最大输入分区数钉死、运行时不可改。加一个实例、某实例掉线、某线程死掉，都触发一次再平衡，把所有 task 在当前所有实例的所有线程间重新摊分。无状态 task 搬家几乎零成本——换个线程接着消费就行。有状态 task 不同：它在新落点没有本地 RocksDB，必须先回放 changelog 重建状态（或从 standby 追尾）才能处理，这段时间该分区的处理停滞。所以"缩放"在有状态拓扑里从来不是瞬时的——这是机制的必然代价，不是故障。

分配逻辑在哪算，是 4.x 的关键变化。经典协议（classic）由其中一个 client 当 group leader、在客户端计算 task 分配；KIP-1071 Streams 再平衡协议把分配移到 broker 端的 group coordinator，4.2 起新集群默认开（需 broker + client 都 ≥ 4.2，feature streams.version=1）。broker 端分配能拿到全局视图做更稳的 task 放置与 standby 调度，但仍不支持 static membership、运行中在线把 classic 协议迁到 streams 协议、以及运行中修改 topology。

带来的代价

① 有状态搬迁成本：每次再平衡，被重分配的有状态 task 都要重建状态——无 standby 时这就是分钟级停顿。② 并行上限 = 最大输入分区数：num.stream.threads 把线程提到 task 数能用满单实例，再加实例才有意义；线程/实例超过 task 数则纯闲置，要更多并行只能重分区输入 topic（破坏性操作）。③ 无 static membership（KIP-1071 下）：滚动重启 / pod 漂移更容易触发全量再平衡，靠 standby 副本和持久卷缓解，而非靠 group.instance.id 跳过再平衡。

2.3重分区：改 key 就给下游埋了一次往返

改 key 的算子置一个"需重分区"标志，下游第一个 stateful 算子自动建一条 repartition topic 把数据按新 key 重洗。

聚合和等值 join 要求"相同 key 的记录落在同一个 task/分区"（共分区）。但改 key 的操作——selectKey / map / flatMap / groupBy——会让记录的新 key 与它当前所在分区不再一致。Streams 的处理方式：这些算子只是置一个"需重分区"标志，自己不立即重洗；当下游出现第一个依赖 key 的 stateful 算子（aggregate / count / join）时，Streams 在它前面自动插入一条内部 <application.id>-<name>-repartition topic：上游把改了 key 的记录 produce 进这条 topic（按新 key 分区），下游再把它重新消费进来——此时相同新 key 必然同分区，聚合/join 才正确。selectKey 是 lazy 的：单独用它（后面没有依赖 key 的算子）不会建任何 topic，只有真正需要共分区的下游才把重洗物化出来。这是"默认正确"的设计——用户不必手动保证共分区，代价是这条往返被藏在拓扑里。

带来的代价

① 一次完整往返：每个被物化的重分区点，每条记录都要 produce 进 repartition topic 再 re-consume 回来——多一份网络、broker 磁盘、序列化/反序列化和端到端延迟。② 隐藏 topic 成本：这些 topic 不在你的代码里，但占 broker 存储、计入分区配额，排查时容易遗漏。③ 能省则省：不改 key 的操作（mapValues / filter）和 groupByKey（不改 key）不触发重分区；优先它们而非 map / groupBy。用 topology.describe() 打印拓扑、数清实际建了几条 repartition topic（04 章）。

2.4窗口与时间：stream-time 由数据推进，不是墙上时钟

窗口按事件时间切分，由 stream-time（每 task 见过的最大时间戳）驱动关闭——它只在有记录到达时前进。

窗口把无界流切成有界的桶来聚合，四种形状：tumbling 滚动（固定大小、不重叠，一条记录进一个窗）、hopping 跳跃（固定大小 + 更小步进 → 窗口重叠 → 一条记录进多个窗）、sliding 滑动（按记录两两之间的时间差成对落窗）、session 会话（按活动间隙合并，间隙超阈值就开新会话）。关键不在形状，而在驱动它们关闭的时钟。Streams 默认用 event-time（记录自带的时间戳，由 TimestampExtractor 提取），不是 processing-time（记录被处理的墙上时间），也不是 ingestion-time（写入 broker 的时间）。窗口由 stream-time 推进：stream-time = 这个 task 到目前为止见过的最大记录时间戳，单调不减。它只在有记录到达时前进——没有新数据，stream-time 就冻结，依赖它关闭的窗口永远不发结果。这是 event-time 换来可重放确定性（同样的输入重跑得到同样的窗口结果，与运行时刻无关）所付的代价。

迟到事件由 宽限期 grace 控制：窗口在 windowEnd 逻辑上结束后，还允许 stream-time 推进到 windowEnd + grace 之前到达的迟到记录更新该窗结果；一旦 stream-time 越过 windowEnd + grace，更晚到的记录被静默丢弃（只计入 dropped-records 指标，不报错、不进死信）。状态存储的 retention 必须 ≥ 窗口大小 + grace，否则窗口还没关、状态先被清掉。若要"每个窗口只发一个最终结果"而非每次更新都发一条中间结果，用 suppress(untilWindowCloses)（KIP-328）把中间更新压住、只在窗口关闭时发一次。

带来的代价

① 空闲分区冻结：stream-time 只随记录前进，低流量或空闲分区让窗口迟迟不关、结果延迟甚至永不发出——event-time 的根本代价（04 章）。② suppress 的内存压力：suppress 把未关窗口的中间状态缓在内存，缓冲满时行为取决于配置——emitEarlyWhenFull 会在内存压力下提前发出非最终结果（破坏"只发最终"的承诺），要严格"最终"必须用 shutDownWhenFull（宁可崩也不发中间结果）。③ retention 约束：窗口状态与 changelog 的 retention 必须 ≥ 窗口 + grace，配小了窗口没关状态先没。

2.5Join：四种语义，各自的共分区要求与逃生口

四种 join 对应四种数据形态，等值 join 要求两侧共分区，GlobalKTable 与外键 join 是免共分区的逃生口。

join 把两条流/表按 key 关联，语义由两侧是流还是表决定。KStream-KStream=窗口化 join：两侧记录都入状态存储缓冲，在窗口 + grace 内每出现一对匹配就发一条——因为两条都是无界流，必须给一个时间窗口才能界定"什么算同时发生"。KStream-KTable=查表 join：流记录到达时探查表的当前值，表自己的更新不回头重新触发已处理的流记录——典型用于流数据打维表标签。KTable-KTable=维护式 join：任一侧更新都重新计算并发出新结果，维护一个始终最新的连接视图。外键 join（KIP-213, 2.4）：左表按 value 里提取的外键关联右表主键，Streams 在背后建两个隐藏 topic（subscription + response）加一个复合 RocksDB 存储来重新按外键路由——这是唯一不要求左侧用主键 join 的表-表 join。

共分区要求是等值 join 的硬约束：两侧必须相同分区数 + 相同分区方式（同一个 producer 分区器、同样的 key），相同 key 才会落到同一个 task。分区数不一致，Streams 在启动时直接校验失败；分区方式不一致（数量碰巧相同但分区器不同），Streams 不报错却静默丢匹配——这是最难排查的 join 故障。Streams 会对改过 key 的一侧自动重分区来对齐，但两个独立来源 topic 的分区方式得你自己保证。GlobalKTable 是逃生口：它在每个实例全量加载所有分区，所以任何 key 本地都能查到，免共分区，但代价是每实例一份全量副本（磁盘 + 内存 + 启动 bootstrap 延迟），只配小而慢变的维表；外键 join 同样免共分区（它自己用隐藏 topic 重路由）。

带来的代价

① 静默无输出：未共分区时（分区方式不一致）零输出且无任何报错——join "不工作"却查不出原因，必须核对两侧分区数 + 分区器（04 章）。② 外键 join 的内部 topic：FK join 额外建 subscription + response 两条内部 topic 加复合存储，运维面与存储成本随之上升。③ KStream-KStream 缓冲整窗：两侧记录在窗口 + grace 内都要入状态存储，宽窗 + 高流量 = 大状态、大 changelog。④ GlobalKTable 全量代价：每实例一份完整副本，维表一大就磁盘/内存爆炸、启动 bootstrap 拖慢。

2.6EOS v2：一个事务包住 offset、状态、输出

exactly_once_v2 把 offset 提交、状态 changelog 写、输出 produce 绑进一个 Kafka 事务，下游 read_committed 只读已提交。

read-process-write 循环里有三件事必须同生共死，否则就重复或丢数：消费位移的提交、状态的 changelog 写、结果的输出 produce。任意两件之间崩溃都会破坏 exactly-once——比如输出发了但 offset 没提交，重启后重新处理同一批，输出就重复。processing.guarantee=exactly_once_v2 用一个 Kafka 事务把这三者原子化：要么三件一起提交、要么一起回滚。下游消费者设 isolation.level=read_committed 就只读已提交的记录、自动跳过被回滚（aborted）的部分。v2 的关键改进（KIP-447）是一个 StreamThread 一个 producer 覆盖该线程的所有 task；v1（exactly_once，已淘汰）是一个 task 一个 producer，高分区数下 producer 数量爆炸、内存与 broker 端事务协调开销失控。这就是为什么 v2 是 3.0 起的默认、v1 被弃用。

# 生产者侧（Streams 应用）
processing.guarantee=exactly_once_v2
# EOS 下 Streams 把提交间隔默认改成 100ms（非 EOS 是 30000ms）——
# 这才是"开了 EOS 就变慢"的真因，不是事务本身的固定开销
commit.interval.ms=100
# EOS 需要事务，至少 3 个 broker（事务状态 topic 副本因子默认 3）

# 消费者侧（下游应用）：不设这个，EOS 形同虚设
isolation.level=read_committed

带来的代价

① 延迟与吞吐：EOS 把 commit.interval.ms 默认从 30000ms 改到 100ms——这才是"开了 EOS 就变慢"的真因（更频繁提交 + read_committed 消费者 15–30% 吞吐下降），而非事务本身有多重。调大它换吞吐、调小换延迟（04 章）。② 边界止于 Kafka：事务只在单个 Kafka 集群内原子，不覆盖外部副作用——往 DB / HTTP / 发短信，事务回滚也撤不回，仍会重复。要端到端一致得上 outbox 模式（把外部写也变成 Kafka 写，由另一个消费者按幂等落地）。③ 部署门槛：需要至少 3 个 broker（事务状态 topic 副本因子默认 3），下游必须配 read_committed 否则 EOS 形同虚设。这与主教程的投递语义一脉相承——Streams 的 EOS 是消费组 + 事务在流处理上的封装，不是新机制。

2.7串起来：一条记录在有状态聚合里的完整一生

六个机制不是孤立的。把它们串在一个具体场景上看协同：一条订单事件进入一个 EOS v2 的有状态聚合拓扑（按用户 ID 累计消费额），从进入到被下游读到，依次穿过状态、重分区、提交、投递四道关。

这条链解释了一个常被误解的现象：为什么"开了 EOS 的有状态聚合"在崩溃恢复后既不丢数也不重复算。不是因为某个魔法标志，而是因为状态的真相在 changelog、changelog 的写又被事务和 offset 提交绑在一起——恢复时回放到的状态，和已提交给下游的输出，永远对齐在同一个事务边界上。