Abacn commented on code in PR #39465:
URL: https://github.com/apache/beam/pull/39465#discussion_r3798817517
##########
sdks/python/apache_beam/transforms/periodicsequence.py:
##########
@@ -166,9 +166,26 @@ def process(
current_output_timestamp = start + interval * current_output_index
if current_output_timestamp > time.time():
- # we are too ahead of time, let's wait.
- restriction_tracker.defer_remainder(
- timestamp.Timestamp(current_output_timestamp))
+ # We are ahead of time, let's wait. No element will be produced before
+ # the next fire time, so advance the watermark up to that timestamp.
+ # Without this the reported watermark stalls at the last emitted
+ # element's timestamp and lags by up to one fire_interval, producing a
+ # saw-tooth watermark age (regression introduced in 2.67.0 by the
+ # process() rewrite in #35412, see
+ # https://github.com/apache/beam/issues/39026).
+ #
+ # For pre-timestamped data the provided event times may be out of order
+ # (a later element can carry an earlier timestamp, treated as a late
+ # event), so we must not advance the watermark past what has already
+ # been emitted.
Review Comment:
```suggestion
# We are ahead of time, let's wait. No element will be produced
before
# the next fire time, so advance the watermark up to that timestamp.
```
--
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]
