陷阱与失败模式

# 错误：默认 0 个 standby，任何故障转移都从零回放 changelog
# num.standby.replicas=0

# 正确：至少 1 个热备，故障转移从"分钟回放"降到"秒级追尾"
num.standby.replicas=1

# rack-aware standby（KIP-925，3.2+）：把 standby 摆到另一可用区，
# 同时挡住"主实例挂 + 同机架 standby 一起挂"
rack.aware.assignment.tags=zone
client.tag.zone=az-a

# 提高恢复并行度：恢复用独立线程拉 changelog，不占处理线程
# （默认即开，确认未被调低）

# 现状：6 分区输入 → 最多 6 个 task

# 错误：单实例把线程开到 16，超过 6 的 10 个线程全闲置
# num.stream.threads=16

# 正确：先把每实例线程提到接近 task 数，再横向加实例直到 task 摊满
# 例如 2 实例 × 3 线程 = 6 个 slot，正好 1 task/slot
num.stream.threads=3
# 然后部署 2 个实例

# 要超过 6 路并行：只能重分区输入 topic（破坏性，需停机/双写迁移）
# kafka-topics --alter --partitions 12 ...  ← 不可逆地改变 key→分区映射

// 错误：map 改了 key（哪怕 value 没动），再 groupBy 又改一次 →
// 触发一次甚至两次重分区往返
KStream<String, Order> s = orders
    .map((k, v) -> KeyValue.pair(v.userId(), v))   // 改 key → 标记需重分区
    .groupBy((k, v) -> v.userId())                 // 又改 key → 又一次重分区
    .count(...);                                    // 实际建了多余的 repartition topic

// 正确①：value 变换用 mapValues（不碰 key，不触发重分区）
KStream<String, Long> amounts = orders
    .mapValues(Order::amount);                      // key 不变，零重分区

// 正确②：key 已经对了就用 groupByKey（绝不重分区），别用 groupBy
KTable<String, Long> perUser = orders
    .selectKey((k, v) -> v.userId())               // selectKey 是 lazy：自己不建 topic
    .groupByKey(Grouped.with(Serdes.String(), orderSerde))
    .count();                                       // 这里才物化一次重分区，且仅一次

// 正确③：必须重 key 时，用 Repartitioned 命名 + 控分区数，让它可观测可控
KStream<String, Order> rk = orders
    .selectKey((k, v) -> v.userId())
    .repartition(Repartitioned.<String, Order>as("orders-by-user")
        .withNumberOfPartitions(12));

// 错误：clicks 上游被某个外部 producer 用了不同分区器写入，
// 与 impressions 分区数相同但分区方式不同 → join 静默丢匹配
KStream<String, Joined> bad = impressions
    .join(clicks,
          (imp, clk) -> merge(imp, clk),
          JoinWindows.ofTimeDifferenceAndGrace(
              Duration.ofMinutes(5), Duration.ofMinutes(1)));

// 正确①：join 前对至少一侧显式 repartition，强制 Streams 用自己的
// 默认分区器重洗，对齐分区方式（也对齐分区数）
KStream<String, Click> alignedClicks = clicks
    .repartition(Repartitioned.<String, Click>as("clicks-aligned")
        .withNumberOfPartitions(impressionsPartitions));

KStream<String, Joined> good = impressions.join(alignedClicks, ...);

// 正确②：维表小且慢变，用 GlobalKTable —— 每实例全分区副本，
// 天然免共分区，是这条陷阱的逃生口
GlobalKTable<String, User> users = builder.globalTable("users");
KStream<String, Enriched> enriched = events.join(
    users,
    (eventKey, event) -> event.userId(),   // 从流记录提 join key
    (event, user) -> enrich(event, user));

// 错误：依赖全局默认 serde；不同分支类型不同时会用错 serde 炸；
// null key 直接进 groupByKey 被悄悄丢
KStream<String, Payment> payments = builder.stream("payments");  // 用 default serde
KTable<String, Long> bad = payments
    .groupByKey()                                  // null key 记录在此静默消失
    .count();

// 正确：每个 source/sink/store 显式传 Serde；先过滤 null key 再聚合
KStream<String, Payment> typed = builder.stream(
    "payments",
    Consumed.with(Serdes.String(), paymentSerde)); // 显式 key/value serde

KTable<String, Long> good = typed
    .filter((k, v) -> k != null && v != null)      // 显式处理 null，而非让它静默蒸发
    .groupByKey(Grouped.with(Serdes.String(), paymentSerde))
    .count(Materialized.with(Serdes.String(), Serdes.Long())); // store 也显式 serde

// 错误：不管 RocksDB 内存，每个 store 各自一份 block cache + memtable，
// 堆外随 task 数线性膨胀，最终顶破容器上限被 OOMKill

// 正确：自定义 RocksDBConfigSetter，让所有 store 共享一个有上限的
// LRUCache + WriteBufferManager —— 把堆外内存钉成一个常数
public class BoundedRocksDBConfig implements RocksDBConfigSetter {
    // static：进程内所有 task 的所有 store 共用这两个对象
    private static final long TOTAL_OFF_HEAP = 256L * 1024 * 1024; // 256 MiB 总上限
    private static final long TOTAL_MEMTABLE = 128L * 1024 * 1024;

    private static final Cache CACHE = new LRUCache(TOTAL_OFF_HEAP);
    private static final WriteBufferManager WBM =
        new WriteBufferManager(TOTAL_MEMTABLE, CACHE); // memtable 也计入同一池

    @Override
    public void setConfig(String storeName, Options options,
                          Map<String, Object> configs) {
        BlockBasedTableConfig table =
            (BlockBasedTableConfig) options.tableFormatConfig();
        table.setBlockCache(CACHE);                       // 共享 cache
        table.setCacheIndexAndFilterBlocks(true);         // index/filter 也进 cache，纳入上限
        options.setWriteBufferManager(WBM);               // 共享写缓冲池
        options.setTableFormatConfig(table);
    }

    @Override public void close(String storeName, Options options) {}
}

# 挂上 config setter
rocksdb.config.setter=com.example.BoundedRocksDBConfig

# 容器内存 = 堆(-Xmx) + 堆外上限(LRUCache+WBM) + 线程栈/元空间 + 余量
# 例：-Xmx2g + 256MiB cache + ~512MiB 余量 → 容器 limit ≥ 2.75g

# 抑制 glibc 多 arena 造成的 native 内存碎片（容器里尤其明显）
# 作为环境变量设置：MALLOC_ARENA_MAX=2

// 错误：没显式声明 grace，迟到记录何时被丢由默认行为决定，不可控
TimeWindows badWin = TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5));

// 正确：显式声明窗口大小 + grace，把"容忍迟到多久"写成代码里看得见的决策
TimeWindows win = TimeWindows.ofSizeAndGrace(
    Duration.ofMinutes(5),    // 窗口大小
    Duration.ofMinutes(10));  // grace：窗口关后还容忍 10 分钟的迟到

KTable<Windowed<String>, Long> counts = events
    .groupByKey()
    .windowedBy(win)
    .count()
    // suppress：只在窗口彻底关闭后发一个最终结果，避免下游看到中间态
    .suppress(Suppressed.untilWindowCloses(
        Suppressed.BufferConfig.unbounded()));

// 同时：changelog/窗口 store 的 retention 必须 ≥ 窗口 + grace，否则
// 状态先于 grace 被清掉，迟到记录连匹配的窗口都找不到
// Materialized.as(...).withRetention(Duration.ofMinutes(20))

// 配置：EOS 下 commit.interval.ms 默认 100ms，按延迟/吞吐权衡调整
// （properties）
//   processing.guarantee=exactly_once_v2
//   commit.interval.ms=200        # 在延迟和吞吐之间找平衡，不必死守 100
//   # EOS 需要 ≥3 个 broker（事务状态有副本要求）

// 错误：在 EOS 拓扑里直接做外部副作用，以为 EOS 会保证它精确一次
stream.foreach((k, order) -> {
    paymentGateway.charge(order);   // 事务 abort 重试时会重复扣款！EOS 管不到外部
});

// 正确：外部副作用走幂等或 outbox —— 让"重复"在外部侧被吸收
// ① 幂等：用 event-id 做外部 upsert，重复写无副作用
stream.foreach((k, order) ->
    db.upsertByKey(order.eventId(), order));   // 同一 eventId 重复执行结果一致

// ② outbox：Streams 只写一条 Kafka 输出（受 EOS 保护），
//    由独立消费者带去重地投递给外部系统
KStream<String, Charge> outbox = orders
    .mapValues(o -> new Charge(o.eventId(), o.amount()));
outbox.to("charge-outbox");        // 外部投递在另一进程按 eventId 去重