Spring Cloud Stream Kafka Streams入站KTable可预测的内部状态存储主题名称

Question

我们正在使用带有Spring Cloud Stream函数的Kafka Streams。我们有一个典型的示例应用程序，它将用户单击kstream与用户区域ktable结合在一起。

我们知道，在定义拓扑结构时，我们可以通过使用适当的方法来强制内部变更日志或重新分区主题的自定义名称，这些方法接受物化存储的名称：

  @Bean
  public BiFunction<KStream<String, Long>, KTable<String, String>, KStream<String, Long>> bifunctionktable() {
    return (userClicksStream, userRegionsTable) -> userClicksStream
        .leftJoin(userRegionsTable,
            (clicks, region) -> new RegionWithClicks(region == null ? "UNKNOWN" : region, clicks),
            Joined.with(Serdes.String(), Serdes.Long(), null, "bifunctionktable-leftjoin"))
        .map((user, regionWithClicks) -> new KeyValue<>(regionWithClicks.getRegion(), regionWithClicks.getClicks()))
        .groupByKey(Grouped.with(Serdes.String(), Serdes.Long()).withName("bifunctionktable-groupbykey"))
        .reduce((firstClicks, secondClicks) -> firstClicks + secondClicks, Materialized.as("bifunctionktable-reduce"))
        .toStream();
  }

但是对于输入的KTable，我们不能更改它的状态存储内部主题名称，我们总是得到这个主题名称：myapp-id-user-区域-STATE-STORE-0000000001-变更日志

如果我们完全按照代码创建拓扑，那么我们确实有builder。表（最终字符串主题，最终物化）




Answer 1

您可以使用以下属性为传入的KTable添加自定义名称：

spring.cloud.stream.kafka.streams.bindings.bifunctionktable-in-1.consumer.materializedAs: <Your-custom-store-name>

这在参考文件的本节中有记录。