Implement Kafka Streams windowed aggregations with grace period configuration

Steps

1. Group a KStream by key using groupByKey or groupBy with appropriate serdes.
2. Define a window using TimeWindows.ofSizeWithNoGrace(Duration) or, for late-arrival tolerance, TimeWindows.ofSizeAndGrace(windowSize, gracePeriod).
3. Apply an aggregate using windowedBy(windows).aggregate(initializer, aggregator, Materialized.as(...)).
4. The grace period controls how long Kafka Streams accepts late records into already-emitted windows; records arriving after the grace period are dropped.
5. Configure suppress(Suppressed.untilWindowCloses(unbounded())) if you want to emit only the final result after the window and grace period both close.
6. Verify late record handling by producing out-of-order records to the input topic and inspecting whether they are included in the window aggregate.