Fix PeriodicImpulse/PeriodicSequence watermark regression (#39026)

[uditjainstjis]

Fixes #39026

Problem

After upgrading from 2.66.0 to 2.74.0, streaming jobs using a PeriodicImpulse side input that fires at long intervals saw the job's watermark age follow a saw-tooth pattern, climbing up to the fire interval before snapping back. Users with watermark-age alerts were paged spuriously.

Root cause

process() in ImpulseSeqGenDoFn was rewritten between 2.66.0 and 2.74.0. The old code, after emitting the element at index k, recomputed the timestamp for index k+1 and set the watermark to that next fire time before deferring — so the watermark sat ~one interval ahead of the last emitted element and the reported age stayed flat.

The rewritten code sets the watermark to the just-emitted element's timestamp and never advances it on the defer path. Between fires the watermark is therefore held at the last element's event time, so downstream watermark age grows to fire_interval and resets on each fire.

Fix

When process() defers because the next scheduled output is in the future, advance the watermark to that next fire time (monotonic-guarded). No element can be produced before it, so it is a valid watermark, and this restores the pre-2.74.0 behavior.

The advance is gated on not self._is_pre_timestamped: for pre-timestamped data, event times may be intentionally out of order (a later element can carry an earlier timestamp, treated as a late event per the PeriodicImpulse docstring), so we must not declare future late events droppable.

Testing

Added two unit tests that drive process() directly with a RestrictionTrackerView + ManualWatermarkEstimator:

• test_watermark_advances_to_next_fire_on_defer — fails on unpatched master (watermark stuck at last emitted element) and passes with the fix. This is the regression reproduced as a unit test (the Dataflow-side symptom cannot be reproduced in a unit test).
• test_watermark_not_advanced_past_emitted_for_pre_timestamped — asserts the watermark is not advanced past emitted event times for pre-timestamped data.

Full periodicsequence_test.py passes locally (26 passed, 1 skipped). yapf==0.43.0 clean, pylint 10.00/10.

• Update CHANGES.md with noteworthy changes.

[gemini-code-assist]

Caution

The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased.

[github-actions]

Assigning reviewers:

R: @jrmccluskey for label python.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[Abacn]

We would need end-to-end pipeline evidence to verify this fix worked, like the screenshot in #39026

[uditjainstjis]

Thanks @Abacn, fair request. I don't have a Dataflow project to reproduce the exact screenshot, so instead I reproduced the SDK-reported output watermark itself — the ManualWatermarkEstimator value ImpulseSeqGenDoFn sends upstream, which is exactly what Dataflow renders as data freshness = now − watermark. That value is a pure function of ImpulseSeqGenDoFn.process(), so master vs this PR is a deterministic apples-to-apples comparison with no runner timing/flakiness.

Setup: fire_interval = 60s; advance a simulated wall-clock now and, at each point, read the watermark the DoFn reports to the runner.

now (s)	master wm	master freshness	this PR wm	this PR freshness
0	0	0	60	0
30	0	30	60	0
58	0	58	60	0
60	60	0	120	0
90	60	30	120	0
118	60	58	120	0
120	120	0	180	0

(freshness shown the way the runner reports it: max(0, now − watermark))

• master: the watermark is pinned to the last emitted element, so freshness climbs to a full fire_interval (60s) then snaps back — the saw-tooth reported in this issue.
• this PR: on deferral the watermark is advanced to the next fire time, so it tracks wall-clock and freshness stays flat at 0 — the pre-2.74.0 behavior.

This is the same thing the unit test in the PR asserts (it fails on master with the watermark stuck at the last emitted element, and passes with the fix).

Repro — no Dataflow needed, run against a checkout of each branch (full script):

from unittest import mock
import apache_beam.transforms.periodicsequence as ps
from apache_beam.io.restriction_trackers import OffsetRange, OffsetRestrictionTracker
from apache_beam.io.watermark_estimators import ManualWatermarkEstimator
from apache_beam.runners.sdf_utils import RestrictionTrackerView, ThreadsafeRestrictionTracker
from apache_beam.transforms.periodicsequence import ImpulseSeqGenDoFn

def reported_watermark(now, start=0.0, interval=60.0):
    tracker = ThreadsafeRestrictionTracker(OffsetRestrictionTracker(OffsetRange(0, 10**9)))
    est = ManualWatermarkEstimator(None)
    with mock.patch.object(ps.time, 'time', return_value=now):
        list(ImpulseSeqGenDoFn().process(
            (start, start + 10**9, interval),
            restriction_tracker=RestrictionTrackerView(tracker),
            watermark_estimator=est))
    return float(est.current_watermark())

for now in range(0, 121, 30):
    wm = reported_watermark(now)
    print(now, wm, max(0.0, now - wm))
# master: 0/30/58 -> freshness 0/30/58 (saw-tooth);  this PR: freshness stays 0

For completeness I also ran a bounded streaming PeriodicImpulse pipeline on the local Prism runner: the fix's output watermark leads the last-emitted element further than master there as well, though Prism damps the magnitude — it derives part of the residual watermark hold from the SDF resume time (which master already sets via defer_remainder), whereas Dataflow uses the estimator value this PR corrects. Happy to have anyone with a Dataflow project confirm against the exact repro in the issue description.