cloud-fan commented on code in PR #57202:
URL: https://github.com/apache/spark/pull/57202#discussion_r3566748522
##########
sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/SparkConnectClient.scala:
##########
@@ -1048,6 +1099,10 @@ object SparkConnectClient {
sessionId: Option[String] = None,
grpcMaxMessageSize: Int = ConnectCommon.CONNECT_GRPC_MAX_MESSAGE_SIZE,
grpcMaxRecursionLimit: Int =
ConnectCommon.CONNECT_GRPC_MARSHALLER_RECURSION_LIMIT,
+ grpcKeepAliveEnabled: Boolean =
ConnectCommon.CONNECT_GRPC_KEEPALIVE_ENABLED,
+ grpcKeepAliveTimeMs: Long =
ConnectCommon.CONNECT_GRPC_KEEPALIVE_TIME_SECONDS * 1000,
+ grpcKeepAliveTimeoutMs: Long =
ConnectCommon.CONNECT_GRPC_KEEPALIVE_TIMEOUT_SECONDS * 1000,
+ grpcKeepAliveWithoutCalls: Boolean = true,
Review Comment:
Minor consistency: the other three keepalive defaults
(`grpcKeepAliveEnabled`/`TimeMs`/`TimeoutMs`) come from `ConnectCommon` — a
single JVM/Python source of truth — but `grpcKeepAliveWithoutCalls` defaults to
a bare literal `true` here and again in `core.py`
(`GRPC_DEFAULT_KEEPALIVE_WITHOUT_CALLS = True`). Consider adding
`CONNECT_GRPC_KEEPALIVE_WITHOUT_CALLS` to `ConnectCommon` so a future default
change has one home, matching the `grpcMaxMessageSize` pattern. Non-blocking.
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectService.scala:
##########
@@ -408,7 +408,21 @@ object SparkConnectService extends Logging {
case _ => NettyServerBuilder.forPort(port)
}
sb.maxInboundMessageSize(SparkEnv.get.conf.get(CONNECT_GRPC_MAX_INBOUND_MESSAGE_SIZE).toInt)
- .addService(sparkConnectService)
+ if (SparkEnv.get.conf.get(CONNECT_GRPC_KEEPALIVE_ENABLED)) {
+ val keepAliveTimeSeconds =
SparkEnv.get.conf.get(CONNECT_GRPC_KEEPALIVE_TIME)
+ // Detects a silently-dead client connection (e.g. a NAT gateway/load
balancer dropping
+ // an idle connection mapping without sending a TCP RST/FIN) via
gRPC/HTTP2 keepalive
+ // PINGs. permitKeepAliveTime must be <= the client's keepAliveTime
(matched here),
+ // otherwise the server rejects the client's keepalive PINGs as
"too_many_pings" and
+ // tears down every long-lived connection, not just dead ones.
+ sb.keepAliveTime(keepAliveTimeSeconds, TimeUnit.SECONDS)
+ sb.keepAliveTimeout(
+ SparkEnv.get.conf.get(CONNECT_GRPC_KEEPALIVE_TIMEOUT),
+ TimeUnit.SECONDS)
+ sb.permitKeepAliveTime(keepAliveTimeSeconds, TimeUnit.SECONDS)
Review Comment:
**Decouple `permitKeepAliveTime` from the server's own `keepAlive.time`.**
These are two independent axes (gRFC A8 is explicit: `PERMIT_KEEPALIVE_TIME`
"operates independently from the client's `KEEPALIVE_TIME`"):
- `keepAliveTime` (above) is the server's own *detection* interval — how
often the server probes for dead clients.
- `permitKeepAliveTime` is a *tolerance ceiling* on how fast the server will
let a **client** ping before it strikes the connection and GOAWAYs it as
`too_many_pings` (gRPC closes after `ping_strikes > 2`).
Setting `permitKeepAliveTime = keepAliveTimeSeconds` couples them and
creates an unstated, unenforced cross-config invariant: **client
`grpc_keepalive_time_ms` >= server `spark.connect.grpc.keepAlive.time`.** It
holds at matched 60s defaults but breaks two documented ways:
1. The PR's own faster-detection example, `grpc_keepalive_time_ms=30000`
(`client-connection-string.md`), against a default (60s) server → every healthy
connection torn down as `too_many_pings`.
2. An operator raising the server's *detection* interval above 60s silently
tightens *client tolerance* → default clients struck.
Even at exact 60s/60s there's zero margin: the strike algorithm penalizes a
ping arriving sooner than the permit since the last one, so normal jitter/GC on
a healthy idle connection accrues strikes.
**Suggested fix:** don't derive the permit from the detection interval — set
it to a low fixed floor well below any supported client rate (gRPC's
health-check guidance cites ~10s), which leaves the 60s client default a wide
(6x) margin and preserves fast client detection as an option. Keep
`permitKeepAliveWithoutCalls(true)`. Since gRPC offers no client/server
negotiation, also document the "client interval >= server permit" contract on
both the server conf and the client params. A fixed constant is the pragmatic
choice; a dedicated `spark.connect.grpc.keepAlive.permitTime` conf is the more
orthodox one (gRFC A8: "service owners decide") — your DoS-posture call,
especially since Connect clients are authenticated session peers rather than
anonymous callers.
Also: the comment says `permitKeepAliveTime must be <= the client's
keepAliveTime (matched here)`, but it's matched to the *server's* own conf, not
the client's — please correct it.
Refs: [gRPC keepalive guide](https://grpc.io/docs/guides/keepalive/), [gRFC
A8](https://github.com/grpc/proposal/blob/master/A8-client-side-keepalive.md).
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectServiceKeepAliveSuite.scala:
##########
@@ -0,0 +1,268 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.connect.service
+
+import java.util.UUID
+import java.util.concurrent.TimeUnit
+
+import scala.collection.mutable
+import scala.concurrent.duration.FiniteDuration
+
+import org.scalatest.concurrent.Eventually
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.connect.SparkConnectServerTest
+import org.apache.spark.sql.connect.client.{RetryPolicy, SparkConnectClient}
+import org.apache.spark.sql.connect.config.Connect
+
+/**
+ * End-to-end test that the *real* `SparkConnectService.startGRPCService()`
wiring actually
+ * applies the configured gRPC keepalive settings (SPARK-58094).
+ *
+ * Unlike the tests in `SparkConnectClientSuite` (sql/connect/client/jvm),
which hand-build their
+ * own `NettyServerBuilder` that only mirrors the production server
construction, this suite
+ * starts the real `SparkConnectService` (via `SparkConnectServerTest`) with
the keepalive confs
+ * tuned to short test values, puts a freezable TCP relay in front of it, and
confirms a blocked
+ * client call fails with a keepalive-triggered `UNAVAILABLE` within a bounded
time -- so a future
+ * refactor of `startGRPCService()` that silently drops the keepalive wiring
would be caught here,
+ * not just in the client-side unit tests.
+ */
+class SparkConnectServiceKeepAliveSuite extends SparkConnectServerTest {
Review Comment:
This suite (which exercises the real production `startGRPCService` wiring)
covers the dead-connection and disabled paths, but not a **healthy long-lived
connection** through the real server — that test lives in
`SparkConnectClientSuite` against a hand-built `NettyServerBuilder` double. So
a refactor that dropped only the production `permitKeepAliveTime` call would
pass every server-side test here: the frozen-relay test surfaces "Keepalive
failed" from the client before any ping-rate teardown can occur.
Consider adding a real-service healthy-connection test with **client
`keepalive_time_ms` < server `keepAlive.time`** and several spaced calls. It
would both guard the `permitKeepAliveTime` wiring against a silent-drop
refactor and directly exercise the coupling flagged in
`SparkConnectService.scala` — today nothing catches that scenario. Non-blocking.
--
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]
