Kafka Streams流连接问题：仅读取topic-a数据，未实现消息合并

1. 确保两个Topic的消息Key完全一致 #

流-流窗口连接的核心要求是相同Key的消息才能在窗口内匹配关联。你仅提供了topic-a的发送代码，需要确认topic-b的消息发送时使用和topic-a完全相同的Key：

// 参照topic-a的发送逻辑，topic-b需用相同内容作为Key
@PostMapping("publish-b/{word}")
fun publishEventB(@PathVariable(value = "word") event: String){
   kafkaTemplate.send("topic-b", event, event)
}

如果topic-b的Key与topic-a不一致（比如使用默认null或其他值），两个流的消息永远无法匹配，最终只会得到topic-a的原始数据。

2. 修正Serde序列化配置错误 #

你使用JsonSerde<String>处理字符串，但发送的是原始字符串（如"Hi"），而JsonSerde要求输入为合法JSON格式（如带双引号的"Hi"），这会导致反序列化失败，消息被Kafka Streams静默丢弃。

直接替换为Serdes.String()即可，无需JSON序列化：

// 替换原有JsonSerde配置
Serde<String> stringSerde = Serdes.String();
KStream<String, String> stream = builder
        .stream("topic-a", Consumed.with(Serdes.String(), stringSerde));

return stream.join(
                // 给topic-b也指定一致的Consumed配置
                builder.stream("topic-b", Consumed.with(Serdes.String(), stringSerde)),
                (w1, w2) -> w1+w2,
                JoinWindows.of(Duration.ofSeconds(10)),
                StreamJoined.with(Serdes.String(), stringSerde, stringSerde))
        .toTable(Materialized.<String, String>as(store)
        .withKeySerde(Serdes.String())
        .withValueSerde(stringSerde));

3. 验证消息时间戳是否在窗口范围内 #

窗口连接要求两条匹配消息的时间戳差不超过10秒。可通过Kafka命令行工具查看消息时间戳：

kafka-console-consumer.sh --bootstrap-server <你的Kafka地址> --topic topic-a --property print.timestamp=true
kafka-console-consumer.sh --bootstrap-server <你的Kafka地址> --topic topic-b --property print.timestamp=true

确保两条消息的时间戳间隔在10秒以内，否则会因超出窗口范围无法匹配。

4. 检查状态存储的读取逻辑 #

如果通过交互式查询读取状态存储，需确认：

• 查询的Key为匹配的目标Key（如"Hi"）
• 应用已完成两条消息的处理，状态存储已完成更新

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