windowTimeout(maxSize, duration, fairBackpressure=true) does not propagate backpressure when window closes by timeout #4205
Description
Motivation
When using windowTimeout with fairBackpressure=true (introduced in #3054), backpressure is correctly propagated to the upstream only when the window closes because maxSize is reached.
However, when the window closes because the Duration timeout elapses, the operator continues to request elements from upstream unboundedly, effectively breaking the backpressure chain.
This seems an issue for pipelines where a fast source (e.g. a Kafka consumer) feeds into a slow or intermittently unavailable sink (e.g. a message broker under pressure).
In such scenarios, the intent of using fairBackpressure=true is to slow down the source when the downstream cannot keep up. But if the timeout triggers before maxSize is reached (which is exactly what happens when the sink is slow — fewer elements are processed per window), the operator falls back to unbounded requesting behavior, defeating the purpose of the flag.
Desired solution
When fairBackpressure=true, the operator should apply the same backpressure behavior regardless of whether the window closes due to maxSize being reached or the Duration timeout elapsing.
Specifically, when a window closes by timeout and there is no demand from downstream for the next window, the operator should stop requesting elements from upstream — exactly as it already does when the window closes by maxSize.