Kafka Streams为何为聚合与Join操作自动创建主题？

背景 #

近期我基于spring-cloud-stream-kafka-binding搭建了首个学习用Kafka Streams应用，这是一个简单电商系统——读取products主题的商品入库数据，聚合计算商品总库存。当时有两种方案可选：

• 将聚合结果（KTable）发送至aggregated-products主题
• 物化聚合数据

我选了第二种方案，结果发现应用自动创建了一个Kafka主题，消费这个主题也能拿到聚合结果。

用到的核心代码片段如下：

.peek((k,v) -> LOGGER.info("Received product with key [{}] and value [{}]",k, v))
.groupByKey()
.aggregate(Product::new,
        (key, value, aggregate) -> aggregate.process(value),
        Materialized.<String, Product, KeyValueStore<Bytes, byte[]>>as(PRODUCT_AGGREGATE_STATE_STORE).withValueSerde(productEventSerde)//.withKeySerde(keySerde)
        // because keySerde is configured in application.properties
);

另外，我可以通过InteractiveQueryService访问状态存储，直接查询商品总库存。现在有三个疑问需要解答：

1. 应用为何自动创建新的Kafka主题？ #

这是Kafka Streams的状态存储持久化机制导致的——当用Materialized指定状态存储时，Kafka Streams会自动生成一个changelog主题。它的核心作用是同步状态存储的变化，用于故障恢复：如果应用重启或实例扩容，新实例可以通过消费这个changelog主题，快速恢复出完整的状态数据，保证聚合结果不丢失、不重复。

2. 这种自动创建主题存储聚合数据的方式，和手动发送到指定主题有何区别？ #

• 用途定位不同：自动生成的changelog主题是Kafka Streams内部容错用的，不是给外部系统消费的；手动发送到指定主题的聚合结果，是专门给下游业务系统（比如库存查询服务、报表系统）使用的。
• 数据逻辑不同：changelog主题里的是状态存储的增量更新（每次聚合结果变化就发一条），格式遵循Kafka Streams内部约定；手动发送的主题可自定义消息格式，还能控制发送时机（比如只发最终结果、按固定频率发送）。
• 生命周期不同：changelog主题的生命周期和状态存储绑定，状态存储被删除后通常会被清理；手动指定的主题是独立的，由开发者自行维护生命周期。

3. RocksDB在此场景中起到什么作用？ #

RocksDB是Kafka Streams默认的本地状态存储引擎，作用是把聚合后的商品库存数据存在本地磁盘：

• 解决内存瓶颈：如果聚合的商品数量庞大，全放内存会撑爆JVM，RocksDB用磁盘存数据、内存做缓存，平衡性能和存储容量。
• 加速查询：通过InteractiveQueryService查库存时，直接从本地RocksDB读取，比远程查Kafka主题快得多。
• 配合故障恢复：应用重启时，先从changelog主题同步数据，恢复RocksDB里的状态，确保聚合结果和故障前完全一致。

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