Eliaaazzz opened a new pull request, #39849:
URL: https://github.com/apache/beam/pull/39849
Fixes #39848.
### The bug
`PersistBundle` decoded the SDK's requested resume delay into
`engine.Residual.Delay` and then returned every residual to the pending queue
immediately, at the TODO "actually reschedule based on the residuals delay". A
polling SDF therefore busy-spins: on a master build of Prism, a Python `Watch`
poll loop requesting 3 second pacing via `defer_remainder` executed 7,034 poll
rounds in 9 seconds of wall time.
### The change
`PersistBundle` partitions residuals by delay. A residual with no delay
returns to pending as before. A delayed residual is parked on its stage in
`stageState.delayedResiduals`, keyed by the processing time it becomes
schedulable, and that time is scheduled through the machinery processing-time
timers already use: `ptRefreshes` → `em.processTimeEvents.Schedule` →
`em.wakeUpAt`. The watermark evaluation loop releases due parked residuals back
to pending before checking bundle readiness.
Parked elements pin the stage's input watermark by being included in
`minPendingTimestampLocked`, the same way pending elements do. An earlier
revision used `watermarkHolds` instead; those clamp only the output watermark,
and since the input watermark is monotonic, a bounded pipeline's global-window
aggregation could fire early with partial data once the input watermark ran
ahead during a park. The `TestSeparation` ProcessContinuations subtests catch
exactly this and pass with the final design.
Under the fast-forward clock (`EnableRTC=false`), `processingTimeNow` peeks
the event queue, so delayed residuals still fire immediately in synthetic time
and test pipelines stay fast. Under the default real-time clock the delay is
honored; `--experiments=prism_disable_rtc` remains the opt-out. The
ProcessContinuations subtests of `TestSeparation` now take real time (~15s
instead of ~4s locally) because the 1 second resume delays they request are
honored.
### Verification
- New `TestPersistBundle_ResidualResumeDelay` engine tests: a delayed
residual is parked rather than pending, pins `minPendingTimestampLocked`, and
schedules a processing time event; under a real-time clock it fires only after
the delay; under the fast-forward clock it fires without real waiting.
Reverting the parking fails the first two subtests.
- Full `runners/prism/...` Go tests pass.
- End to end with the Python SDK against a rebuilt prism binary: the
3s-pacing probe went from 7,034 polls in 9 seconds to 4 polls at 3.01 second
spacing; a `MatchContinuously` pipeline that previously missed files added
mid-run on Prism now passes; `Watch` pipelines in both deduplication modes
still pass.
------------------------
Thank you for your contribution! Follow this checklist to help us
incorporate your contribution quickly and easily:
- [x] Mention the appropriate issue in your description (for example:
`addresses #123`), if applicable. This will automatically add a link to the
pull request in the issue. If you would like the issue to automatically close
on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [x] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual
Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more
tips on [how to make review process
smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
--
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]
