uditjainstjis commented on PR #39465: URL: https://github.com/apache/beam/pull/39465#issuecomment-5071955524
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](https://github.com/uditjainstjis/beam/blob/evidence-39026/evidence/drive_sim.py)): ```python 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. -- 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]
