Kishor Radhakrishnan created SPARK-58846:
--------------------------------------------
Summary: 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
{{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]
