feat: support Spark 3.5.8

Author: sweb

CHANGELOG.md

@@ -8,7 +8,7 @@ All notable changes to this project will be documented in this file.
 
 - BREAKING: Add required CLI argument and env var to set the image repository used to construct final product image names: `IMAGE_REPOSITORY` (`--image-repository`), eg. `oci.example.org/my/namespace` ([#684]).
 - Support for Spark `4.1.2` ([#708])
-- Add `spec.openLineage` to enable OpenLineage lineage emission: injects the OpenLineage Spark listener, HTTP transport (endpoint resolved from a discovery ConfigMap), a stable job name, and the required `--add-opens` JVM flag ([#XXXX]).
+- Add `spec.openLineage` to enable OpenLineage lineage emission: injects the OpenLineage Spark listener, HTTP transport (endpoint resolved from a discovery ConfigMap) and a stable job name. The Scala build of the listener jar and the `--add-opens java.base/java.security` JVM flag are selected by the Spark version: the `--add-opens` flag is emitted only on Spark 4.x (it is unnecessary on the Spark 3.5.x images and must not be set there) ([#XXXX]).
 
 ### Fixed
 

docs/modules/spark-k8s/pages/usage-guide/openlineage.adoc

@@ -6,10 +6,10 @@ The Spark operator can automatically emit lineage events for a `SparkApplication
 When enabled, the operator injects everything required to make the https://openlineage.io/docs/integrations/spark/[OpenLineage Spark listener] work:
 
 * the OpenLineage Spark listener (appended to `spark.extraListeners`, so any listener you configure yourself is preserved),
-* the OpenLineage jar baked into the Spark image, referenced via `spark.jars` so it shares the same classloader as connectors pulled in through xref:usage-guide/job-dependencies.adoc[`deps.packages`] (for example Apache Iceberg),
+* the OpenLineage jar baked into the Spark image, referenced via `spark.jars` so it shares the same classloader as connectors pulled in through xref:usage-guide/job-dependencies.adoc[`deps.packages`] (for example Apache Iceberg). The operator selects the jar built for the image's Scala version automatically (Scala 2.13 for Spark 4.x, Scala 2.12 for Spark 3.5.x),
 * an HTTP transport pointing at the backend endpoint,
 * a stable lineage namespace and job name (see <<job-name>>), and
-* the `--add-opens java.base/java.security=ALL-UNNAMED` JVM flag the listener requires on the driver and executors.
+* on Spark 4.x only, the `--add-opens java.base/java.security=ALL-UNNAMED` JVM flag the listener requires on the driver and executors. This flag is not added on the Spark 3.5.x images, where it is unnecessary (and, on their JVM, potentially harmful).
 
 == Enabling OpenLineage
 

rust/operator-binary/src/config/jvm.rs

@@ -1,11 +1,13 @@
 use stackable_operator::crd::s3;
 
 use crate::crd::{
+    Error,
     constants::{
         JVM_SECURITY_PROPERTIES_FILE, OPENLINEAGE_ADD_OPENS, STACKABLE_TLS_STORE_PASSWORD,
         STACKABLE_TRUST_STORE, VOLUME_MOUNT_PATH_LOG_CONFIG,
     },
     logdir::ResolvedLogDir,
+    spark_major_version,
     tlscerts::tls_secret_names,
     v1alpha1::SparkApplication,
 };
@@ -16,11 +18,15 @@ use crate::crd::{
 ///
 /// Returns `(driver, executor)`: the operator-generated base arguments with the role's
 /// `jvmArgumentOverrides` applied on top.
+///
+/// `product_version` is the resolved Spark product version (e.g. `4.1.2`); it gates the
+/// version-specific OpenLineage `--add-opens` flag.
 pub fn construct_extra_java_options(
     spark_application: &SparkApplication,
     s3_conn: &Option<s3::v1alpha1::ConnectionSpec>,
     log_dir: &Option<ResolvedLogDir>,
-) -> (String, String) {
+    product_version: &str,
+) -> Result<(String, String), Error> {
     // Note (@sbernauer): As of 2025-03-04, we did not set any heap related JVM arguments, so I
     // kept the implementation as is. We can always re-visit this as needed.
 
@@ -39,12 +45,10 @@ pub fn construct_extra_java_options(
     // OpenLineage on Spark 4.x needs `java.base/java.security` opened to the unnamed module,
     // otherwise the driver throws a non-fatal `InaccessibleObjectException` and silently degrades
     // extension-interface lineage (MVP §7). Added to both driver and executor.
-    if spark_application
-        .spec
-        .open_lineage
-        .as_ref()
-        .is_some_and(|open_lineage| open_lineage.enabled)
-    {
+    //
+    // This is scoped to Spark 4.x: the flag is a no-op — and on some JVMs a startup error — on the
+    // JDK 11/17 Spark 3.x images the operator also ships, so it must not be emitted there.
+    if spark_application.openlineage_enabled() && spark_major_version(product_version)? >= 4 {
         jvm_args.push(OPENLINEAGE_ADD_OPENS.to_string());
     }
 
@@ -68,16 +72,24 @@ pub fn construct_extra_java_options(
         None => jvm_args.clone(),
     };
 
-    (driver.join(" "), executor.join(" "))
+    Ok((driver.join(" "), executor.join(" ")))
 }
 
 #[cfg(test)]
 mod tests {
+    use rstest::rstest;
+
     use super::*;
 
+    fn spark_app_from_yaml(input: &str) -> SparkApplication {
+        let deserializer = serde_yaml::Deserializer::from_str(input);
+        serde_yaml::with::singleton_map_recursive::deserialize(deserializer).unwrap()
+    }
+
     #[test]
     fn test_construct_jvm_arguments_defaults() {
-        let input = r#"
+        let spark_app = spark_app_from_yaml(
+            r#"
             apiVersion: spark.stackable.tech/v1alpha1
             kind: SparkApplication
             metadata:
@@ -86,14 +98,11 @@ mod tests {
               mode: cluster
               mainApplicationFile: test.py
               sparkImage:
-                productVersion: 1.2.3
-        "#;
-
-        let deserializer = serde_yaml::Deserializer::from_str(input);
-        let spark_app: SparkApplication =
-            serde_yaml::with::singleton_map_recursive::deserialize(deserializer).unwrap();
+                productVersion: 4.1.2
+        "#,
+        );
         let (driver_extra_java_options, executor_extra_java_options) =
-            construct_extra_java_options(&spark_app, &None, &None);
+            construct_extra_java_options(&spark_app, &None, &None, "4.1.2").unwrap();
 
         assert_eq!(
             driver_extra_java_options,
@@ -107,7 +116,8 @@ mod tests {
 
     #[test]
     fn test_construct_jvm_argument_overrides() {
-        let input = r#"
+        let spark_app = spark_app_from_yaml(
+            r#"
             apiVersion: spark.stackable.tech/v1alpha1
             kind: SparkApplication
             metadata:
@@ -116,7 +126,7 @@ mod tests {
               mode: cluster
               mainApplicationFile: test.py
               sparkImage:
-                productVersion: 1.2.3
+                productVersion: 4.1.2
               driver:
                 jvmArgumentOverrides:
                   add:
@@ -127,13 +137,10 @@ mod tests {
                     - -Dhttps.proxyHost=from-executor
                   removeRegex:
                     - -Djava.security.properties=.*
-        "#;
-
-        let deserializer = serde_yaml::Deserializer::from_str(input);
-        let spark_app: SparkApplication =
-            serde_yaml::with::singleton_map_recursive::deserialize(deserializer).unwrap();
+        "#,
+        );
         let (driver_extra_java_options, executor_extra_java_options) =
-            construct_extra_java_options(&spark_app, &None, &None);
+            construct_extra_java_options(&spark_app, &None, &None, "4.1.2").unwrap();
 
         assert_eq!(
             driver_extra_java_options,
@@ -145,29 +152,71 @@ mod tests {
         );
     }
 
-    #[test]
-    fn test_construct_jvm_arguments_openlineage_add_opens() {
-        let input = r#"
-            apiVersion: spark.stackable.tech/v1alpha1
-            kind: SparkApplication
-            metadata:
-              name: spark-example
-            spec:
-              mode: cluster
-              mainApplicationFile: test.py
-              sparkImage:
-                productVersion: 1.2.3
-              openLineage:
-                enabled: true
-        "#;
-
-        let deserializer = serde_yaml::Deserializer::from_str(input);
-        let spark_app: SparkApplication =
-            serde_yaml::with::singleton_map_recursive::deserialize(deserializer).unwrap();
+    const OPENLINEAGE_ENABLED: &str = r#"
+        apiVersion: spark.stackable.tech/v1alpha1
+        kind: SparkApplication
+        metadata:
+          name: spark-example
+        spec:
+          mode: cluster
+          mainApplicationFile: test.py
+          sparkImage:
+            productVersion: PLACEHOLDER
+          openLineage:
+            enabled: true
+    "#;
+
+    const OPENLINEAGE_DISABLED: &str = r#"
+        apiVersion: spark.stackable.tech/v1alpha1
+        kind: SparkApplication
+        metadata:
+          name: spark-example
+        spec:
+          mode: cluster
+          mainApplicationFile: test.py
+          sparkImage:
+            productVersion: PLACEHOLDER
+          openLineage:
+            enabled: false
+    "#;
+
+    const OPENLINEAGE_ABSENT: &str = r#"
+        apiVersion: spark.stackable.tech/v1alpha1
+        kind: SparkApplication
+        metadata:
+          name: spark-example
+        spec:
+          mode: cluster
+          mainApplicationFile: test.py
+          sparkImage:
+            productVersion: PLACEHOLDER
+    "#;
+
+    /// `--add-opens` is emitted only on Spark 4.x with OpenLineage enabled — never on the Scala
+    /// 2.12 / JDK 17 Spark 3.5.x images, and never when OpenLineage is disabled or absent.
+    #[rstest]
+    #[case::enabled_spark_3(OPENLINEAGE_ENABLED, "3.5.8", false)]
+    #[case::enabled_spark_4(OPENLINEAGE_ENABLED, "4.1.2", true)]
+    #[case::disabled_spark_4(OPENLINEAGE_DISABLED, "4.1.2", false)]
+    #[case::absent_spark_4(OPENLINEAGE_ABSENT, "4.1.2", false)]
+    fn test_openlineage_add_opens_is_version_gated(
+        #[case] yaml_template: &str,
+        #[case] product_version: &str,
+        #[case] expect_add_opens: bool,
+    ) {
+        let spark_app = spark_app_from_yaml(&yaml_template.replace("PLACEHOLDER", product_version));
         let (driver_extra_java_options, executor_extra_java_options) =
-            construct_extra_java_options(&spark_app, &None, &None);
+            construct_extra_java_options(&spark_app, &None, &None, product_version).unwrap();
 
-        assert!(driver_extra_java_options.contains(OPENLINEAGE_ADD_OPENS));
-        assert!(executor_extra_java_options.contains(OPENLINEAGE_ADD_OPENS));
+        assert_eq!(
+            driver_extra_java_options.contains(OPENLINEAGE_ADD_OPENS),
+            expect_add_opens,
+            "driver --add-opens presence mismatch for Spark {product_version}"
+        );
+        assert_eq!(
+            executor_extra_java_options.contains(OPENLINEAGE_ADD_OPENS),
+            expect_add_opens,
+            "executor --add-opens presence mismatch for Spark {product_version}"
+        );
     }
 }

rust/operator-binary/src/crd/constants.rs

@@ -108,15 +108,25 @@ pub const DEFAULT_SUBMIT_JOB_RETRY_ON_FAILURE_COUNT: u16 = 0;
 /// The OpenLineage Spark listener, appended to `spark.extraListeners` when OpenLineage is enabled.
 pub const OPENLINEAGE_LISTENER_CLASS: &str = "io.openlineage.spark.agent.OpenLineageSparkListener";
 
+/// Version of the OpenLineage Spark listener jar baked into the Spark images.
+///
+/// IMPORTANT: MUST stay in sync with the `openlineage-spark-version` build-argument in
+/// `docker-images/spark-k8s/boil-config.toml` (the image is what places the jar on disk).
+pub const OPENLINEAGE_SPARK_JAR_VERSION: &str = "1.51.0";
+
 /// `local://` URI of the OpenLineage jar baked into the Spark image, referenced via `spark.jars` so
 /// it shares the (child) classloader with `--packages` connectors. Delivering it this way — rather
 /// than on `extraClassPath` (system classloader) — is what lets OpenLineage's connector probes
 /// succeed; see the classpath discussion in the OpenLineage usage guide.
 ///
-/// IMPORTANT: the version MUST stay in sync with the `openlineage-spark-version` build-argument in
-/// `docker-images/spark-k8s/boil-config.toml` (the image is what places the jar at this path).
-pub const OPENLINEAGE_JAR_LOCAL_URI: &str =
-    "local:///stackable/spark/openlineage/openlineage-spark_2.13-1.51.0.jar";
+/// The jar's Scala suffix depends on the Spark image: Stackable ships Spark 4.x as Scala 2.13 and
+/// Spark 3.5.x as Scala 2.12, so the operator selects the matching jar by Spark major version.
+/// `scala_binary_version` is e.g. `"2.13"` or `"2.12"`.
+pub fn openlineage_jar_local_uri(scala_binary_version: &str) -> String {
+    format!(
+        "local:///stackable/spark/openlineage/openlineage-spark_{scala_binary_version}-{OPENLINEAGE_SPARK_JAR_VERSION}.jar"
+    )
+}
 
 /// The key the OpenLineage discovery ConfigMap must hold, carrying the backend base URL. Mirrors the
 /// `ADDRESS` contract of the existing `vectorAggregatorConfigMapName` discovery ConfigMap.