title: How does Kafka Streams windowing work?
body: <p>I am having a hard time comprehending how Windowing works in Kafka Streams. The results don't seem to align with what I have read and understood so far. </p>
  
  <p>I have created a KSQL Stream with a backing topic. one of the 'columns' in the KSQL SELECT statement has been designated as the TIMESTAMP for the topic.</p>
  
  <pre><code>CREATE STREAM my_stream WITH (KAFKA_topic='my-stream-topic', VALUE_FORMAT='json', TIMESTAMP='recorded_timestamp') AS select &lt;select list&gt; PARTITION BY PARTITION_KEY;
  </code></pre>
  
  <p>Records in my-stream-topic are grouped by the key (PARTITION_KEY) and windowed with a hopping window</p>
  
  <pre><code>val dataWindowed: TimeWindowedKStream[String, TopicValue] = builder.stream('my-stream-topic', consumed) 
      .groupByKey(Serialized.`with`(Serdes.String(), valueSerde))
      .windowedBy(TimeWindows.`of`(TimeUnit.MINUTES.toMillis(5)).advanceBy(TimeUnit.MINUTES.toMillis(1)).until(TimeUnit.MINUTES.toMillis(5)))
  </code></pre>
  
  <p>Records are aggregated via</p>
  
  <pre><code>val dataAgg: KTable[Windowed[String], ValueStats] = dataWindowed
      .aggregate(
        new Initializer[TopicStats] {&lt;code omitted&gt;}},
        new Aggregator[String, TopicValue, TopicStats] {&lt;code omitted&gt;}},
        Materialized.`as`[String, TopicStats, WindowStore[Bytes, Array[Byte]]]("time-windowed-aggregated-stream-store")
          .withValueSerde(new JSONSerde[TopicStats])
      )
  
    val topicStats: KStream[String, TopicValueStats] = dataAgg
      .toStream()
      .map( &lt;code omitted for brevity&gt;)
  </code></pre>
  
  <p>I then print to console via</p>
  
  <pre><code>dataAgg.print()
  topicStats.print()
  </code></pre>
  
  <p>The first window that's in the group translates to 7:00 - 7:05</p>
  
  <p>When I examine the records in my-stream-topic via a console consumer I see that there are 2 records that should fall within the above window. However, only 1 of them is picked up by the aggregator.</p>
  
  <p>I thought that the dataAgg windowed KTable would contain 1 record for the grouped key but the aggregate would have used the 2 records to compute the aggregate. The aggregate value printed is incorrect. </p>
  
  <p>What am I missing? </p>