title: Confluent Kafka JSON Deserialization Error
body: <p>I have a.NET application which is writing a JSON message to Kafka TOPIC using the Confluent JSON schema aware serializer.</p>
  <p>I have another spring boot JAVA application which is reading from that topic and creating a KTABLE and grouping by Keys and writing the value as JSON to another output TOPIC.</p>
  <p>The code for spring boot JAVA is as follows:</p>
  <p><strong>application.properties</strong></p>
  <pre><code>server.port=9081
  #stream config
  spring.kafka.streams.bootstrap-servers=localhost:9092
  spring.kafka.streams.properties.schema.registry.url=http://localhost:8081
  spring.kafka.streams.application-id=KTable-aggregations
  spring.kafka.streams.properties.group-id=stream-group
  spring.kafka.streams.replication-factor=1
  spring.kafka.streams.properties.commit.interval.ms=30000
  spring.kafka.streams.properties.cache.max.bytes.buffering=10485760
  spring.kafka.streams.properties.metadata.max.age.ms=10000
  spring.kafka.streams.properties.num.stream.threads=1
  spring.kafka.streams.properties.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
  spring.kafka.streams.properties.default.value.serde=io.confluent.kafka.streams.serdes.json.KafkaJsonSchemaSerde
  spring.kafka.streams.properties.spring.json.value.default.type=com.kafkastreams.aggregate.model.Output
  spring.kafka.streams.properties.default.deserialization.exception.handler=org.apache.kafka.streams.errors.LogAndContinueExceptionHandler
  spring.kafka.streams.properties.compression-type=gzip
  #spring.kafka.streams.properties.state.dir=/var/lib/kafka-streams
  #spring.kafka.streams.properties.state.cleanup.delay.ms=600000
  spring.kafka.streams.properties.auto-offset-reset=latest
  spring.kafka.streams.properties.timestamp.extractor=org.apache.kafka.streams.processor.WallclockTimestampExtractor
  
  #default queue
  kafka.topic.input=input-topic
  kafka.topic.output=output-topic
  </code></pre>
  <p><strong>myAPI.java</strong></p>
  <pre><code>@JsonInclude(JsonInclude.Include.NON_NULL)
  @JsonPropertyOrder({
      &quot;id&quot;,
      &quot;name&quot;,
      &quot;address&quot;,
  })
  public class MyAPI{
  
      @JsonProperty(&quot;id&quot;)
      private long id;
      @JsonProperty(&quot;name&quot;)
      private String name;
      @JsonProperty(&quot;address&quot;)
      private String address;
  
      @JsonProperty(&quot;id&quot;)
      public long getId() {
          return id;
      }
  
      @JsonProperty(&quot;id&quot;)
      public void setId(long id) {
          this.id = id;
      }
  
      @JsonProperty(&quot;name&quot;)
      public String getName() {
          return name;
      }
  
      @JsonProperty(&quot;name&quot;)
      public void setName(String name) {
          this.name = name;
      }
  
      @JsonProperty(&quot;address&quot;)
      public String getAddress() {
          return address;
      }
  
      @JsonProperty(&quot;address&quot;)
      public void setAddress(String address) {
          this.address = address;
  }
  </code></pre>
  <p><strong>Output.java</strong></p>
  <pre><code>@JsonPropertyOrder({
      &quot;id&quot;,
      &quot;name&quot;,
      &quot;address&quot;,
  })
  public class Output{
  
      @JsonProperty(&quot;id&quot;)
      private long id;
      @JsonProperty(&quot;name&quot;)
      private String name;
  
  
      @JsonProperty(&quot;id&quot;)
      public long getId() {
          return id;
      }
  
      @JsonProperty(&quot;id&quot;)
      public void setId(long id) {
          this.id = id;
      }
  
      @JsonProperty(&quot;name&quot;)
      public String getName() {
          return name;
      }
  
      @JsonProperty(&quot;name&quot;)
      public void setName(String name) {
          this.name = name;
      }
  
  }
  </code></pre>
  <p><strong>Processor.java</strong></p>
  <pre><code>@Configuration
  @EnableKafka
  @EnableKafkaStreams
  public class KafkaStreamConfig {
      private final static Logger logger = LoggerFactory.getLogger(KafkaStreamConfig.class);
      @Autowired
      private KafkaProperties kafkaProperties;
  
      @Value(&quot;${kafka.topic.input}&quot;)
      private String inputTopic;
  
      @Value(&quot;${kafka.topic.output}&quot;)
      private String outputTopic;
  
      @Bean 
      public KStream&lt;String, Api&gt; kStream(StreamsBuilder kStreamBuilder) 
      {
          Map&lt;String, String&gt; serdeConfig = Collections.singletonMap(&quot;schema.registry.url&quot;,
                  &quot;http://localhost:8081&quot;);
      
            Serde&lt;MyApi&gt; myapiSerde = new KafkaJsonSchemaSerde&lt;&gt;(); //Not sure if this is the right way of creating a Serde
            myapiSerde.configure(serdeConfig, false);
            Serde&lt;Output&gt; outputSerde = new KafkaJsonSchemaSerde&lt;&gt;();//Not sure if this is the right way of creating a Serde
            outputSerde.configure(serdeConfig, false);
            Serde&lt;String&gt; stringSerde = Serdes.String();
  
            
            KTable&lt;String, Output&gt; cdcTbl = kStreamBuilder.stream(inputTopic,
            Consumed.with(stringSerde,myapiSerde ) .withOffsetResetPolicy(EARLIEST))
            .mapValues(v -&gt; Api.newBuilder(v).build()) .groupByKey() .reduce((aggValue,
            newValue) -&gt; newValue /*adder*/ );
            
            cdcTbl.toStream().peek((k, v) -&gt; logger.info(&quot;API Topic key and value {} {}&quot;,
            k, v)) .to(outputTopic, Produced.with(stringSerde, outputSerde));
            
            return kStreamBuilder.stream(outputTopic); 
       }
  
  
  }
  </code></pre>
  <p>I am getting the following error message.</p>
  <p>*Caused by: org.apache.kafka.streams.errors.StreamsException: ClassCastException invoking Processor. 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: java.lang.String, and value: java.util.LinkedHashMap.
  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.</p>
  <p>Caused by: java.lang.ClassCastException: java.util.LinkedHashMap cannot be cast to MyApi class*</p>
  <p>Please help. Thanks a lot in advance. Please let me know, if you need more details.</p>