fix: use sparkImage.pullPolicy in all containers

CHANGELOG.md

@@ -51,6 +51,12 @@ 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]).
+- BREAKING (behaviour): `sparkImage.pullPolicy` is no longer ignored by the driver and executor
+  containers, the Spark Connect executors and the truststore and `job` init containers, and it now
+  covers the user-supplied `spec.image` too. With the default `Always`, these containers pull on
+  every start instead of using the node's image cache. The operator also emits the policy as
+  `spark.kubernetes.container.image.pullPolicy`, which Spark previously left at its default when
+  rebuilding the containers from the pod templates ([#764]).
 
 [#721]: https://github.com/stackabletech/spark-k8s-operator/pull/721
 [#727]: https://github.com/stackabletech/spark-k8s-operator/pull/727
@@ -62,6 +68,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
+[#764]: https://github.com/stackabletech/spark-k8s-operator/pull/764
 
 ## [26.7.0] - 2026-07-21
 

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

@@ -63,6 +63,12 @@ spec:
 <1> Reference to your custom image..
 <2> Apache Spark version bundled in your custom image.
 
+`sparkImage.pullPolicy` governs every container that the operator selects an image for: the submit, driver and executor containers, the `job`, `requirements` and `tls` init containers, and the Spark Connect server and its executors.
+This includes the user-supplied `spec.image`, which the `job` init container runs.
+
+NOTE: Under a mutable image tag, `IfNotPresent` lets a node serve a stale cached image, so `Always` is the minimum for driver and executors to match the submit pod.
+It is not a guarantee, because each container resolves the tag when it starts: pin a digest or a unique tag in `sparkImage.custom` for that.
+
 === 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

@@ -148,7 +148,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);
@@ -230,6 +230,10 @@ pub(crate) fn executor_properties(
             "spark.kubernetes.executor.container.image".to_string(),
             Some(spark_image),
         ),
+        (
+            "spark.kubernetes.container.image.pullPolicy".to_string(),
+            Some(resolved_product_image.image_pull_policy.clone()),
+        ),
         (
             "spark.executor.defaultJavaOptions".to_string(),
             Some(executor_jvm_args(
@@ -371,7 +375,45 @@ mod tests {
     };
 
     use super::*;
-    use crate::connect::controller::build::test_support::minimal_validated_cluster;
+    use crate::connect::controller::build::test_support::{
+        PULL_POLICY_NEVER, minimal_validated_cluster, validated_cluster_with_s3_tls,
+    };
+
+    #[test]
+    fn image_pull_policy_is_set_on_every_container_spark_does_not_rebuild() {
+        let validated = validated_cluster_with_s3_tls();
+        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())
+            .filter(|container| container.name != SparkConnectContainer::Spark.to_string())
+            .map(|container| {
+                (
+                    container.name.as_str(),
+                    container.image_pull_policy.as_deref(),
+                )
+            })
+            .collect();
+
+        assert_eq!(
+            vec![("tls-truststore-init", Some(PULL_POLICY_NEVER))],
+            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

@@ -222,6 +222,14 @@ pub(crate) mod test_support {
             productVersion: 4.1.2
         "#};
 
+    pub const PULL_POLICY_NEVER: &str = "Never";
+
+    /// [`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: &str) -> String {
+        format!("{CONNECT_YAML}    pullPolicy: {pull_policy}\n")
+    }
+
     /// 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 +247,22 @@ pub(crate) mod test_support {
         )
         .expect("validate should succeed for the test fixture")
     }
+
+    pub fn validated_cluster_with_s3_tls() -> ValidatedSparkConnectServer {
+        let scs: v1alpha1::SparkConnectServer =
+            yaml_from_str_singleton_map(&connect_yaml_with_pull_policy(PULL_POLICY_NEVER))
+                .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")
+    }
 }

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

@@ -317,7 +317,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);
@@ -532,7 +532,50 @@ mod tests {
     };
 
     use super::*;
-    use crate::connect::controller::build::test_support::minimal_validated_cluster;
+    use crate::connect::controller::build::test_support::{
+        PULL_POLICY_NEVER, minimal_validated_cluster, validated_cluster_with_s3_tls,
+    };
+
+    #[test]
+    fn image_pull_policy_is_set_on_every_server_container() {
+        let validated = validated_cluster_with_s3_tls();
+        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_NEVER)),
+                ("spark", Some(PULL_POLICY_NEVER)),
+            ],
+            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/s3.rs

@@ -1,10 +1,14 @@
-use std::collections::{BTreeMap, BTreeSet};
+use std::{
+    collections::{BTreeMap, BTreeSet},
+    str::FromStr,
+};
 
 use snafu::{OptionExt, ResultExt, Snafu};
 use stackable_operator::{
     commons::product_image_selection::ResolvedProductImage,
     crd::s3::{self, v1alpha1::S3AccessStyle},
-    k8s_openapi::api::core::v1::{Volume, VolumeMount},
+    k8s_openapi::api::core::v1::{Container, Volume, VolumeMount},
+    v2::{builder::pod::container::new_container_builder, types::kubernetes::ContainerName},
 };
 
 use crate::{
@@ -17,6 +21,8 @@ use crate::{
     },
 };
 
+const TRUSTSTORE_INIT_CONTAINER_NAME: &str = "tls-truststore-init";
+
 #[derive(Snafu, Debug)]
 #[allow(clippy::enum_variant_names)]
 pub enum Error {
@@ -42,6 +48,11 @@ pub enum Error {
         source: s3::v1alpha1::ConnectionError,
     },
 
+    #[snafu(display("failed to add a volume mount to the truststore init container"))]
+    AddVolumeMount {
+        source: stackable_operator::builder::pod::container::Error,
+    },
+
     #[snafu(display("failed to get volumes and mounts for S3 connection"))]
     ConnectionVolumesAndMounts {
         source: s3::v1alpha1::ConnectionError,
@@ -64,6 +75,33 @@ impl ResolvedS3 {
         }
     }
 
+    #[cfg(test)]
+    pub(crate) fn tls_connection() -> Self {
+        use stackable_operator::commons::tls_verification::{
+            CaCert, Tls, TlsClientDetails, TlsServerVerification, TlsVerification,
+        };
+
+        Self {
+            s3_buckets: Vec::new(),
+            s3_connection: Some(s3::v1alpha1::ConnectionSpec {
+                host: "my-s3-endpoint.com".parse().expect("a valid host"),
+                port: None,
+                region: s3::v1alpha1::Region {
+                    name: "us-east-1".to_string(),
+                },
+                access_style: S3AccessStyle::Path,
+                credentials: None,
+                tls: TlsClientDetails {
+                    tls: Some(Tls {
+                        verification: TlsVerification::Server(TlsServerVerification {
+                            ca_cert: CaCert::SecretClass("tls-ca-secret-class".to_string()),
+                        }),
+                    }),
+                },
+            }),
+        }
+    }
+
     pub(crate) async fn resolve(
         client: &stackable_operator::client::Client,
         connect_server: &crd::v1alpha1::SparkConnectServer,
@@ -269,25 +307,27 @@ impl ResolvedS3 {
 
     pub(crate) fn truststore_init_container(
         &self,
-        image: ResolvedProductImage,
-    ) -> Result<Option<stackable_operator::k8s_openapi::api::core::v1::Container>, Error> {
+        image: &ResolvedProductImage,
+    ) -> Result<Option<Container>, Error> {
         if let Some(command) = self.truststore_init_container_command() {
             let (_, volume_mounts) = self.volumes_and_mounts()?;
+            let name = ContainerName::from_str(TRUSTSTORE_INIT_CONTAINER_NAME)
+                .expect("TRUSTSTORE_INIT_CONTAINER_NAME is a valid container name");
+
             Ok(Some(
-                stackable_operator::k8s_openapi::api::core::v1::Container {
-                    name: "tls-truststore-init".to_string(),
-                    image: Some(image.image),
-                    command: Some(vec![
+                new_container_builder(&name)
+                    .image_from_product_image(image)
+                    .command(vec![
                         "/bin/bash".to_string(),
                         "-x".to_string(),
                         "-euo".to_string(),
                         "pipefail".to_string(),
                         "-c".to_string(),
                         command,
-                    ]),
-                    volume_mounts: Some(volume_mounts),
-                    ..Default::default()
-                },
+                    ])
+                    .add_volume_mounts(volume_mounts)
+                    .context(AddVolumeMountSnafu)?
+                    .build(),
             ))
         } else {
             Ok(None)