diff --git a/CHANGELOG.md b/CHANGELOG.md index 3a8770f6..c91dd145 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -48,6 +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]). +- 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]). @@ -61,6 +62,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 ## [26.7.0] - 2026-07-21 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 ffbd29dd..2f6470c5 100644 --- a/rust/operator-binary/src/connect/controller/build/executor.rs +++ b/rust/operator-binary/src/connect/controller/build/executor.rs @@ -230,6 +230,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( 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(), 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"),