[Prism] Honor the resume delay of self-checkpointing SDF residuals

[Eliaaazzz]

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:

• 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.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

[codecov]

Codecov Report

❌ Patch coverage is 91.22807% with 5 lines in your changes missing coverage. Please review.
✅ Project coverage is 59.69%. Comparing base (c101795) to head (f85f0cb).
⚠️ Report is 49 commits behind head on master.

Files with missing lines	Patch %	Lines
...am/runners/prism/internal/engine/elementmanager.go	91.22%	3 Missing and 2 partials ⚠️

Additional details and impacted files

@@            Coverage Diff            @@
##             master   #39849   +/-   ##
=========================================
  Coverage     59.69%   59.69%           
+ Complexity    16818    16813    -5     
=========================================
  Files          2863     2863           
  Lines        297877   297972   +95     
  Branches      14679    14679           
=========================================
+ Hits         177805   177875   +70     
- Misses       112468   112486   +18     
- Partials       7604     7611    +7     

Flag	Coverage Δ
go	29.09% <91.22%> (+0.06%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[github-actions]

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

[Eliaaazzz]

Run GoPrism PreCommit

[Eliaaazzz]

Run Java_PVR_Prism_Loopback PreCommit

[github-actions]

Assigning reviewers:

R: @shunping for label go.

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]

In fact I suspect all runners have same issue, if checking the log of previous job: 2026-08-14_04_42_03-12041500729043885033. Effectively busy polling is consistent throughout runners. We need to be more careful and understand the implementation of this fix, and if it indeeds align with spec.

I think a proper fix should be done in SDK side, such that the watch transform should not initiate poll function and output when the time isn't advanced to next poll time:

beam/sdks/python/apache_beam/transforms/periodicsequence.py

Line 168 in 466cf6d

if current_output_timestamp > time.time():

============

EDIT: tested on Dataflow runner that defer_remainder(timestamp) is honored. This is indeed a feature gap of Prism then.

[Abacn]

Tested (1) PeriodicImpulse and (2) MatchContinuously, both pipeline eagerly runs process element prior to this change (as well as prior to #39572), and now the interval is honored.

[shunping]

I like the idea of splitting the originally pending elements from residual into immediate schedulable and delayed schedulable (based on processing time).

I am a little concerned about the linear scanning of the delayed map every time a watermark refresh is needed, but given the size of the delayed map would not be too big in our current use case, I think it is ok and we could improve that in the future.

Overall the PR looks great. Thanks!

[shunping]

We disabled a few java tests for Prism runner at https://github.com/apache/beam/blob/master/runners/prism/java/build.gradle. It would be great if you can enable some of the SDF and timer related ones (for example GroupIntoBatches ones) to see if your PR here can also fix those tests.

This is not a blocker for this PR though.

[Abacn]

Found the following tests in sickbay suite now passing:

MetricsTest: testCommittedCounterMetrics, testCommittedDistributionMetrics, testCommittedGaugeMetrics, testCommittedStringSetMetrics (only boundedTrie not supported)

GroupIntoBatchesTest: testInStreamingMode, testWithShardedKeyInGlobalWindow

[Abacn]

testInStreamingMode still failing

2026-08-25T18:20:41.6121621Z GroupIntoBatchesTest > testInStreamingMode FAILED
2026-08-25T18:20:41.6122801Z     java.lang.AssertionError at GroupIntoBatchesTest.java:438
2026-08-25T18:20:41.6124060Z         Caused by: java.lang.RuntimeException at GroupIntoBatchesTest.java:438
2026-08-25T18:20:41.6125802Z             Caused by: java.util.concurrent.ExecutionException at GroupIntoBatchesTest.java:438
2026-08-25T18:20:41.6127271Z                 Caused by: java.lang.RuntimeException at JobServicePipelineResult.java:176