title: SerializationException: Size of data received by IntegerDeserializer is not 4
body: <p>I am getting this error when executing kstreams to get the aggregated count.</p>
  
  <pre><code>Exception in thread "KStreamThread-1" org.apache.kafka.streams.errors.ProcessorStateException: task [1_2] Failed to flush state store KSTREAM-AGGREGATE-STATE-STORE-0000000002
  at org.apache.kafka.streams.processor.internals.ProcessorStateManager.flush(ProcessorStateManager.java:220)
  at org.apache.kafka.streams.processor.internals.AbstractTask.flushState(AbstractTask.java:204)
  at org.apache.kafka.streams.processor.internals.StreamTask.flushState(StreamTask.java:491)
  at org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:443)
  at org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:431)
  at org.apache.kafka.streams.processor.internals.AssignedTasks.commit(AssignedTasks.java:346)
  at org.apache.kafka.streams.processor.internals.TaskManager.commitAll(TaskManager.java:405)
  at org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:1029)
  at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:883)
  at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:777)
  at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:747)
  Caused by: org.apache.kafka.common.errors.SerializationException: Size of data received by IntegerDeserializer is not 4
  </code></pre>
  
  <p>Here is the code I am executing</p>
  
  <pre><code> final KStream&lt;String, EventsAvro&gt; stream = builder.stream("events_topic");
  
      KStream&lt;Integer, Long&gt; events = stream.map((k, v) -&gt; new KeyValue&lt;Integer, Long&gt;(v.getPageId(), v.getUserId()));
  
      KGroupedStream&lt;Integer, Long&gt; groupedStream = events.groupByKey(Grouped.with(Serdes.Integer(), Serdes.Long()));
      KTable&lt;Windowed&lt;Integer&gt;, Long&gt; windowedCount = groupedStream
                  .windowedBy(TimeWindows.of(Duration.ofMinutes(10)).grace(ofMillis(5L)))
                  .count()
                  .suppress(Suppressed.untilWindowCloses(unbounded()));
  
          windowedCount.toStream()
                  .map((key, value) -&gt; new KeyValue&lt;&gt;(key.key().intValue(),value.longValue()))
                  .to("test_topic",Produced.with(Serdes.Integer(),Serdes.Long()));
  </code></pre>
  
  <p>It used to run fine before I added this suppress code.Any idea?</p>