我正在尝试使用springcloudkafkastreams绑定器来使用来自该主题的avro消息，但无法修复这个classcast异常。
这是我的密码：

@Bean
  public Consumer<KStream<EventKey, Event>> process(){
    return input -> {
      input.peek(((key, value) -> logger.info("key value: "+ key.toString()+" value: "+value.toString())));
      logger.info("Received:" + input);
    };
  }

  @Bean
  public Serde<EventKey> avroInSerde(){
    final SpecificAvroSerde<EventKey> avroInSerde = new SpecificAvroSerde<>();
    Map<String, Object> serdeProperties = new HashMap<>();
    return avroInSerde;
  }

  @Bean
  public Serde<Event> avroOutSerde(){
    final SpecificAvroSerde<Event> avroOutSerde = new SpecificAvroSerde<>();
    return avroOutSerde;
  }

活页夹：

spring:
  application:
    name: ${applicaton-name}
  cloud:
    stream:
      function:
        definition: process
      bindings:
        process-in-0:
          destination: ${input-topic-name}
          contentType: application/Avro
        process-out-0:
          destination: ${enriched-topic-name}
          contentType: application/Avro
      binding-retry-interval: 30
      kafka:
        streams:
          binder:
            brokers: ${kafka-broker}
            application-id: ${consumer-group-name}
            auto-create-topics: false
            auto-add-partitions: false
            configuration:
              processing.guarantee: at_least_once
              auto.offset.reset: earliest
              schema.registry.url: ${kafka-schema-registry}
              auto-register-schema: false
              security.protocol: SSL
              useNativeEncoding: true
              specific.avro.reader: true

错误：

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:146)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:236)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:216)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:168)
    at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:96)
    at org.apache.kafka.streams.processor.internals.StreamTask.lambda$process$1(StreamTask.java:679)
    at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:836)
    at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:679)
    at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:1033)
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:690)
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:551)
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:510)
Caused by: java.lang.ClassCastException: EventKey cannot be cast to EventKey
    at org.apache.kafka.streams.kstream.internals.KStreamPeek$KStreamPeekProcessor.process(KStreamPeek.java:42)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.lambda$process$2(ProcessorNode.java:142)
    at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:836)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:142)

我尝试了这两种方法在这个链接中提到https://spring.io/blog/2019/12/04/stream-processing-with-spring-cloud-stream-and-apache-kafka-streams-part-3-data-deserialization-and-serialization 但运气不好
我错过什么了吗？。