This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 11ea856513a8 [SPARK-57972][SS] Preserve the cause when
StreamingPythonRunner initialization fails
11ea856513a8 is described below
commit 11ea856513a821168662457c31133c70366c1982
Author: gnanda <[email protected]>
AuthorDate: Thu Jul 9 08:58:19 2026 +0900
[SPARK-57972][SS] Preserve the cause when StreamingPythonRunner
initialization fails
### What changes were proposed in this pull request?
After a python worker initialization exception, preserve the original
exception as the cause of the thrown exception
### Why are the changes needed?
Improve observability after failures
### Does this PR introduce _any_ user-facing change?
Yes - additional error context after a python runner initialization failure.
### How was this patch tested?
Existing CI. No behavior changes
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #57050 from gnanda/stack/preserveinitfailurecause.
Lead-authored-by: gnanda <[email protected]>
Co-authored-by: Gurpreet Nanda <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>
(cherry picked from commit e9d2378b5a27f71b7d63078692bdc3168e43f87f)
Signed-off-by: Hyukjin Kwon <[email protected]>
---
.../spark/api/python/StreamingPythonRunner.scala | 18 ++++++++++++------
1 file changed, 12 insertions(+), 6 deletions(-)
diff --git
a/core/src/main/scala/org/apache/spark/api/python/StreamingPythonRunner.scala
b/core/src/main/scala/org/apache/spark/api/python/StreamingPythonRunner.scala
index 7b659941c5ef..c1f3df98f73c 100644
---
a/core/src/main/scala/org/apache/spark/api/python/StreamingPythonRunner.scala
+++
b/core/src/main/scala/org/apache/spark/api/python/StreamingPythonRunner.scala
@@ -113,9 +113,9 @@ private[spark] class StreamingPythonRunner(
dataIn.readInt()
} catch {
case e: java.net.SocketTimeoutException =>
- throw new
StreamingPythonRunnerInitializationTimeoutException(e.getMessage)
+ throw new
StreamingPythonRunnerInitializationTimeoutException(e.getMessage, e)
case e: Exception =>
- throw new
StreamingPythonRunnerInitializationCommunicationException(e.getMessage)
+ throw new
StreamingPythonRunnerInitializationCommunicationException(e.getMessage, e)
}
// Set timeout back to the original timeout
@@ -132,15 +132,21 @@ private[spark] class StreamingPythonRunner(
(dataOut, dataIn)
}
- class StreamingPythonRunnerInitializationCommunicationException(errMessage:
String)
+ class StreamingPythonRunnerInitializationCommunicationException(
+ errMessage: String,
+ cause: Throwable = null)
extends SparkPythonException(
errorClass =
"STREAMING_PYTHON_RUNNER_INITIALIZATION_COMMUNICATION_FAILURE",
- messageParameters = Map("msg" -> errMessage))
+ messageParameters = Map("msg" -> errMessage),
+ cause = cause)
- class StreamingPythonRunnerInitializationTimeoutException(errMessage: String)
+ class StreamingPythonRunnerInitializationTimeoutException(
+ errMessage: String,
+ cause: Throwable = null)
extends SparkPythonException(
errorClass = "STREAMING_PYTHON_RUNNER_INITIALIZATION_TIMEOUT_FAILURE",
- messageParameters = Map("msg" -> errMessage))
+ messageParameters = Map("msg" -> errMessage),
+ cause = cause)
class StreamingPythonRunnerInitializationException(resFromPython: Int,
errMessage: String)
extends SparkPythonException(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
