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

【问题标题】：Spring Cloud Stream Kafka Streams inbound KTable predictable internal state-store topic namesSpring Cloud Stream Kafka Streams 入站 KTable 可预测的内部状态存储主题名称
【发布时间】：2022-01-25 05:41:33
【问题描述】：

我们将 Kafka Streams 与 Spring Cloud Stream Functions 一起使用。我们有一个典型的示例应用程序，它将用户点击 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，我们不能改变它的 state-store 内部主题名称，我们总是得到这个主题名称：myapp-id-user-regions-STATE-STORE-0000000001-changelog

如果我们完全通过代码创建拓扑，我们确实有builder.table(final String topic, final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized) 方法，但是使用函数......在这种情况下，有没有办法为输入 KTable 自定义内部主题名称？

【问题讨论】：

标签： spring-boot apache-kafka apache-kafka-streams spring-cloud-stream spring-cloud-stream-binder-kafka

【解决方案1】：

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

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

这在参考文档的this section 中有记录。

【讨论】：

谢谢@sobychacko！抱歉，我在文档中没有看到。我的错