Kafka Streams自定义State Store变更日志分区不匹配异常排查

这个问题我之前也碰到过，核心原因是你使用的Kafka Streams旧版本（2018年的版本，比如1.0.x及更早）自动创建状态存储变更日志（changelog）的默认规则和预期不一致，导致changelog的分区数、副本数和源topic不匹配，进而引发StreamTask初始化失败。

为什么会出现这个异常？ #

你的源topic test 有2个分区，但自动创建的 test_01-HOUSE-changelog 只有1个分区、1个副本：

1. 当Kafka Streams处理源topic的分区1时，对应的StreamTask（0_1）需要访问changelog的分区1，但这个分区根本不存在，所以抛出 Store HOUSE's change log ... does not contain partition 1 的异常。
2. 禁用自动创建topic后，因为没有提前创建符合要求的changelog，应用找不到对应的topic分区，自然会抛出新的异常。

在早期Kafka Streams版本中，自动创建的changelog默认分区数是1，副本数也是1，并不会自动匹配源topic的配置——这就是问题的根源。

解决办法 #

这里给你三个可行的方案，按推荐程度排序：

1. 升级Kafka Streams到1.1.0及以上版本（最省心）

从Kafka 1.1.0开始，Kafka Streams优化了自动创建changelog的逻辑：

• 自动匹配源topic的分区数创建changelog分区
• 副本数默认使用集群的全局默认值（或你配置的default.replication.factor）
• 自动添加cleanup.policy=compact配置（状态存储的changelog需要日志压缩来控制大小）

升级后，你不需要手动干预changelog的创建，应用会自动生成符合要求的topic，彻底避免分区不匹配的问题。

2. 手动创建符合要求的changelog topic

如果你暂时无法升级版本，可以提前手动创建changelog：

• 名称格式必须是：{application.id}-{store.name}-changelog，你的场景就是test_01-HOUSE-changelog
• 分区数必须和源topic一致（2个）
• 副本数建议和源topic保持一致（3个）
• 必须配置cleanup.policy=compact（否则changelog会无限增长）

创建命令示例：

./kafka-topics.sh --zookeeper localhost:2181 --create \
  --topic test_01-HOUSE-changelog \
  --partitions 2 \
  --replication-factor 3 \
  --config cleanup.policy=compact

创建完成后，记得在应用配置中关闭自动创建topic：auto.create.topics.enable=false，再重启应用即可。

3. 在代码中显式配置changelog参数

如果你既不能升级也不想手动创建topic，可以在定义StateStore时直接指定changelog的配置：

// 定义changelog的配置参数
Map<String, String> changelogConfigs = new HashMap<>();
changelogConfigs.put("cleanup.policy", "compact");
changelogConfigs.put("num.partitions", "2"); // 和源topic分区数一致
changelogConfigs.put("replication.factor", "3"); // 和源topic副本数一致

// 创建StateStore时传入配置
StateStoreSupplier houseStore = Stores.create("HOUSE")
    .withKeys(Serdes.String())
    .withValues(houseSerde)
    .persistent()
    .withLoggingEnabled(changelogConfigs) // 显式指定changelog配置
    .build();

这样当应用自动创建changelog时，就会使用你指定的分区数和副本数，和源topic保持一致。

额外注意事项 #

• 确保你的application.id是唯一的，不同应用不能共享同一个application.id，否则会导致changelog被混用引发异常。
• 如果调整了topic分区数，记得对应的changelog分区数也要同步调整（旧版本需要手动处理，新版本会自动适配，但前提是重新创建应用）。

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