Eliaaazzz opened a new pull request, #39572:
URL: https://github.com/apache/beam/pull/39572
Fixes #39446.
### The bug
An unbounded source SDF that returns `ProcessContinuation.resume()` has its
residual re-queued by `reElementResiduals` as a pending element carrying the
*input element's* event time, which for an Impulse rooted source is
`mtime.MinTimestamp`. That pins `minPendingTimestampLocked`, 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 had just been handed pending elements was never inserted into
`em.changedStages`, `bundleReady` was 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. `checkForQuiescence` cannot 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:
```
round 0: src out=-inf | sink pending=1 scheduled=0
round 4: src out=-inf | sink pending=5 scheduled=0
```
### The fix
1. `PersistBundle` records the consumers that accepted data into
`stageState.consumersWithNewData`, and `updateWatermarks` hands that set to the
scheduler on the `newOut <= ss.output` path, clearing it on the advancing path
where the normal return already lists every consumer. Routing the wake-up
through `updateWatermarks` is what keeps a pipeline whose watermark advances on
exactly its previous schedule.
2. The recording is gated on `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 broke
`TestRunner_Pipelines/flatten_to_sideInput` at `-count=20`.
3. `bundleReady` lets a stateful stage with no side inputs run on pending
data alone, since `statefulStageKind.buildEventTimeBundle` already takes data
at any watermark and gates only timers. Its `stillSchedulable` now requires
buildable work so a key holding only a future timer cannot spin against the
relaxed gate, and such a key no longer consumes the `OneKeyPerBundle` slot, is
not marked in progress, and does not hold the bundle's minimum timestamp.
### What is deliberately unchanged
- **The watermark value.** `BundleApplication.output_watermarks` says an
unreported estimate means `MIN_TIMESTAMP`, and Java's `ProcessFn` holds at
`futureOutputWatermark ?? elementTimestamp`. A pinned `MIN_TIMESTAMP` output
watermark for a source that reports no estimate is the spec, so the TODO the
issue points at is not where the defect is.
- **Aggregate stages** stay on the watermark gate. A GBK genuinely needs
window closure or a trigger.
- **Any stage that reads a side input** stays on the watermark gate, so such
a consumer still waits under a residual pinned source. Side input readiness is
derived from the main watermark, and `GetSideData` filters 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`, `src` self
checkpoints, `mid` returns no residual of its own, `sink` two stages down.
- `TestPersistBundle_ContinuationResidualWatermark`, pins the
`output_watermarks` contract in both directions.
- `TestStatefulBuildEventTimeBundle_OneKeyPerBundle`, a key headed by a
future timer must not consume the single key slot, and must not enter `newKeys`
or hold `minTs`.
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:
- [ ] [**Choose
reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and
mention them in a comment (`R: @username`).
- [ ] 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](https://cla.apache.org/).
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).
To check the build health, please visit
[https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
--
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]
