如何在没有“更改日志主题”的情况下加入 KStream 和 KTable答案

【问题标题】：How to join KStream and KTable without 'changelog topic'如何在没有“更改日志主题”的情况下加入 KStream 和 KTable
【发布时间】：2021-07-05 06:19:14
【问题描述】：

我加入了 KStream 和 KTable。下面是我的代码。

StreamsBuilder builder = new StreamsBuilder();
KTable<String, String> addressTable = builder.table(ADDRESS_TABLE);
KStream<String, String> orderStream = builder.stream(ORDER_STREAM);
orderStream.join(addressTable, 
        (order, address) -> order + " send to " + address)
        .to(ORDER_JOIN_STREAM);

KafkaStreams streams;
streams = new KafkaStreams(builder.build(), props);
streams.start();

执行此代码后，将创建一个新主题。

$ ./kafka-topics.sh --zookeeper localhost:2181 --list
__consumer_offsets
order-join-application3-address-store-name-changelog
address
order

没有'~~~~changlog'主题如何执行代码。

代理版本：0.11.0.2
流版本：2.7.0

【问题讨论】：

你能解释一下为什么你不想要这个话题吗？
@OneCricketeer 我试图阻止集群拥有更多主题。

标签： java apache-kafka apache-kafka-streams

【解决方案1】：

我不确定是否有可能，因为

在 Kafka Streams DSL 中，聚合操作的输入流可以是 KStream 或 KTable，但输出流始终是 KTable。

从Kafka aggregations 中提取，它也像Kafka join 一样使用Streams DSL。基本上，Kafka join 正在做的是在 KTable 上执行 KStream 的查找，他们从here 调用“changelog 流”。

KStream-KTable 连接始终是非窗口连接。他们允许你对 KTable（更改日志流）执行表查找从 KStream（记录流）接收新记录。一个例子用例将丰富用户活动流（KStream）最新的用户个人资料信息（KTable）。

【讨论】：