twalthr commented on code in PR #27962:
URL: https://github.com/apache/flink/pull/27962#discussion_r3116533952
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/AbstractProcessTableOperator.java:
##########
@@ -145,11 +146,17 @@ public void open() throws Exception {
FunctionUtils.openFunction(processTableRunner,
DefaultOpenContext.INSTANCE);
}
+ @Override
+ public final boolean useInterruptibleTimers(ReadableConfig config) {
+ return true;
+ }
+
@Override
public void processWatermark(Watermark mark) throws Exception {
- super.processWatermark(mark);
- // TODO this line has issues with interruptible timers, see FLINK-39437
+ // Update the runner's watermark before firing timers to keep it
consistent with the
+ // timer service watermark, which is also advanced before any timer
fires.
processTableRunner.ingestCurrentWatermarkEvent(mark.getTimestamp());
Review Comment:
I don't think that this line swap is sufficient. The docs in
MailboxWatermarkProcessor state that there might be a chance that new rows
enter the PTF while the current watermark has not fully been processed. We
should call `ingestCurrentWatermarkEvent` only after the watermark has been
fully processed. Otherwise a PTF that implements late data handling manually
and relies on `TimeContext.currentWatermark` would not behave correctly. If you
check `MailboxPartialWatermarkProcessor` it only updates the current watermark
upon completion of processing.
--
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]
