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]
