title: How to use custom serializers with KTable in Kafka streams?
body: <p>I am facing some errors during serialization of <code>KStream</code> to <code>KTable</code> when I introduced the <code>groupBy</code> to my <code>KStream</code>. As I understood, once you have an <code>aggregate</code> or <code>reduce</code> on a <code>KStream</code>, Kafka tries to transform it to a <code>KTable</code> due to necessary shuffle and because of this Kafka has to serialize the records again. So, my original <code>KStream</code> was just mapping the records from <code>JSON</code> to <code>AVRO</code> like this, and it is working fine.</p>
  <pre><code>    @StreamListener(&quot;notification-input-channel&quot;)
      @SendTo(&quot;notification-output-avro-channel&quot;)
      public KStream&lt;String, NotificationAvro&gt; process(KStream&lt;String, PosInvoice&gt; input) {
          log.info(&quot;received PosInvoice JSON: {}&quot;, input);
          KStream&lt;String, NotificationAvro&gt; notificationAvroKStream = input
                  .filter((k, v) -&gt; v.getCustomerType().equalsIgnoreCase(PRIME))
                  .mapValues(v -&gt; recordBuilder.getNotificationAvro(v));
          notificationAvroKStream.foreach((k, v) -&gt; log.info(String.format(&quot;Notification avro - key: %s, value: %s&quot;, k, v)));
          return notificationAvroKStream;
      }
  </code></pre>
  <p>then I introduced the <code>groupByKey</code> and <code>reduce</code> and I realized that it transforms to a <code>KTable</code> and hence it needed <code>Serdes</code> on the <code>application.yaml</code> file. But unfortunately I cannot configure the default <code>Serdes</code> because I have other types of serialization. Hence I decided to serialize on the <code>KTable</code> topology. I am trying to implement this solution <a href="https://stackoverflow.com/q/54062509/2096986">based on this answer</a>.</p>
  <p>The part of the code that I try to materialize with my custom serdes is not working properly (<code>Materialized.with(CustomSerdes.String(), CustomSerdes.NotificationAvro())</code>). First, I don't think I need <code>KeyValueBytesStoreSupplier storeSupplier = Stores.inMemoryKeyValueStore(&quot;mystore&quot;);</code>, but without this it also does not work and I cannot find a materialized that is not <code>KeyValueBytes...</code> where I can define my serdes <code>CustomSerdes.String(), CustomSerdes.NotificationAvro()</code>.</p>
  <p>According to the answer that I mentioned on the link, they also use a <code>final StreamsBuilder builder = new StreamsBuilder();</code>. But since I am computing it using <code>spring-kafka</code> I don't have this option, or if I have I don't know how to use.</p>
  <pre><code>@Service
  @Slf4j
  @EnableBinding(PosListenerJsonAvroBinding.class)
  public class NotificationJsonAvroProcessorService {
      @Autowired
      RecordBuilder recordBuilder;
  
      @StreamListener(&quot;notification-input-channel&quot;)
      @SendTo(&quot;notification-output-avro-channel&quot;)
      public KStream&lt;String, NotificationAvro&gt; process(KStream&lt;String, PosInvoice&gt; input) {
          log.info(&quot;received PosInvoice JSON: {}&quot;, input);
          KStream&lt;String, NotificationAvro&gt; notificationAvroKStream = input
                  .filter((k, v) -&gt; v.getCustomerType().equalsIgnoreCase(PRIME))
                  .map((k, v) -&gt; new KeyValue&lt;&gt;(v.getCustomerCardNo(), recordBuilder.getNotificationAvro(v)));
          notificationAvroKStream.foreach((k, v) -&gt; log.info(String.format(&quot;Notification avro - key: %s, value: %s&quot;, k, v)));
  
          // *********************************************
          // IS THERE A KeyValueStoreSupplier THAT I CAN PASS ALSO MY SERDES INSTEAD OF Bytes?
          // KeyValueBytesStoreSupplier storeSupplier = Stores.inMemoryKeyValueStore(&quot;mystore&quot;);
          KTable&lt;String, NotificationAvro&gt; convertedTable = notificationAvroKStream
                  .toTable(
                          // *********************************************
                          // HOW TO MATERIALIZE KTABLE VALUES WITH SERDES ?
                          Materialized
                                  // .as(storeSupplier) // this is not necessary
                                  .with(CustomSerdes.String(), CustomSerdes.NotificationAvro())
                          // *********************************************
                  )
                  .groupBy((cardNo, notificationAvro) -&gt; KeyValue.pair(cardNo, notificationAvro))
                  .reduce(
                          (aggValue, newValue) -&gt; {
                              newValue.setTotalLoyaltyPoints(newValue.getEarnedLoyaltyPoints() + aggValue.getTotalLoyaltyPoints());
                              return newValue;
                          },
                          (aggValue, oldValue) -&gt; oldValue
                  );
          KStream&lt;String, NotificationAvro&gt; notificationAggAvroKStream = convertedTable.toStream();
          notificationAggAvroKStream.foreach((k, v) -&gt; log.info(String.format(&quot;Notification agg avro - key: %s, value: %s&quot;, k, v)));
  
          return notificationAggAvroKStream;
      }
  }
  </code></pre>
  <p>the custom serdes:</p>
  <pre><code>@Service
  public class CustomSerdes extends Serdes {
      private static final String schema_registry_url = &quot;http://localhost:8081&quot;;
      private final static Map&lt;String, String&gt; serdeConfig = Collections
              .singletonMap(&quot;schema.registry.url&quot;, schema_registry_url);
      public static Serde&lt;NotificationAvro&gt; NotificationAvro() {
          final Serde&lt;NotificationAvro&gt; notificationAvroSerde = new SpecificAvroSerde&lt;&gt;();
          notificationAvroSerde.configure(serdeConfig, false);
          return notificationAvroSerde;
      }
  }
  </code></pre>
  <p>and the error:</p>
  <blockquote>
  <p>Exception in thread
  &quot;NotificationJsonAvroProcessorService-process-applicationId-3e262d96-19ca-438d-a2b8-9d3c2e9bb4ab-StreamThread-1&quot;
  org.apache.kafka.streams.errors.StreamsException: ClassCastException
  while producing data to topic
  NotificationJsonAvroProcessorService-process-applicationId-KTABLE-AGGREGATE-STATE-STORE-0000000010-repartition.
  A serializer (key:
  org.apache.kafka.common.serialization.StringSerializer / value:
  org.apache.kafka.streams.kstream.internals.ChangedSerializer) is not
  compatible to the actual key or value type (key type: java.lang.String
  / value type: org.apache.kafka.streams.kstream.internals.Change).
  Change the default Serdes in StreamConfig or provide correct Serdes
  via method parameters (for example if using the DSL, <code>#to(String topic, Produced&lt;K, V&gt; produced)</code> with
  <code>Produced.keySerde(WindowedSerdes.timeWindowedSerdeFrom(String.class))</code>).
  ...
  ...
  Caused by: java.lang.ClassCastException: class
  com.github.felipegutierrez.explore.spring.model.NotificationAvro
  cannot be cast to class java.lang.String
  (com.github.felipegutierrez.explore.spring.model.NotificationAvro is
  in unnamed module of loader 'app'; java.lang.String is in module
  java.base of loader 'bootstrap')</p>
  </blockquote>