Skip to content

[#32222] Actually maintain the heap invariant for timers. #33270

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lostluck merged 5 commits into from
Dec 4, 2024
Merged

[#32222] Actually maintain the heap invariant for timers. #33270

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
bd2271b
[#32222] Actually maintain the heap invariant for timers.
lostluck Dec 3, 2024
9bf8fe5
Include in CHANGES.md
lostluck Dec 3, 2024
2e799bf
Extend GC by one instant.
lostluck Dec 4, 2024
7516b98
Fix the python test instead.
lostluck Dec 4, 2024
41e92e8
pyfmt
lostluck Dec 4, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGES.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,7 @@
## Bugfixes

* Fixed X (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
* Fixed EventTimeTimer ordering in Prism. ([#32222](https://github.com/apache/beam/issues/32222)).

## Security Fixes
* Fixed (CVE-YYYY-NNNN)[https://www.cve.org/CVERecord?id=CVE-YYYY-NNNN] (Java/Python/Go) ([#X](https://github.com/apache/beam/issues/X)).
Expand Down
9 changes: 1 addition & 8 deletions runners/prism/java/build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -178,10 +178,7 @@ def sickbayTests = [
'org.apache.beam.sdk.transforms.ParDoTest$StateTests.testTwoRequiresTimeSortedInputWithLateData',
'org.apache.beam.sdk.transforms.ParDoTest$StateTests.testRequiresTimeSortedInputWithLateData',

// Timer race condition/ordering issue in Prism.
'org.apache.beam.sdk.transforms.ParDoTest$TimerTests.testTwoTimersSettingEachOtherWithCreateAsInputUnbounded',

// Missing output due to timer skew.
// Missing output due to processing time timer skew.
'org.apache.beam.sdk.transforms.ParDoTest$TimestampTests.testProcessElementSkew',

// TestStream + BundleFinalization.
Expand Down Expand Up @@ -241,10 +238,6 @@ def createPrismValidatesRunnerTask = { name, environmentType ->
excludeCategories 'org.apache.beam.sdk.testing.UsesMultimapState'
}
filter {
// Hangs forever with prism. Put here instead of sickbay to allow sickbay runs to terminate.
// https://github.com/apache/beam/issues/32222
excludeTestsMatching 'org.apache.beam.sdk.transforms.ParDoTest$TimerTests.testEventTimeTimerOrderingWithCreate'

for (String test : sickbayTests) {
excludeTestsMatching test
}
Expand Down
7 changes: 5 additions & 2 deletions sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1159,7 +1159,7 @@ func (ss *stageState) AddPending(newPending []element) int {
}
ss.pendingByKeys[string(e.keyBytes)] = dnt
}
dnt.elements.Push(e)
heap.Push(&dnt.elements, e)
Copy link
Contributor Author

@lostluck lostluck Dec 4, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This fixes maintaining the heap invariant.


if e.IsTimer() {
if lastSet, ok := dnt.timers[timerKey{family: e.family, tag: e.tag, window: e.window}]; ok {
Expand Down Expand Up @@ -1576,6 +1576,8 @@ func (ss *stageState) updateWatermarks(em *ElementManager) set[string] {
// They'll never be read in again.
for _, wins := range ss.sideInputs {
for win := range wins {
// TODO(#https://github.com/apache/beam/issues/31438):
// Adjust with AllowedLateness
// Clear out anything we've already used.
if win.MaxTimestamp() < newOut {
delete(wins, win)
Expand All @@ -1584,7 +1586,8 @@ func (ss *stageState) updateWatermarks(em *ElementManager) set[string] {
}
for _, wins := range ss.state {
for win := range wins {
// Clear out anything we've already used.
// TODO(#https://github.com/apache/beam/issues/31438):
// Adjust with AllowedLateness
if win.MaxTimestamp() < newOut {
delete(wins, win)
}
Expand Down
9 changes: 6 additions & 3 deletions sdks/python/apache_beam/runners/portability/fn_api_runner/fn_runner_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -776,7 +776,8 @@ def process(
state.clear()
yield buffer
else:
timer.set(ts + 1)
# Set the timer to fire within it's window.
timer.set(ts + (1 - timestamp.Duration(micros=1000)))

@userstate.on_timer(timer_spec)
def process_timer(self, state=beam.DoFn.StateParam(state_spec)):
Expand All @@ -790,8 +791,10 @@ def is_buffered_correctly(actual):
# Acutal should be a grouping of the inputs into batches of size
# at most buffer_size, but the actual batching is nondeterministic
# based on ordering and trigger firing timing.
self.assertEqual(sorted(sum((list(b) for b in actual), [])), elements)
self.assertEqual(max(len(list(buffer)) for buffer in actual), buffer_size)
self.assertEqual(
Copy link
Contributor Author

@lostluck lostluck Dec 4, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added the actual output to the asserts help with debugging, hence the line reflows.

sorted(sum((list(b) for b in actual), [])), elements, actual)
self.assertEqual(
max(len(list(buffer)) for buffer in actual), buffer_size, actual)
if windowed:
# Elements were assigned to windows based on their parity.
# Assert that each grouping consists of elements belonging to the
Expand Down
Loading