[SPARK-53759][PYTHON][3.5] Fix missing close in simple-worker path

[anblanco]

What changes were proposed in this pull request?

Backport of the SPARK-53759 fix to branch-3.5 (parent PR: #55201, targeting branch-4.1).

Add an explicit sock_file.close() in a finally block wrapping the main() call in all 3 worker files' __main__ blocks on this branch (worker.py, foreach_batch_worker.py, listener_worker.py — the only worker files that exist on branch-3.5).

The simple-worker codepath (used on Windows and when spark.python.use.daemon=false) calls main(sock_file, sock_file) as the last statement. When main() returns, the process exits without flushing the BufferedRWPair write buffer. On Python 3.12+, changed GC finalization ordering (cpython#97922) causes the underlying socket to close before BufferedRWPair.__del__ can flush, resulting in data loss and EOFException on the JVM side.

On master, SPARK-55665 refactored worker.py to use a get_sock_file_to_executor() context manager with close() in the finally block, which covers this structurally. Since branch-3.5 lacks that refactoring, this backport adds explicit sock_file.close() calls instead. BufferedRWPair.close() flushes internally, so this is equivalent.

Regression tests are in a separate PR targeting master: #55223.

Why are the changes needed?

The simple-worker path crashes on Python 3.12+ with EOFException:

• Windows: Always uses simple-worker (os.fork() unavailable) — this crash is reproducible on every worker-dependent operation with Python 3.12+
• Linux/macOS: Reproducible when spark.python.use.daemon=false

Every worker-dependent operation (rdd.map(), createDataFrame(), UDFs) fails with:

org.apache.spark.SparkException: Python worker exited unexpectedly (crashed)
Caused by: java.io.EOFException

This crash reproduces on all currently supported PySpark releases (3.5.8, 4.0.2, 4.1.1) with Python 3.12 and 3.13.

Root cause: The daemon path (daemon.py) wraps worker_main() in try/finally with outfile.flush(). The simple-worker path (worker.py __main__) does not — it relies on BufferedRWPair.__del__ during interpreter shutdown, which is not guaranteed and no longer works as expected on Python 3.12+ due to changed GC finalization ordering.

Does this PR introduce any user-facing change?

Yes. Resolves a crash that prevents PySpark from functioning on Windows with Python 3.12+, and on Linux/macOS with spark.python.use.daemon=false on Python 3.12+.

How was this patch tested?

• Red/green verified in local WSL build (Python 3.12.3): unfixed build fails with EOFException, fixed build passes 9/9 SimpleWorkerTests
• Regression tests are in a separate PR targeting master: [SPARK-56399][PYTHON] Add SimpleWorkerTests for non-daemon worker path #55223
• Verification matrix from standalone reproducer (anblanco/spark53759-reproducer):

Platform	Python	Unpatched	Patched
Windows 11	3.11.9	PASS	PASS (harmless)
Windows 11	3.12.10	FAIL	PASS
Windows 11	3.13.3	FAIL	PASS
Linux (Ubuntu 24.04)	3.12.3	FAIL	PASS

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Opus 4.6 (Anthropic)

[Shrividya]

Thanks for this PR!

[dongjoon-hyun]

I'm blocking this PR according to the community backporting policy. The branch-4.0/4.1 PRs are not merged yet to master, @allisonwang-db .

• #55223

[dongjoon-hyun]

+1, LGTM. Thank you, @anblanco and all.

[dongjoon-hyun]

Merged to branch-3.5.