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

Question

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?.

Nothing seems to be obvious from your code/config shared. See Gary's answer below and investigate if that is the root of the issue. Some minor suggestions: If you are using the 3.0.x versions of the binder, you don't need to setup useNativeEncoding or useNativeDecoding properties. You also don't need to set contentType in the case of Kafka Streams applications. — sobychacko

Gary Russell Gary Russell · Accepted Answer · 2021-02-08T16:45:20

Caused by: java.lang.ClassCastException: EventKey cannot be cast to EventKey

This is probably a class loader problem; with the deserializer and consumer bean being loaded by different class loaders, are you using Spring DevTools?

With spring-kafka, this can be avoided by explicitly creating the consumer factory and injecting the deserializer into it.

With spring-cloud-stream (starting with version 3.0.6) you can provide a ClientFactoryCustomizer bean and inject the deserializer instances (defined as @Beans, so that they use the same class loader).

Or, stop using DevTools.

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

1 Answers