I'm looking into making a feature for flink related to watermarks and am digging into the inner watermark mechanisms, specifically with idleness. I'm familiar with idleness, but digging into the root code I can only get to where idlenessTimeout gets set in WatermarkStrategyWithIdleness.java.
But what I'm looking for the pieces beyond that. If I set the idleness to 500 milliseconds, where in the code does it actually go "I haven't seen a message in 500 milliseconds. I'm setting this stream to idle."? The reason being that what I'm thinking of would need to be able to see if any streams are marked idle, and if so react accordingly. Thanks for any help in advance.
