Hyukjin Kwon created SPARK-57998:
------------------------------------
Summary: Deflake 'python foreachBatch process: process terminates
after query is stopped'
Key: SPARK-57998
URL: https://issues.apache.org/jira/browse/SPARK-57998
Project: Spark
Issue Type: Bug
Components: Connect, Tests
Affects Versions: 5.0.0
Reporter: Hyukjin Kwon
The `python foreachBatch process: process terminates after query is stopped`
test in `SparkConnectSessionHolderSuite` still flakes after SPARK-56586 (~1/8
branch-4.2 Non-ANSI runs), e.g.
https://github.com/apache/spark/actions/runs/28841324286/job/85537586435 :
{code}
- python foreachBatch process: process terminates after query is stopped ***
FAILED ***
java.util.concurrent.TimeoutException: Stream Execution thread ... failed to
stop within 30000 milliseconds (specified by spark.sql.streaming.stopTimeout).
Cause: ... at
StreamingForeachBatchHelper$.$anonfun$pythonForeachBatchWrapper$3(StreamingForeachBatchHelper.scala:172)
// dataIn.readInt()
{code}
Root cause: `query.stop()` interrupts the stream thread and joins for
`stopTimeout`, but that thread is parked in the *non-interruptible* native
`dataIn.readInt()` waiting for the in-flight foreachBatch Python callback to
return. The interrupt is a no-op until the batch's socket round-trip completes.
On a busy CI host that round-trip occasionally exceeds the tight 30s
`stopTimeout` that SPARK-56586 introduced, turning a merely-slow (not stuck)
stop into a hard `TimeoutException`.
The real hang guard is the outer machinery SPARK-56586 also added:
`awaitTestBodyInNewThread` caps each attempt at 2 minutes and, on timeout,
closes the worker sockets via `cleanUpAll()` to unblock `readInt()`, then
retries. The 30s `stopTimeout` is therefore redundant with -- and tighter than
-- that 2-minute cap, and is the sole source of the spurious failure.
Fix (test-only): raise `spark.sql.streaming.stopTimeout` to sit comfortably
above the 2-minute per-attempt cap (300s) so the outer bound is always the
deterministic guard, while keeping `stopTimeout` finite (never the default
0/forever).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
