vinodkc opened a new pull request, #57202:
URL: https://github.com/apache/spark/pull/57202
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR:
https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g.,
'[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a
faster review.
7. If you want to add a new configuration, please read the guideline first
for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the
guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section
is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster
reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class
hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other
DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
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?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
`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?
<!--
Note that it means *any* user-facing change including all aspects such as
new features, bug fixes, or other behavior changes. Documentation-only updates
are not considered user-facing changes.
If yes, please clarify the previous behavior and the change this PR proposes
- provide the console output, description and/or an example to show the
behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to
the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
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?
<!--
If tests were added, say they were added here. Please make sure to add some
test cases that check the changes thoroughly including negative and positive
cases if possible.
If it was tested in a way different from regular unit tests, please clarify
how you tested step by step, ideally copy and paste-able, so that other
reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why
it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions
for the consistent environment, and the instructions could accord to:
https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
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?
<!--
If generative AI tooling has been used in the process of authoring this
patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling
Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
Yes. Generated-by: Claude Code (Sonnet 5).
--
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]
