Spring cloud Kafka streams Deserialiation issue wtih Caused by: java.lang.ClassCastException: ClassName cannot be cast to ClassName

I'm trying to use the spring cloud Kafka streams binder to consume Avro messages from the topic but not able to fix this classCast exception.

Here is my code:

  @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;
  }

Binder:

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

Error:

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)

i tried both the approaches as mentioned in this link https://spring.io/blog/2019/12/04/stream-processing-with-spring-cloud-stream-and-apache-kafka-streams-part-3-data-deserialization-and-serialization but no luck

Did I missed anything?.