fhueske commented on code in PR #27962:
URL: https://github.com/apache/flink/pull/27962#discussion_r3123208140
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/process/ProcessRowTableOperator.java:
##########
@@ -82,11 +83,21 @@ public void processElement(StreamRecord<RowData> element)
throws Exception {
0,
element.getValue(),
inputSemantics.timeColumn(),
- processTableRunner.getCurrentWatermark());
+ combinedWatermark.getCombinedWatermark());
}
processTableRunner.processEval();
}
+ @Override
+ public void processWatermark(Watermark mark) throws Exception {
+ // Advance combinedWatermark before super.processWatermark(mark):
super.processWatermark()
+ // fires due timers, which read the watermark via timeContext(). V2's
OneInputStreamOperator
+ // watermark path doesn't update combinedWatermark itself — only
reportWatermark()
+ // (multi-input) does — so we do it here.
+ combinedWatermark.updateWatermark(0, mark.getTimestamp());
+ super.processWatermark(mark);
Review Comment:
I tried this but it doesn't really work.
`reportWatermark()` first updates the `combinedWatermark` but its call to
`processWatermark()` again (recursively) is directed to the overriding
`ProcessRowTableOperator.processWatermark()` which calls `reportWatermark()`
again, but this time the `combinedWatermark` is already updated and we don't
call `processWatermark()` again (otherwise we would run into a stack-overflow
situation).
The consequence is that we never call
`AbstractStreamOperatorV2.processWatermark()`.
--
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]
