Spring Cloud Stream Kafka Streams绑定与Schema Registry配置求助

我使用Debezium实现CDC，按Debezium文档配置了Apicurio Schema Registry，已成功将Avro消息发送至Kafka Topic，且消息Schema已注册到Apicurio中。尝试按Spring Cloud Kafka文档指定的属性配置后，消费者仍默认以JSON格式读取消息。

消费者代码 #

@Bean
public Consumer<KStream<Key, Value>> capacityExport() {
  return message -> message.filter((k, v) -> v != null)
      .foreach((k, v) -> processEvent(v));
}

其中Key和Value为自定义普通Java POJO。

调整后的Kafka Streams绑定配置 #

spring:
  cloud:
    stream:
      kafka:
        streams:
          bindings:
            capacityExport-in-0:
              consumer:
                configuration:
                  key.deserializer: io.apicurio.registry.serde.avro.AvroKafkaDeserializer
                  value.deserializer: io.apicurio.registry.serde.avro.AvroKafkaDeserializer
                  apicurio.registry.url: http://172.64.5.254:8080
                  schema.registry.url: http://172.64.5.254:8080
                  specific.avro.reader: true
                key-serde:  io.apicurio.registry.serde.avro.AvroSerde
                value-serde: io.apicurio.registry.serde.avro.AvroSerde

报错信息 #

org.apache.kafka.streams.errors.StreamsException: ClassCastException invoking processor: KSTREAM-FILTER-0000000001. 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: org.apache.avro.generic.GenericData$Record, and value: org.apache.avro.generic.GenericData$Record.
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.
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:150) ~[kafka-streams-3.1.1.jar:na]
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forwardInternal(ProcessorContextImpl.java:253) ~[kafka-streams-3.1.1.jar:na]
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:232) ~[kafka-streams-3.1.1.jar:na]
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:191) ~[kafka-streams-3.1.1.jar:na]
    at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:84) ~[kafka-streams-3.1.1.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamTask.lambda$process$1(StreamTask.java:731) ~[kafka-streams-3.1.1.jar:na]
    at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:809) ~[kafka-streams-3.1.1.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:731) ~[kafka-streams-3.1.1.jar:na]
    at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:1296) ~[kafka-streams-3.1.1.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:784) ~[kafka-streams-3.1.1.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:604) ~[kafka-streams-3.1.1.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:576) ~[kafka-streams-3.1.1.jar:na]

需求说明 #

根据错误信息，反序列化结果为GenericRecord而非自定义POJO，导致类型转换失败。我不想使用Maven Avro插件生成Java类，希望直接将Registry中的Avro Schema消息反序列化为普通Java POJO，也在考虑是否切换至Spring Schema Registry会有更好的集成效果。本人为Kafka新手，恳请帮助。

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