[
https://issues.apache.org/jira/browse/SPARK-57998?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Hyukjin Kwon resolved SPARK-57998.
----------------------------------
Resolution: Duplicate
Duplicate of SPARK-57963 (PR https://github.com/apache/spark/pull/57041), which
fixes the real root cause: Hadoop FileContext.rename() clears the interrupt
status set by StreamExecution.stop(), so the micro-batch stream thread never
observes the interrupt and stop() times out. Closing this in favor of that
product-code fix; raising the test stopTimeout would only mask the symptom.
> 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
> Priority: Minor
>
> 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]
