A windowed state store associates each record key with a time window. In Kafka Streams, windowed aggregations and stream-stream joins materialize their intermediate data this way, so the key identifies the entity while the window identifies the relevant time range. section 7.3.1 section 9.1.1
The public key type is Windowed<K>. It contains the underlying K and a Window object with start and end times. An aggregation therefore produces separate results for the same key in different windows rather than one undifferentiated total. section 9.1.1 section 9.1.6
One-minute temperature windows
A sensor stream can group records by key, create one-minute windows that advance every ten seconds, and aggregate each window: KStream<String, Double> sensors = builder.stream("heat-sensor-input"); sensors.groupByKey() .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(1)) .advanceBy(Duration.ofSeconds(10))) .aggregate(() -> new IotSensorAggregation(tempThreshold), aggregator, Materialized.with(stringSerde, aggregationSerde)); Because the window advances by less than its size, results can overlap. The resulting KTable can be converted with toStream(), then inspected with peek to read windowedKey.window().startTime() and endTime(). section 9.1.1 section 9.1.6
Choose the output representation deliberately. Map windowedKey to windowedKey.key() when downstream records should use the original key, or retain Windowed<String> and configure WindowedSerdes.timeWindowedSerdeFrom. Stream-stream joins likewise require key and value Serdes because join participants are materialized in windowed state stores. section 7.3.1 section 9.1.1
Listing 9.3 Adding windowing to the aggregation for the Iot temperature readings
KStream<String,Double> iotHeatSensorStream =
builder.stream("heat-sensor-input",
Consumed.with(stringSerde, doubleSerde));
iotHeatSensorStream.groupByKey() #1
.windowedBy(TimeWindows.ofSizeWithNoGrace(
Duration.ofMinutes(1)) #2
.advanceBy(Duration.ofSeconds(10)))
.aggregate(
() -> new IotSensorAggregation(tempThreshold), #3
aggregator,
Materialized.with(stringSerde, aggregationSerde))
.toStream().to("sensor-agg-output",
Produced.with(serdeString,
sensorAggregationSerde)) #4
Use the complete temperature-aggregation listing to trace grouping, window creation, aggregation, materialization, and output serialization. section 9.1.1
