关于Kafka Stream默认状态存储位置及自动创建机制的咨询

为什么没指定状态存储代码还能正常运行？ #

• 你用的count()属于Kafka Streams的有状态操作，当调用这类API时，系统会自动创建默认的状态存储。虽然你没显式指定存储类型，但Kafka Streams默认就用RocksDB作为状态存储实现（这和文档里的说明一致）。
• 你代码里的Materialized.with()其实已经触发了状态存储的创建逻辑，它指定了键值对的序列化器，系统会基于这个配置自动生成对应的RocksDB存储。
• 你看到的changelog主题，是Kafka Streams自动为这个状态存储创建的状态备份主题，用来在应用重启、故障恢复时恢复状态，保证数据的一致性和持久性。

自动创建的RocksDB文件在文件系统中的位置 #

• 默认存储路径由配置项state.dir控制，默认值是/tmp/kafka-streams。
• 每个应用的状态存储会放在独立的子目录下，路径结构大致是：[application.id]/[线程ID]/[存储名称]
  • application.id是你的Kafka Streams应用唯一标识（如果没手动配置，系统会生成一个默认值，但建议显式指定，避免不同应用的状态目录混乱）
  • 线程ID是应用内部的处理线程编号
  • 存储名称是系统自动生成的（如果在Materialized里用as("自定义名称")指定了名称，就会用你定义的名字）

你的示例代码 #

builder.stream("my-topic-1")
        .groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
        .count(Materialized.with(Serdes.String(), Serdes.Long()))
        .toStream()
        .to("my-topic-2");

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