pnowojski commented on code in PR #28752:
URL: https://github.com/apache/flink/pull/28752#discussion_r3603838893
##########
flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/InternalTimeServiceManagerImpl.java:
##########
@@ -215,15 +220,51 @@ public void advanceWatermark(Watermark watermark) throws
Exception {
}
}
+ @Override
+ public void configureIntermediateWatermarkInterval(Duration interval) {
+ this.intermediateWatermarkIntervalMs = interval.toMillis();
+ }
+
@Override
public boolean tryAdvanceWatermark(
Watermark watermark, ShouldStopAdvancingFn shouldStopAdvancingFn)
throws Exception {
+ maybeScheduleIntermediateWatermarkNudge();
+ boolean fullyAdvanced = true;
for (InternalTimerServiceImpl<?, ?> service : timerServices.values()) {
- if (!service.tryAdvanceWatermark(watermark.getTimestamp(),
shouldStopAdvancingFn)) {
- return false;
+ // Once one service is interrupted, stop attempting to fire on the
remaining ones this
+ // round, but still fold their (possibly stale, from an earlier
round) reachedWatermark
+ // into the min below: a service we don't retry this round may be
even further behind.
+ if (fullyAdvanced) {
+ fullyAdvanced =
+ service.tryAdvanceWatermark(
+ watermark.getTimestamp(),
shouldStopAdvancingFn);
}
}
Review Comment:
After giving it the second thought, instead of starting from the same timer
service always, I will implement some round robin here. One interruption will
only handle one service, without advancing the `minReachedWatermark`, but every
interruption will be alternating between different timer services and
subsequently advancing the `minReachedWatermark` across multiple calls. Given
that the interruptions will happen every ~200ms (default auto watermarking
interval), that should be perfectly fine. WDYT?
--
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]
