[Bug]: Spark streaming fires processing-time timers in reverse timestamp order #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
• 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
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner