title: Kafka Stream creates new aggregation even before window is closed
body: <p>I am using Kafka stream for windowed aggregation. Here is the logic:</p>
  
  <pre class="lang-java prettyprint-override"><code>    KStream&lt;String, String&gt; stream = builder.stream(topics, Consumed.with(stringSerde, stringSerde));
      Stream.groupByKey()
      .windowedBy(TimeWindows.of(Duration.ofMinutes(2)).grace(Duration.ZERO))
      .aggregate(()-&gt; "",
           (key, word, aggregate) -&gt; {
             logger.info("(key, word, aggregate): ({}, {}, {})" , key, word, aggregate);
             aggregate = aggregate+ "-" + word;
             return aggregate;
           }, Materialized.with(stringSerde, stringSerde))
       .toStream((k, v) -&gt; k.key())
       .foreach((k, v) -&gt; logger.info("Aggregated to Stream. (k,v): ({}, {})" , k, v));
  </code></pre>
  
  <p>Though this works for most of the times, I observed these issues:</p>
  
  <ol>
  <li>Aggregation is prematurely flushed</li>
  <li>New aggregation bucket gets created even before the window is closed</li>
  </ol>
  
  <p>These issues are evident by these logs (marked lines):</p>
  
  <pre><code>[2019-08-14 14:10:38,855] [ INFO] [prc-client-StreamThread-1] Aggregation (118) - (key, word, aggregate): (1, a, )
  
  (1)[2019-08-14 14:11:24,503] [ INFO] [prc-client-StreamThread-1] Aggregation (124) - Aggregated to Stream. (k,v): (1, -a)
  
  [2019-08-14 14:11:27,735] [ INFO] [prc-client-StreamThread-1] Aggregation (118) - (key, word, aggregate): (1, b, -a)
  [2019-08-14 14:11:43,298] [ INFO] [prc-client-StreamThread-1] Aggregation (118) - (key, word, aggregate): (1, f, -a-b)
  [2019-08-14 14:11:59,373] [ INFO] [prc-client-StreamThread-1] Aggregation (118) - (key, word, aggregate): (1, b, -a-b-f)
  
  (2)[2019-08-14 14:12:14,196] [ INFO] [prc-client-StreamThread-1] Aggregation (118) - (key, word, aggregate): (1, r, )
  
  [2019-08-14 14:13:24,808] [ INFO] [prc-client-StreamThread-1] Aggregation (124) - Aggregated to Stream. (k,v): (1, -a-b-f-b)
  [2019-08-14 14:13:24,808] [ INFO] [prc-client-StreamThread-1] Aggregation (124) - Aggregated to Stream. (k,v): (1, -r)
  </code></pre>
  
  <p>Is there is anyway to address these issues?</p>