A tumbling window is a fixed-duration, non-overlapping stream-processing window. Its advance time equals its size, so each event belongs to one distinct time interval rather than overlapping results. section 9.1.2
In Kafka Streams, create one with TimeWindows.ofSizeWithNoGrace. Omitting advanceBy makes the advance time equal the window size; retaining advanceBy with the same duration is equivalent but less explicit. section 9.1.2
One-minute sensor aggregation
KStream<String,Double> sensors = builder.stream("heat-sensor-input", Consumed.with(stringSerde, doubleSerde)); sensors.groupByKey() .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(1))) .aggregate(() -> new IotSensorAggregation(tempThreshold), aggregator, Materialized.with(stringSerde, aggregationSerde)) .toStream() .to("sensor-agg-output", Produced.with(windowedSerdes, sensorAggregationSerde)); This produces distinct one-minute aggregation windows for each sensor key. The resulting key is Windowed<String> unless the stream maps it back to the underlying key. section 9.1.2 section 9.1.1
Keep these implementation details in view:
- Use a WindowedSerdes.timeWindowedSerdeFrom(String.class, 60_000L) when retaining the windowed key in the output topic. section 9.1.1
- Map windowedKey to windowedKey.key() when downstream consumers need the original sensor key without window metadata. section 9.1.1
- Treat TimeWindows as epoch-aligned logical intervals; record timestamps determine window membership, not each window's start or end. section 9.1.5
Listing 9.6 Specifying a tumbling window by removing the advanceBy method
Serde<Windowed<String>> windowedSerdes =
WindowedSerdes.timeWindowedSerdeFrom(String.class,
60_000L
);
KStream<String,Double> iotHeatSensorStream =
builder.stream("heat-sensor-input",
Consumed.with(stringSerde, doubleSerde));
iotHeatSensorStream.groupByKey()
.windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(1)) #1
aggregate(() -> new IotSensorAggregation(tempThreshold),
aggregator,
Materialized.with(stringSerde, aggregationSerde))
.toStream()
.to("sensor-agg-output",
Produced.with(windowedSerdes, sensorAggregationSerde))
Inspect the complete Kafka Streams configuration for a one-minute tumbling window with the advanceBy call removed. section 9.1.2
