Kafka KStream 到 KStream 加入不适用于 Avro SpecificRecords答案

【问题标题】：Kafka KStream to KStream join not working for Avro SpecificRecordsKafka KStream 到 KStream 加入不适用于 Avro SpecificRecords
【发布时间】：2022-10-24 03:13:44
【问题描述】：

我有两个要加入的 Avro SpecificRecord 流。我可以独立处理两个流，但我无法理解可以用来加入这两个流的代码。到目前为止，这是我的代码：

    KStream<String, RecordOne> recOneStream = streamsBuilder.stream(recOneTopic, Consumed.with(Serdes.String(), recOneSpecificSerde));
    KStream<String, RecordTwo> recTwoStream = streamsBuilder.stream(recTwoTopic, Consumed.with(Serdes.String(), recTwoSpecificSerde));


    // change key to match key of rec two
    KStream<String, RecordOne> recOneChangedKeyStream = recOneStream.selectKey((k, v) -> v.getKeyValue().toString());

    // folowing works
    recOneChangedKeyStream.peek((k, v) -> System.out.println("Key : " + k + " Value : " + v)); // output is as expected here

    // trying to make following work?
    KStream<String, JoinedRecord> joinedRecord = recOneChangedKeyStream.join(recTwoStream, (recOn, recTwo) -> {
                JoinedRecord jr = new JoinedRecord();
                jr.setFieldOne...
                return jr;
            },
            JoinWindows.of(Duration.ofSeconds(60)),
            // if I add following line the code breaks at compile time, if I don't add it then it breaks at runtime
            Joined.with(Serdes.String(), recOneSpecificSerde, recTwoSpecificSerde)
    );

所以问题出在Joined.with中。我想我已经正确地遵循了这个例子： confluent-example 因为它也以与我类似的方式使用 Joined.with。但是，就我而言，我在 IDE 中看到以下异常：

Cannot resolve method 'join(org.apache.kafka.streams.kstream.KStream<java.lang.String, RecordTwo>, <lambda expression>, org.apache.kafka.streams.kstream.JoinWindows, org.apache.kafka.streams.kstream.Joined<K,V,VO>)'

【问题讨论】：

标签： java apache-kafka-streams avro

【解决方案1】：

如下面的屏幕截图所示，文档状态应使用 Joined.with。但是在阅读了 IDE 中的错误消息然后继续 GitHub examples 之后，我注意到该文档没有更新，因为所需参数的类型是 StreamJoined 而不是 Joined

【讨论】：