title: Kafka Streams - java.lang.ClassCastException:
body: <p>I have written a Kafka streams application to calculate sum up count of transaction per person.</p>
  <p>Definition of Transaction POJO:</p>
  <pre><code>public class Transaction implements Serializable {
  private static final long serialVersionUID = 4392626942674353834L;
  
  private String name;
  private double amount;
  private Date date;
  
  public String getName() {
      return name;
  }
  
  public void setName(String name) {
      this.name = name;
  }
  
  public double getAmount() {
      return amount;
  }
  
  public void setAmount(double amount) {
      this.amount = amount;
  }
  
  public Date getDate() {
      return date;
  }
  
  public void setDate(Date date) {
      this.date = date;
  }
  
  @Override
  public String toString() {
      return &quot;Transaction{&quot; +
              &quot;name='&quot; + name + '\'' +
              &quot;, amount=&quot; + amount +
              &quot;, date=&quot; + date +
              '}';
  }
  
  @Override
  public boolean equals(Object o) {
      if (this == o) return true;
      if (!(o instanceof Transaction)) return false;
      Transaction that = (Transaction) o;
      return Double.compare(that.getAmount(), getAmount()) == 0 &amp;&amp;
              Objects.equals(getName(), that.getName()) &amp;&amp;
              Objects.equals(getDate(), that.getDate());
  }
  
  @Override
  public int hashCode() {
      return Objects.hash(getName(), getAmount(), getDate());
  }
  }
  </code></pre>
  <p>My Kafka Stream Application looks like below</p>
  <pre><code>public class Deposit {
  public static final Gson gson = new Gson();
  
  public static void main(String[] args) {
  
      var properties = new Properties();
      properties.put(StreamsConfig.APPLICATION_ID_CONFIG, &quot;streams-starter-app&quot;);
      properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, &quot;localhost:9092&quot;);
      properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, &quot;earliest&quot;);
      properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
      properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
      properties.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE);
  
      StreamsBuilder streamsBuilder = new StreamsBuilder();
  
      KStream&lt;String, String&gt; wordCountPipe = streamsBuilder.stream(&quot;balance1&quot;, Consumed.with(Serdes.String(), Serdes.String()));
  
      wordCountPipe.filter((key, value) -&gt; StringUtils.isNoneBlank(value))
              .map((key, value) -&gt; KeyValue.pair(key, toTrans(value)))
              .filter((k, v) -&gt; StringUtils.isNoneBlank(v.getName()) &amp;&amp; Objects.nonNull(v.getAmount()))
              .selectKey((k, v) -&gt; v.getName())
              .groupByKey()
              .count()
              .toStream()
              .to(&quot;result&quot;, Produced.with(Serdes.String(), Serdes.Long()));
  
  
      KafkaStreams kafkaStreams = new KafkaStreams(streamsBuilder.build(), properties);
      kafkaStreams.start();
      Runtime.getRuntime().addShutdownHook(new Thread(kafkaStreams::close));
  
  }
  
  private static Transaction toTrans(String val) {
      try {
          return gson.fromJson(val, Transaction.class);
      } catch (Exception ex) {
          return new Transaction();
      }
  }
  }
  </code></pre>
  <p>Having defined serdes for both key and value during consumption and production. While error comes indicates that Transaction cannot cast to String. A sample message which my stream app consumes with both key and value in String format: <code>Key -&gt; Alex</code>  <code>Value -&gt; {&quot;name&quot;:&quot;Alex&quot;,&quot;amount&quot;:57.0,&quot;date&quot;:&quot;Dec 23, 2020, 9:34:42 PM&quot;}</code>. Feel confused why it's throwing this error, could anyone shed some light upon it?</p>
  <pre><code>    Exception in thread &quot;streams-starter-app-819ba257-564f-44f2-b433-2bbd18cdae0d-StreamThread-1&quot; org.apache.kafka.streams.errors.StreamsException: Exception caught in process. taskId=0_0, processor=KSTREAM-SOURCE-0000000000, topic=balance1, partition=0, offset=68, stacktrace=org.apache.kafka.streams.errors.StreamsException: ClassCastException while producing data to a sink topic. A serializer (key: org.apache.kafka.common.serialization.StringSerializer / value: org.apache.kafka.common.serialization.StringSerializer) is not compatible to the actual key or value type (key type: java.lang.String / value type: pojo.Transaction). Change the default Serdes in StreamConfig or provide correct Serdes via method parameters (for example if using the DSL, `#to(String topic, Produced&lt;K, V&gt; produced)` with `Produced.keySerde(WindowedSerdes.timeWindowedSerdeFrom(String.class))`).
      at org.apache.kafka.streams.processor.internals.SinkNode.process(SinkNode.java:94)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:201)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:180)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:133)
      at org.apache.kafka.streams.kstream.internals.KStreamFilter$KStreamFilterProcessor.process(KStreamFilter.java:43)
      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:806)
      at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:142)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:201)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:180)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:133)
      at org.apache.kafka.streams.kstream.internals.KStreamMap$KStreamMapProcessor.process(KStreamMap.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:806)
      at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:142)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:201)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:180)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:133)
      at org.apache.kafka.streams.kstream.internals.KStreamFilter$KStreamFilterProcessor.process(KStreamFilter.java:43)
      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:806)
      at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:142)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:201)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:180)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:133)
      at org.apache.kafka.streams.kstream.internals.KStreamMap$KStreamMapProcessor.process(KStreamMap.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:806)
      at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:142)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:201)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:180)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:133)
      at org.apache.kafka.streams.kstream.internals.KStreamFilter$KStreamFilterProcessor.process(KStreamFilter.java:43)
      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:806)
      at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:142)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:201)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:180)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:133)
      at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:101)
      at org.apache.kafka.streams.processor.internals.StreamTask.lambda$process$3(StreamTask.java:383)
      at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:806)
      at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:383)
      at org.apache.kafka.streams.processor.internals.AssignedStreamsTasks.process(AssignedStreamsTasks.java:475)
      at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:550)
      at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:802)
      at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:697)
      at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:670)
  Caused by: java.lang.ClassCastException: class pojo.Transaction cannot be cast to class java.lang.String (pojo.Transaction is in unnamed module of loader 'app'; java.lang.String is in module java.base of loader 'bootstrap')
      at org.apache.kafka.common.serialization.StringSerializer.serialize(StringSerializer.java:28)
      at org.apache.kafka.common.serialization.Serializer.serialize(Serializer.java:62)
      at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:176)
      at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:111)
      at org.apache.kafka.streams.processor.internals.SinkNode.process(SinkNode.java:89)
      ... 47 more
  
      at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:400)
      at org.apache.kafka.streams.processor.internals.AssignedStreamsTasks.process(AssignedStreamsTasks.java:475)
      at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:550)
      at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:802)
      at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:697)
      at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:670)
  Caused by: org.apache.kafka.streams.errors.StreamsException: ClassCastException while producing data to a sink topic. A serializer (key: org.apache.kafka.common.serialization.StringSerializer / value: org.apache.kafka.common.serialization.StringSerializer) is not compatible to the actual key or value type (key type: java.lang.String / value type: pojo.Transaction). Change the default Serdes in StreamConfig or provide correct Serdes via method parameters (for example if using the DSL, `#to(String topic, Produced&lt;K, V&gt; produced)` with `Produced.keySerde(WindowedSerdes.timeWindowedSerdeFrom(String.class))`).
      at org.apache.kafka.streams.processor.internals.SinkNode.process(SinkNode.java:94)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:201)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:180)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:133)
      at org.apache.kafka.streams.kstream.internals.KStreamFilter$KStreamFilterProcessor.process(KStreamFilter.java:43)
      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:806)
      at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:142)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:201)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:180)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:133)
      at org.apache.kafka.streams.kstream.internals.KStreamMap$KStreamMapProcessor.process(KStreamMap.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:806)
      at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:142)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:201)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:180)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:133)
      at org.apache.kafka.streams.kstream.internals.KStreamFilter$KStreamFilterProcessor.process(KStreamFilter.java:43)
      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:806)
      at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:142)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:201)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:180)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:133)
      at org.apache.kafka.streams.kstream.internals.KStreamMap$KStreamMapProcessor.process(KStreamMap.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:806)
      at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:142)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:201)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:180)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:133)
      at org.apache.kafka.streams.kstream.internals.KStreamFilter$KStreamFilterProcessor.process(KStreamFilter.java:43)
      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:806)
      at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:142)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:201)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:180)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:133)
      at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:101)
      at org.apache.kafka.streams.processor.internals.StreamTask.lambda$process$3(StreamTask.java:383)
      at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:806)
      at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:383)
      ... 5 more
  Caused by: java.lang.ClassCastException: class pojo.Transaction cannot be cast to class java.lang.String (pojo.Transaction is in unnamed module of loader 'app'; java.lang.String is in module java.base of loader 'bootstrap')
      at org.apache.kafka.common.serialization.StringSerializer.serialize(StringSerializer.java:28)
      at org.apache.kafka.common.serialization.Serializer.serialize(Serializer.java:62)
      at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:176)
      at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:111)
      at org.apache.kafka.streams.processor.internals.SinkNode.process(SinkNode.java:89)
      ... 47 more
  Disconnected from the target VM, address: '127.0.0.1:56761', transport: 'socket'
  
  Process finished with exit code 0
  </code></pre>