fix: mount spark-env.sh and security.properties

CHANGELOG.md

@@ -31,6 +31,8 @@ All notable changes to this project will be documented in this file.
 - Environment variable overrides (`envOverrides`) are now applied after all environment variables
   set by the operator. In particular, `SPARK_CONF_DIR` of a SparkApplication can now be overridden,
   whereas previously the operator's values always took precedence ([#753]).
+- BREAKING (behaviour): Keys of the `spark-env.sh` `configOverrides` must be valid shell
+  identifiers (matching `[a-zA-Z_][a-zA-Z0-9_]*`) and are now rejected if they are not ([#761]).
 
 ### Fixed
 
@@ -51,6 +53,8 @@ All notable changes to this project will be documented in this file.
 - 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]).
+- The `configOverrides` for `spark-env.sh` and `security.properties` of a SparkApplication now take
+  effect in the submit, driver and executor Pods ([#761]).
 
 [#721]: https://github.com/stackabletech/spark-k8s-operator/pull/721
 [#727]: https://github.com/stackabletech/spark-k8s-operator/pull/727
@@ -62,6 +66,7 @@ All notable changes to this project will be documented in this file.
 [#753]: https://github.com/stackabletech/spark-k8s-operator/pull/753
 [#754]: https://github.com/stackabletech/spark-k8s-operator/pull/754
 [#757]: https://github.com/stackabletech/spark-k8s-operator/pull/757
+[#761]: https://github.com/stackabletech/spark-k8s-operator/pull/761
 
 ## [26.7.0] - 2026-07-21
 

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

@@ -74,6 +74,9 @@ NOTE: The given properties are written to `spark-env.sh` in the form `export KEY
 Make sure to escape the value already in the specification.
 Be aware that some environment variables may already be set, so prepend or append a reference to them in the value, as it is done in the example.
 
+IMPORTANT: Because the file is sourced by the shell, keys must be valid shell identifiers, matching `[a-zA-Z_][a-zA-Z0-9_]*`.
+A key such as `MY-VAR` is rejected, because `export MY-VAR="..."` is not something the shell can execute.
+
 === The security.properties file
 
 The `security.properties` file is used to configure JVM security properties.

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

@@ -3,7 +3,7 @@ use stackable_operator::crd::s3;
 use crate::crd::{
     constants::{
         JVM_SECURITY_PROPERTIES_FILE, STACKABLE_TLS_STORE_PASSWORD, STACKABLE_TRUST_STORE,
-        VOLUME_MOUNT_PATH_LOG_CONFIG,
+        VOLUME_MOUNT_PATH_CONFIG,
     },
     logdir::ResolvedLogDir,
     tlscerts::tls_secret_names,
@@ -25,7 +25,7 @@ pub fn construct_extra_java_options(
     // kept the implementation as is. We can always re-visit this as needed.
 
     let mut jvm_args = vec![format!(
-        "-Djava.security.properties={VOLUME_MOUNT_PATH_LOG_CONFIG}/{JVM_SECURITY_PROPERTIES_FILE}"
+        "-Djava.security.properties={VOLUME_MOUNT_PATH_CONFIG}/{JVM_SECURITY_PROPERTIES_FILE}"
     )];
 
     if tls_secret_names(s3_conn, log_dir).is_some() {
@@ -85,11 +85,11 @@ mod tests {
 
         assert_eq!(
             driver_extra_java_options,
-            "-Djava.security.properties=/stackable/log_config/security.properties"
+            "-Djava.security.properties=/stackable/spark/conf/security.properties"
         );
         assert_eq!(
             executor_extra_java_options,
-            "-Djava.security.properties=/stackable/log_config/security.properties"
+            "-Djava.security.properties=/stackable/spark/conf/security.properties"
         );
     }
 
@@ -125,7 +125,7 @@ mod tests {
 
         assert_eq!(
             driver_extra_java_options,
-            "-Djava.security.properties=/stackable/log_config/security.properties -Dhttps.proxyHost=from-driver"
+            "-Djava.security.properties=/stackable/spark/conf/security.properties -Dhttps.proxyHost=from-driver"
         );
         assert_eq!(
             executor_extra_java_options,

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

@@ -10,7 +10,7 @@ use constants::*;
 use history::LogFileDirectorySpec;
 use logdir::ResolvedLogDir;
 use serde::{Deserialize, Serialize};
-use snafu::{OptionExt, ResultExt, Snafu};
+use snafu::{OptionExt, ResultExt, Snafu, ensure};
 use stackable_operator::{
     builder::pod::volume::{
         SecretFormat, SecretOperatorVolumeSourceBuilder, SecretOperatorVolumeSourceBuilderError,
@@ -120,14 +120,20 @@ pub enum Error {
         source: stackable_operator::v2::builder::pod::container::Error,
     },
 
+    #[snafu(display(
+        "invalid key [{key}] in the {SPARK_ENV_SH_FILE_NAME} configOverrides: the file is sourced \
+        by the shell, so keys must be valid shell identifiers matching [a-zA-Z_][a-zA-Z0-9_]*"
+    ))]
+    InvalidSparkEnvShKey { key: String },
+
     #[snafu(display("failed to configure log directory"))]
     ConfigureLogDir { source: logdir::Error },
 }
 
 // `_STACKABLE_PRE_HOOK` is evaluated by the entrypoint script (run-spark.sh) in the Spark images
 // before the actual JVM process is started; the operator uses it to run `containerdebug` in the
 // background of every `spark` container.
-constant!(STACKABLE_PRE_HOOK: EnvVarName = "_STACKABLE_PRE_HOOK");
+constant!(pub STACKABLE_PRE_HOOK: EnvVarName = "_STACKABLE_PRE_HOOK");
 constant!(PYTHONPATH: EnvVarName = "PYTHONPATH");
 // The environment variable holding the trust store password; its value is the
 // `STACKABLE_TLS_STORE_PASSWORD` string constant.
@@ -1058,18 +1064,28 @@ fn resources_to_executor_props(
     Ok(())
 }
 
-/// Create the content of the file spark-env.sh.
-/// The properties are serialized in the form 'export {k}="{v}"',
-/// escaping neither the key nor the value. The user is responsible for
-/// providing escaped values.
-pub fn to_spark_env_sh_string<'a, T>(properties: T) -> String
+/// Create the content of the file spark-env.sh, serializing the properties as 'export {k}="{v}"'.
+/// The file is sourced by the shell, so keys that are not valid shell identifiers are rejected.
+/// Values are left unescaped on purpose, so that they can reference other variables.
+pub fn to_spark_env_sh_string<'a, T>(properties: T) -> Result<String, Error>
 where
     T: Iterator<Item = (&'a String, &'a String)>,
 {
     properties
-        .map(|(k, v)| format!("export {k}=\"{v}\""))
-        .collect::<Vec<String>>()
-        .join("\n")
+        .map(|(k, v)| {
+            ensure!(is_shell_identifier(k), InvalidSparkEnvShKeySnafu { key: k });
+            Ok(format!("export {k}=\"{v}\""))
+        })
+        .collect::<Result<Vec<String>, Error>>()
+        .map(|lines| lines.join("\n"))
+}
+
+fn is_shell_identifier(key: &str) -> bool {
+    let mut chars = key.chars();
+    chars
+        .next()
+        .is_some_and(|first| first.is_ascii_alphabetic() || first == '_')
+        && chars.all(|char| char.is_ascii_alphanumeric() || char == '_')
 }
 
 #[cfg(test)]
@@ -1099,6 +1115,49 @@ mod tests {
         let _ = *STACKABLE_TLS_STORE_PASSWORD_ENV;
     }
 
+    #[rstest]
+    #[case("SPARK_HISTORY_OPTS")]
+    #[case("_LEADING_UNDERSCORE")]
+    #[case("TRAILING_DIGITS_123")]
+    #[case("lowercase")]
+    fn to_spark_env_sh_string_accepts_shell_identifiers(#[case] key: &str) {
+        let overrides = BTreeMap::from([(key.to_string(), "value".to_string())]);
+
+        assert_eq!(
+            to_spark_env_sh_string(overrides.iter()).expect("the key is a valid shell identifier"),
+            format!("export {key}=\"value\"")
+        );
+    }
+
+    #[rstest]
+    #[case("TEST_SPARK-ENV-SH")]
+    #[case("1_LEADING_DIGIT")]
+    #[case("WITH SPACE")]
+    #[case("with.dot")]
+    #[case("")]
+    fn to_spark_env_sh_string_rejects_other_keys(#[case] key: &str) {
+        let overrides = BTreeMap::from([(key.to_string(), "value".to_string())]);
+
+        assert!(matches!(
+            to_spark_env_sh_string(overrides.iter()),
+            Err(Error::InvalidSparkEnvShKey { key: invalid }) if invalid == key
+        ));
+    }
+
+    /// Values are shell-evaluated by design, so references to other variables must survive.
+    #[test]
+    fn to_spark_env_sh_string_leaves_values_untouched() {
+        let overrides = BTreeMap::from([(
+            "SPARK_HISTORY_OPTS".to_string(),
+            "$SPARK_HISTORY_OPTS -Dsome.token=$SAS_TOKEN".to_string(),
+        )]);
+
+        assert_eq!(
+            to_spark_env_sh_string(overrides.iter()).expect("the key is a valid shell identifier"),
+            r#"export SPARK_HISTORY_OPTS="$SPARK_HISTORY_OPTS -Dsome.token=$SAS_TOKEN""#
+        );
+    }
+
     #[test]
     fn test_default_resource_limits() {
         let spark_application = serde_yaml::from_str::<v1alpha1::SparkApplication>(indoc! {"

rust/operator-binary/src/history/controller/build/resource/config_map.rs

@@ -48,6 +48,14 @@ pub enum Error {
     InvalidSparkDefaults {
         source: stackable_operator::v2::config_file_writer::PropertiesWriterError,
     },
+
+    #[snafu(display(
+        "History server : failed to serialize [{SPARK_ENV_SH_FILE_NAME}] for group {rolegroup}"
+    ))]
+    SparkEnvSh {
+        source: crate::crd::Error,
+        rolegroup: String,
+    },
 }
 
 type Result<T, E = Error> = std::result::Result<T, E>;
@@ -95,7 +103,10 @@ pub(crate) fn build_config_map(
         .add_data(SPARK_DEFAULTS_FILE_NAME, spark_defaults)
         .add_data(
             SPARK_ENV_SH_FILE_NAME,
-            to_spark_env_sh_string(rg.config.config_overrides.spark_env_sh.overrides.iter()),
+            to_spark_env_sh_string(rg.config.config_overrides.spark_env_sh.overrides.iter())
+                .with_context(|_| SparkEnvShSnafu {
+                    rolegroup: role_group_name.to_string(),
+                })?,
         )
         .add_data(
             JVM_SECURITY_PROPERTIES_FILE,

rust/operator-binary/src/spark_k8s_controller/build/mod.rs

@@ -203,7 +203,20 @@ pub(crate) fn recommended_labels_for_component_resources(
 
 #[cfg(test)]
 mod tests {
+    use indoc::indoc;
+    use stackable_operator::{
+        cli::OperatorEnvironmentOptions,
+        k8s_openapi::api::core::v1::{PodSpec, PodTemplateSpec},
+    };
+
     use super::*;
+    use crate::{
+        crd::{
+            constants::{POD_TEMPLATE_FILE, VOLUME_MOUNT_NAME_CONFIG, VOLUME_MOUNT_PATH_CONFIG},
+            v1alpha1,
+        },
+        spark_k8s_controller::{dereference::DereferencedSparkApplication, validate::validate},
+    };
 
     #[test]
     fn test_constants() {
@@ -214,4 +227,154 @@ mod tests {
         let _ = *SPARK_JOB_TEMPLATE_COMPONENT_NAME;
         let _ = *SPARK_SUBMIT_COMPONENT_NAME;
     }
+
+    /// The Pod specs of the submit Job and of the driver and executor pod templates, each with the
+    /// name of the resource it was taken from.
+    fn pod_specs(enable_vector_agent: bool) -> Vec<(String, PodSpec)> {
+        let yaml = format!(
+            indoc! {r#"
+                apiVersion: spark.stackable.tech/v1alpha1
+                kind: SparkApplication
+                metadata:
+                  name: spark-example
+                  namespace: default
+                  uid: 12345678-1234-1234-1234-123456789012
+                spec:
+                  mode: cluster
+                  mainApplicationFile: test.py
+                  sparkImage:
+                    productVersion: 1.2.3
+                  image: oci.example.org/jobs/spark-example:1.0.0
+                  vectorAggregatorConfigMapName: vector-aggregator-discovery
+                  deps:
+                    requirements:
+                      - tabulate==0.8.9
+                    packages:
+                      - org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.11.0
+                  driver:
+                    config:
+                      logging:
+                        enableVectorAgent: {enable_vector_agent}
+                  executor:
+                    config:
+                      logging:
+                        enableVectorAgent: {enable_vector_agent}
+            "#},
+            enable_vector_agent = enable_vector_agent
+        );
+        let deserializer = serde_yaml::Deserializer::from_str(&yaml);
+        let spark_application: v1alpha1::SparkApplication =
+            serde_yaml::with::singleton_map_recursive::deserialize(deserializer)
+                .expect("invalid test SparkApplication YAML");
+
+        let validated = validate(
+            DereferencedSparkApplication {
+                spark_application,
+                resolved_template_refs: Vec::new(),
+                s3_connection: None,
+                log_dir: None,
+            },
+            &OperatorEnvironmentOptions {
+                operator_namespace: "stackable-operators".to_string(),
+                operator_service_name: "spark-k8s-operator".to_string(),
+                image_repository: "oci.example.org/sdp".to_string(),
+            },
+        )
+        .expect("the fixture validates");
+
+        let resources = build(&validated).expect("the resources can be built");
+
+        let mut pod_specs = vec![(
+            "spark-submit Job".to_string(),
+            resources.jobs[0]
+                .spec
+                .clone()
+                .expect("the Job has a spec")
+                .template
+                .spec
+                .expect("the Job has a pod spec"),
+        )];
+        for config_map in &resources.config_maps {
+            let Some(template) = config_map
+                .data
+                .as_ref()
+                .and_then(|data| data.get(POD_TEMPLATE_FILE))
+            else {
+                continue;
+            };
+            let template: PodTemplateSpec =
+                serde_yaml::from_str(template).expect("the pod template deserializes");
+            pod_specs.push((
+                config_map.metadata.name.clone().unwrap_or_default(),
+                template.spec.expect("the pod template has a spec"),
+            ));
+        }
+
+        assert_eq!(pod_specs.len(), 3);
+
+        let vector_containers = pod_specs
+            .iter()
+            .flat_map(|(_, pod_spec)| &pod_spec.containers)
+            .filter(|container| container.name == "vector")
+            .count();
+        assert_eq!(vector_containers, if enable_vector_agent { 2 } else { 0 });
+
+        pod_specs
+    }
+
+    #[test]
+    fn every_declared_volume_is_mounted() {
+        for enable_vector_agent in [false, true] {
+            for (name, pod_spec) in pod_specs(enable_vector_agent) {
+                let PodSpec {
+                    containers,
+                    init_containers,
+                    volumes,
+                    ..
+                } = pod_spec;
+                let mounted: Vec<&str> = containers
+                    .iter()
+                    .chain(init_containers.iter().flatten())
+                    .flat_map(|container| container.volume_mounts.iter().flatten())
+                    .map(|volume_mount| volume_mount.name.as_str())
+                    .collect();
+                let unmounted: Vec<&str> = volumes
+                    .iter()
+                    .flatten()
+                    .map(|volume| volume.name.as_str())
+                    .filter(|volume_name| !mounted.contains(volume_name))
+                    .collect();
+
+                assert!(
+                    unmounted.is_empty(),
+                    "{name} declares volumes that no container mounts: {unmounted:?}"
+                );
+            }
+        }
+    }
+
+    #[test]
+    fn spark_containers_mount_the_config_volume() {
+        for enable_vector_agent in [false, true] {
+            for (name, pod_spec) in pod_specs(enable_vector_agent) {
+                let spark_container = pod_spec
+                    .containers
+                    .iter()
+                    .find(|container| container.name == "spark" || container.name == "spark-submit")
+                    .unwrap_or_else(|| panic!("{name} has a Spark container"));
+                let mount_path = spark_container
+                    .volume_mounts
+                    .iter()
+                    .flatten()
+                    .find(|volume_mount| volume_mount.name == VOLUME_MOUNT_NAME_CONFIG.as_ref())
+                    .map(|volume_mount| volume_mount.mount_path.as_str());
+
+                assert_eq!(
+                    mount_path,
+                    Some(VOLUME_MOUNT_PATH_CONFIG),
+                    "the Spark container of {name} must mount the config Volume at {VOLUME_MOUNT_PATH_CONFIG}"
+                );
+            }
+        }
+    }
 }