rkhachatryan commented on code in PR #28752:
URL: https://github.com/apache/flink/pull/28752#discussion_r3597780794
##########
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:
Sorry but I don't understand this loop :)
First, once fullyAdvance is false, we still iterate, but just do nothing?
Why not just break?
Second, if the iteration order is the same next round (likely) then we won't
update min wm if there are more services with lower wm after the interrupted
one.
I think in practice renders the feature as no-op with high probability.
--
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]
