[Prism] Schedule consumers of a self checkpointing source - #39572
Conversation
|
Caution The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased. |
|
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment |
Fixes apache#39446. An unbounded source SDF that returns a process continuation residual has that residual re-queued as a pending element carrying the input element's event time, MinTimestamp for an Impulse rooted source, so the stage's watermark never advances. updateWatermarks returns no refreshes when the output watermark does not move, and PersistBundle marked only the producing stage as changed, so a consumer that was just handed pending elements was never surfaced to the scheduler. Its elements accumulated forever. checkForQuiescence cannot catch this, because the source stays schedulable, so the job live locks instead of failing fast. PersistBundle now records the consumers that accepted data, and updateWatermarks hands them to the scheduler on the path where the output watermark does not advance. Recording is gated on any bundle having returned a residual, so a pipeline that never self checkpoints keeps its previous bundle scheduling. bundleReady additionally lets a stateful stage with no side inputs run on pending data alone, under the same gate, since statefulStageKind.buildEventTimeBundle takes data at any watermark and gates only timers. Its stillSchedulable now requires buildable work, and a key that supplies nothing no longer consumes the OneKeyPerBundle slot, is not marked in progress, and does not hold the bundle's minimum timestamp. A consumer that reads a side input still waits on the watermark, since side input readiness is derived from it.
d147583 to
feddf87
Compare
|
Assigning reviewers: R: @jrmccluskey for label go. Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |
|
Reminder, please take a look at this pr: @jrmccluskey @shunping |
|
Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment R: @lostluck for label go. Available commands:
|
Abacn
left a comment
Abacn
left a comment
There was a problem hiding this comment.
Tested locally and on CI: https://github.com/apache/beam/actions/runs/32273187172/attempts/1 . Local test found previously failing test now passing consistently. However, same commit running on CI still has a failing test. Likely there are other flakiness
| // hasBuildableDataLocked reports whether a key that isn't in progress heads its | ||
| // heap with data, or with a timer the watermark has reached. Callers hold ss.mu. | ||
| func (ss *stageState) hasBuildableDataLocked(watermark mtime.Time) bool { | ||
| for k, dnt := range ss.pendingByKeys { |
There was a problem hiding this comment.
Here it iterates over pendingByKeys again. Would there be any performance concern?
There was a problem hiding this comment.
It stops at the first key it could build, so the full pass over pendingByKeys only happens when the answer is no, which is exactly the case that previously rescheduled empty bundles without bound; one scan to park the stage replaces that loop. On the productive path it usually stops at the first key that heads with data. The bundleReady call site is behind em.sawResidual, so pipelines without a self checkpointing source never evaluate it, and the builder call site is short circuited by timerCleared.
The cost that remains is one scan per unproductive build on a stateful stage with many pending keys headed by future timers. An index of buildable keys would remove it, but buildability depends on the current watermark, since a key headed by a future timer becomes buildable when the watermark moves, so the index needs its own bookkeeping on every watermark advance. I would leave that until a profile shows this scan, and can file it as a follow up if you prefer.
There was a problem hiding this comment.
which is exactly the case that previously rescheduled empty bundles...
Is that correct that full scan only happens when there is no new data? In this case the performance concern would be minimum then
There was a problem hiding this comment.
Yes, that's right. It only runs to the end when nothing is buildable, and returns at the first buildable key otherwise.
|
Looked into the run. Attempt 1 failed on ActiveMQJmsIOTest.test_xlang_jms_write_read_queue with 0 != 100 and the rerun of the same commit passed. That suite runs the Python DirectRunner with a Java expansion service, so prism is not on the execution path. The same workflow failed twice on master in its last twenty runs without this change, both in the same file with the same 0 != 100 shape, IbmMqJmsIOTest.test_xlang_jms_write_read_queue_client_ack on Aug 16 and Aug 17. So I read it as pre existing flakiness in the new JMS suite rather than something this change introduced. Happy to file a flaky test issue for it. |
(Sorry for replying to wrong thread previously) These tests do use prism runner: My comment was about this does fix the flakiness seen in the JmsIO test to some extent. Since this PR closes #39446, can we apply f274b1b here (remove 5s delay) and trigger PostCommit Python Xlang Messaging Direct? Thanks! |
…ontinuation-watermark
The test pre-published part of the records before running the pipeline to work around the starvation this change fixes. All records are now published after the pipeline starts, and the duplicate-tolerant assertion tightens back to an exact count. Also bumps the messaging postcommit trigger file.
Thanks for the good call. I have done that! |
2 participants
Fixes #39446.
The bug
An unbounded source SDF that returns
ProcessContinuation.resume()has its residual re-queued byreElementResidualsas a pending element carrying the input element's event time, which for an Impulse rooted source ismtime.MinTimestamp. That pinsminPendingTimestampLocked, so the stage's watermark never advances.updateWatermarksreturns no refreshes when the output watermark does not move, andPersistBundlemarked only the producing stage as changed. So a consumer that had just been handed pending elements was never inserted intoem.changedStages,bundleReadywas never called for it, and its elements accumulated forever.This is worse than the issue describes: it starves plain ordinary consumers too, not only stateful ones, and it survives transitively through a chain of stages.
checkForQuiescencecannot catch it, because the source stage stays schedulable, so the job live locks rather than failing fast.Reproduced at the element manager level before fixing. Pre-fix, across 5 source bundles:
The fix
PersistBundlerecords the consumers that accepted data intostageState.consumersWithNewData, andupdateWatermarkshands that set to the scheduler on thenewOut <= ss.outputpath, clearing it on the advancing path where the normal return already lists every consumer. Routing the wake-up throughupdateWatermarksis what keeps a pipeline whose watermark advances on exactly its previous schedule.em.sawResidual, latched the first time any bundle returns a residual. A pipeline that never self checkpoints never populates the set, so its bundle scheduling is unchanged. This matters: an earlier revision that recorded unconditionally changed bundle packing and brokeTestRunner_Pipelines/flatten_to_sideInputat-count=20.bundleReadylets a stateful stage with no side inputs run on pending data alone, sincestatefulStageKind.buildEventTimeBundlealready takes data at any watermark and gates only timers. ItsstillSchedulablenow requires buildable work so a key holding only a future timer cannot spin against the relaxed gate, and such a key no longer consumes theOneKeyPerBundleslot, is not marked in progress, and does not hold the bundle's minimum timestamp.What is deliberately unchanged
BundleApplication.output_watermarkssays an unreported estimate meansMIN_TIMESTAMP, and Java'sProcessFnholds atfutureOutputWatermark ?? elementTimestamp. A pinnedMIN_TIMESTAMPoutput watermark for a source that reports no estimate is the spec, so the TODO the issue points at is not where the defect is.GetSideDatafilters non global windows by the bundle watermark, so relaxing this would let a bundle read a partial or empty side input. That is worse than waiting. Fixing it properly means changing how side inputs are materialized for every stage, which I would rather do separately with your input.Testing
New
elementmanager_continuation_test.go, all teeth checked by reverting each half of the fix independently:TestPersistBundle_ContinuationResidualConsumers, ordinary and stateful consumers of a source that self checkpoints forever.TestPersistBundle_ContinuationResidualTransitive,srcself checkpoints,midreturns no residual of its own,sinktwo stages down.TestPersistBundle_ContinuationResidualWatermark, pins theoutput_watermarkscontract in both directions.TestStatefulBuildEventTimeBundle_OneKeyPerBundle, a key headed by a future timer must not consume the single key slot, and must not enternewKeysor holdminTs.Green:
go test ./pkg/beam/runners/prism/... -count=1,-run "TestRunner_Pipelines/flatten" -count=20, and the new tests at-count=50.Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
R: @username).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, commentfixes #<ISSUE NUMBER>instead.CHANGES.mdwith noteworthy changes.See the Contributor Guide for more tips on how to make review process smoother.
To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md