Eliaaazzz opened a new issue, #39824: URL: https://github.com/apache/beam/issues/39824
### What happened? In the classic Spark runner's streaming mode, stateful ParDo fires processing-time timers through `SparkTimerInternals.getNextProcessingTimer()`, which picks the eligible timer with the latest timestamp: https://github.com/apache/beam/blob/6a8eee94aa4d3a95e42345eb60219884da3d7031/runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkTimerInternals.java#L219-L228 When several timers for a key are due in the same micro-batch, which the default 500 ms batch interval makes routine, their `@OnTimer` callbacks run latest-first. A callback scheduled for T2 observes state that the callback for an earlier T1 has not written yet, and the callbacks see time going backwards. Timers elsewhere in Beam fire in timestamp order: `InMemoryTimerInternals` drains an ordered set, and the runners keep the same order. `TimerUtils.getExpiredTimers` has the same problem on the expired-timer sweep: it collects from an unordered `Set`, so the sweep fires in arbitrary order. Reproducible with a unit test: set two eligible processing-time timers on a `SparkTimerInternals` and drain it the way `ParDoStateUpdateFn.SparkTimerInternalsIterator` does; the later timer comes out first. I have a fix ready: return the earliest eligible timer, and sort the expired-timer sweep by timestamp. ### Issue Priority Priority: 2 (default / most bugs should be filed as P2) ### Issue Components - [ ] Component: Python SDK - [X] Component: Java SDK - [ ] Component: Go SDK - [ ] Component: Typescript SDK - [ ] Component: IO connector - [ ] Component: Beam YAML - [ ] Component: Beam examples - [ ] Component: Beam playground - [ ] Component: Beam katas - [ ] Component: Website - [ ] Component: Infrastructure - [X] Component: Spark Runner - [ ] Component: Flink Runner - [ ] Component: Samza Runner - [ ] Component: Twister2 Runner - [ ] Component: Hazelcast Jet Runner - [ ] Component: Google Cloud Dataflow Runner -- 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]
