What happened?
A splittable DoFn that self-checkpoints with a requested resume delay (Python tracker.defer_remainder(Duration(...)), Go sdf.ResumeProcessingIn(...), Java ProcessContinuation.resume().withResumeDelay(...)) is resumed by Prism immediately. The delay is decoded from the bundle response into engine.Residual.Delay, but nothing consumes that field: PersistBundle returns every residual straight to the pending queue at the existing TODO
|
// TODO(https://github.com/apache/beam/issues/39446) |
|
// Return unprocessed to this stage's pending |
|
// TODO sort out pending element watermark holds for process continuation residuals. |
|
unprocessedElements := reElementResiduals(residuals.Data, inputInfo, rb) |
|
|
|
// Add unprocessed back to the pending stack. |
|
if len(unprocessedElements) > 0 { |
|
// TODO actually reschedule based on the residuals delay... |
|
count := stage.AddPending(em, unprocessedElements) |
|
em.addPending(count) |
so a polling SDF busy-spins instead of pacing its poll rounds.
Measured on a current master build of Prism, driving the Python SDK in LOOPBACK: a Watch poll loop requesting a 3 second poll interval via defer_remainder executed 7,034 poll rounds in 9 seconds of wall time. Expected: 4 rounds.
This breaks transforms whose semantics assume the requested pacing. MatchContinuously derives a bounded number of polls from stop_timestamp and interval; on Prism the polling window collapses in wall-clock terms, so files added while the pipeline runs are missed. Reproduced with a pipeline matching a directory with interval=3 and stop_timestamp=now+24 while a file is added 8 seconds in: only the pre-existing file is matched. The same pipeline passes on the DirectRunner.
The engine already has the machinery a fix needs: em.processTimeEvents plus em.wakeUpAt schedule processing-time timers under the real-time clock, and checkForQuiescence treats scheduled processing-time events as pending work.
Issue Priority
Priority: 2 (default / most bugs should be filed as P2)
Issue Components
What happened?
A splittable DoFn that self-checkpoints with a requested resume delay (Python
tracker.defer_remainder(Duration(...)), Gosdf.ResumeProcessingIn(...), JavaProcessContinuation.resume().withResumeDelay(...)) is resumed by Prism immediately. The delay is decoded from the bundle response intoengine.Residual.Delay, but nothing consumes that field:PersistBundlereturns every residual straight to the pending queue at the existing TODObeam/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go
Lines 943 to 952 in c101795
so a polling SDF busy-spins instead of pacing its poll rounds.
Measured on a current master build of Prism, driving the Python SDK in LOOPBACK: a
Watchpoll loop requesting a 3 second poll interval viadefer_remainderexecuted 7,034 poll rounds in 9 seconds of wall time. Expected: 4 rounds.This breaks transforms whose semantics assume the requested pacing.
MatchContinuouslyderives a bounded number of polls fromstop_timestampandinterval; on Prism the polling window collapses in wall-clock terms, so files added while the pipeline runs are missed. Reproduced with a pipeline matching a directory withinterval=3andstop_timestamp=now+24while a file is added 8 seconds in: only the pre-existing file is matched. The same pipeline passes on the DirectRunner.The engine already has the machinery a fix needs:
em.processTimeEventsplusem.wakeUpAtschedule processing-time timers under the real-time clock, andcheckForQuiescencetreats scheduled processing-time events as pending work.Issue Priority
Priority: 2 (default / most bugs should be filed as P2)
Issue Components