feat: support Spark 3 as well

Author: sweb

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

@@ -9,7 +9,7 @@ When configured, the operator injects everything required to make the https://op
 * 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),
 * 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,7 +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.is_some() {
+    //
+    // This is scoped to Spark 4.x: the flag is unnecessary — and on the JDK 17 Spark 3.5.x images the
+    // operator also ships, potentially a startup error — 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());
     }
 
@@ -63,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:
@@ -81,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,
@@ -102,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:
@@ -111,7 +126,7 @@ mod tests {
               mode: cluster
               mainApplicationFile: test.py
               sparkImage:
-                productVersion: 1.2.3
+                productVersion: 4.1.2
               driver:
                 jvmArgumentOverrides:
                   add:
@@ -122,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,
@@ -140,32 +152,59 @@ 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:
-                connection:
-                  inline:
-                    host: marquez
-                    port: 5000
-        "#;
-
-        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:
+            connection:
+              inline:
+                host: marquez
+                port: 5000
+    "#;
+
+    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 absent.
+    #[rstest]
+    #[case::enabled_spark_3(OPENLINEAGE_ENABLED, "3.5.8", false)]
+    #[case::enabled_spark_4(OPENLINEAGE_ENABLED, "4.1.2", true)]
+    #[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

@@ -113,10 +113,13 @@ pub const OPENLINEAGE_LISTENER_CLASS: &str = "io.openlineage.spark.agent.OpenLin
 /// 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).
+/// This is a stable, version- and Scala-independent symlink the image maintains (mirroring the
+/// `jmx_prometheus_javaagent.jar` pattern): each Spark image bakes the correct
+/// `openlineage-spark_<scala>-<version>.jar` for its build (Scala 2.13 for Spark 4.x, Scala 2.12 for
+/// Spark 3.5.x) and symlinks it here. Referencing the symlink decouples the operator from the jar's
+/// version and Scala suffix, so bumping either only touches `docker-images/spark-k8s`.
 pub const OPENLINEAGE_JAR_LOCAL_URI: &str =
-    "local:///stackable/spark/openlineage/openlineage-spark_2.13-1.51.0.jar";
+    "local:///stackable/spark/openlineage/openlineage-spark.jar";
 
 /// Java module-system flag OpenLineage requires on Spark 4.x: without it the driver throws a
 /// non-fatal `InaccessibleObjectException` and silently degrades extension-interface lineage.