如何控制Kafka Streams应用中leftJoin操作的内部主题名称？

对于KStream之间的leftJoin操作生成的内部变更日志主题，除了application.id前缀外，完全可以通过API控制主题名称的中间部分，核心是给join关联的状态存储指定自定义名称，替代系统自动生成的默认标识。

具体操作步骤：

• 调用leftJoin()时，选用支持Materialized参数的重载方法，通过Materialized.as("自定义存储名")为join的状态存储设置自定义名称。
• 自定义名称会直接替换主题名中KSTREAM-JOINTHIS这类自动生成的部分，最终主题格式变为{application-id}-{自定义存储名}-store-changelog。

示例代码：

// 左侧流与右侧流执行leftJoin
leftStream.leftJoin(
    rightStream,
    (leftVal, rightVal) -> /* 此处编写你的join业务逻辑 */,
    JoinWindows.of(Duration.ofMinutes(5)), // 根据业务设置窗口时长
    Materialized.as("my-custom-left-join-store") // 指定自定义存储名称
);

注意事项 #

• 提前创建主题时，必须保证主题的分区数、副本数、清理策略等配置与Kafka Streams的预期一致，否则应用启动会抛出配置不匹配的异常。
• 如果join操作需要对数据流重分区（比如左右流的Key不一致），可以继续使用你已掌握的Named.as()为重分区主题指定自定义名称。

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