pnowojski commented on code in PR #28752:
URL: https://github.com/apache/flink/pull/28752#discussion_r3602386784
##########
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:
Yes, this is doesn't work if there is more than one timer service - which is
rare. Only I think Fabian's new operator is using two timer services?
--
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]
