You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "Eliaaazzz (via GitHub)" <gi...@apache.org> on 2026/07/31 15:47:30 UTC
[PR] [Prism] Schedule consumers of a self checkpointing source [beam]
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Prism] Schedule consumers of a self checkpointing source [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on code in PR #39572:
URL: https://github.com/apache/beam/pull/39572#discussion_r3814860485
##########
sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:
##########
@@ -1874,22 +1898,51 @@ keysPerBundle:
break
}
}
+ if len(toProcessForKey) > 0 {
+ newKeys.insert(k)
+ // Track the min-timestamp for later watermark handling. Elements pop
+ // in timestamp order, so the first selected one is the earliest.
+ if ts := toProcessForKey[0].timestamp; ts < minTs {
+ minTs = ts
+ }
+ }
toProcess = append(toProcess, toProcessForKey...)
if dnt.elements.Len() == 0 {
delete(ss.pendingByKeys, k)
}
- if OneKeyPerBundle {
+ // A key that yielded nothing, such as one headed by a timer above the
+ // watermark, must not consume the single key slot, or the bundle is empty
+ // and the stage keeps rescheduling on it.
+ if OneKeyPerBundle && len(toProcessForKey) > 0 {
break keysPerBundle
}
}
- // If we're out of data, and timers were not cleared then the watermark is accurate.
- stillSchedulable := !(len(ss.pendingByKeys) == 0 && !timerCleared)
+ // Reschedule only when a later bundle could build something, or a cleared
+ // timer may have held back the minimum pending timestamp.
+ stillSchedulable := timerCleared || ss.hasBuildableDataLocked(watermark)
return toProcess, minTs, newKeys, holdsInBundle, nil, stillSchedulable, 0
}
+// 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 {
Review Comment:
Here it iterates over pendingByKeys again. Would there be any performance concern?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Prism] Schedule consumers of a self checkpointing source [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #39572:
URL: https://github.com/apache/beam/pull/39572#issuecomment-5148670502
Assigning reviewers:
R: @jrmccluskey 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).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Prism] Schedule consumers of a self checkpointing source [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn merged PR #39572:
URL: https://github.com/apache/beam/pull/39572
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Prism] Schedule consumers of a self checkpointing source [beam]
Posted by "Eliaaazzz (via GitHub)" <gi...@apache.org>.
Eliaaazzz commented on PR #39572:
URL: https://github.com/apache/beam/pull/39572#issuecomment-5353366472
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.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Prism] Schedule consumers of a self checkpointing source [beam]
Posted by "Eliaaazzz (via GitHub)" <gi...@apache.org>.
Eliaaazzz commented on code in PR #39572:
URL: https://github.com/apache/beam/pull/39572#discussion_r3826722282
##########
sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:
##########
@@ -1874,22 +1898,51 @@ keysPerBundle:
break
}
}
+ if len(toProcessForKey) > 0 {
+ newKeys.insert(k)
+ // Track the min-timestamp for later watermark handling. Elements pop
+ // in timestamp order, so the first selected one is the earliest.
+ if ts := toProcessForKey[0].timestamp; ts < minTs {
+ minTs = ts
+ }
+ }
toProcess = append(toProcess, toProcessForKey...)
if dnt.elements.Len() == 0 {
delete(ss.pendingByKeys, k)
}
- if OneKeyPerBundle {
+ // A key that yielded nothing, such as one headed by a timer above the
+ // watermark, must not consume the single key slot, or the bundle is empty
+ // and the stage keeps rescheduling on it.
+ if OneKeyPerBundle && len(toProcessForKey) > 0 {
break keysPerBundle
}
}
- // If we're out of data, and timers were not cleared then the watermark is accurate.
- stillSchedulable := !(len(ss.pendingByKeys) == 0 && !timerCleared)
+ // Reschedule only when a later bundle could build something, or a cleared
+ // timer may have held back the minimum pending timestamp.
+ stillSchedulable := timerCleared || ss.hasBuildableDataLocked(watermark)
return toProcess, minTs, newKeys, holdsInBundle, nil, stillSchedulable, 0
}
+// 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 {
Review Comment:
Yes, with one nuance: the full pass happens exactly when no key outside a running bundle heads with data or a ripe timer, which is the round that parks the stage. So it is nothing buildable rather than no new data in the strict sense, since a ready timer also counts. A productive round stops at the first buildable key it meets.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Prism] Schedule consumers of a self checkpointing source [beam]
Posted by "Eliaaazzz (via GitHub)" <gi...@apache.org>.
Eliaaazzz commented on PR #39572:
URL: https://github.com/apache/beam/pull/39572#issuecomment-5364270710
You are right, I had it backwards: messagingCrossLanguage runs on the Prism runner, so those failures are this starvation showing up in the suite, not unrelated flakiness. Applied your commit without the 5s delay, so the pre-publish workaround is gone, all records are published after the pipeline starts, and the assertion tightens back to an exact distinct count. Also merged master and bumped the messaging postcommit trigger file so the suite runs here. Commit ac3978b.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Prism] Schedule consumers of a self checkpointing source [beam]
Posted by "Eliaaazzz (via GitHub)" <gi...@apache.org>.
Eliaaazzz commented on code in PR #39572:
URL: https://github.com/apache/beam/pull/39572#discussion_r3827813385
##########
sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:
##########
@@ -1874,22 +1898,51 @@ keysPerBundle:
break
}
}
+ if len(toProcessForKey) > 0 {
+ newKeys.insert(k)
+ // Track the min-timestamp for later watermark handling. Elements pop
+ // in timestamp order, so the first selected one is the earliest.
+ if ts := toProcessForKey[0].timestamp; ts < minTs {
+ minTs = ts
+ }
+ }
toProcess = append(toProcess, toProcessForKey...)
if dnt.elements.Len() == 0 {
delete(ss.pendingByKeys, k)
}
- if OneKeyPerBundle {
+ // A key that yielded nothing, such as one headed by a timer above the
+ // watermark, must not consume the single key slot, or the bundle is empty
+ // and the stage keeps rescheduling on it.
+ if OneKeyPerBundle && len(toProcessForKey) > 0 {
break keysPerBundle
}
}
- // If we're out of data, and timers were not cleared then the watermark is accurate.
- stillSchedulable := !(len(ss.pendingByKeys) == 0 && !timerCleared)
+ // Reschedule only when a later bundle could build something, or a cleared
+ // timer may have held back the minimum pending timestamp.
+ stillSchedulable := timerCleared || ss.hasBuildableDataLocked(watermark)
return toProcess, minTs, newKeys, holdsInBundle, nil, stillSchedulable, 0
}
+// 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 {
Review Comment:
Yes, that's right. It only runs to the end when nothing is buildable, and returns at the first buildable key otherwise.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Prism] Schedule consumers of a self checkpointing source [beam]
Posted by "Eliaaazzz (via GitHub)" <gi...@apache.org>.
Eliaaazzz commented on code in PR #39572:
URL: https://github.com/apache/beam/pull/39572#discussion_r3827670667
##########
sdks/python/apache_beam/io/external/xlang_jmsio_it_test.py:
##########
@@ -143,11 +137,11 @@ def subscribe():
connection_param),
queue=sink_queue,
expansion_service=self.expansion_service))
- publisher.start()
result = p.run()
+ publisher.start()
Review Comment:
Sorry Yi, I was in the meeting the whole time and didn't see it. All the CIs have passed now.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Prism] Schedule consumers of a self checkpointing source [beam]
Posted by "Eliaaazzz (via GitHub)" <gi...@apache.org>.
Eliaaazzz commented on code in PR #39572:
URL: https://github.com/apache/beam/pull/39572#discussion_r3819854589
##########
sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:
##########
@@ -1874,22 +1898,51 @@ keysPerBundle:
break
}
}
+ if len(toProcessForKey) > 0 {
+ newKeys.insert(k)
+ // Track the min-timestamp for later watermark handling. Elements pop
+ // in timestamp order, so the first selected one is the earliest.
+ if ts := toProcessForKey[0].timestamp; ts < minTs {
+ minTs = ts
+ }
+ }
toProcess = append(toProcess, toProcessForKey...)
if dnt.elements.Len() == 0 {
delete(ss.pendingByKeys, k)
}
- if OneKeyPerBundle {
+ // A key that yielded nothing, such as one headed by a timer above the
+ // watermark, must not consume the single key slot, or the bundle is empty
+ // and the stage keeps rescheduling on it.
+ if OneKeyPerBundle && len(toProcessForKey) > 0 {
break keysPerBundle
}
}
- // If we're out of data, and timers were not cleared then the watermark is accurate.
- stillSchedulable := !(len(ss.pendingByKeys) == 0 && !timerCleared)
+ // Reschedule only when a later bundle could build something, or a cleared
+ // timer may have held back the minimum pending timestamp.
+ stillSchedulable := timerCleared || ss.hasBuildableDataLocked(watermark)
return toProcess, minTs, newKeys, holdsInBundle, nil, stillSchedulable, 0
}
+// 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 {
Review 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.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Prism] Schedule consumers of a self checkpointing source [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #39572:
URL: https://github.com/apache/beam/pull/39572#issuecomment-5315894308
Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment `assign to next reviewer`:
R: @lostluck for label go.
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)
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Prism] Schedule consumers of a self checkpointing source [beam]
Posted by "Eliaaazzz (via GitHub)" <gi...@apache.org>.
Eliaaazzz commented on code in PR #39572:
URL: https://github.com/apache/beam/pull/39572#discussion_r3827813385
##########
sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:
##########
@@ -1874,22 +1898,51 @@ keysPerBundle:
break
}
}
+ if len(toProcessForKey) > 0 {
+ newKeys.insert(k)
+ // Track the min-timestamp for later watermark handling. Elements pop
+ // in timestamp order, so the first selected one is the earliest.
+ if ts := toProcessForKey[0].timestamp; ts < minTs {
+ minTs = ts
+ }
+ }
toProcess = append(toProcess, toProcessForKey...)
if dnt.elements.Len() == 0 {
delete(ss.pendingByKeys, k)
}
- if OneKeyPerBundle {
+ // A key that yielded nothing, such as one headed by a timer above the
+ // watermark, must not consume the single key slot, or the bundle is empty
+ // and the stage keeps rescheduling on it.
+ if OneKeyPerBundle && len(toProcessForKey) > 0 {
break keysPerBundle
}
}
- // If we're out of data, and timers were not cleared then the watermark is accurate.
- stillSchedulable := !(len(ss.pendingByKeys) == 0 && !timerCleared)
+ // Reschedule only when a later bundle could build something, or a cleared
+ // timer may have held back the minimum pending timestamp.
+ stillSchedulable := timerCleared || ss.hasBuildableDataLocked(watermark)
return toProcess, minTs, newKeys, holdsInBundle, nil, stillSchedulable, 0
}
+// 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 {
Review Comment:
Yes, that's right. It only runs to the end when nothing is buildable, and returns at the first buildable key otherwise (first in map iteration order, so not strictly O(1), but one pass over pendingByKeys at worst).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Prism] Schedule consumers of a self checkpointing source [beam]
Posted by "Eliaaazzz (via GitHub)" <gi...@apache.org>.
Eliaaazzz commented on PR #39572:
URL: https://github.com/apache/beam/pull/39572#issuecomment-5365952037
> > 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:
>
> https://github.com/apache/beam/blob/93f3e051c3ef47238711eb4f313d7e01477c42c3/sdks/python/test-suites/direct/common.gradle#L454
>
> 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](https://github.com/apache/beam/commit/f274b1b35a14123afb17547cae9e2e5683864ab2) here (remove 5s delay) and trigger PostCommit Python Xlang Messaging Direct? Thanks!
Thanks for the good call. I have done that!
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Prism] Schedule consumers of a self checkpointing source [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on code in PR #39572:
URL: https://github.com/apache/beam/pull/39572#discussion_r3822866335
##########
sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:
##########
@@ -1874,22 +1898,51 @@ keysPerBundle:
break
}
}
+ if len(toProcessForKey) > 0 {
+ newKeys.insert(k)
+ // Track the min-timestamp for later watermark handling. Elements pop
+ // in timestamp order, so the first selected one is the earliest.
+ if ts := toProcessForKey[0].timestamp; ts < minTs {
+ minTs = ts
+ }
+ }
toProcess = append(toProcess, toProcessForKey...)
if dnt.elements.Len() == 0 {
delete(ss.pendingByKeys, k)
}
- if OneKeyPerBundle {
+ // A key that yielded nothing, such as one headed by a timer above the
+ // watermark, must not consume the single key slot, or the bundle is empty
+ // and the stage keeps rescheduling on it.
+ if OneKeyPerBundle && len(toProcessForKey) > 0 {
break keysPerBundle
}
}
- // If we're out of data, and timers were not cleared then the watermark is accurate.
- stillSchedulable := !(len(ss.pendingByKeys) == 0 && !timerCleared)
+ // Reschedule only when a later bundle could build something, or a cleared
+ // timer may have held back the minimum pending timestamp.
+ stillSchedulable := timerCleared || ss.hasBuildableDataLocked(watermark)
return toProcess, minTs, newKeys, holdsInBundle, nil, stillSchedulable, 0
}
+// 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 {
Review 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
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Prism] Schedule consumers of a self checkpointing source [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on code in PR #39572:
URL: https://github.com/apache/beam/pull/39572#discussion_r3822866335
##########
sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:
##########
@@ -1874,22 +1898,51 @@ keysPerBundle:
break
}
}
+ if len(toProcessForKey) > 0 {
+ newKeys.insert(k)
+ // Track the min-timestamp for later watermark handling. Elements pop
+ // in timestamp order, so the first selected one is the earliest.
+ if ts := toProcessForKey[0].timestamp; ts < minTs {
+ minTs = ts
+ }
+ }
toProcess = append(toProcess, toProcessForKey...)
if dnt.elements.Len() == 0 {
delete(ss.pendingByKeys, k)
}
- if OneKeyPerBundle {
+ // A key that yielded nothing, such as one headed by a timer above the
+ // watermark, must not consume the single key slot, or the bundle is empty
+ // and the stage keeps rescheduling on it.
+ if OneKeyPerBundle && len(toProcessForKey) > 0 {
break keysPerBundle
}
}
- // If we're out of data, and timers were not cleared then the watermark is accurate.
- stillSchedulable := !(len(ss.pendingByKeys) == 0 && !timerCleared)
+ // Reschedule only when a later bundle could build something, or a cleared
+ // timer may have held back the minimum pending timestamp.
+ stillSchedulable := timerCleared || ss.hasBuildableDataLocked(watermark)
return toProcess, minTs, newKeys, holdsInBundle, nil, stillSchedulable, 0
}
+// 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 {
Review 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
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Prism] Schedule consumers of a self checkpointing source [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on code in PR #39572:
URL: https://github.com/apache/beam/pull/39572#discussion_r3822822050
##########
sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:
##########
@@ -1874,22 +1898,51 @@ keysPerBundle:
break
}
}
+ if len(toProcessForKey) > 0 {
+ newKeys.insert(k)
+ // Track the min-timestamp for later watermark handling. Elements pop
+ // in timestamp order, so the first selected one is the earliest.
+ if ts := toProcessForKey[0].timestamp; ts < minTs {
+ minTs = ts
+ }
+ }
toProcess = append(toProcess, toProcessForKey...)
if dnt.elements.Len() == 0 {
delete(ss.pendingByKeys, k)
}
- if OneKeyPerBundle {
+ // A key that yielded nothing, such as one headed by a timer above the
+ // watermark, must not consume the single key slot, or the bundle is empty
+ // and the stage keeps rescheduling on it.
+ if OneKeyPerBundle && len(toProcessForKey) > 0 {
break keysPerBundle
}
}
- // If we're out of data, and timers were not cleared then the watermark is accurate.
- stillSchedulable := !(len(ss.pendingByKeys) == 0 && !timerCleared)
+ // Reschedule only when a later bundle could build something, or a cleared
+ // timer may have held back the minimum pending timestamp.
+ stillSchedulable := timerCleared || ss.hasBuildableDataLocked(watermark)
return toProcess, minTs, newKeys, holdsInBundle, nil, stillSchedulable, 0
}
+// 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 {
Review Comment:
These tests do use prism runner:
https://github.com/apache/beam/blob/93f3e051c3ef47238711eb4f313d7e01477c42c3/sdks/python/test-suites/direct/common.gradle#L454
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 https://github.com/apache/beam/commit/f274b1b35a14123afb17547cae9e2e5683864ab2 here (remove 5s delay) and trigger PostCommit Python Xlang Messaging Direct? Thanks!
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Prism] Schedule consumers of a self checkpointing source [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #39572:
URL: https://github.com/apache/beam/pull/39572#issuecomment-5266775798
Reminder, please take a look at this pr: @jrmccluskey @shunping
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Prism] Schedule consumers of a self checkpointing source [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on PR #39572:
URL: https://github.com/apache/beam/pull/39572#issuecomment-5144763727
> [!CAUTION]
> The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Prism] Schedule consumers of a self checkpointing source [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on code in PR #39572:
URL: https://github.com/apache/beam/pull/39572#discussion_r3822822050
##########
sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:
##########
@@ -1874,22 +1898,51 @@ keysPerBundle:
break
}
}
+ if len(toProcessForKey) > 0 {
+ newKeys.insert(k)
+ // Track the min-timestamp for later watermark handling. Elements pop
+ // in timestamp order, so the first selected one is the earliest.
+ if ts := toProcessForKey[0].timestamp; ts < minTs {
+ minTs = ts
+ }
+ }
toProcess = append(toProcess, toProcessForKey...)
if dnt.elements.Len() == 0 {
delete(ss.pendingByKeys, k)
}
- if OneKeyPerBundle {
+ // A key that yielded nothing, such as one headed by a timer above the
+ // watermark, must not consume the single key slot, or the bundle is empty
+ // and the stage keeps rescheduling on it.
+ if OneKeyPerBundle && len(toProcessForKey) > 0 {
break keysPerBundle
}
}
- // If we're out of data, and timers were not cleared then the watermark is accurate.
- stillSchedulable := !(len(ss.pendingByKeys) == 0 && !timerCleared)
+ // Reschedule only when a later bundle could build something, or a cleared
+ // timer may have held back the minimum pending timestamp.
+ stillSchedulable := timerCleared || ss.hasBuildableDataLocked(watermark)
return toProcess, minTs, newKeys, holdsInBundle, nil, stillSchedulable, 0
}
+// 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 {
Review Comment:
These tests do use prism runner:
https://github.com/apache/beam/blob/93f3e051c3ef47238711eb4f313d7e01477c42c3/sdks/python/test-suites/direct/common.gradle#L454
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 https://github.com/apache/beam/commit/f274b1b35a14123afb17547cae9e2e5683864ab2 here (remove 5s delay) and trigger PostCommit Python Xlang Messaging Direct? Thanks!
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Prism] Schedule consumers of a self checkpointing source [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on PR #39572:
URL: https://github.com/apache/beam/pull/39572#issuecomment-5357866278
> 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:
https://github.com/apache/beam/blob/93f3e051c3ef47238711eb4f313d7e01477c42c3/sdks/python/test-suites/direct/common.gradle#L454
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 https://github.com/apache/beam/commit/f274b1b35a14123afb17547cae9e2e5683864ab2 here (remove 5s delay) and trigger PostCommit Python Xlang Messaging Direct? Thanks!
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Prism] Schedule consumers of a self checkpointing source [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #39572:
URL: https://github.com/apache/beam/pull/39572#issuecomment-5145499738
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment `assign set of reviewers`
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Prism] Schedule consumers of a self checkpointing source [beam]
Posted by "Eliaaazzz (via GitHub)" <gi...@apache.org>.
Eliaaazzz commented on code in PR #39572:
URL: https://github.com/apache/beam/pull/39572#discussion_r3826722282
##########
sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:
##########
@@ -1874,22 +1898,51 @@ keysPerBundle:
break
}
}
+ if len(toProcessForKey) > 0 {
+ newKeys.insert(k)
+ // Track the min-timestamp for later watermark handling. Elements pop
+ // in timestamp order, so the first selected one is the earliest.
+ if ts := toProcessForKey[0].timestamp; ts < minTs {
+ minTs = ts
+ }
+ }
toProcess = append(toProcess, toProcessForKey...)
if dnt.elements.Len() == 0 {
delete(ss.pendingByKeys, k)
}
- if OneKeyPerBundle {
+ // A key that yielded nothing, such as one headed by a timer above the
+ // watermark, must not consume the single key slot, or the bundle is empty
+ // and the stage keeps rescheduling on it.
+ if OneKeyPerBundle && len(toProcessForKey) > 0 {
break keysPerBundle
}
}
- // If we're out of data, and timers were not cleared then the watermark is accurate.
- stillSchedulable := !(len(ss.pendingByKeys) == 0 && !timerCleared)
+ // Reschedule only when a later bundle could build something, or a cleared
+ // timer may have held back the minimum pending timestamp.
+ stillSchedulable := timerCleared || ss.hasBuildableDataLocked(watermark)
return toProcess, minTs, newKeys, holdsInBundle, nil, stillSchedulable, 0
}
+// 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 {
Review Comment:
Yes, with one nuance: the full pass happens exactly when no key outside a running bundle heads with data or a ripe timer, which is the round that parks the stage. So it is nothing buildable rather than no new data in the strict sense, since a ready timer also counts. A productive round stops at the first buildable key it meets.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Prism] Schedule consumers of a self checkpointing source [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on code in PR #39572:
URL: https://github.com/apache/beam/pull/39572#discussion_r3827055085
##########
sdks/python/apache_beam/io/external/xlang_jmsio_it_test.py:
##########
@@ -143,11 +137,11 @@ def subscribe():
connection_param),
queue=sink_queue,
expansion_service=self.expansion_service))
- publisher.start()
result = p.run()
+ publisher.start()
Review Comment:
```suggestion
publisher.start()
result = p.run()
```
revert this change to trigger tests again
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org