viirya opened a new pull request, #56995:
URL: https://github.com/apache/spark/pull/56995

   ### What changes were proposed in this pull request?
   
   The pipelined Python UDF path (SPARK-56642) switches a borrowed worker's 
`SocketChannel` from non-blocking to blocking mode in `createPipelinedDataIn` 
and never restores it. With worker reuse enabled (the default), the worker is 
returned to the idle pool with its channel still in blocking mode.
   
   `PythonWorkerFactory.create()` now normalizes a reused daemon worker's 
channel back to non-blocking before calling `refresh()`, so the worker is 
always handed to the next task in the same (non-blocking) mode as a freshly 
created one.
   
   ### Why are the changes needed?
   
   `PythonWorker.refresh()` only opens a selector when the channel is 
non-blocking. A pooled worker left in blocking mode therefore comes back with a 
null selector and selection key. If the next task that reuses it runs on the 
non-pipelined (single-threaded NIO selector) path, it dereferences 
`worker.selector` / `worker.selectionKey` and throws a `NullPointerException`. 
This is cross-task state pollution: one task using pipelined mode corrupts the 
channel state for any later task that reuses the same pooled worker.
   
   The fix is applied at the pool boundary (`create()`) rather than at the 
borrow site's task-completion listener, because the worker is released back to 
the pool from the reader iterator when it reaches `END_OF_STREAM`, which 
happens *before* the task-completion listener runs. Normalizing in `create()` 
is therefore correct regardless of ordering, and robust against any code path 
that leaves a pooled channel in blocking mode.
   
   ### Does this PR introduce _any_ user-facing change?
   
   No. This fixes a bug in an opt-in feature 
(`spark.python.udf.pipelined.enabled`) that has not been released yet.
   
   ### How was this patch tested?
   
   New test in `PythonWorkerFactorySuite` that creates a daemon worker, 
simulates the blocking channel state the pipelined path leaves behind, returns 
it to the idle pool, and asserts the reused worker comes back in non-blocking 
mode with a live selector.
   
   ### Was this patch authored or co-authored using generative AI tooling?
   
   Generated-by: Claude Code
   


-- 
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]

Reply via email to