Jonathan Gao created SPARK-58751:
------------------------------------
Summary: TransformWithState leaks Python workers when
initialization fails
Key: SPARK-58751
URL: https://issues.apache.org/jira/browse/SPARK-58751
Project: Spark
Issue Type: Bug
Components: Structured Streaming, PySpark
Affects Versions: 4.0.0
Reporter: Jonathan Gao
`TransformWithStateInPySpark` can create a Python worker during driver-side
initialization before initialization completes. If initialization fails after
worker creation, the previous lifecycle does not reach `stop()`, so the worker
and associated resources can remain alive for the driver lifetime.
Repeated streaming restarts can accumulate leaked workers and eventually
prevent subsequent isolated workers from starting.
The proposed fix moves the driver-side pre-initialization runner lifecycle
(`init`, `process`, and `stop`) into a cleanup-protected path so `stop()` runs
on initialization and processing failures. It also preserves the original
initialization failure when cleanup itself throws, and null-guards the
state-server daemon thread because that thread is assigned only near the end of
initialization.
Regression coverage uses a stubbed pre-init runner, so no real Python worker is
required. The suite covers:
* initialization failure after worker creation still stops the runner
* repeated initialization failures do not accumulate live workers
* cleanup failure is attached as suppressed rather than masking the
initialization error
* process failure is still wrapped as a driver-worker crash and still stops the
runner
* success path stops the runner exactly once
* stopping before state-server startup does not throw a null-pointer exception
Pull request: https://github.com/apache/spark/pull/57941
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
