fix: spark connect extra class path

CHANGELOG.md

@@ -59,6 +59,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]).
+- 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]).
 - The history and connect controllers now watch all resources that they create, the missing RBAC
   `watch` permissions were added, and all controllers early-exit the reconcile action when the object
   is marked for deletion ([#757]).
@@ -81,6 +84,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
 [#757]: https://github.com/stackabletech/spark-k8s-operator/pull/757
 [#758]: https://github.com/stackabletech/spark-k8s-operator/pull/758
 [#761]: https://github.com/stackabletech/spark-k8s-operator/pull/761

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 {

rust/operator-binary/src/connect/controller/build/executor.rs

@@ -238,6 +238,13 @@ pub(crate) fn executor_properties(
             "spark.kubernetes.executor.container.image".to_string(),
             Some(spark_image),
         ),
+        // Must be the same value as `spark.driver.extraClassPath`, see `common::extra_class_path`.
+        (
+            "spark.executor.extraClassPath".to_string(),
+            Some(common::extra_class_path(
+                &resolved_product_image.product_version,
+            )),
+        ),
         (
             "spark.executor.defaultJavaOptions".to_string(),
             Some(executor_jvm_args(

rust/operator-binary/src/connect/controller/build/server.rs

@@ -403,7 +403,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()
@@ -447,9 +446,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(),

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