[jira] [Updated] (SPARK-57931) Restore worker channel blocking mode after pipelined Python UDF execution

["ASF GitHub Bot (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/SPARK-57931?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

ASF GitHub Bot updated SPARK-57931:
-----------------------------------
    Labels: pull-request-available  (was: )

> Restore worker channel blocking mode after pipelined Python UDF execution
> -------------------------------------------------------------------------
>
>                 Key: SPARK-57931
>                 URL: https://issues.apache.org/jira/browse/SPARK-57931
>             Project: Spark
>          Issue Type: Bug
>          Components: SQL
>    Affects Versions: 5.0.0
>            Reporter: L. C. Hsieh
>            Priority: Major
>              Labels: pull-request-available
>
> SPARK-56642 added an opt-in pipelined Python UDF path. When enabled,
> createPipelinedDataIn() switches the shared worker's SocketChannel from
> non-blocking to blocking mode (channel.configureBlocking(true) + worker.refresh())
> so the writer thread and the task thread can do full-duplex blocking I/O.
> However, the channel mode is never restored. With worker reuse enabled
> (spark.python.worker.reuse=true, the default), the task completion listener only
> cancels the writer future; the worker is then released back into the idle pool
> (PythonWorkerFactory.releaseWorker -> idleWorkers.enqueue) with its channel still
> in blocking mode.
> When a subsequent task pulls that worker from the pool, PythonWorkerFactory.create()
> only calls worker.refresh(). Because refresh() decides whether to open a selector
> based on the current channel mode, a channel left in blocking mode results in
> selectorOpt = None / selectionKeyOpt = None. If that next task runs on the
> non-pipelined (single-threaded NIO selector) path, it dereferences worker.selector
> / worker.selectionKey and hits a NullPointerException.
> This is cross-task state pollution: a pooled worker borrowed once by the pipelined
> path corrupts the channel state for every task that later reuses it, regardless of
> whether that task uses pipelined mode. It also breaks setups that rely on pooled
> workers being in a known (non-blocking) state.
> Fix: capture the channel's original blocking mode before createPipelinedDataIn()
> changes it, and restore it in the task completion listener (after the writer has
> exited and only if the channel is still open), so a pooled worker is always
> returned in the exact mode it was borrowed in.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org