将 Spring Cloud Stream Kafka Streams 与带有 nativeEncoding/decoding=false 的 Avro 输入/输出一起使用

Question

我们正在通过带有 Avro 输入/输出记录的 Spring Cloud Stream 功能支持测试 Kafka Streams 的使用，但设置 nativeEncoding=false 和 nativeDecoding=false 以便在我们进行 Avro 转换时使用自定义 MessageConverter。

默认 serdes 是 StringSerde 用于键和 ByteArraySerde 用于值。

当我们只使用 KStream to KStream 函数时一切正常，例如：

    @Bean
    public Function<KStream<String, DataRecordAvro>, KStream<String, DataRecordAvro>> wordsCount() {
      return input -> input
          .flatMapValues(value -> Arrays.asList(value.getName().toString().toLowerCase().split("\\W+")))
          .map((key, value) -> new KeyValue<>(value, value))
          .groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
          .windowedBy(TimeWindows.of(Duration.ofSeconds(5)).grace(Duration.ofMillis(0)))
          .count()
          .toStream()
          .map((key, value) -> new KeyValue<>(key.key(), new DataRecordAvro(key.key(), value)));
    }

但是当我们尝试一个涉及输入 KTable 的更复杂的示例时：

    @Bean
    public BiFunction<KStream<String, DataRecordAvro>, KTable<String, DataRecordAvro>, KStream<String, DataRecordAvro>> userClicksRegionKTableAvro() {
      return (userClicksStream, usersRegionKTable) -> userClicksStream
          .leftJoin(usersRegionKTable,
              (clicks, region) -> new RegionWithClicks(region == null ? "UNKNOWN" : region.getName().toString(), clicks.getCount()))
          .map((user, regionWithClicks) -> new KeyValue<>(regionWithClicks.getRegion(), regionWithClicks.getClicks()))
          .groupByKey(Grouped.with(Serdes.String(), Serdes.Long()))
          .reduce(Long::sum)
          .mapValues((key, value) -> new DataRecordAvro(key, value))
          .toStream();
    }

（DataRecordAvro 类只有两个成员：CharSequence name; Long count;）

当收到第一条记录时抛出此异常：

ClassCastException invoking Processor. Do the Processor's input types match the deserialized types? Check the Serde setup and change the default Serdes in StreamConfig or provide correct Serdes via method parameters. Make sure the Processor can accept the deserialized input of type key: java.lang.String, and value: com.xxxx.kstreams.fixtures.avro.DataRecordAvro.
Note that although incorrect Serdes are a common cause of error, the cast exception might have another cause (in user code, for example). For example, if a processor wires in a store, but casts the generics incorrectly, a class cast exception could be raised during processing, but the cause would not be wrong Serdes.

抛出异常的处理器似乎是：

KSTREAM-LEFTJOIN-0000000011:
    states:     [user-regions-avro-STATE-STORE-0000000008]

我们不知道为什么它在这种情况下不起作用。也许leftJoin 操作将信息保留到内部主题，而useNativeEncoding/Decoding=false 没有被考虑在内？但是为什么上面的 kstream->kstream 示例确实有效？我们认为 Avro 转换只在拓扑的开始和结束时完成，为什么在使用 leftJoin 时会出现这种转换异常？

这是另一个可以正常工作的示例（没有输入 Avro 记录，将消费者 useNativeDecoding 保留为默认 true）：

    @Bean
    public BiFunction<KStream<String, Long>, KTable<String, String>, KStream<String, DataRecordAvro>> userClicksRegionKTable() {
      return (userClicksStream, usersRegionKTable) -> userClicksStream
          .leftJoin(usersRegionKTable,
              (clicks, region) -> new RegionWithClicks(region == null ? "UNKNOWN" : region, clicks))
          .map((user, regionWithClicks) -> new KeyValue<>(regionWithClicks.getRegion(), regionWithClicks.getClicks()))
          .groupByKey(Grouped.with(Serdes.String(), Serdes.Long()))
          .reduce(Long::sum)
          .mapValues((key, value) -> new DataRecordAvro(key, value))
          .toStream();
    }

请帮忙！

Answer 1

对于 Spring Cloud Stream 中的 Kafka Streams binder，我们建议使用带有 Serdes 的本机解码/编码，除非您有充分的理由依赖消息转换方法。什么是迫使您在这里使用消息转换器的用例？在实践中，在 Spring Cloud Stream 的 Kafka Streams 应用程序中使用消息转换器进行序列化会在您的拓扑中添加一个额外的层并使其更深，因此建议使用本机解码/编码。

正如您所指出的，对于 KTable，活页夹始终使用本机解码 - 目前，在那里无法使用消息转换器。当您在 KTable 绑定上关闭 useNativeDecoding 时，绑定程序会忽略它并仅使用默认字节 serde。我建议在 KTable 绑定上使用默认值，然后在您的应用程序配置中添加以下 bean。

@Bean
public Serde< DataRecordAvro> dataRecordAvroSerde() {
   // return Serde
}

这样，binder 将检测到这个 bean 并意识到 Serde 类型与函数签名中的类型匹配，然后在这些输入上使用它。

如果您对此应用还有其他问题，请随时分享MCRE。那我们可以进一步看看。