damccorm commented on PR #41137: URL: https://github.com/apache/beam/pull/41137#issuecomment-6831006582
> Thanks for chasing this down, the diagnosis is right and one `StatefulProcessor` per query thread is the invariant we need. Two things before I can approve: > > 1. `SerializableUtils.clone` also copies the `MetricsAccumulator` that `ParDoStatefulProcessor` holds. Spark only merges the accumulator the driver registered (see the note in `MetricsAccumulator.getInstance`), so a stateful ParDo in every leaf after the first would stop reporting metrics. > 2. The copies for leaves 1..n are taken after query 0 is started on the original instance, and `StatefulProcessor` keeps its handle in a non transient field that query 0 sets and clears while planning its first batch. That narrows the race, it doesn't remove it. > > Both go away if `BeamStatefulProcessor` gets a copy method that the two subclasses implement with their constructor and the same final fields, instead of a Java serialization round trip. It keeps the registered accumulator and reads no Spark handle, so the order against query 0 no longer matters. > > Smaller points: could you align `ofRows` with the existing `Dataset$.MODULE$` call sites (`BoundedDatasetFactory`, `UnboundedSourceDataset`), add a short comment on why leaf 0 keeps the original, and mention #41198 as the long term fix (one query per pipeline would remove this code). Does the new test fail without the change? A stateful ParDo with a counter and two leaves would also cover point 1. And could you please bump the Spark 4 ValidatesRunner trigger file so the PostCommit runs here? Thanks - I think I addressed the feedback. I am a little out of my wheelhouse here (this is part of a broader effort I'm working on to automate away some of our test flakes/toil around that), so there may be pieces I'm missing. If so, feel free to either help me through it or try to fork/fix yourself as you have time available - whatever is easiest for you depending on your availability. Regardless, it would be good to clean up the failures introduced here. -- 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]
