Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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]).
Expand All @@ -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
Expand Down
15 changes: 15 additions & 0 deletions rust/operator-binary/src/connect/common.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
7 changes: 7 additions & 0 deletions rust/operator-binary/src/connect/controller/build/executor.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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(),

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't know enough about how our operators work: Can this still be overridden/appended to by users? That is a relatively "common" property.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We can still override but not extend extraClassPath in all roles independently.

Some(common::extra_class_path(
&resolved_product_image.product_version,
)),
),
(
"spark.executor.defaultJavaOptions".to_string(),
Some(executor_jvm_args(
Expand Down
6 changes: 4 additions & 2 deletions rust/operator-binary/src/connect/controller/build/server.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down Expand Up @@ -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(),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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"),
Expand Down
Loading