title: Getting Kafka Streams Class Cast Exception on the same Class while applying map function
body: <p><strong>UserRecord.java</strong> (autogenerated by Maven Avro plugin) <br/></p>
  
  <pre><code>UserRecord extends SpecificRecordBase implement SpecificRecord
  </code></pre>
  
  <p><strong>UserRecordSerde.java</strong></p>
  
  <pre><code>UserRecordSerde extends SpecificAvroSerde
  </code></pre>
  
  <p><strong>application.yml</strong> </p>
  
  <pre><code>spring.cloud.stream.bindings.input.destination: userTopic
  spring.cloud.stream.bindings.input.consumer.useNativeDecoding: true
  spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde:UserRecordSerde
  spring.cloud.stream.kafka.streams.bindings.input.consumer.keySerde: LongSerdespring.cloud.stream.kafka.streams.binder.configuration.default.key.serde: LongSerde
  spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde: SpecificAvroSerde
  </code></pre>
  
  <h2><strong>Class - StreamListener - raw stream comes with null key and UserRecord object in avro</strong></h2>
  
  <pre><code>@StreamListener
          public KStream&lt;Long, ArrayList&lt;UserRecord&gt;&gt; handleUserRecords (@Input KStream&lt;?, UserRecord&gt; userRecordStream) { &lt;br/&gt;
          Map&lt;String, Object&gt; serdeConfig = new HashMap();
          serdeConfig.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");
          serdeConfig.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true); &lt;br/&gt;
          Serde&lt;ArrayList&lt;UserRecord&gt;&gt; userRecordListSerde = new SpecificAvroSerde();
          userRecordListSerde.configure(serdeConfig, false); &lt;br/&gt;
          return userRecordStream
              .map((key, value) -&gt; new KeyValue(value.getUserID, value)
              .groupByKey(Serialized.with(Serdes.Long(), userRecordSerde))
              .aggregate(ArrayList::new, Long key, UserRecord value, ArrayList agg -&gt;
              {
                 agg.add(value);
                 return agg;
              }, userRecordListSerde)
          .toStream();
      }
  </code></pre>
  
  <p><strong>exception</strong></p>
  
  <pre><code>java.lang.ClassCastException: com.example.UserRecord cannot be cast to com.example.UserRecord
      at org.apache.kafka.streams.kstream.internals.KStreamMap$KStreamMapProcessor.process(KStreamMap.java:41)
      at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:46)
      at org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:208)
      at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:124)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:85)
      at org.apache.kafka.streams.kstream.internals.KStreamMap$KStreamMapProcessor.process(KStreamMap.java:42)
      at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:46)
      at org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:208)
  </code></pre>