Kafka Streams 7.2.2-ccs状态存储、监控及RocksDB配置问题咨询

拓扑说明 #

我在Kafka Streams 7.2.2-ccs中构建的拓扑代码实现如下：

val groupedStream = StreamsBuilder().stream<String, Quote>("quotes").groupByKey()
for (windowSize in windows()) {
    groupedStream
        .windowedBy(TimeWindows.ofSizeWithNoGrace(windowSize))
        .aggregate({ Aggregator() },{ _, quote, aggregator -> aggregator.execute(quote) })
        .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()))
        .toStream()
        .to("outputTopic")
}

问题解答 #

1. 为何不存在unbounded suppressed store相关的监控指标？ #

Kafka Streams 7.2.2-ccs版本中，io.micrometer.core.instrument.binder.kafka.KafkaStreamsMetrics默认未暴露Unbounded Suppress Store的专属指标，原因如下：

• Unbounded Suppress的底层依赖窗口存储的RocksDB状态后端，其状态操作会复用窗口存储的指标体系，你看到的带rocksdb_window_state_id标签的指标已经包含了Suppress阶段的状态统计。
• 若需单独监控Suppress行为，可通过自定义MetricsReporter监听org.apache.kafka.streams.processor.internals.SuppressProcessor的内部指标，或升级到更高版本的Kafka Streams（部分新版本补充了Suppress专属监控维度）。

2. 输入主题"quotes"有3个分区时，将创建多少个RocksDB实例？窗口存储的段数量如何查看？ #

• RocksDB实例数量：每个窗口聚合操作会为每个输入分区创建独立的RocksDB实例。假设windows()返回N个窗口大小，总实例数为 3（输入分区数） × N（窗口数量）——因为每个窗口聚合都是独立的状态存储，Kafka Streams会为每个任务（对应输入分区）分配独立的状态存储实例。
• 窗口存储段数量：段数量由retention.ms和segment.ms配置决定（默认segment.ms为7天），每个段对应一个时间范围的窗口数据。查看方式：
  • 直接查看状态存储目录下的RocksDB子目录，每个段对应带时间戳的子文件夹；
  • 通过JMX指标kafka.streams:type=stream-state-metrics,state-store-type=window-store,state-store-name=*,scope=partition,partition=*,name=num-segments获取（需开启JMX监控）。

3. 是否可配置RocksDB将已关闭窗口的所有键刷新至磁盘？如何解决堆外内存持续增长问题？ #

可以通过以下配置优化：

• 强制刷新已关闭窗口数据：设置state.cleanup.delay.ms为较小值（如30000ms），该配置控制Kafka Streams清理过期窗口状态的延迟时间，触发清理时会将已关闭窗口的键值对从内存刷盘并删除。
• RocksDB内存调优：
  • 配置rocksdb.block.cache.size限制块缓存大小；
  • 调整rocksdb.write.buffer.size和rocksdb.max.write.buffer.number控制写缓冲的内存占用；
  • 开启rocksdb.cache-index-and-filter-blocks减少内存消耗。
• 窗口生命周期校验：确认TimeWindows.ofSizeWithNoGrace(windowSize)的窗口大小符合业务需求，避免不必要的窗口数据留存；若业务允许，可考虑添加Grace Period（使用ofSizeAndGrace方法）。

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