zifeif2 opened a new pull request, #59449: URL: https://github.com/apache/spark/pull/59449
### What changes were proposed in this pull request? Add regression tests that reproduce Scala Spark Connect losing structured server errors whose conditions are absent from the client's core error catalog. This PR changes only `SparkConnectClientSuite`; it contains no production fix. The tests cover unknown conditions and subclasses across all 18 mapped exception types, enriched Kafka errors with causes/query contexts/server stack traces, and the ErrorInfo fallback with truncated parameters. The existing known-condition test also checks the actual message and every mapped type. The implementation is reviewed separately in https://github.com/apache/spark/pull/59409. ### Why are the changes needed? The server sends its rendered message, but the Scala converter reconstructs some exception types using constructors that render the condition again with the client's core catalog. Connector-specific conditions and conditions from newer servers can be absent there. Reconstruction then throws `INTERNAL_ERROR`, hiding the original failure. The regression tests intentionally assert the desired behavior and fail on unfixed Spark. This reproduction PR stays draft while the implementation is reviewed separately. Merge the implementation before these tests. ### Does this PR introduce _any_ user-facing change? No. This PR adds tests only. ### How was this patch tested? Compiled the seven affected production files from current master and from the fix, plus this test suite, using Scala 2.13.18, Java 17, and cached dependencies. Ran the 10 tests matching `error` through ScalaTest: - Unfixed code: 7 passed, 3 failed. All three new regression tests fail because client reconstruction throws `INTERNAL_ERROR` for an unknown condition. - Fixed code from #59409: all 10 passed. - `git diff --check` passed. The normal SBT invocation could not load the project because Maven Central was unreachable from the local environment; the direct compiler/ScalaTest runs above used the existing cached test classpath. No source changes were made to work around dependency resolution. To run the focused tests through the normal build: ```bash build/sbt 'connect-client-jvm/testOnly *SparkConnectClientSuite -- -z error' ``` The released Spark 4.1.1 reproduction below also demonstrates the existing issue with real Kafka option validation. It does not require #59219 or a Kafka broker. The recorded released-version runs are separate from the regression-test validation above. #### Released Spark 4.1.1 reproduction This is reproducible with existing Kafka conditions in released Spark 4.1.1, without applying #59219. Both the Connect client/server and the classic comparison used Spark 4.1.1, the Kafka 4.1.1 connector, Java 17, and Scala 2.13.17. The Connect server ran with `local[2]` on a Linux development machine. A separate Scala client JVM connected over loopback gRPC. Classic Scala ran in its own local Spark driver. No Databricks workspace or remote cluster was used. No Kafka broker, topic, or data is required: these invalid offset options fail validation before broker access. Two existing conditions were reproduced: | Input | Existing Kafka condition | | --- | --- | | Start at offset 10 and end at offset 5 for `repro_topic`, partition 0 | `UNRESOLVED_START_OFFSET_GREATER_THAN_END_OFFSET` | | Start offsets name partition 0; end offsets name partition 1 | `MISMATCHED_TOPIC_PARTITIONS_BETWEEN_START_OFFSET_AND_END_OFFSET` | For both inputs: | Where the error was observed | Result | | --- | --- | | Connect server log | `KafkaIllegalArgumentException` with the useful Kafka explanation | | Scala Connect client | `SparkException`, condition `INTERNAL_ERROR`, SQLSTATE `XX000`; original explanation lost | | Classic Scala | `KafkaIllegalArgumentException`, original Kafka condition and explanation preserved | Recorded Scala Connect output: ```text Client: released Scala Spark Connect 4.1.1 Server Spark version: 4.1.1 Session implementation: org.apache.spark.sql.connect.SparkSession Actual client exception: org.apache.spark.SparkException Actual client condition: INTERNAL_ERROR Actual client message: [INTERNAL_ERROR] Cannot find main error class 'UNRESOLVED_START_OFFSET_GREATER_THAN_END_OFFSET' SQLSTATE: XX000 Actual client exception: org.apache.spark.SparkException Actual client condition: INTERNAL_ERROR Actual client message: [INTERNAL_ERROR] Cannot find main error class 'MISMATCHED_TOPIC_PARTITIONS_BETWEEN_START_OFFSET_AND_END_OFFSET' SQLSTATE: XX000 ``` The client stack goes through `SparkThrowableHelper.getMessage`, `GrpcExceptionConverter.errorFactory`, `errorsToThrowable`, and `fetchEnrichedError`. The server log contains the original explanations: ```text org.apache.spark.sql.kafka010.KafkaIllegalArgumentException: The specified start offset 10 is greater than the end offset 5 for topic repro_topic partition 0. org.apache.spark.sql.kafka010.KafkaIllegalArgumentException: The specified start offset and end offset should have the same topic-partitions. ``` Recorded classic Scala output: ```text Spark version: 4.1.1 Session implementation: org.apache.spark.sql.classic.SparkSession SparkContext master: local[2] Actual exception: org.apache.spark.sql.kafka010.KafkaIllegalArgumentException Actual condition: UNRESOLVED_START_OFFSET_GREATER_THAN_END_OFFSET Actual message: The specified start offset 10 is greater than the end offset 5 for topic repro_topic partition 0. Actual exception: org.apache.spark.sql.kafka010.KafkaIllegalArgumentException Actual condition: MISMATCHED_TOPIC_PARTITIONS_BETWEEN_START_OFFSET_AND_END_OFFSET Actual message: The specified start offset and end offset should have the same topic-partitions. Topic-partitions for start offset: repro_topic-0, topic-partitions for end offset: repro_topic-1. ``` These recorded runs demonstrate the bug in the released binaries. The focused tests listed above validate this PR's fix; the released-version comparison is not an end-to-end test of a patched client binary. The standalone Scala sources below also compiled successfully against the released 4.1.1 dependencies. <details> <summary>Standalone Scala reproduction sources</summary> Save these two files together in a working directory. The Connect program asserts the existing bug; the classic program asserts that both original errors survive. ##### ScalaConnectKafkaErrorRepro.scala ```scala import org.apache.spark.SparkThrowable import org.apache.spark.sql.connect.SparkSession import scala.util.control.NonFatal /** * Existing Kafka errors masked by Scala Spark Connect 4.1.1. * Run with the released spark-connect-client-jvm_2.13:4.1.1 client in a separate JVM. * The Connect server needs the matching Kafka connector. No Kafka broker is required. */ object ScalaConnectKafkaErrorRepro { case class ReproCase( condition: String, startOffsets: String, endOffsets: String, expectedMessage: String) val cases = Seq( ReproCase( "UNRESOLVED_START_OFFSET_GREATER_THAN_END_OFFSET", """{"repro_topic":{"0":10}}""", """{"repro_topic":{"0":5}}""", "The specified start offset 10 is greater than the end offset 5 " + "for topic repro_topic partition 0."), ReproCase( "MISMATCHED_TOPIC_PARTITIONS_BETWEEN_START_OFFSET_AND_END_OFFSET", """{"repro_topic":{"0":10}}""", """{"repro_topic":{"1":5}}""", "The specified start offset and end offset should have the same topic-partitions.")) def main(args: Array[String]): Unit = { require(args.length == 1, "Usage: ScalaConnectKafkaErrorRepro sc://localhost:15002") val spark = SparkSession.builder().remote(args(0)).getOrCreate() try { println("Client: released Scala Spark Connect 4.1.1") println(s"Server Spark version: ${spark.version}") println(s"Session implementation: ${spark.getClass.getName}") cases.foreach { testCase => println(s"\nExisting Kafka condition: ${testCase.condition}") println(s"Expected Kafka message: ${testCase.expectedMessage}") val caught = try { spark.read .format("kafka") .option("kafka.bootstrap.servers", "127.0.0.1:1") .option("assign", """{"repro_topic":[0,1]}""") .option("startingOffsets", testCase.startOffsets) .option("endingOffsets", testCase.endOffsets) .option("kafka.request.timeout.ms", "1000") .option("kafka.default.api.timeout.ms", "1000") .load() .collect() None } catch { case NonFatal(error) => Some(error) } val error = caught.getOrElse(sys.error("Expected the invalid Kafka options to fail.")) val condition = error match { case structured: SparkThrowable => structured.getCondition case _ => "<not a SparkThrowable>" } println(s"Actual client exception: ${error.getClass.getName}") println(s"Actual client condition: $condition") println(s"Actual client message: ${error.getMessage}") println("Relevant client stack frames:") error.getStackTrace.filter { frame => frame.getClassName.contains("GrpcExceptionConverter") || frame.getClassName.contains("ErrorClassesJSONReader") || frame.getClassName.contains("SparkThrowableHelper") }.take(8).foreach(frame => println(s" at $frame")) require(condition == "INTERNAL_ERROR", "Did not reproduce the original converter bug.") require(error.getMessage.contains("Cannot find main error class")) require(error.getMessage.contains(testCase.condition)) require(error.getStackTrace.exists(_.getClassName.contains("GrpcExceptionConverter"))) println("REPRODUCED: Scala Connect replaced the Kafka error with INTERNAL_ERROR.") } println("\nBoth existing Kafka errors reproduced through real Scala Connect RPCs.") } finally { spark.stop() } } } ``` ##### ScalaClassicKafkaErrorRepro.scala ```scala import org.apache.spark.SparkThrowable import org.apache.spark.sql.SparkSession import scala.util.control.NonFatal /** * Classic Scala counterpart of ScalaConnectKafkaErrorRepro. * Uses the same Kafka options and existing conditions on released Spark 4.1.1. * The Kafka connector is required; no Kafka broker is needed. */ object ScalaClassicKafkaErrorRepro { case class ReproCase( condition: String, startOffsets: String, endOffsets: String, expectedMessage: String) val cases = Seq( ReproCase( "UNRESOLVED_START_OFFSET_GREATER_THAN_END_OFFSET", """{"repro_topic":{"0":10}}""", """{"repro_topic":{"0":5}}""", "The specified start offset 10 is greater than the end offset 5 " + "for topic repro_topic partition 0."), ReproCase( "MISMATCHED_TOPIC_PARTITIONS_BETWEEN_START_OFFSET_AND_END_OFFSET", """{"repro_topic":{"0":10}}""", """{"repro_topic":{"1":5}}""", "The specified start offset and end offset should have the same topic-partitions.")) // Notebook callers can reuse their existing classic session without stopping it. def verify(spark: SparkSession): Unit = { require(spark.isInstanceOf[org.apache.spark.sql.classic.SparkSession], "Use a classic Scala SparkSession for this reproduction.") println(s"Spark version: ${spark.version}") println(s"Session implementation: ${spark.getClass.getName}") println(s"SparkContext master: ${spark.sparkContext.master}") cases.foreach { testCase => println(s"\nExisting Kafka condition: ${testCase.condition}") println(s"Expected Kafka message: ${testCase.expectedMessage}") val caught = try { spark.read .format("kafka") .option("kafka.bootstrap.servers", "127.0.0.1:1") .option("assign", """{"repro_topic":[0,1]}""") .option("startingOffsets", testCase.startOffsets) .option("endingOffsets", testCase.endOffsets) .option("kafka.request.timeout.ms", "1000") .option("kafka.default.api.timeout.ms", "1000") .load() .collect() None } catch { case NonFatal(error) => Some(error) } val error = caught.getOrElse(sys.error("Expected the invalid Kafka options to fail.")) val condition = error match { case structured: SparkThrowable => structured.getCondition case _ => "<not a SparkThrowable>" } val message = error.getMessage println(s"Actual exception: ${error.getClass.getName}") println(s"Actual condition: $condition") println(s"Actual message: $message") require(error.isInstanceOf[IllegalArgumentException]) require(condition == testCase.condition, "The original Kafka condition was not preserved.") require(message.contains(testCase.expectedMessage), "The Kafka explanation was lost.") require(!message.contains("INTERNAL_ERROR")) require(!error.getStackTrace.exists(_.getClassName.contains("GrpcExceptionConverter"))) println("PASS: classic Scala preserved the Kafka condition and message; no INTERNAL_ERROR.") } println("\nBoth existing Kafka errors remain actionable in classic Scala Spark.") } def main(args: Array[String]): Unit = { val spark = SparkSession.builder() .master("local[2]") .appName("classic-scala-kafka-error-repro") .config("spark.api.mode", "classic") .getOrCreate() try { require(spark.version == "4.1.1", "Use Spark 4.1.1 for the same-version comparison.") spark.sparkContext.setLogLevel("ERROR") verify(spark) } finally { spark.stop() } } } ``` </details> <details> <summary>Build and run the standalone Scala reproductions</summary> Prerequisites: a Spark 4.1.1 binary distribution, Java 17 JDK, and Maven. Set `SPARK_HOME` to that distribution and `JAVA_HOME` to the JDK; make `mvn` available on `PATH`. Run these Bash commands in the directory containing the two Scala files. Maven resolves public artifacts from Maven Central (or your configured mirror). Python and Jupyter are not needed. **1. Resolve dependencies and build the two application JARs.** ```bash set -euo pipefail : "${SPARK_HOME:?Set SPARK_HOME to a Spark 4.1.1 binary distribution}" : "${JAVA_HOME:?Set JAVA_HOME to a Java 17 JDK}" unset SPARK_REMOTE SPARK_CONNECT_MODE_ENABLED SPARK_API_MODE export SPARK_HOME JAVA_HOME export SPARK_LOCAL_IP=127.0.0.1 export KAFKA_REPRO_DIR="$PWD" mkdir -p client-deps kafka-deps connect-classes classic-classes cat > client-deps/pom.xml <<'XML' <project xmlns="http://maven.apache.org/POM/4.0.0";> <modelVersion>4.0.0</modelVersion> <groupId>local.reproduction</groupId><artifactId>connect-client</artifactId><version>1</version> <dependencies><dependency> <groupId>org.apache.spark</groupId><artifactId>spark-connect-client-jvm_2.13</artifactId> <version>4.1.1</version> </dependency></dependencies> </project> XML cat > kafka-deps/pom.xml <<'XML' <project xmlns="http://maven.apache.org/POM/4.0.0";> <modelVersion>4.0.0</modelVersion> <groupId>local.reproduction</groupId><artifactId>kafka-connector</artifactId><version>1</version> <dependencies><dependency> <groupId>org.apache.spark</groupId><artifactId>spark-sql-kafka-0-10_2.13</artifactId> <version>4.1.1</version> </dependency></dependencies> </project> XML mvn -B -q -f client-deps/pom.xml \ org.apache.maven.plugins:maven-dependency-plugin:3.8.1:copy-dependencies \ -DincludeScope=runtime -DoutputDirectory="$KAFKA_REPRO_DIR/client-deps/jars" mvn -B -q -f kafka-deps/pom.xml \ org.apache.maven.plugins:maven-dependency-plugin:3.8.1:copy-dependencies \ -DincludeScope=runtime \ -DincludeArtifactIds=spark-sql-kafka-0-10_2.13,spark-token-provider-kafka-0-10_2.13,kafka-clients,commons-pool2 \ -DoutputDirectory="$KAFKA_REPRO_DIR/kafka-deps/jars" "$JAVA_HOME/bin/java" -cp "$KAFKA_REPRO_DIR/client-deps/jars/*" \ scala.tools.nsc.Main -usejavacp -d connect-classes ScalaConnectKafkaErrorRepro.scala "$JAVA_HOME/bin/jar" --create --file connect-repro.jar -C connect-classes . "$JAVA_HOME/bin/java" -cp "$SPARK_HOME/jars/*" \ scala.tools.nsc.Main -usejavacp -d classic-classes ScalaClassicKafkaErrorRepro.scala "$JAVA_HOME/bin/jar" --create --file classic-repro.jar -C classic-classes . printf -v KAFKA_REPRO_JARS '%s,' "$KAFKA_REPRO_DIR"/kafka-deps/jars/*.jar export KAFKA_REPRO_JARS="${KAFKA_REPRO_JARS%,}" ``` **2. Start the Connect server in the same terminal.** The Kafka connector belongs on the server classpath. Leave this process running and wait for it to report that it is listening on port 15002. ```bash "$SPARK_HOME/bin/spark-submit" \ --master 'local[2]' \ --class org.apache.spark.sql.connect.service.SparkConnectServer \ --conf spark.ui.enabled=false \ --conf spark.driver.bindAddress=127.0.0.1 \ --conf spark.connect.grpc.binding.address=127.0.0.1 \ --conf spark.connect.grpc.binding.port=15002 \ --conf spark.port.maxRetries=0 \ --jars "$KAFKA_REPRO_JARS" \ "$SPARK_HOME/jars/spark-connect_2.13-4.1.1.jar" ``` **3. Run the Scala Connect client in a second terminal.** Set `JAVA_HOME` in that terminal too, and change to the same reproduction directory. The client classpath contains the Connect client and its dependencies. Expect two `REPRODUCED` messages, each with `INTERNAL_ERROR` and the missing condition name. ```bash unset SPARK_REMOTE SPARK_CONNECT_MODE_ENABLED SPARK_API_MODE "$JAVA_HOME/bin/java" \ --add-opens=java.base/java.nio=ALL-UNNAMED \ -cp "$PWD/connect-repro.jar:$PWD/client-deps/jars/*" \ ScalaConnectKafkaErrorRepro sc://127.0.0.1:15002 ``` **4. Stop the server with Ctrl+C, then run classic Scala in the first terminal.** Expect two `PASS` messages, each retaining the original Kafka condition and useful message. Both modes need the Kafka dependency JARs; omitting them produces `Failed to find data source: kafka`, which is a setup failure. ```bash "$SPARK_HOME/bin/spark-submit" \ --master 'local[2]' \ --class ScalaClassicKafkaErrorRepro \ --conf spark.api.mode=classic \ --conf spark.ui.enabled=false \ --conf spark.driver.bindAddress=127.0.0.1 \ --jars "$KAFKA_REPRO_JARS" \ "$KAFKA_REPRO_DIR/classic-repro.jar" ``` </details> ### Was this patch authored or co-authored using generative AI tooling? Generated-by: OpenAI Codex (GPT-6) -- 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]
