[SPARK-58094][ Connect] Blocking RPCs hang forever when the connection dies mid-wait

[vinodkc]

What changes were proposed in this pull request?

This PR configures gRPC/HTTP2 keepalive by default on both sides of Spark Connect, for both the JVM and Python (PySpark) clients:

• Client (JVM) — SparkConnectClient.scala's Configuration.createChannel(): sets keepAliveTime/keepAliveTimeout/keepAliveWithoutCalls on the gRPC channel builder. Configurable via new connection-string params grpc_keepalive_enabled / grpc_keepalive_time_ms / grpc_keepalive_timeout_ms / grpc_keepalive_without_calls (SparkConnectClientParser.scala) and matching Builder.grpcKeepAliveEnabled / grpcKeepAliveTimeMs / grpcKeepAliveTimeoutMs / grpcKeepAliveWithoutCalls methods. Defaults: enabled / 60s / 20s / enabled (ConnectCommon.CONNECT_GRPC_KEEPALIVE_ENABLED/_TIME_SECONDS/_TIMEOUT_SECONDS).
• Client (Python) — core.py's ChannelBuilder: the same four options (grpc_keepalive_enabled/_time_ms/_timeout_ms/_without_calls), applied via a new _effective_channel_options() that both _insecure_channel/_secure_channel funnel through (so any custom ChannelBuilder subclass gets the fix too, not just DefaultChannelBuilder), same defaults as the JVM client. An explicitly-set channelOptions/setChannelOption() value for one of these keys is never overwritten.
• Server — SparkConnectService.scala's startGRPCService(): sets keepAliveTime/keepAliveTimeout/permitKeepAliveTime/permitKeepAliveWithoutCalls(true) on the NettyServerBuilder, applied only when enabled. New static confs spark.connect.grpc.keepAlive.enabled/.time/.timeout (Connect.scala), same enabled/60s/20s defaults. permitKeepAliveTime is set equal to the client's default keepAliveTime (60s) — otherwise the server's gRPC-library default 5-minute permitKeepAliveTime would reject the client's more frequent pings as "too_many_pings" and tear down every long-lived healthy connection, not just dead ones.
• Both sides can be fully disabled independently (spark.connect.grpc.keepAlive.enabled=false server-side, grpc_keepalive_enabled=false/Builder.grpcKeepAliveEnabled(false) client-side), as an escape hatch for environments where this default-on behavior change is undesirable (e.g. a server/client environment prone to long GC pauses or other stalls that could otherwise trip a false-positive disconnect).
• Documentation updated: docs/configuration.md (new server confs) and sql/connect/docs/client-connection-string.md (new client connection-string params).

This does not change awaitTermination()'s (or any other blocking call's) semantics — blocking indefinitely while a request is genuinely still in flight on a healthy connection is correct. The gap was purely at the transport layer: once keepalive detects a connection that has gone silently dark, the channel fails with UNAVAILABLE, and the existing bounded retry policy (RetryPolicy.scala, maxRetries=15, capped exponential backoff) takes over and surfaces a real exception, instead of the RPC waiting forever for a response that will never arrive.

Why are the changes needed?

StreamingQuery.awaitTermination() (and, more generally, any blocking Spark Connect RPC) can hang indefinitely and never throw if the network path between client and server goes silently dark while the call is blocked — e.g. a NAT gateway, load balancer, or corporate proxy silently evicting an idle connection mapping without sending a TCP RST/FIN. This is a well-known operational gotcha for gRPC deployments behind such middleboxes, and Spark Connect configured no keepalive anywhere (client or server) to guard against it.

Root cause: with no keepalive configured, neither endpoint has any way to detect that the underlying TCP connection has gone silently dark — both sides simply continue believing the RPC is still legitimately in flight, forever. The streaming query can (and did, in the reported incident) terminate independently on the server in the meantime, but there is no live channel left to deliver that information back to the client.

Reproduced end-to-end with two genuinely separate JVM processes (a real Spark Connect server and a standalone client app) and a small TCP proxy in between, frozen mid-RPC with SIGSTOP to simulate a silently-dropped connection without either endpoint's socket seeing a close/reset — see "How was this patch tested?" below. Pre-fix, the client's main thread parks forever in ArrayBlockingQueue.take() (via ExecutePlanResponseReattachableIterator); post-fix, keepalive detects the dead connection and the client eventually surfaces a real, bounded exception.

Does this PR introduce any user-facing change?

Yes.

• New configuration, all optional and defaulted to preserve the fixed behavior:
  • Server (SQLConf): spark.connect.grpc.keepAlive.enabled (default true), spark.connect.grpc.keepAlive.time (default 60s), spark.connect.grpc.keepAlive.timeout (default 20s).
  • Client (connection-string params, JVM Builder methods, and Python ChannelBuilder connection-string params): grpc_keepalive_enabled (default true), grpc_keepalive_time_ms (default 60000), grpc_keepalive_timeout_ms (default 20000), grpc_keepalive_without_calls (default true).
• Behavior change (default-on): both the Spark Connect client and server now send gRPC/HTTP2 keepalive PINGs periodically. Previously, a Spark Connect connection that went silently dark mid-RPC would leave the client blocked forever with no exception; now, such a connection is detected within roughly keepAliveTime + keepAliveTimeout (60s + 20s by default) and the in-flight call fails with UNAVAILABLE: Keepalive failed. The connection is likely gone, after which Spark Connect's existing (unmodified) retry policy takes over.
• This is a behavior change from all previously-released Spark versions (keepalive was never configured before); it can be fully reverted via the .enabled/grpc_keepalive_enabled flags above for environments where it is undesirable.

How was this patch tested?

Unit and end-to-end tests :

• SparkConnectClientBuilderParseTestSuite (JVM): connection-string/CLI parsing of the new keepalive params (including an explicit grpc_keepalive_enabled=true case, not just the default and the disabled case), and that defaults apply when unset — 20/20.
• SparkConnectClientSuite (JVM), two new tests: a FreezableTcpRelay test helper (an in-process byte-forwarding relay that can be frozen without closing either socket) sits between a real test client and server; once a blocking RPC is in flight, the relay is frozen and the test asserts the client's call fails within a bounded time with UNAVAILABLE: Keepalive failed. The connection is likely gone instead of hanging past the test's own timeout; a second test confirms keepalive does not disrupt a healthy long-lived connection (guards against a permitKeepAliveTime mistuning regression) .
• SparkConnectServiceKeepAliveSuite (new, server side): starts the real SparkConnectService (not a hand-built test double) with short keepalive confs, puts a FreezableTcpRelay in front of it, and confirms both that (a) a blocked call fails with the keepalive UNAVAILABLE when keepalive is enabled, and (b) disabling spark.connect.grpc.keepAlive.enabled reverts to the pre-fix hang (the call stays genuinely blocked well past the window that would have triggered detection) .
• Python test_connect_channel.py: defaults, connection-string overrides

Manual end-to-end reproduction, using two genuinely separate JVM/OS processes (not an in-process test) plus a small Python TCP proxy in between, to faithfully reproduce the client/server RPC boundary and a silently-dropped connection:

1. A real Spark Connect server
2. A standalone Scala client app running a rate-source streaming query that fails via assert_true at a chosen row value (simulating a non-retryable source error), with a StreamingQueryListener and a watchdog thread, calling query.awaitTermination().
3. The TCP proxy frozen with SIGSTOP shortly after awaitTermination() was called (no TCP RST/FIN sent to either endpoint — the same observable behavior as a NAT gateway/load balancer silently evicting an idle connection mapping).

Confirmed:

• Pre-fix: the server terminates the query independently a few seconds later (confirming the hang is not an exception-propagation bug), but the client's awaitTermination() never returns; a thread dump shows the main thread permanently parked in ArrayBlockingQueue.take() via ExecutePlanResponseReattachableIterator/GrpcRetryHandler$Retrying.retry, with no path out.
• Post-fix: the same scenario instead surfaces a bounded exception. With fast (1s/1s) keepalive settings, a thread dump within ~1-2 minutes shows the client has alreadymoved off ArrayBlockingQueue.take() into GrpcRetryHandler$Retrying.waitAfterAttempt (keepalive detected the dead connection); the call eventually throws UNAVAILABLE:Keepalive failed. The connection is likely gone (observed consistently at ~630-770s end-to-end, governed by the existing, unmodified retry policy's own backoff budget, not by this fix). This was verified identically whether spark.connect.grpc.keepAlive.enabled/grpc_keepalive_enabled was left at its default (true) or set explicitly totrue, and the escape-hatch (enabled=false on both sides) was separately confirmed to genuinely revert to the pre-fix hang signature rather than just changing its timing.
• The same manual reproduction was repeated against the Python client (grpc.insecure_channel/grpc.secure_channel via ChannelBuilder), confirming both the original gap(Python builds its own gRPC channel and initially had none of this fix) and the fix once applied: awaitTermination() threw SparkConnectGrpcException / UNAVAILABLE:Stream removed (ping timeout) — grpcio's equivalent wording for the same keepalive-triggered failure.

A minimal, runnable, standalone reproduction (client + server + proxy scripts, independent of this Spark checkout's test suite) is attached to SPARK-58094 for reviewers who want to reproduce the issue.

Was this patch authored or co-authored using generative AI tooling?

Yes. Generated-by: Claude Code (Sonnet 5).

[cloud-fan]

1 blocking, 2 non-blocking, 0 nits.
Well-tested, well-documented fix diagnosing the transport-layer keepalive gap and following the established grpcMaxMessageSize builder pattern. The one blocking item is a footgun in the server's permitKeepAliveTime coupling that the PR's own docs would trip.

Design / architecture (2)

• SparkConnectService.scala:422: server permitKeepAliveTime is tied to the server's own keepAlive.time, not the client's ping rate — creates an undocumented invariant (client grpc_keepalive_time_ms >= server keepAlive.time) that the PR's own documented grpc_keepalive_time_ms=30000 example violates against a default server, tearing down healthy connections as too_many_pings — see inline
• SparkConnectClient.scala:1105: grpcKeepAliveWithoutCalls default is a scattered literal, not a shared ConnectCommon constant like the other three — see inline

Correctness (1)

• SparkConnectServiceKeepAliveSuite.scala:44: no healthy-long-lived-connection test through the real server; a regression dropping only production permitKeepAliveTime wouldn't be caught — see inline

Verification

Confirmed the fix itself is sound: no keepalive existed pre-PR (base state); the enable-gated channel wiring, the strict connection-string/CLI parsing, and the Python _effective_channel_options explicit-option-precedence + no-mutation logic all behave as documented and are tested. The load-bearing concern is the permitKeepAliveTime↔keepAlive.time coupling below: with matched 60s defaults it is correct, but per gRFC A8 these are independent axes, and the coupling is reachable through documented knobs — not a defect in the happy path.