Eliaaazzz opened a new issue, #39822: URL: https://github.com/apache/beam/issues/39822
### What happened? In the classic Spark runner's streaming mode, `GlobalWatermarkHolder.advance()` rebuilds the global watermark map from only the sources that have a queued update for the completed batch, and then overwrites the stored map with the result: https://github.com/apache/beam/blob/6a8eee94aa4d3a95e42345eb60219884da3d7031/runners/spark/src/main/java/org/apache/beam/runners/spark/util/GlobalWatermarkHolder.java#L148-L160 `computeNewWatermarks` skips every source whose queue is empty, so a source that reports no progress in that batch disappears from the map entirely, together with its last known watermark. Downstream, `SparkTimerInternals.forStreamFromSources` takes the slowest watermark over the source ids it finds in the map and silently skips ids that are missing. Once an idle source has been dropped, the input watermark of every stage fed by it jumps ahead to the watermark of the remaining sources. Timers then fire as if the idle source had caught up, and elements it produces later are dropped as late. The queues do not stay in lockstep between sources, so this is easy to hit in any pipeline with more than one source. For example `CreateStream` enqueues its whole pre-baked watermark script at translation time, so with two of them the shorter script drains first and that source's watermark vanishes a batch later. I believe this is one deterministic cause of the watermark races described in #23129. Reproducible with a unit test against `GlobalWatermarkHolderTest`: source 1 and 2 both report in the first batch, only source 1 reports in the second. After the second `advance()`, `get()` no longer contains source 2 at all. I have a fix ready: merge the per-source updates over the stored map instead of replacing it, so an idle source keeps its last watermark until it reports again. ### 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]
