2

我希望为 3 个 ktable 执行 Ktable-Ktable 连接，类似这样

var joinedStream =
        person
            .join(
                address,
                Person::getAddress,
                valueJoiner,
                Materialized.with(Serdes.Integer(), jsonSerde))
            .leftJoin(addressType, this::getAddressType,valueJoiner);

虽然 jsonSerde 在使用 customAvroServde 实现主题时有效，但最终连接失败并出现序列化异常。

在流连接中执行此操作时，我能够使用 StreamJoined.with 功能执行它，该功能有助于为内部流设置序列化程序，但找不到任何类似 ktables 的东西。

我应该如何使用 Ktables 执行连接？

4

0 回答 0