What problem does suppress(Suppressed.untilWindowCloses(...)) solve in windowed aggregations, and what are its operational requirements and caveats?

## The problem A windowed aggregation returns a KTable keyed by `Windowed<K>` and, like any KTable, emits an update on **every** input record. So a 1-minute count window for a key receiving 50 records emits ~50 intermediate values, all but the last being non-final 'partial counts'. For use cases that want **one row per window** — hourly reports, threshold alerts, billing — these intermediates are noise and can trigger false alerts or duplicate downstream work. The record cache reduces but doesn't eliminate them (it flushes on size/time, not on window close). ## What suppress(untilWindowCloses) does `KTable.suppress(Suppressed.untilWindowCloses(bufferConfig))` is a dedicated operator that **buffers** every update for a window and forwards **only the final value** once the window is provably complete — i.e., when **stream time** (the max observed event timestamp) passes **window end + grace period**. After that point, no more records can legally enter the window (late ones are already dropped), so the buffered value is final and is emitted exactly once. All intermediate updates are swallowed. ## BufferConfig — mandatory and a real trade-off You must pass a `Suppressed.BufferConfig`: - `unbounded()` (a.k.a. `maxBytes(Long.MAX_VALUE)`): never drops, but unbounded memory — risk of OOM with many open windows/keys. - `maxBytes(n)` / `maxRecords(n)`: cap the buffer, then choose overflow behavior: - `.emitEarlyWhenFull()`: when full, emit some buffered (possibly non-final) results to free space — bounds memory but **breaks the strict final-only guarantee**. - `.shutDownWhenFull()`: keep strict semantics; if the buffer overflows, the application **throws and stops** (you must provision more memory or reduce cardinality). ## The stream-time trap Suppression releases results based on **event/stream time, not wall-clock**. Stream time only advances when **new records arrive**. So if a key/partition goes idle, the final result for its last window is **never emitted** until more data bumps stream time past the close threshold. This surprises people expecting timely emission on low-traffic topics. Mitigations: ensure steady traffic, use idle-partition handling, or accept the latency. (`max.task.idle.ms` affects how stream time advances across partitions but doesn't manufacture time from nothing.) ## State, restart, and recovery The suppression buffer is a stateful store backed by a changelog, so buffered-but-not-yet-emitted windows survive restarts — but they add to memory footprint and restore time, and the buffer must be sized for peak open-window cardinality. ## untilTimeLimit vs untilWindowCloses `Suppressed.untilTimeLimit(Duration, bufferConfig)` is a **rate limiter**: it emits at most one update per key per time bound, still emitting intermediates — it does NOT give final-only semantics and works on non-windowed KTables too. Only `untilWindowCloses(...)` gives 'final result per window', and it requires a windowed KTable. ## Edge cases - suppress sits **after** the aggregation in the topology; it does not change the aggregate, only when results are emitted. - Combining suppress with `emitEarlyWhenFull()` means downstream may still see a non-final value, so consumers should treat keys idempotently. - Session windows can be suppressed too, but merges complicate which final value emits; reason carefully.