spark-test commented on code in PR #59860:
URL: https://github.com/apache/spark/pull/59860#discussion_r4404098577
##########
python/pyspark/worker.py:
##########
@@ -76,6 +78,33 @@ def report_metrics(outfile: BinaryIO, metrics: dict[str,
int]) -> None:
write_with_length(payload, outfile)
+def report_init_response(
+ outfile: BinaryIO, error: Optional[BaseException] = None,
is_payload_error: bool = False
+) -> None:
+ """
+ Write the init response as a length-prefixed JSON object after
START_OF_INIT_RESPONSE.
+
+ It is sent exactly once per task, before any output: ``{}`` once the task
is initialized, or
+ ``{"error": {...}}`` if initialization failed, in which case no output
follows. An init error
+ carries the exception message, the traceback as the JVM reports it, the
exception class name,
+ and whether the worker failed to load the UDF payload (``isPayloadError``)
rather than to set
+ up the task. Must be called while handling ``error``.
+ """
+ response: dict[str, Any] = {}
+ if error is not None:
+ response["error"] = {
+ "message": str(error),
Review Comment:
**Non-blocking (P2):** `str(error)` runs before any byte of the init
response is written, and nothing guards it. If an initialization-phase
exception cannot be converted to a string, for example a payload whose
unpickling raises an exception whose `__str__` raises or returns a non-string,
this line raises. The new fallback in `invoke_udf` then only prints to stderr,
and the worker exits without writing any frame, while the JVM is blocked in
`handleInitResponse` waiting for one:
- With the default `spark.python.use.daemon=true`, the forked worker keeps
draining the socket until the JVM closes it. The JVM has already written all
input and only waits to read, and `spark.python.worker.idleTimeoutSeconds`
defaults to 0, so the task hangs until it is cancelled.
- With the daemon disabled (or on Windows), the worker process exits and the
task fails with the generic "Python worker exited unexpectedly (crashed)"
error; the user's traceback reaches only executor stderr.
Before this change, `handle_worker_exception` formatted the same exception
through `TracebackException`, which renders a failing `__str__` as `<exception
str() failed>`, so the failure still surfaced as a `PythonException` with the
traceback. `format_worker_exception` on the next line is already tolerant in
the same way; deriving `message` defensively too (for example, falling back to
that placeholder when `str()` fails) would keep the init response, and the
user-facing error, intact. Only init-phase exceptions whose string conversion
fails are affected.
**Verification:**
- **Regression:** Run the worker on a payload whose loading raises an
exception with a failing string conversion, for example one whose __str__
raises or returns a non-string. The worker must emit an init response whose
error object has a string message, a traceback naming the exception class, the
class name and isPayloadError true, and then no further output.
- **Behavior:** The existing init-response scenarios keep producing the same
responses: success, payload-loading error, task-setup error, and a data-phase
error that follows a successful init response.
- **Compatibility:** The error object produced for the failing-conversion
case still decodes through the JVM init-response decoder: message, traceback
and errorClass remain JSON strings and isPayloadError remains a boolean.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
