title: KafkaStreams How to specify Serdes in stream aggregation?
body: <p>I am working on a Kafka streams application and I have some trouble figuring out how to make an aggregation work.</p>
  
  <p>I have a KStream <code>bankTransactions</code> where the keys are of type <code>String</code> and the values of type <code>JsonNode</code> so I configured my app's Serdes with </p>
  
  <pre><code>// Definition of the different Serdes used in the streams
  final Serde&lt;String&gt; stringSerde = Serdes.String();
  final Serde&lt;JsonNode&gt; jsonSerde = new JsonSerde();
  final Serde&lt;Long&gt; longSerde = Serdes.Long();
  
  config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, stringSerde.getClass().getName());
  config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, jsonSerde.getClass().getName());
  </code></pre>
  
  <p>I want to aggregate the values in a <code>KTable&lt;String, Long&gt;</code> where the keys will be the same but the values will be <code>Long</code>s extracted from my Json.</p>
  
  <p>So firstly I wrote:</p>
  
  <pre><code>KTable&lt;String, Long&gt; totalBalances = bankTransactions
          .groupByKey()
          .aggregate(
                  () -&gt; 0L,
                  (key, transaction, balance) -&gt; (Long)((Long)balance + transaction.get("amount").asLong()),
                  Materialized.as("bank-total-balance")
          );
  </code></pre>
  
  <p>And I get the following error at runtime:</p>
  
  <pre><code>Caused by: org.apache.kafka.streams.errors.StreamsException:
  A serializer (value: org.apache.kafka.connect.json.JsonSerializer) is not compatible to
  the actual value type (value type: java.lang.Long).
  Change the default Serdes in StreamConfig or provide correct Serdes via method parameters.
  </code></pre>
  
  <p>I understand that Kafka is complaining because I'm trying to use the default Json serdes to serialize a <code>Long</code>. So reading from <a href="https://docs.confluent.io/current/streams/developer-guide/dsl-api.html#aggregating" rel="nofollow noreferrer">confluent's doc</a> I tried this</p>
  
  <pre><code>KTable&lt;String, Long&gt; totalBalances = bankTransactions
          .groupByKey()
          .aggregate(
                  () -&gt; 0L,
                  (key, transaction, balance) -&gt; (Long)((Long)balance + transaction.get("amount").asLong()),
                  Materialized.as("bank-total-balance").withValueSerde(Serdes.Long())
          );
  </code></pre>
  
  <p>But then I get an error at compilation:</p>
  
  <pre><code>Error:(121, 89) java: incompatible types:
  org.apache.kafka.common.serialization.Serde&lt;java.lang.Long&gt; cannot be converted
  to org.apache.kafka.common.serialization.Serde&lt;java.lang.Object&gt;
  </code></pre>
  
  <p>I tried different way to write this code (e.g. using <code>Serdes.long()</code> instead of my <code>longSerdes</code>, trying to parametrize the types of <code>Materialize</code> and even trying to write my initializer and aggregator as function, Java 7 style) but I can't figure out what I am doing wrong.</p>
  
  <p>So my question is simple: How to I properly specify the Serdes that <code>aggregate</code> should use when they are not the defaults Serdes?</p>