如何基于滚动方式使用Kafka Streams清理Kafka及状态存储？

一、设置一周主题保留期是否安全？ #

这取决于你的流处理场景，核心判断依据是Kafka Streams的状态恢复需求和业务逻辑：

• 安全场景：如果你的应用不需要频繁从头消费数据，且流处理中没有超过一周的窗口（比如小时级/天级聚合），同时重启时消费者能在一周内追上最新偏移量，那一周的保留期完全安全。
• 风险场景：
  • 若业务需要全量重跑历史数据（比如调试、重新计算季度指标），一周保留期会导致部分数据丢失，无法完成全量计算。
  • 若使用了超过一周的窗口逻辑（比如周度/月度统计），输入主题保留期必须长于窗口时长+宽限期，否则窗口内的旧数据被删除后，迟到的数据无法被处理，会导致聚合结果不一致。
  • 若应用重启或扩容时，消费者滞后量过大，超过一周还没追上，会丢失中间数据，导致状态恢复失败，出现数据不一致。

建议：先评估业务的最大窗口时长和重启恢复需求，再设置保留期；同时监控消费者滞后量，确保重启后能在数据过期前完成状态恢复。

二、安全清理Kafka Streams状态存储的旧数据 #

Kafka Streams提供了多种内置机制来自动清理状态，无需手动暴力删除，关键是配置到位：

1. 窗口状态自动清理 #

针对窗口聚合产生的状态，通过以下参数控制：

• retention.ms：配置窗口状态的保留时长，必须大于window.size.ms + grace.period.ms（窗口时长+数据迟到宽限期），确保迟到的数据仍能被处理。例如窗口为1天、宽限期12小时，retention.ms至少设为54000000（1.5天）。
• 状态存储对应的changelog主题，默认cleanup.policy=delete，会自动删除过期的状态记录，无需额外配置。

2. 键值状态的TTL配置 #

对于普通键值存储（如KeyValueStore），创建存储时可直接指定TTL，自动清理过期键值对：

// 示例：创建一个TTL为7天的键值存储
StoreBuilder<KeyValueStore<String, Long>> storeBuilder = Stores.keyValueStoreBuilder(
    Stores.persistentKeyValueStore("user-score-store"),
    Serdes.String(),
    Serdes.Long())
    .withRetention(Duration.ofDays(7)) // 设置键值对TTL为7天
    .withLoggingEnabled(Collections.singletonMap("retention.ms", "604800000")); // 同步设置changelog主题保留期为7天
streamsBuilder.addStateStore(storeBuilder);

3. 手动清理（谨慎使用） #

如果需要临时清理状态，可通过以下方式：

• 使用kafka-streams-application-reset工具重置应用状态，该工具会删除本地状态存储并重置消费者偏移量，让应用从头消费数据。注意：执行前必须停止应用，且确认不需要现有状态数据。
• 通过AdminClient删除状态对应的changelog主题，但仅适用于应用已停止、且确认状态无保留价值的场景，否则会导致状态丢失。

4. 关键注意事项 #

• 清理动作必须避开业务的关键处理阶段，确保不会影响正在进行的窗口计算或数据更新。
• 定期监控状态存储大小（通过JMX指标kafka.streams:type=stream-state-metrics,state-store=*,topic=*下的Size），及时调整TTL或保留期配置。
• 对changelog主题启用备份（如利用Kafka的镜像复制功能），避免清理失误导致的数据无法恢复。

内容的提问来源于stack exchange，提问作者AndCode