From 0c9852a8e53ce8160c0782654a65f0587f785714 Mon Sep 17 00:00:00 2001 From: Maxi Wittich Date: Mon, 24 Aug 2026 08:57:11 +0200 Subject: [PATCH 1/4] fix: spark connect extra class path --- .../src/connect/controller/build/executor.rs | 17 +++++++++++++++++ .../spark-connect/20-run-connect-client.yaml.j2 | 9 +++++++++ 2 files changed, 26 insertions(+) diff --git a/rust/operator-binary/src/connect/controller/build/executor.rs b/rust/operator-binary/src/connect/controller/build/executor.rs index ffbd29dd..ef81481d 100644 --- a/rust/operator-binary/src/connect/controller/build/executor.rs +++ b/rust/operator-binary/src/connect/controller/build/executor.rs @@ -224,12 +224,29 @@ pub(crate) fn executor_properties( let config = &validated.executor_config; let resolved_product_image = &validated.resolved_product_image; let spark_image = resolved_product_image.image.clone(); + let spark_version = resolved_product_image.product_version.clone(); let mut result: BTreeMap> = [ ( "spark.kubernetes.executor.container.image".to_string(), Some(spark_image), ), + // Must mirror `spark.driver.extraClassPath` on the server. + // + // Spark Connect jar is not in `/stackable/spark/jars`. + // Without it, the executors cannot deserialize the closures that Connect + // ships with every task that returns rows to a client, and any `count`, `collect` or + // `toPandas` fails with: + // + // java.lang.ClassCastException: cannot assign instance of + // java.lang.invoke.SerializedLambda to field + // org.apache.spark.rdd.MapPartitionsRDD.f of type scala.Function3 + ( + "spark.executor.extraClassPath".to_string(), + Some(format!( + "/stackable/spark/extra-jars/*:/stackable/spark/connect/spark-connect-{spark_version}.jar" + )), + ), ( "spark.executor.defaultJavaOptions".to_string(), Some(executor_jvm_args( diff --git a/tests/templates/kuttl/spark-connect/20-run-connect-client.yaml.j2 b/tests/templates/kuttl/spark-connect/20-run-connect-client.yaml.j2 index 86ca7848..3af44758 100644 --- a/tests/templates/kuttl/spark-connect/20-run-connect-client.yaml.j2 +++ b/tests/templates/kuttl/spark-connect/20-run-connect-client.yaml.j2 @@ -67,6 +67,15 @@ data: print(f"Reading data back from {ingest_bucket}") data = spark.read.parquet(ingest_bucket) + # Regression guard. `count` returns rows to the client through Spark Connect's Arrow + # conversion, and the closure that shipped to the executors is defined in the Connect jar. + # + # Without this assertion the test passes against a Connect server that cannot return a single row to a client. + print("Counting rows read back...") + row_count = data.count() + print(f"Row count: {row_count}") + assert row_count == 1000, f"expected 1000 rows, got {row_count}" + print("Computing statistics...") stats = data.groupBy(fn.month(fn.col("date_field")).alias("month")).agg( fn.count("*").alias("count"), From 4206dca1ede16c13f83c193e4baa055f508c22b2 Mon Sep 17 00:00:00 2001 From: Maxi Wittich Date: Mon, 24 Aug 2026 08:59:47 +0200 Subject: [PATCH 2/4] add CHANGELOG.md --- CHANGELOG.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 509f7903..3f79c2a4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -48,6 +48,9 @@ All notable changes to this project will be documented in this file. labels without the version label, so that the labels stay stable across upgrades. Existing connect and history-server StatefulSets must be deleted once before the new operator can reconcile them (connect server: [#750], history server: [#753]). +- Adds Spark-Connect `spark.executor.extraClassPath`, without which the Connect jar + and `/stackable/spark/extra-jars` were only on the driver and any query returning rows to a + client failed to deserialize its task ([#755]). [#721]: https://github.com/stackabletech/spark-k8s-operator/pull/721 [#727]: https://github.com/stackabletech/spark-k8s-operator/pull/727 @@ -58,6 +61,7 @@ All notable changes to this project will be documented in this file. [#750]: https://github.com/stackabletech/spark-k8s-operator/pull/750 [#753]: https://github.com/stackabletech/spark-k8s-operator/pull/753 [#754]: https://github.com/stackabletech/spark-k8s-operator/pull/754 +[#755]: https://github.com/stackabletech/spark-k8s-operator/pull/755 ## [26.7.0] - 2026-07-21 From 5a51dd7be0c868e0913ef1ffca3436ad6226f8ab Mon Sep 17 00:00:00 2001 From: Maxi Wittich Date: Mon, 31 Aug 2026 14:03:23 +0200 Subject: [PATCH 3/4] fix: moving extra_class_path() to common.rs --- rust/operator-binary/src/connect/common.rs | 15 +++++++++++++++ .../src/connect/controller/build/executor.rs | 16 +++------------- .../src/connect/controller/build/server.rs | 6 ++++-- 3 files changed, 22 insertions(+), 15 deletions(-) diff --git a/rust/operator-binary/src/connect/common.rs b/rust/operator-binary/src/connect/common.rs index 426c80ac..c43c4638 100644 --- a/rust/operator-binary/src/connect/common.rs +++ b/rust/operator-binary/src/connect/common.rs @@ -59,6 +59,21 @@ pub(crate) fn object_name(stacklet_name: &str, role: SparkConnectRole) -> String } } +// Returns the extra class path shared by the Connect server and its executors. +// +// The product image keeps the Spark Connect jars out of `/stackable/spark/jars` to +// avoid class path conflicts with regular Spark applications, so both roles have to add the Connect +// jar explicitly. +// +// `spark.driver.extraClassPath` and `spark.executor.extraClassPath` must be set to this same value: +// the Connect server ships closures defined in the Connect jar with every task that returns rows to +// a client. +pub(crate) fn extra_class_path(product_version: &str) -> String { + format!( + "/stackable/spark/extra-jars/*:/stackable/spark/connect/spark-connect-{product_version}.jar" + ) +} + // Returns the operator-generated jvm arguments with the user-provided overrides applied on top. pub(crate) fn jvm_args(jvm_args: &[String], user_java_config: Option<&JavaCommonConfig>) -> String { match user_java_config { diff --git a/rust/operator-binary/src/connect/controller/build/executor.rs b/rust/operator-binary/src/connect/controller/build/executor.rs index ef81481d..2f6470c5 100644 --- a/rust/operator-binary/src/connect/controller/build/executor.rs +++ b/rust/operator-binary/src/connect/controller/build/executor.rs @@ -224,27 +224,17 @@ pub(crate) fn executor_properties( let config = &validated.executor_config; let resolved_product_image = &validated.resolved_product_image; let spark_image = resolved_product_image.image.clone(); - let spark_version = resolved_product_image.product_version.clone(); let mut result: BTreeMap> = [ ( "spark.kubernetes.executor.container.image".to_string(), Some(spark_image), ), - // Must mirror `spark.driver.extraClassPath` on the server. - // - // Spark Connect jar is not in `/stackable/spark/jars`. - // Without it, the executors cannot deserialize the closures that Connect - // ships with every task that returns rows to a client, and any `count`, `collect` or - // `toPandas` fails with: - // - // java.lang.ClassCastException: cannot assign instance of - // java.lang.invoke.SerializedLambda to field - // org.apache.spark.rdd.MapPartitionsRDD.f of type scala.Function3 + // Must be the same value as `spark.driver.extraClassPath`, see `common::extra_class_path`. ( "spark.executor.extraClassPath".to_string(), - Some(format!( - "/stackable/spark/extra-jars/*:/stackable/spark/connect/spark-connect-{spark_version}.jar" + Some(common::extra_class_path( + &resolved_product_image.product_version, )), ), ( diff --git a/rust/operator-binary/src/connect/controller/build/server.rs b/rust/operator-binary/src/connect/controller/build/server.rs index ae074a6b..740cd9ac 100644 --- a/rust/operator-binary/src/connect/controller/build/server.rs +++ b/rust/operator-binary/src/connect/controller/build/server.rs @@ -395,7 +395,6 @@ pub(crate) fn server_properties( let config = &validated.server_config; let resolved_product_image = &validated.resolved_product_image; let spark_image = resolved_product_image.image.clone(); - let spark_version = resolved_product_image.product_version.clone(); let service_account_name = validated.cluster_resource_names().service_account_name(); let namespace = driver_service .namespace() @@ -435,9 +434,12 @@ pub(crate) fn server_properties( config, )), ), + // Must be the same value as `spark.executor.extraClassPath`, see `common::extra_class_path`. ( "spark.driver.extraClassPath".to_string(), - Some(format!("/stackable/spark/extra-jars/*:/stackable/spark/connect/spark-connect-{spark_version}.jar")), + Some(common::extra_class_path( + &resolved_product_image.product_version, + )), ), ( "spark.metrics.conf".to_string(), From 73359c2546e77c92e304350da78c26f9ee18116e Mon Sep 17 00:00:00 2001 From: Maximilian Wittich <56642549+Maleware@users.noreply.github.com> Date: Mon, 31 Aug 2026 14:07:25 +0200 Subject: [PATCH 4/4] Update CHANGELOG.md Co-authored-by: Lars Francke --- CHANGELOG.md | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3f79c2a4..fea778c8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -48,9 +48,7 @@ All notable changes to this project will be documented in this file. labels without the version label, so that the labels stay stable across upgrades. Existing connect and history-server StatefulSets must be deleted once before the new operator can reconcile them (connect server: [#750], history server: [#753]). -- Adds Spark-Connect `spark.executor.extraClassPath`, without which the Connect jar - and `/stackable/spark/extra-jars` were only on the driver and any query returning rows to a - client failed to deserialize its task ([#755]). +- Previously the Spark Connect executors ran without `spark.executor.extraClassPath`, so the Connect jar and `/stackable/spark/extra-jars` were on the driver only and any query returning rows to a client failed to deserialize its tasks. Fixed by setting the property on the executor role to the same value as `spark.driver.extraClassPath` ([#755]). [#721]: https://github.com/stackabletech/spark-k8s-operator/pull/721 [#727]: https://github.com/stackabletech/spark-k8s-operator/pull/727