This is an automated email from the ASF dual-hosted git repository.
gaogaotiantian pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 08462255a8a4 [SPARK-59912][PYTHON][TEST] Fix flaky WorkerReuseTest and
WorkerPoolCrashTest by pinning spark.python.use.daemon
08462255a8a4 is described below
commit 08462255a8a440667e19282327e77637020405e9
Author: Tian Gao <[email protected]>
AuthorDate: Thu Oct 1 12:02:17 2026 -0700
[SPARK-59912][PYTHON][TEST] Fix flaky WorkerReuseTest and
WorkerPoolCrashTest by pinning spark.python.use.daemon
### What changes were proposed in this pull request?
Pin `spark.python.use.daemon=true` in `WorkerReuseTest` and
`WorkerPoolCrashTest` (`python/pyspark/tests/test_worker.py`) by switching them
to `ReusedPySparkTestCase` and overriding `conf()`, so they no longer rely on
the gateway JVM's ambient default for `spark.python.use.daemon`.
### Why are the changes needed?
`WorkerReuseTest.test_reuse_worker_of_parallelize_range` and
`WorkerPoolCrashTest.test_worker_crash` assert that Python worker PIDs are
reused across consecutive jobs. Worker reuse (and thus stable PIDs) only
happens on the daemon path: the forked daemon child loops and serves task after
task in the same process. The simple-worker path
(`spark.python.use.daemon=false`) runs `python -m pyspark.worker` once per
task, so every task gets a fresh process and a fresh PID, and the reuse asse
[...]
PySpark tests share one gateway JVM per process. The first `SparkContext`
to launch the gateway propagates its conf into the JVM's system properties (via
`JavaMainApplication`), and every later `SparkConf(loadDefaults=True)` inherits
them (same mechanism as SPARK-59885). Sibling suites in this file set
`spark.python.use.daemon=false` (`SimpleWorkerTests`,
`WorkerSegfaultNonDaemonTest`). When one of those launches the gateway first,
the leaked `use.daemon=false` makes the two reuse sui [...]
Pinning `spark.python.use.daemon=true` on each suite's own `SparkContext`
overrides the leaked default (the flag is read from the context's own conf via
`SparkEnv.get.conf`), so the assertions are independent of test ordering.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Existing `WorkerReuseTest` / `WorkerPoolCrashTest`. Test-only change (a
config pin); no new tests needed.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #59172 from gaogaotiantian/SPARK-59912-worker-reuse-daemon.
Authored-by: Tian Gao <[email protected]>
Signed-off-by: Tian Gao <[email protected]>
---
python/pyspark/tests/test_worker.py | 26 +++++++++++++++++++++++---
1 file changed, 23 insertions(+), 3 deletions(-)
diff --git a/python/pyspark/tests/test_worker.py
b/python/pyspark/tests/test_worker.py
index 402821a75288..a7468ff0ff9e 100644
--- a/python/pyspark/tests/test_worker.py
+++ b/python/pyspark/tests/test_worker.py
@@ -31,7 +31,7 @@ except ImportError:
from py4j.protocol import Py4JJavaError
from pyspark import SparkConf, SparkContext
-from pyspark.testing.utils import PySparkTestCase, QuietTest,
ReusedPySparkTestCase, eventually
+from pyspark.testing.utils import QuietTest, ReusedPySparkTestCase, eventually
class WorkerTests(ReusedPySparkTestCase):
@@ -205,7 +205,19 @@ class WorkerTests(ReusedPySparkTestCase):
self.assertRegex(str(e), "exception with δΈ")
-class WorkerReuseTest(PySparkTestCase):
+class WorkerReuseTest(ReusedPySparkTestCase):
+ @classmethod
+ def conf(cls):
+ # Pin spark.python.use.daemon=true. Worker reuse (hence stable worker
PIDs) only
+ # happens on the daemon path; the simple-worker path spawns a fresh
process per
+ # task. Sibling suites here run with use.daemon=false and, because all
suites in
+ # this file share one gateway JVM, whichever launches it first leaks
its conf into
+ # the JVM system properties (SPARK-59885). Pinning it keeps this reuse
assertion
+ # independent of test ordering.
+ conf = super().conf()
+ conf.set("spark.python.use.daemon", "true")
+ return conf
+
@eventually(catch_assertions=True)
def test_reuse_worker_of_parallelize_range(self):
rdd = self.sc.parallelize(range(20), 8)
@@ -307,7 +319,15 @@ class WorkerSegfaultNonDaemonTest(WorkerSegfaultTest):
return _conf
-class WorkerPoolCrashTest(PySparkTestCase):
+class WorkerPoolCrashTest(ReusedPySparkTestCase):
+ @classmethod
+ def conf(cls):
+ # Pin daemon so worker reuse (stable PIDs) is independent of test
ordering; see
+ # WorkerReuseTest.conf for the shared-gateway leak rationale
(SPARK-59885).
+ conf = super().conf()
+ conf.set("spark.python.use.daemon", "true")
+ return conf
+
def test_worker_crash(self):
# SPARK-47565: Kill a worker that is currently idling
rdd = self.sc.parallelize(range(20), 4)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
