SEPURI-SAI-KRISHNA commented on PR #29039:
URL: https://github.com/apache/flink/pull/29039#issuecomment-5662202012
Small one on the lower bound.
The countdown cannot start until a periodic probe sees silence, so detection
is never within one interval of the timeout either. Driving
`WatermarksWithIdleness` with a manual clock and recording when `markIdle()`
fires, across timeout/interval ratios from 0.2 to 12 and every probe phase:
```
earliest markIdle: idleTimeout + 1.00 x interval (timeout=999)
latest markIdle: idleTimeout + 3.00 x interval (timeout=1000)
```
Your upper bound matches exactly. "no earlier than idleTimeout" is correct
but a full interval loose, "no earlier than idleTimeout plus one watermark
interval" held everywhere I measured.
<details>
<summary>Harness</summary>
`javac -cp flink-core/target/classes IdlenessTiming.java && java -cp
.:flink-core/target/classes IdlenessTiming`
```java
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.util.clock.RelativeClock;
import java.time.Duration;
/** Measures when WatermarksWithIdleness actually calls markIdle() after the
last event. */
public class IdlenessTiming {
static long detect(long timeout, long interval, long firstProbeOffset) {
long[] now = {0};
RelativeClock clock = new RelativeClock() {
public long relativeTimeNanos() { return now[0]; }
public long relativeTimeMillis() { return now[0] / 1_000_000L; }
};
boolean[] idle = {false};
WatermarkOutput out = new WatermarkOutput() {
public void emitWatermark(Watermark w) {}
public void markIdle() { idle[0] = true; }
public void markActive() { idle[0] = false; }
};
WatermarksWithIdleness<Long> g = new WatermarksWithIdleness<>(
new AscendingTimestampsWatermarks<>(),
Duration.ofNanos(timeout), clock);
long ts = 0;
for (int i = 0; i < 3; i++) { // events flowing, probes
in between
g.onEvent(1L, ++ts, out);
now[0] += interval;
g.onPeriodicEmit(out);
}
g.onEvent(1L, ++ts, out); // last event, at t = 0
long base = now[0];
for (long t = base + firstProbeOffset; ; t += interval) {
now[0] = t;
g.onPeriodicEmit(out);
if (idle[0]) return t - base; // delay after the last
event
}
}
public static void main(String[] args) {
final long I = 1000; // watermark interval
double min = Double.MAX_VALUE, max = -1;
long tMin = 0, tMax = 0;
for (long timeout = 200; timeout <= 12000; timeout++) {
for (long phase = 1; phase <= I; phase++) {
double over = (detect(timeout, I, phase) - timeout) /
(double) I;
if (over < min) { min = over; tMin = timeout; }
if (over > max) { max = over; tMax = timeout; }
}
}
System.out.printf("earliest markIdle: idleTimeout + %.2f x interval
(timeout=%d)%n", min, tMin);
System.out.printf("latest markIdle: idleTimeout + %.2f x interval
(timeout=%d)%n", max, tMax);
}
}
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
