[Spark] Fire processing-time timers in timestamp order - #39825
Conversation
SparkTimerInternals.getNextProcessingTimer() picked the eligible timer with the latest timestamp, so timers due in the same micro-batch fired in reverse order and a later callback observed state its predecessors had not written yet. It now returns the earliest one, and the expired timers fired by TimerUtils.triggerExpiredTimers are sorted the same way. Earliest-first also surfaces superseded settings of a re-set timer, so timers are now stored in a map keyed by namespace, id and family: a later setting replaces the prior one, per the TimerInternals contract, and timers restored from state that predates this change collapse to the setting with the latest target.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #39825 +/- ##
============================================
+ Coverage 55.95% 58.14% +2.19%
- Complexity 2284 13139 +10855
============================================
Files 1116 2524 +1408
Lines 175741 265686 +89945
Branches 1475 10822 +9347
============================================
+ Hits 98332 154495 +56163
- Misses 74901 105353 +30452
- Partials 2508 5838 +3330
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment |
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
|
Assigning reviewers: R: @chamikaramj added as fallback since no labels match configuration 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). |
SparkTimerInternals.getNextProcessingTimer()picked the eligible processing-time timer with the latest timestamp, so when several timers for a key were due in the same micro-batch their@OnTimercallbacks ran latest-first: a callback scheduled for T2 observed state that the callback for an earlier T1 had not written yet.TimerUtils.getExpiredTimerscollected the expired-timer sweep from an unordered set, so that path had no defined order either.Timers now fire earliest-first, matching
InMemoryTimerInternalsand the other runners, and the timers fired bytriggerExpiredTimersare sorted by timestamp; the deletion-only sweep stays unsorted.Earliest-first would also have surfaced superseded settings of a re-set timer, which the old set-based store kept alongside the replacement. Timers are therefore stored in a map keyed by namespace, id and family, so a later setting replaces the prior one, per the
TimerInternalscontract. Timers restored from state that predates this change collapse to the setting with the latest target; a checkpoint written before it cannot be recovered by an upgraded application anyway (Spark's DStream checkpoints do not survive application upgrades), so the merge only normalizes bytes Spark already declares unrecoverable.testProcessingTimersFireInTimestampOrderfails on master with the later timer drained first;testSettingATimerAgainClearsThePriorSettingandtestAddTimersKeepsTheLatestSettingOfATimerpin the replacement semantics.Fixes #39824.
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
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
GitHub Actions Tests Status (on master branch)
See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.