fix: spark connect extra class path

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
 

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<String, Option<String>> = [
         (
             "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(

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"),