fix: use sparkImage.pullPolicy in all containers

CHANGELOG.md

@@ -56,6 +56,12 @@ All notable changes to this project will be documented in this file.
   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]).
+- BREAKING (behaviour): The image pull policy is no longer ignored by several containers.
+  `sparkImage.pullPolicy` of a SparkApplication now covers the driver and executor containers, the
+  `job`, `requirements` and `tls` init containers and the user-supplied `spec.image`, and
+  `image.pullPolicy` of a SparkConnectServer now covers its truststore init container and its
+  executors. With the default `Always`, these containers pull on every start instead of using the
+  node's image cache ([#764]).
 
 [#721]: https://github.com/stackabletech/spark-k8s-operator/pull/721
 [#727]: https://github.com/stackabletech/spark-k8s-operator/pull/727
@@ -68,6 +74,7 @@ All notable changes to this project will be documented in this file.
 [#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
+[#764]: https://github.com/stackabletech/spark-k8s-operator/pull/764
 [#766]: https://github.com/stackabletech/spark-k8s-operator/pull/766
 
 ## [26.7.0] - 2026-07-21

docs/modules/spark-k8s/pages/usage-guide/job-dependencies.adoc

@@ -63,6 +63,13 @@ spec:
 <1> Reference to your custom image..
 <2> Apache Spark version bundled in your custom image.
 
+`sparkImage.pullPolicy` governs every container of a SparkApplication for which the operator selects an image: the submit, driver and executor containers, the `job`, `requirements` and `tls` init containers.
+It also applies to the user-supplied `spec.image`, that the `job` init container runs, so with `pullPolicy: Never` that image has to be pre-loaded on the nodes as well.
+The history and Spark Connect server have no `sparkImage` - their pull policy comes from `spec.image.pullPolicy` on their own resource.
+
+NOTE: With a mutable image tag, `IfNotPresent` lets each node serve whatever it has cached, so the driver and its executors can end up running different builds of the same tag.
+`Always` (the default) narrows that window but does not close it, because each container resolves the tag when it starts: pin a digest or a unique tag in `sparkImage.custom` if all containers of a job must run the same image.
+
 === Dependency volumes
 
 With this method, the job dependencies are provisioned from a `PersistentVolume` as shown in this example:

rust/operator-binary/src/connect/controller/build/executor.rs

@@ -173,7 +173,7 @@ pub fn executor_pod_template(
 
     // S3: Add truststore init container for S3 endpoint communication with TLS.
     if let Some(truststore_init_container) = resolved_s3
-        .truststore_init_container(resolved_product_image.clone())
+        .truststore_init_container(resolved_product_image)
         .context(TrustStoreInitContainerSnafu)?
     {
         template.add_init_container(truststore_init_container);
@@ -374,12 +374,54 @@ pub(crate) fn executor_config_map(
 
 #[cfg(test)]
 mod tests {
-    use stackable_operator::k8s_openapi::{
-        api::core::v1::EnvVar, apimachinery::pkg::apis::meta::v1::ObjectMeta,
+    use stackable_operator::{
+        commons::product_image_selection::PullPolicy,
+        k8s_openapi::{api::core::v1::EnvVar, apimachinery::pkg::apis::meta::v1::ObjectMeta},
     };
 
     use super::*;
-    use crate::connect::controller::build::test_support::minimal_validated_cluster;
+    use crate::connect::controller::build::test_support::{
+        minimal_validated_cluster, validated_cluster_with_s3_tls,
+    };
+
+    #[test]
+    fn image_pull_policy_is_set_on_every_container_spark_does_not_rebuild() {
+        let pull_policy = PullPolicy::Never;
+        let validated = validated_cluster_with_s3_tls(&pull_policy);
+        let config_map = ConfigMap {
+            metadata: ObjectMeta {
+                name: Some("my-connect-executor".to_string()),
+                ..ObjectMeta::default()
+            },
+            ..ConfigMap::default()
+        };
+
+        let pod_spec = executor_pod_template(&validated, &config_map)
+            .expect("the executor pod template can be built")
+            .spec
+            .expect("the executor pod template has a spec");
+
+        let policies: Vec<(&str, Option<&str>)> = pod_spec
+            .init_containers
+            .iter()
+            .flatten()
+            .chain(pod_spec.containers.iter())
+            .map(|container| {
+                (
+                    container.name.as_str(),
+                    container.image_pull_policy.as_deref(),
+                )
+            })
+            .collect();
+
+        assert_eq!(
+            vec![
+                ("tls-truststore-init", Some(pull_policy.as_ref())),
+                ("spark", None),
+            ],
+            policies
+        );
+    }
 
     /// `envOverrides` must be applied after all operator-set environment variables, so a user
     /// override replaces the operator-set value instead of duplicating it or being ignored.

rust/operator-binary/src/connect/controller/build/mod.rs

@@ -193,7 +193,10 @@ pub(crate) fn role_selector(server: &ValidatedSparkConnectServer, role_name: &Ro
 #[cfg(test)]
 pub(crate) mod test_support {
     use indoc::indoc;
-    use stackable_operator::{cli::OperatorEnvironmentOptions, utils::yaml_from_str_singleton_map};
+    use stackable_operator::{
+        cli::OperatorEnvironmentOptions, commons::product_image_selection::PullPolicy,
+        utils::yaml_from_str_singleton_map,
+    };
 
     use crate::connect::{
         controller::{
@@ -222,6 +225,12 @@ pub(crate) mod test_support {
             productVersion: 4.1.2
         "#};
 
+    /// [`CONNECT_YAML`] with an explicit `pullPolicy`, appended to the `spec.image` block that
+    /// [`CONNECT_YAML`] ends with.
+    fn connect_yaml_with_pull_policy(pull_policy: &PullPolicy) -> String {
+        format!("{CONNECT_YAML}    pullPolicy: {}\n", pull_policy.as_ref())
+    }
+
     /// Runs the real validate step against the minimal fixture.
     pub fn minimal_validated_cluster() -> ValidatedSparkConnectServer {
         let scs: v1alpha1::SparkConnectServer = yaml_from_str_singleton_map(CONNECT_YAML)
@@ -239,4 +248,74 @@ pub(crate) mod test_support {
         )
         .expect("validate should succeed for the test fixture")
     }
+
+    pub fn validated_cluster_with_s3_tls(pull_policy: &PullPolicy) -> ValidatedSparkConnectServer {
+        let scs: v1alpha1::SparkConnectServer =
+            yaml_from_str_singleton_map(&connect_yaml_with_pull_policy(pull_policy))
+                .expect("invalid test SparkConnectServer YAML");
+        validate(
+            &scs,
+            DereferencedSparkConnectServer {
+                resolved_s3: ResolvedS3::tls_connection(),
+            },
+            &OperatorEnvironmentOptions {
+                operator_namespace: "stackable-operators".to_string(),
+                operator_service_name: "spark-k8s-operator".to_string(),
+                image_repository: "oci.example.org/sdp".to_string(),
+            },
+        )
+        .expect("validate should succeed for the test fixture")
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use rstest::*;
+    use stackable_operator::commons::product_image_selection::PullPolicy;
+
+    use super::*;
+    use crate::{connect::common::object_name, crd::constants::SPARK_DEFAULTS_FILE_NAME};
+
+    const PULL_POLICY_PROPERTY: &str = "spark.kubernetes.container.image.pullPolicy";
+
+    #[rstest]
+    #[case::from_the_resolved_image(None, None, "Never")]
+    #[case::server_config_override_wins(Some("Always"), None, "Always")]
+    #[case::executor_config_override_wins(None, Some("Always"), "Always")]
+    fn config_overrides_override_the_operator_set_pull_policy(
+        #[case] server_override: Option<&str>,
+        #[case] executor_override: Option<&str>,
+        #[case] expected: &str,
+    ) {
+        let mut validated = test_support::validated_cluster_with_s3_tls(&PullPolicy::Never);
+        for (overrides, value) in [
+            (&mut validated.server_overrides, server_override),
+            (&mut validated.executor_overrides, executor_override),
+        ] {
+            if let Some(value) = value {
+                overrides
+                    .config_overrides
+                    .spark_defaults_conf
+                    .overrides
+                    .insert(PULL_POLICY_PROPERTY.to_string(), value.to_string());
+            }
+        }
+
+        let resources = build(&validated, &[]).expect("the resources can be built");
+        let server_cm_name = object_name(&validated.name_any(), SparkConnectRole::Server);
+        let spark_defaults = resources
+            .config_maps
+            .iter()
+            .find(|cm| cm.name_any() == server_cm_name)
+            .and_then(|cm| cm.data.as_ref())
+            .and_then(|data| data.get(SPARK_DEFAULTS_FILE_NAME))
+            .expect("the server ConfigMap contains spark-defaults.conf");
+
+        assert!(
+            spark_defaults
+                .lines()
+                .any(|line| line == format!("{PULL_POLICY_PROPERTY}={expected}")),
+            "expected {PULL_POLICY_PROPERTY}={expected} in\n {spark_defaults}"
+        )
+    }
 }

rust/operator-binary/src/connect/controller/build/server.rs

@@ -325,7 +325,7 @@ pub(crate) fn build_stateful_set(
 
     // S3: Add truststore init container for S3 endpoint communication with TLS.
     if let Some(truststore_init_container) = resolved_s3
-        .truststore_init_container(resolved_product_image.clone())
+        .truststore_init_container(resolved_product_image)
         .context(TrustStoreInitContainerSnafu)?
     {
         pb.add_init_container(truststore_init_container);
@@ -427,6 +427,10 @@ pub(crate) fn server_properties(
             "spark.kubernetes.driver.container.image".to_string(),
             Some(spark_image.clone()),
         ),
+        (
+            "spark.kubernetes.container.image.pullPolicy".to_string(),
+            Some(resolved_product_image.image_pull_policy.clone()),
+        ),
         ("spark.kubernetes.namespace".to_string(), Some(namespace)),
         (
             "spark.kubernetes.authenticate.driver.serviceAccountName".to_string(),
@@ -535,12 +539,57 @@ pub(crate) fn build_listener(
 
 #[cfg(test)]
 mod tests {
-    use stackable_operator::k8s_openapi::{
-        api::core::v1::EnvVar, apimachinery::pkg::apis::meta::v1::ObjectMeta,
+    use stackable_operator::{
+        commons::product_image_selection::PullPolicy,
+        k8s_openapi::{api::core::v1::EnvVar, apimachinery::pkg::apis::meta::v1::ObjectMeta},
     };
 
     use super::*;
-    use crate::connect::controller::build::test_support::minimal_validated_cluster;
+    use crate::connect::controller::build::test_support::{
+        minimal_validated_cluster, validated_cluster_with_s3_tls,
+    };
+
+    #[test]
+    fn image_pull_policy_is_set_on_every_server_container() {
+        let pull_policy = PullPolicy::Never;
+        let validated = validated_cluster_with_s3_tls(&pull_policy);
+        let config_map = ConfigMap {
+            metadata: ObjectMeta {
+                name: Some("my-connect-server".to_string()),
+                ..ObjectMeta::default()
+            },
+            ..ConfigMap::default()
+        };
+
+        let pod_spec = build_stateful_set(&validated, &config_map, "my-connect-server", vec![])
+            .expect("the StatefulSet can be built")
+            .spec
+            .expect("the StatefulSet has a spec")
+            .template
+            .spec
+            .expect("the StatefulSet has a pod spec");
+
+        let policies: Vec<(&str, Option<&str>)> = pod_spec
+            .init_containers
+            .iter()
+            .flatten()
+            .chain(pod_spec.containers.iter())
+            .map(|container| {
+                (
+                    container.name.as_str(),
+                    container.image_pull_policy.as_deref(),
+                )
+            })
+            .collect();
+
+        assert_eq!(
+            vec![
+                ("tls-truststore-init", Some(pull_policy.as_ref())),
+                ("spark", Some(pull_policy.as_ref())),
+            ],
+            policies
+        );
+    }
 
     /// `envOverrides` must be applied after all operator-set environment variables, so a user
     /// override replaces the operator-set value instead of duplicating it or being ignored.