title: OutOfMemoryError when restart my Kafka Streams appplication
body: <p>I have a Kafka Streams app(Kafka Streams 2.1 + Kafka broker 2.0) which does a aggregation based on TimeWindows, and I use the suppress operator to supress the result's output. </p>
  
  <p>Everything works well until I restart my app, it will reset the offset of KTABLE-SUPPRESS-STATE-STORE to 0 to restore the suppression state, as expected. But each time I restart it, it will throw an <code>OutOfMemoryError</code>, I thought maybe the heap size is not enough, so I use a larger <code>Xmx/Xms</code>, it works one or two restart, and then the <code>OutOfMemoryError</code> comes back again. Now the <code>Xmx</code> is about 20G now, I think something is not right here.</p>
  
  <p>The code snippet:</p>
  
  <pre><code>TimeWindows windows = TimeWindows.of(windowSize).until(retentionHours.toMillis()).grace(graceHours);
  
  KTable&lt;Windowed&lt;String&gt;, MyStatistics&gt; kTable = groupedBySerialNumber
                  .windowedBy(windows)
                  .aggregate(MyStatistics::new,
                      (sn, resList, stats) -&gt; stats.addResources(resList).updateSN(sn),
                      Materialized.with(Serdes.String(), ArchiveSerdes.resourceStatistics()))
                  .suppress(Suppressed.untilTimeLimit(timeToWait, Suppressed.BufferConfig.maxBytes(bufferMaxBytes)));
  </code></pre>
  
  <p>And I find that the key of record in KTABLE-SUPPRESS-STATE-STORE is something like 1234567j�P, which is not readable, but I guess it's generated by combine the SN and window, I think this will make KTABLE-SUPPRESS-STATE-STORE redundent, because each SN will have multi records for each window.</p>
  
  <p>I have two questions:</p>
  
  <ol>
  <li>If the <code>OutOfMemoryError</code> indicates a small heap size or not, if so, how to limit the rate, if not, what does it means?</li>
  <li>The key for KTABLE-SUPPRESS-STATE-STORE is defined by which API, how or should can I control it?</li>
  </ol>
  
  <p>Thanks!</p>
  
  <p><strong>Edit in 2019/4/16</strong></p>
  
  <p>The error stacktrace is:</p>
  
  <pre><code>java.lang.OutOfMemoryError: Java heap space        
          at java.nio.HeapByteBuffer.&lt;init&gt;(HeapByteBuffer.java:57)
          at java.nio.ByteBuffer.allocate(ByteBuffer.java:335)        
          at org.apache.kafka.common.memory.MemoryPool$1.tryAllocate(MemoryPool.java:30)        
          at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:112)        
          at org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:381)
          at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:342)        
          at org.apache.kafka.common.network.Selector.attemptRead(Selector.java:609)        
          at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:541)
          at org.apache.kafka.common.network.Selector.poll(Selector.java:467)        
          at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:535)        
          at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:265)
          at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:236)
          at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1243)
          at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1188)
          at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1164)
          at org.apache.kafka.streams.processor.internals.StoreChangelogReader.restore(StoreChangelogReader.java:88)
          at org.apache.kafka.streams.processor.internals.TaskManager.updateNewAndRestoringTasks(TaskManager.java:321)
          at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:839)
          at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:777)
          at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:747)
  </code></pre>