[
https://issues.apache.org/jira/browse/SPARK-58846?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated SPARK-58846:
-----------------------------------
Labels: pull-request-available (was: )
> Spark Connect foreachBatch can hang indefinitely when the
> RemoveRemoteCachedRelation release RPC never returns
> --------------------------------------------------------------------------------------------------------------
>
> Key: SPARK-58846
> URL: https://issues.apache.org/jira/browse/SPARK-58846
> Project: Spark
> Issue Type: Bug
> Components: Connect, Structured Streaming
> Affects Versions: 5.0.0
> Reporter: Kishor Radhakrishnan
> Priority: Major
> Labels: pull-request-available
> Attachments: spark_connect_foreachbatch_release_hang_repro.zip
>
>
> {{CachedRemoteRelation._{_}del{_}_}} (python/pyspark/sql/connect/plan.py)
> issues a blocking, non-reattachable {{ExecutePlan}} RPC (a
> {{RemoveRemoteCachedRelation}} cleanup command) through a raw {{unary_unary}}
> channel, with {*}no client-side deadline{*}:
> {code:python}
> for attempt in session.client._retrying():
> with attempt:
> channel = session.client._channel.unary_unary(
> "/spark.connect.SparkConnectService/ExecutePlan", ...)
> metadata = session.client._builder.metadata()
> channel(req, metadata=metadata) # <-- blocking, no timeout=
> {code}
> If the response is never delivered, this finalizer blocks forever. On the
> *foreachBatch Spark Connect path* it runs on the per-batch critical path: the
> Python worker
> (python/pyspark/sql/connect/streaming/worker/foreach_batch_worker.py) creates
> the batch DataFrame as a local of {{{}process(){}}}, so when {{process()}}
> returns the {{CachedRemoteRelation}} finalizer fires *before* the worker
> writes its completion signal ({{{}write_int(0){}}}). If that finalizer
> blocks, the completion signal is never sent, and the driver JVM
> ({{{}StreamingForeachBatchHelper{}}}) stays blocked on its per-batch
> {{dataIn.readInt()}} – which also has no timeout. The streaming query then
> stalls indefinitely with {*}no error and no timeout at any layer{*}, and
> recovers only on a manual restart.
> There is no bounded timeout anywhere on this handshake: the JVM sets a socket
> timeout during worker init only and restores it to infinity afterward; the
> Python worker calls {{{}sock.settimeout(None){}}}; and the release RPC above
> has no deadline. {{_retrying()_}} \{_}does not help – it only re-runs the
> call when it _raises{_}, and a blocked call never raises.
> *Impact:* an intermittent, indefinite streaming-query stall (observed as
> multi-hour hangs) with no failure surfaced, requiring manual intervention to
> recover.
> h3. Proposed fix
> Give the release RPC a bounded client-side deadline via the existing
> {{RpcDeadlines}} framework – add a {{release_relation}} field (default 60s)
> and pass it as the gRPC {{{}timeout{}}}. Unlike a query {{{}ExecutePlan{}}},
> a timeout here kills nothing recoverable: the eviction is best effort and the
> server performs it independently (and, on the foreachBatch path, the JVM's
> {{finally}} also evicts the same relation id), so on timeout the call is
> caught, logged, and the finalizer proceeds. {{DefaultPolicy.can_retry}}
> already returns {{false}} for {{DEADLINE_EXCEEDED}} on non-reattachable RPCs,
> so the release call is not retried – worst case becomes one bounded wait
> (~60s) instead of forever.
> h3. Reproduction
> A standalone script (attached) stands up an in-process
> {{SparkConnectService}} whose {{ExecutePlan}} never responds, points a real
> {{SparkConnectClient}} at it, and triggers
> {{{}CachedRemoteRelation._{_}del{_}_{}}}. With no deadline the finalizer
> blocks; with a {{release_relation}} deadline it returns within the deadline.
> No cluster, streaming query, or build required.
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
