如何避免Kafka Streams生成changelog主题？内存窗口存储配置问题

首先，你遇到的问题核心原因是状态存储类型与聚合操作不匹配：你创建了WindowBytesStoreSupplier（窗口存储），但却用在了非窗口聚合（没有调用windowedBy()的aggregate操作）中。这种不匹配会让Kafka Streams忽略你自定义的内存存储配置，自动回退到默认的持久化RocksDB存储，进而创建changelog主题——哪怕你调用了withLoggingDisabled()也没用。

接下来分两种情况给你对应的解决方案：

情况1：你确实需要窗口聚合 #

如果你的业务逻辑是基于时间窗口的聚合（比如统计每X秒内的数据），那你需要在groupByKey()之后添加windowedBy()操作，将聚合绑定到时间窗口上，这样才能正确使用你的窗口内存存储：

WindowBytesStoreSupplier storeSupplier = Stores.inMemoryWindowStore(
    "in-mem-store-" + index,
    Duration.ofSeconds(windowRetentionPeriodInSeconds),
    Duration.ofSeconds(aggregationWindowSizeInSeconds),
    false
);

myStream.filter((key, val) -> val != null)
    .selectKey((key, val) -> val.getId())
    .groupByKey(Grouped.as("key-grouper").with(Serdes.String(), new MyDtoSerde()))
    // 新增windowedBy，将聚合关联到时间窗口
    .windowedBy(TimeWindows.of(Duration.ofSeconds(aggregationWindowSizeInSeconds))
        .until(Duration.ofSeconds(windowRetentionPeriodInSeconds)))
    .aggregate(MyDto::new, new MyUpdater(), 
        Materialized.as(storeSupplier)
            .withCachingDisabled()
            .withLoggingDisabled()
            .with(Serdes.String(), new MyDtoSerde())
    );

情况2：你需要的是全局聚合（无窗口） #

如果你的聚合不需要时间窗口，只是对所有数据做全局聚合，那你应该使用内存键值存储而非窗口存储，将Stores.inMemoryWindowStore替换为Stores.inMemoryKeyValueStore：

KeyValueBytesStoreSupplier storeSupplier = Stores.inMemoryKeyValueStore(
    "in-mem-store-" + index
);

myStream.filter((key, val) -> val != null)
    .selectKey((key, val) -> val.getId())
    .groupByKey(Grouped.as("key-grouper").with(Serdes.String(), new MyDtoSerde()))
    .aggregate(MyDto::new, new MyUpdater(), 
        Materialized.as(storeSupplier)
            .withCachingDisabled()
            .withLoggingDisabled()
            .with(Serdes.String(), new MyDtoSerde())
    );

额外验证点 #

除了上面的核心问题，你还可以检查这两个地方：

• 确认你的Kafka Streams应用配置中没有全局强制启用状态存储日志的配置（比如default.state.store.logging.enabled=true），不过这个配置优先级低于Materialized.withLoggingDisabled()，所以大概率不是问题。
• 确保withLoggingDisabled()是在Materialized对象链的正确位置调用（你当前的调用顺序是对的，先指定存储，再禁用缓存和日志）。

这样调整后，Kafka Streams就会正确使用你定义的内存存储，不会再创建changelog主题了。

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