uditjainstjis opened a new pull request, #39465: URL: https://github.com/apache/beam/pull/39465
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. - [x] Update `CHANGES.md` with noteworthy changes. -- 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]
