rkhachatryan commented on code in PR #28752:
URL: https://github.com/apache/flink/pull/28752#discussion_r3597744210
##########
flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java:
##########
@@ -611,6 +611,29 @@ public class CheckpointingOptions {
+ " For this feature to be enabled, it
must be also supported by the operator."
+ " Currently this is supported by all
TableStreamOperators and CepOperator.");
+ /**
+ * Controls whether an intermediate watermark is emitted while a watermark
advance is
+ * interrupted before completing (see {@link
#ENABLE_UNALIGNED_INTERRUPTIBLE_TIMERS}). Has no
+ * effect unless interruptible timers are enabled. The emission interval
is governed by {@link
+ * PipelineOptions#AUTO_WATERMARK_INTERVAL}, the same as regular periodic
watermark generation,
+ * since both have comparable performance implications.
+ */
+ @Experimental
+ public static final ConfigOption<Boolean>
+ UNALIGNED_INTERRUPTIBLE_TIMERS_EMIT_INTERMEDIATE_WATERMARKS =
+ ConfigOptions.key(
+
"execution.checkpointing.unaligned.interruptible-timers.emit-intermediate-watermarks")
+ .booleanType()
+ .defaultValue(true)
Review Comment:
I think this should be disabled by default for 1 release to minimize
potential impact.
##########
flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java:
##########
@@ -395,9 +396,17 @@ public void open() throws Exception {
&& areInterruptibleTimersConfigured()
&& getTimeServiceManager().isPresent()) {
LOG.info("Interruptible timers enabled for {}",
getClass().getSimpleName());
+ InternalTimeServiceManager<?> timeServiceManager =
getTimeServiceManager().get();
+ if (getContainingTask()
+ .getJobConfiguration()
+ .get(
+ CheckpointingOptions
+
.UNALIGNED_INTERRUPTIBLE_TIMERS_EMIT_INTERMEDIATE_WATERMARKS)) {
+ timeServiceManager.configureIntermediateWatermarkInterval(
+
Duration.ofMillis(getExecutionConfig().getAutoWatermarkInterval()));
+ }
Review Comment:
What about V2?
##########
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 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.
##########
flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/MailboxWatermarkProcessor.java:
##########
@@ -74,7 +76,16 @@ private void emitWatermarkInsideMailbox() throws Exception {
maxInputWatermark, mailboxExecutor::shouldInterrupt)) {
// In case output watermark has fully progressed emit it
downstream.
output.emitWatermark(maxInputWatermark);
- } else if (!progressWatermarkScheduled) {
+ return;
+ }
+ long reachedWatermark =
internalTimeServiceManager.getReachedWatermark();
+ if (reachedWatermark > lastEmittedIntermediateWatermark) {
+ // Firing was interrupted before completing; surface the progress
made so far instead
+ // of leaving watermark advancement stalled until the whole drain
finishes.
+ lastEmittedIntermediateWatermark = reachedWatermark;
+ output.emitWatermark(new Watermark(reachedWatermark));
Review Comment:
Can we honor the config option here?
##########
flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/MailboxWatermarkProcessor.java:
##########
@@ -74,7 +76,16 @@ private void emitWatermarkInsideMailbox() throws Exception {
maxInputWatermark, mailboxExecutor::shouldInterrupt)) {
// In case output watermark has fully progressed emit it
downstream.
output.emitWatermark(maxInputWatermark);
- } else if (!progressWatermarkScheduled) {
+ return;
+ }
+ long reachedWatermark =
internalTimeServiceManager.getReachedWatermark();
+ if (reachedWatermark > lastEmittedIntermediateWatermark) {
+ // Firing was interrupted before completing; surface the progress
made so far instead
+ // of leaving watermark advancement stalled until the whole drain
finishes.
+ lastEmittedIntermediateWatermark = reachedWatermark;
Review Comment:
Should we also update `lastEmittedIntermediateWatermark` in the short-cut
branch above?
--
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]
