title: Kafka streams aggregate throwing Exception
body: <p>this is my Kafka streams code which uses sliding window to sum all the integer data in the time window. </p>
  
  <pre><code>public class KafkaWindowingLIS {
      public static void main(String[] args) {
          Properties config = new Properties();
          config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafkahost:9092");
          config.put(StreamsConfig.APPLICATION_ID_CONFIG, "kafka-windowing-lis");
          config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
          config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
          config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
  
          Integer uid = 1;
          long tenSeconds = 1000 * 10;
  
          StreamsBuilder builder = new StreamsBuilder();
          KStream&lt;String, String&gt; dataStream = builder.stream("kafka-windowing-lis");
          KStream&lt;Integer, Integer&gt; integerKStream = dataStream
                  .filter((key, val) -&gt; {               //Filter only numbers from Stream
                      try {
                          Integer.parseInt(val);
                          return true;
                      } catch (NumberFormatException e) {
                          return false;
                      }
                  })
                  .map((key, val) -&gt; new KeyValue&lt;Integer, Integer&gt;(uid, Integer.parseInt(val)));
  
          TimeWindowedKStream&lt;Integer, Integer&gt; timeWindowedKStream = integerKStream
                  .groupBy((k, v) -&gt; k, Serialized.with(Serdes.Integer(), Serdes.Integer()))     //Killed my time
                  .windowedBy(TimeWindows.of(tenSeconds));
  
          timeWindowedKStream.aggregate(
                  () -&gt; 0,
                  (key, value, aggregate) -&gt; value + aggregate)
                  .toStream().print(Printed.toSysOut());
  
  
          KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), config);
  //        kafkaStreams.cleanUp();
          kafkaStreams.start();
          Runtime.getRuntime().addShutdownHook(new Thread(kafkaStreams::close));
      }
  }
  </code></pre>
  
  <p>I am getting the following exception :</p>
  
  <pre><code>Caused by: org.apache.kafka.streams.errors.StreamsException: A serializer (org.apache.kafka.common.serialization.StringSerializer) is not compatible to the actual value type (value type: java.lang.Integer). Change the default Serdes in StreamConfig or provide correct Serdes via method parameters.
      at org.apache.kafka.streams.state.StateSerdes.rawValue(StateSerdes.java:204)
      at org.apache.kafka.streams.state.internals.MeteredWindowStore.put(MeteredWindowStore.java:138)
      at org.apache.kafka.streams.processor.internals.ProcessorContextImpl$WindowStoreReadWriteDecorator.put(ProcessorContextImpl.java:533)
      at org.apache.kafka.streams.kstream.internals.KStreamWindowAggregate$KStreamWindowAggregateProcessor.process(KStreamWindowAggregate.java:138)
      at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:117)
      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:87)
      at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:366)
      ... 5 more
  Caused by: java.lang.ClassCastException: java.lang.Integer cannot be cast to java.lang.String
      at org.apache.kafka.common.serialization.StringSerializer.serialize(StringSerializer.java:28)
      at org.apache.kafka.streams.state.internals.ValueAndTimestampSerializer.serialize(ValueAndTimestampSerializer.java:59)
      at org.apache.kafka.streams.state.internals.ValueAndTimestampSerializer.serialize(ValueAndTimestampSerializer.java:50)
      at org.apache.kafka.streams.state.internals.ValueAndTimestampSerializer.serialize(ValueAndTimestampSerializer.java:27)
      at org.apache.kafka.streams.state.StateSerdes.rawValue(StateSerdes.java:192)
      ... 14 more
  </code></pre>
  
  <p>I looked for the solution in various pages, looks like I am missing something in this.
  Any inputs would be greatly welcomed.
  Thank you.</p>