damccorm opened a new pull request, #41137: URL: https://github.com/apache/beam/pull/41137
## Summary Fixes a driver concurrency race in Spark 4 Structured Streaming when multiple leaf `StreamingQuery` instances share an upstream `TransformWithState` logical plan node (such as `GroupByKey` or stateful `ParDo` consumed by multiple `PAssert` queries in `GroupByKeyTest$WindowTests.testGroupByKeyMultipleWindows`). ## Root Cause Failed master run: https://github.com/apache/beam/actions/runs/42658100679/job/129301949163 (`PostCommit Java ValidatesRunner Spark4`). In Spark 4 (`org.apache.spark.sql.execution.streaming.TransformWithStateExec`), `getDriverProcessorHandle()` and `closeProcessorHandle()` are invoked on the driver by each `StreamingQuery`'s `StreamExecution` thread during micro-batch planning (`IncrementalExecution$StateSchemaAndOperatorMetadataRule` -> `getColFamilySchemas` / `getStateVariableInfos`): 1. `getDriverProcessorHandle()` calls `statefulProcessor.setHandle(driverProcessorHandle)` followed by `statefulProcessor.init(outputMode, timeMode)`. 2. `closeProcessorHandle()` calls `statefulProcessor.close()` followed by `statefulProcessor.setHandle(null)`. Because `StatefulProcessor.setHandle` and `StatefulProcessor.getHandle` are `public final` methods that mutate a single non-thread-safe `private StatefulProcessorHandle statefulProcessorHandle` field on the `StatefulProcessor` instance, sharing the same `TransformWithState` logical plan node across sibling leaf `StreamingQuery`s causes their `StreamExecution` threads to concurrently set and clear `statefulProcessorHandle = null` mid-`init()`, throwing: ``` org.apache.spark.SparkRuntimeException: [STATE_STORE_HANDLE_NOT_INITIALIZED] State store handle is not initialized. at org.apache.spark.sql.streaming.StatefulProcessor.getHandle(StatefulProcessor.scala:146) at org.apache.beam.runners.spark.structuredstreaming.translation.streaming.BeamStatefulProcessor.init(BeamStatefulProcessor.java:110) at org.apache.spark.sql.execution.streaming.TransformWithStateExec.getDriverProcessorHandle(TransformWithStateExec.scala:155) ``` ## Fix - In `StreamingEvaluationContext.startQuery`, clone `StatefulProcessor` instances inside `TransformWithState` logical plan nodes for additional leaf queries (`leafIndex > 0`) via `SerializableUtils.clone(tws.statefulProcessor())` so each leaf `StreamingQuery` has its own isolated `StatefulProcessor` instance on the driver. - Add `WindowedGroupByKeyStreamingTest.testGroupByKeySharedAcrossMultipleLeaves` to verify multiple `PAssert` leaf queries consuming a shared `GroupByKey`. ### Verification - Triggered `beam_PostCommit_Java_ValidatesRunner_Spark4.yml` workflow run on branch: [`42696643750`](https://github.com/apache/beam/actions/runs/42696643750) -- 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]
