You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ch...@apache.org on 2026/08/03 08:19:08 UTC
(spark) branch branch-4.x updated: [SPARK-57933][CONNECT] Map Spark Connect client failures to SQLException in the JDBC driver
This is an automated email from the ASF dual-hosted git repository.
pan3793 pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 17bef863f743 [SPARK-57933][CONNECT] Map Spark Connect client failures to SQLException in the JDBC driver
17bef863f743 is described below
commit 17bef863f7436718e2bf3dc8f41dd7f7eafb3131
Author: Jiwon Park <jp...@outlook.kr>
AuthorDate: Mon Aug 3 16:18:41 2026 +0800
[SPARK-57933][CONNECT] Map Spark Connect client failures to SQLException in the JDBC driver
### What changes were proposed in this pull request?
Two commits:
1. `GrpcExceptionConverter` now preserves the original `StatusRuntimeException` as the cause of fallback errors (errors without `ErrorInfo`), generalizing what the `DEADLINE_EXCEEDED` branch already did for the same reason.
2. The Connect JDBC driver maps client failures to `SQLException` at the two points where they cross the JDBC boundary: `SparkConnectStatement.execute` (`executeQuery`/`executeUpdate` route through it) and `SparkConnectResultSet.next` (rows are fetched lazily, so errors during result streaming surface there).
| Error | Mapped to | SQLState |
|---|---|---|
| `INVALID_HANDLE.SESSION_*` (session is gone) | `SQLNonTransientConnectionException` | `08003` |
| gRPC `UNAVAILABLE` | `SQLTransientConnectionException` | `08006` |
| gRPC `DEADLINE_EXCEEDED` | `SQLTimeoutException` | `HYT00` |
| Anything else | `SQLException` | server-provided |
The gRPC status is read only from the `StatusRuntimeException` preserved in the cause chain (commit 1), never from message text, so a server-side error that merely quotes a gRPC exception cannot be mistaken for a transport failure. The original exception is always kept as the cause.
Notes on the mapping:
- Operation-level `INVALID_HANDLE` subconditions (`OPERATION_*`, `FORMAT`) are not connection errors: the session is still healthy.
- `DEADLINE_EXCEEDED` is a timeout, not a connection failure: a slow query fires it on a healthy connection, and the client's own `RetryPolicy` deliberately excludes it from retryable errors.
Other JDBC entry points (`Connection.setCatalog`, `DatabaseMetaData` query methods, `Statement.cancel`) will be covered in a follow-up.
### Why are the changes needed?
Currently unchecked exceptions (e.g. `SparkException`) leak across the JDBC boundary: standard JDBC code catching `SQLException` misses them, and there is no SQLState to act on. SQLState class 08 is how connection pools (e.g. HikariCP) and BI tools detect a dead connection and reconnect; without it they keep handing out a connection whose server-side session is gone.
### Does this PR introduce _any_ user-facing change?
Yes, for the unreleased Connect JDBC driver:
stmt.execute("SELECT * FROM nonexistent")
// before: throws AnalysisException (unchecked)
// after: throws SQLException with getSQLState() == "42P01"
For other Connect clients, the only observable change is that fallback `SparkException`s now carry the original `StatusRuntimeException` as their cause (previously often null); messages are unchanged.
### How was this patch tested?
- `SparkConnectClientSuite`: new test that transport errors preserve the gRPC exception as the cause; existing deadline tests still pass.
- New `JdbcErrorUtilsSuite`: 13 unit tests covering the mapping table, precedence, cause-chain traversal, and the quoted-gRPC-text negative case.
- Integration tests: a failing query throws `SQLException`/`42P01` from `execute`; a mid-stream failure (ANSI divide-by-zero) throws `SQLException`/`22012` from `next`.
Ran `build/sbt "connect-client-jdbc/test"` and scalastyle.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (claude-fable-5)
Closes #57009 from j1wonpark/SPARK-57933.
Authored-by: Jiwon Park <jp...@outlook.kr>
Signed-off-by: Cheng Pan <ch...@apache.org>
(cherry picked from commit 3d9d7df61d4c39563c783bc538fb42fadb2c16b4)
Signed-off-by: Cheng Pan <ch...@apache.org>
---
sql/connect/client/jdbc/pom.xml | 8 +
.../client/jdbc/SparkConnectResultSet.scala | 20 ++-
.../client/jdbc/SparkConnectStatement.scala | 26 ++--
.../connect/client/jdbc/util/JdbcErrorUtils.scala | 67 +++++++++
.../client/jdbc/SparkConnectResultSetSuite.scala | 14 ++
.../client/jdbc/SparkConnectStatementSuite.scala | 13 ++
.../client/jdbc/util/JdbcErrorUtilsSuite.scala | 163 +++++++++++++++++++++
.../connect/client/SparkConnectClientSuite.scala | 33 ++++-
.../connect/client/GrpcExceptionConverter.scala | 21 ++-
.../SparkConnectServiceKeepAliveSuite.scala | 5 +-
10 files changed, 333 insertions(+), 37 deletions(-)
diff --git a/sql/connect/client/jdbc/pom.xml b/sql/connect/client/jdbc/pom.xml
index 52a79fdd9012..7e00574996b1 100644
--- a/sql/connect/client/jdbc/pom.xml
+++ b/sql/connect/client/jdbc/pom.xml
@@ -47,6 +47,14 @@
<classifier>tests</classifier>
<scope>test</scope>
</dependency>
+ <!-- Compile-time only: the shade plugin relocates io.grpc references to the
+ copies embedded in the spark-connect-client-jvm jar, so consumers must
+ not pull the unshaded artifact transitively. -->
+ <dependency>
+ <groupId>io.grpc</groupId>
+ <artifactId>grpc-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
</dependencies>
<build>
<outputDirectory>target/scala-${scala.binary.version}/classes</outputDirectory>
diff --git a/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectResultSet.scala b/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectResultSet.scala
index 6b02f655f04c..12bd6cd133ab 100644
--- a/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectResultSet.scala
+++ b/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectResultSet.scala
@@ -54,17 +54,21 @@ class SparkConnectResultSet(
override def next(): Boolean = {
checkOpen()
- val hasNext = iterator.hasNext
- if (hasNext) {
- currentRow = iterator.next()
- cursor += 1
- } else {
- currentRow = null
- if (cursor > 0 && cursor == sparkResult.length) {
+ // rows are fetched lazily, so an error raised while the query is still
+ // streaming results surfaces here rather than in execute()
+ JdbcErrorUtils.mapToSQLException {
+ val hasNext = iterator.hasNext
+ if (hasNext) {
+ currentRow = iterator.next()
cursor += 1
+ } else {
+ currentRow = null
+ if (cursor > 0 && cursor == sparkResult.length) {
+ cursor += 1
+ }
}
+ hasNext
}
- hasNext
}
@volatile private var closed: Boolean = false
diff --git a/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectStatement.scala b/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectStatement.scala
index 245832087268..66f9cc8fbec4 100644
--- a/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectStatement.scala
+++ b/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectStatement.scala
@@ -99,18 +99,20 @@ class SparkConnectStatement(conn: SparkConnectConnection) extends Statement {
resultSet = null
resultsExhausted = false
- var df = conn.spark.sql(sql)
- if (maxRows > 0) {
- df = df.limit(maxRows)
- }
- val sparkResult = df.collectResult()
- operationId = sparkResult.operationId
- if (hasResultSet(sparkResult)) {
- resultSet = new SparkConnectResultSet(sparkResult, this)
- true
- } else {
- sparkResult.close()
- false
+ JdbcErrorUtils.mapToSQLException {
+ var df = conn.spark.sql(sql)
+ if (maxRows > 0) {
+ df = df.limit(maxRows)
+ }
+ val sparkResult = df.collectResult()
+ operationId = sparkResult.operationId
+ if (hasResultSet(sparkResult)) {
+ resultSet = new SparkConnectResultSet(sparkResult, this)
+ true
+ } else {
+ sparkResult.close()
+ false
+ }
}
}
diff --git a/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/util/JdbcErrorUtils.scala b/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/util/JdbcErrorUtils.scala
index 6480c5d768f3..72d007c1304a 100644
--- a/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/util/JdbcErrorUtils.scala
+++ b/sql/connect/client/jdbc/src/main/scala/org/apache/spark/sql/connect/client/jdbc/util/JdbcErrorUtils.scala
@@ -19,6 +19,12 @@ package org.apache.spark.sql.connect.client.jdbc.util
import java.sql.{Array => _, _}
+import scala.util.control.NonFatal
+
+import io.grpc.{Status, StatusRuntimeException}
+
+import org.apache.spark.SparkThrowable
+
private[jdbc] object JdbcErrorUtils {
def stringifyTransactionIsolationLevel(level: Int): String = level match {
@@ -53,4 +59,65 @@ private[jdbc] object JdbcErrorUtils {
case _ =>
throw new IllegalArgumentException(s"Invalid fetch direction: $direction")
}
+
+ // SQLState class 08 is "connection exception"; HYT00 is the conventional
+ // (ODBC-derived) state for an elapsed timeout.
+ private val CONNECTION_DOES_NOT_EXIST = "08003"
+ private val CONNECTION_FAILURE = "08006"
+ private val TIMEOUT_EXPIRED = "HYT00"
+
+ private val SESSION_GONE_CONDITION_PREFIX = "INVALID_HANDLE.SESSION_"
+
+ /**
+ * Maps the unchecked exceptions raised by the Spark Connect client (a
+ * [[SparkThrowable]] once GrpcExceptionConverter has converted the gRPC error) to
+ * the [[SQLException]] a JDBC method is required to throw:
+ *
+ * - `INVALID_HANDLE.SESSION_*` means the server-side session is gone (e.g. it
+ * timed out); the connection is unusable and retrying on it is pointless, so it
+ * maps to a [[SQLNonTransientConnectionException]] with SQLState 08003
+ * ("connection does not exist"). The operation-level subconditions
+ * (`OPERATION_*`, `FORMAT`) concern a single operation on a healthy session
+ * and are deliberately not treated as connection errors.
+ * - a gRPC UNAVAILABLE (e.g. a server restart or a network blip) maps to a
+ * [[SQLTransientConnectionException]] with SQLState 08006 ("connection
+ * failure"), since a fresh connection can succeed.
+ * - a gRPC DEADLINE_EXCEEDED means the RPC deadline elapsed, which a slow query
+ * fires on a perfectly healthy connection, so it maps to a
+ * [[SQLTimeoutException]] rather than a connection error.
+ * - any other error keeps the server-provided SQLState when one is available.
+ *
+ * The gRPC status is read from the [[StatusRuntimeException]] that
+ * GrpcExceptionConverter preserves in the cause chain, never from message text,
+ * so a server-side error merely quoting a gRPC exception cannot be mistaken for
+ * a transport failure. SQLState class 08 is how connection pools and BI tools
+ * detect a dead connection and reconnect.
+ */
+ def toSQLException(t: Throwable): SQLException = t match {
+ case e: SQLException => e
+ case e =>
+ val chain = causeChain(e)
+ val sparkThrowableOpt = chain.collectFirst { case st: SparkThrowable => st }
+ val condition = sparkThrowableOpt.flatMap(st => Option(st.getCondition))
+ val grpcCode = chain.collectFirst { case sre: StatusRuntimeException =>
+ sre.getStatus.getCode
+ }
+ if (condition.exists(_.startsWith(SESSION_GONE_CONDITION_PREFIX))) {
+ new SQLNonTransientConnectionException(e.getMessage, CONNECTION_DOES_NOT_EXIST, e)
+ } else if (grpcCode.contains(Status.Code.UNAVAILABLE)) {
+ new SQLTransientConnectionException(e.getMessage, CONNECTION_FAILURE, e)
+ } else if (grpcCode.contains(Status.Code.DEADLINE_EXCEEDED)) {
+ new SQLTimeoutException(e.getMessage, TIMEOUT_EXPIRED, e)
+ } else {
+ new SQLException(e.getMessage, sparkThrowableOpt.map(_.getSqlState).orNull, e)
+ }
+ }
+
+ /** Runs `body`, rethrowing any non-fatal failure as the mapped [[SQLException]]. */
+ def mapToSQLException[T](body: => T): T =
+ try body catch { case NonFatal(e) => throw toSQLException(e) }
+
+ // The take(20) caps the walk in case of a cause cycle.
+ private def causeChain(t: Throwable): Seq[Throwable] =
+ Iterator.iterate(t)(_.getCause).takeWhile(_ != null).take(20).toSeq
}
diff --git a/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectResultSetSuite.scala b/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectResultSetSuite.scala
index 21b8e261aef4..8470a9151533 100644
--- a/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectResultSetSuite.scala
+++ b/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectResultSetSuite.scala
@@ -186,4 +186,18 @@ class SparkConnectResultSetSuite extends ConnectFunSuite with RemoteSparkSession
assert(!rs.next())
}
}
+
+ test("next throws SQLException when the query fails during result streaming") {
+ // The server sends the schema response before any arrow batch, so executeQuery
+ // returns successfully and the runtime error surfaces while iterating rows.
+ withStatement { stmt =>
+ val rs = stmt.executeQuery("SELECT 10 / (5 - id) FROM range(10)")
+ val e = intercept[SQLException] {
+ while (rs.next()) {}
+ }
+ // DIVIDE_BY_ZERO under ANSI mode
+ assert(e.getSQLState === "22012")
+ assert(e.getCause != null)
+ }
+ }
}
diff --git a/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectStatementSuite.scala b/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectStatementSuite.scala
index 6ace1fa9cf78..1163feaf5e90 100644
--- a/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectStatementSuite.scala
+++ b/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectStatementSuite.scala
@@ -173,4 +173,17 @@ class SparkConnectStatementSuite extends ConnectFunSuite with RemoteSparkSession
}
}
}
+
+ test("execute throws SQLException with server-provided SQLState on query failure") {
+ withStatement { stmt =>
+ val e = intercept[SQLException] {
+ stmt.execute("SELECT * FROM this_table_does_not_exist")
+ }
+ // TABLE_OR_VIEW_NOT_FOUND
+ assert(e.getSQLState === "42P01")
+ assert(!e.isInstanceOf[SQLNonTransientConnectionException])
+ assert(!e.isInstanceOf[SQLTransientConnectionException])
+ assert(e.getCause != null)
+ }
+ }
}
diff --git a/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/util/JdbcErrorUtilsSuite.scala b/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/util/JdbcErrorUtilsSuite.scala
new file mode 100644
index 000000000000..fd434dcbb206
--- /dev/null
+++ b/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/util/JdbcErrorUtilsSuite.scala
@@ -0,0 +1,163 @@
+/*
+ * 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.client.jdbc.util
+
+import java.sql.{SQLException, SQLNonTransientConnectionException, SQLTimeoutException, SQLTransientConnectionException}
+
+import io.grpc.{Status, StatusRuntimeException}
+
+import org.apache.spark.SparkThrowable
+import org.apache.spark.sql.connect.test.ConnectFunSuite
+
+/**
+ * Tests for [[JdbcErrorUtils.toSQLException]]. A RuntimeException mixing in
+ * [[SparkThrowable]] stands in for converted server errors; a real
+ * [[StatusRuntimeException]] cause models transport errors the way
+ * GrpcExceptionConverter preserves them.
+ */
+class JdbcErrorUtilsSuite extends ConnectFunSuite {
+
+ private def sparkError(
+ condition: String,
+ msg: String,
+ cause: Throwable = null): RuntimeException =
+ new RuntimeException(msg, cause) with SparkThrowable {
+ override def getCondition: String = condition
+ }
+
+ private def grpcError(status: Status): StatusRuntimeException =
+ new StatusRuntimeException(status)
+
+ test("INVALID_HANDLE.SESSION_CLOSED maps to a connection exception with SQLState 08003") {
+ val e = JdbcErrorUtils.toSQLException(
+ sparkError("INVALID_HANDLE.SESSION_CLOSED", "Session was closed"))
+ assert(e.isInstanceOf[SQLNonTransientConnectionException])
+ assert(e.getSQLState === "08003")
+ assert(e.getMessage === "Session was closed")
+ }
+
+ test("other session-level INVALID_HANDLE subconditions also map to 08003") {
+ Seq("INVALID_HANDLE.SESSION_NOT_FOUND", "INVALID_HANDLE.SESSION_CHANGED").foreach {
+ condition =>
+ val e = JdbcErrorUtils.toSQLException(sparkError(condition, "gone"))
+ assert(e.isInstanceOf[SQLNonTransientConnectionException], condition)
+ assert(e.getSQLState === "08003", condition)
+ }
+ }
+
+ test("operation-level INVALID_HANDLE subconditions do not map to a connection exception") {
+ // the session, and thus the connection, is still healthy: no SQLState class 08
+ Seq(
+ "INVALID_HANDLE.OPERATION_NOT_FOUND",
+ "INVALID_HANDLE.OPERATION_ABANDONED",
+ "INVALID_HANDLE.OPERATION_ALREADY_EXISTS",
+ "INVALID_HANDLE.FORMAT").foreach { condition =>
+ val e = JdbcErrorUtils.toSQLException(sparkError(condition, "operation gone"))
+ assert(!e.isInstanceOf[SQLNonTransientConnectionException], condition)
+ assert(!e.isInstanceOf[SQLTransientConnectionException], condition)
+ }
+ }
+
+ test("a non-connection Spark error keeps the server-provided SQLState") {
+ val e = JdbcErrorUtils.toSQLException(sparkError("DIVIDE_BY_ZERO", "boom"))
+ assert(!e.isInstanceOf[SQLNonTransientConnectionException])
+ assert(!e.isInstanceOf[SQLTransientConnectionException])
+ assert(e.getSQLState === "22012")
+ assert(e.getMessage === "boom")
+ }
+
+ test("an UNAVAILABLE StatusRuntimeException maps to a transient connection exception") {
+ val e = JdbcErrorUtils.toSQLException(
+ grpcError(Status.UNAVAILABLE.withDescription("Channel shutdown invoked")))
+ assert(e.isInstanceOf[SQLTransientConnectionException])
+ assert(e.getSQLState === "08006")
+ }
+
+ test("an UNAVAILABLE cause preserved by GrpcExceptionConverter maps to 08006") {
+ // the shape GrpcExceptionConverter produces for transport errors
+ val wrapped = sparkError(
+ "CONNECT_CLIENT_UNEXPECTED_MISSING_SQL_STATE",
+ "io.grpc.StatusRuntimeException: UNAVAILABLE: io exception",
+ cause = grpcError(Status.UNAVAILABLE.withDescription("io exception")))
+ val e = JdbcErrorUtils.toSQLException(wrapped)
+ assert(e.isInstanceOf[SQLTransientConnectionException])
+ assert(e.getSQLState === "08006")
+ }
+
+ test("a server error merely quoting a gRPC exception is not a connection failure") {
+ // the gRPC status comes from exception instances, never from message text, so a
+ // server error embedding gRPC text (e.g. from a UDF) is not a transport failure
+ val e = JdbcErrorUtils.toSQLException(sparkError(
+ "FAILED_EXECUTE_UDF",
+ "Job aborted: io.grpc.StatusRuntimeException: UNAVAILABLE: backend down"))
+ assert(!e.isInstanceOf[SQLTransientConnectionException])
+ assert(!e.isInstanceOf[SQLNonTransientConnectionException])
+ }
+
+ test("a DEADLINE_EXCEEDED cause maps to a timeout, not a connection failure") {
+ // a slow query fires the RPC deadline on a healthy connection: timeout, not class 08
+ val wrapped = sparkError(
+ "CONNECT_CLIENT_UNEXPECTED_MISSING_SQL_STATE",
+ "io.grpc.StatusRuntimeException: DEADLINE_EXCEEDED: deadline exceeded",
+ cause = grpcError(Status.DEADLINE_EXCEEDED.withDescription("deadline exceeded")))
+ val e = JdbcErrorUtils.toSQLException(wrapped)
+ assert(e.isInstanceOf[SQLTimeoutException])
+ assert(e.getSQLState === "HYT00")
+ assert(!e.isInstanceOf[SQLTransientConnectionException])
+ }
+
+ test("a non-retryable gRPC status does not map to a connection exception") {
+ val e = JdbcErrorUtils.toSQLException(
+ grpcError(Status.INVALID_ARGUMENT.withDescription("bad plan")))
+ assert(!e.isInstanceOf[SQLTransientConnectionException])
+ assert(!e.isInstanceOf[SQLNonTransientConnectionException])
+ assert(!e.isInstanceOf[SQLTimeoutException])
+ }
+
+ test("a closed session takes precedence over a transport code") {
+ // both signals present: the gone session wins
+ val e = JdbcErrorUtils.toSQLException(sparkError(
+ "INVALID_HANDLE.SESSION_CLOSED",
+ "closed",
+ cause = grpcError(Status.UNAVAILABLE.withDescription("x"))))
+ assert(e.isInstanceOf[SQLNonTransientConnectionException])
+ assert(e.getSQLState === "08003")
+ }
+
+ test("the connection error is found through the cause chain") {
+ val root = sparkError("INVALID_HANDLE.SESSION_CLOSED", "closed")
+ val e = JdbcErrorUtils.toSQLException(new RuntimeException("wrapper", root))
+ assert(e.isInstanceOf[SQLNonTransientConnectionException])
+ assert(e.getSQLState === "08003")
+ }
+
+ test("a non-Spark exception becomes a plain SQLException carrying the cause") {
+ val cause = new RuntimeException("raw failure")
+ val e = JdbcErrorUtils.toSQLException(cause)
+ assert(!e.isInstanceOf[SQLNonTransientConnectionException])
+ assert(!e.isInstanceOf[SQLTransientConnectionException])
+ assert(e.getSQLState === null)
+ assert(e.getMessage === "raw failure")
+ assert(e.getCause eq cause)
+ }
+
+ test("an existing SQLException passes through unchanged") {
+ val original = new SQLException("already mapped")
+ assert(JdbcErrorUtils.toSQLException(original) eq original)
+ }
+}
diff --git a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/SparkConnectClientSuite.scala b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/SparkConnectClientSuite.scala
index 24e637a1f215..9d48eb0343d9 100644
--- a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/SparkConnectClientSuite.scala
+++ b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/SparkConnectClientSuite.scala
@@ -898,6 +898,34 @@ class SparkConnectClientSuite extends ConnectFunSuite {
}
}
+ test("transport errors preserve the original gRPC status exception as the cause") {
+ val failingService = new DummySparkConnectService {
+ override def analyzePlan(
+ request: AnalyzePlanRequest,
+ responseObserver: StreamObserver[AnalyzePlanResponse]): Unit = {
+ responseObserver.onError(
+ Status.UNAVAILABLE.withDescription("injected failure").asRuntimeException())
+ }
+ }
+ server = NettyServerBuilder.forPort(0).addService(failingService).build().start()
+ service = failingService
+ client = SparkConnectClient
+ .builder()
+ .connectionString(s"sc://localhost:${server.getPort}")
+ .retryPolicy(RetryPolicy(maxRetries = Some(0), canRetry = _ => false, name = "NoRetry"))
+ .build()
+
+ val ex = intercept[SparkException] {
+ client.analyze(proto.AnalyzePlanRequest.newBuilder().setSessionId("abc123").build())
+ }
+ // The original StatusRuntimeException must survive as the cause so that callers
+ // can programmatically inspect the gRPC status code instead of parsing the message.
+ assert(ex.getCause.isInstanceOf[StatusRuntimeException])
+ assert(
+ ex.getCause.asInstanceOf[StatusRuntimeException].getStatus.getCode ==
+ Status.Code.UNAVAILABLE)
+ }
+
test(
"SPARK-58094: gRPC keepalive surfaces a bounded failure on a silently dropped " +
"connection instead of hanging forever") {
@@ -963,9 +991,8 @@ class SparkConnectClientSuite extends ConnectFunSuite {
scala.concurrent.Await.result(resultPromise.future, FiniteDuration(15, TimeUnit.SECONDS))
}
// scalastyle:on awaitresult
- // A keepalive-triggered UNAVAILABLE carries no wrapped cause (same as DEADLINE_EXCEEDED,
- // see GrpcExceptionConverter.toThrowable), so the status code/description is only in the
- // message.
+ // The status code/description of a keepalive-triggered UNAVAILABLE is part of the
+ // message (see GrpcExceptionConverter.toThrowable).
assert(ex.getMessage.contains("UNAVAILABLE"))
assert(ex.getMessage.contains("Keepalive failed"))
} finally {
diff --git a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/GrpcExceptionConverter.scala b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/GrpcExceptionConverter.scala
index b52261e5efde..f4da00b5694f 100644
--- a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/GrpcExceptionConverter.scala
+++ b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/GrpcExceptionConverter.scala
@@ -170,23 +170,22 @@ private[client] class GrpcExceptionConverter(
}
// If no ErrorInfo is found, create a SparkException based on the StatusRuntimeException.
- val (message, cause) = if (ex.getStatus.getCode == Status.Code.DEADLINE_EXCEEDED) {
- val msg = s"${ex.toString}: RPC deadline exceeded. Deadlines can be configured via " +
+ val message = if (ex.getStatus.getCode == Status.Code.DEADLINE_EXCEEDED) {
+ s"${ex.toString}: RPC deadline exceeded. Deadlines can be configured via " +
"SparkConnectClient.Builder.rpcDeadlines(). To disable all deadlines: " +
"SparkConnectClient.builder().rpcDeadlines(RpcDeadlines.disabled).build()"
- // For DEADLINE_EXCEEDED, we pass `ex` itself as the cause rather than `ex.getCause`.
- // StatusRuntimeException.getCause() returns status.getCause(), which is always null for
- // client-side deadline fires (gRPC constructs the status without a wrapped cause). Using
- // ex.getCause would produce a SparkException with cause = null, losing the gRPC status
- // code and description from the exception chain. Passing ex preserves full context and
- // allows callers to programmatically inspect the status code via getCause().getStatus().
- (msg, ex)
} else {
- (ex.toString, ex.getCause)
+ ex.toString
}
+ // Pass `ex` itself as the cause rather than `ex.getCause`. StatusRuntimeException.getCause()
+ // returns status.getCause(), which is often null (e.g. always for client-side deadline
+ // fires, since gRPC constructs the status without a wrapped cause). Using ex.getCause
+ // would produce a SparkException with cause = null, losing the gRPC status code and
+ // description from the exception chain. Passing ex preserves full context and allows
+ // callers to programmatically inspect the status code via getCause().getStatus().
new SparkException(
message = message,
- cause = cause,
+ cause = ex,
errorClass = Some("CONNECT_CLIENT_UNEXPECTED_MISSING_SQL_STATE"),
messageParameters = Map("message" -> message),
context = Array.empty)
diff --git a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectServiceKeepAliveSuite.scala b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectServiceKeepAliveSuite.scala
index a0b249a002ab..ccf7a00ae1f2 100644
--- a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectServiceKeepAliveSuite.scala
+++ b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectServiceKeepAliveSuite.scala
@@ -108,9 +108,8 @@ class SparkConnectServiceKeepAliveSuite extends SparkConnectServerTest {
FiniteDuration(15, TimeUnit.SECONDS))
}
// scalastyle:on awaitresult
- // A keepalive-triggered UNAVAILABLE carries no wrapped cause (same as
- // DEADLINE_EXCEEDED, see GrpcExceptionConverter.toThrowable), so the status
- // code/description is only in the message.
+ // The status code/description of a keepalive-triggered UNAVAILABLE is part of the
+ // message (see GrpcExceptionConverter.toThrowable).
assert(ex.getMessage.contains("UNAVAILABLE"))
assert(ex.getMessage.contains("Keepalive failed"))
} finally {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org