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).