diff --git a/rust/operator-binary/src/command.rs b/rust/operator-binary/src/command.rs
index 0925b49a..8407cd92 100644
--- a/rust/operator-binary/src/command.rs
+++ b/rust/operator-binary/src/command.rs
@@ -11,7 +11,8 @@ use crate::{
     catalog::config::CatalogConfig,
     controller::{STACKABLE_LOG_CONFIG_DIR, STACKABLE_LOG_DIR},
     crd::{
-        CONFIG_DIR_NAME, Container, LOG_PROPERTIES, RW_CONFIG_DIR_NAME, STACKABLE_CLIENT_TLS_DIR,
+        CONFIG_DIR_NAME, Container, EXCHANGE_MANAGER_PROPERTIES, LOG_PROPERTIES,
+        RW_CONFIG_DIR_NAME, SPOOLING_MANAGER_PROPERTIES, STACKABLE_CLIENT_TLS_DIR,
         STACKABLE_INTERNAL_TLS_DIR, STACKABLE_MOUNT_INTERNAL_TLS_DIR,
         STACKABLE_MOUNT_SERVER_TLS_DIR, STACKABLE_SERVER_TLS_DIR, STACKABLE_TLS_STORE_PASSWORD,
         SYSTEM_TRUST_STORE, SYSTEM_TRUST_STORE_PASSWORD, TrinoRole, client_protocol,
@@ -97,8 +98,6 @@ pub fn container_prepare_args(
 pub fn container_trino_args(
     authentication_config: &TrinoAuthenticationConfig,
     catalogs: &[CatalogConfig],
-    resolved_fte_config: &Option<ResolvedFaultTolerantExecutionConfig>,
-    resolved_spooling_config: &Option<client_protocol::ResolvedClientProtocolConfig>,
 ) -> Vec<String> {
     let mut args = vec![
         // copy config files to a writeable empty folder
@@ -126,19 +125,17 @@ pub fn container_trino_args(
         }
     });
 
-    // Add fault tolerant execution environment variables from files
-    if let Some(resolved_fte) = resolved_fte_config {
-        for (env_name, file) in &resolved_fte.load_env_from_files {
-            args.push(format!("export {env_name}=\"$(cat {file})\""));
-        }
-    }
+    // Resolve credentials for fault tolerant execution exchange manager if needed
+    args.push(format!(
+        "test -f {rw_exchange_manager_config_file} && config-utils template {rw_exchange_manager_config_file}",
+        rw_exchange_manager_config_file = format!("{RW_CONFIG_DIR_NAME}/{EXCHANGE_MANAGER_PROPERTIES}")
+    ));
 
-    // Add client spooling environment variables from files
-    if let Some(resolved_spooling) = resolved_spooling_config {
-        for (env_name, file) in &resolved_spooling.load_env_from_files {
-            args.push(format!("export {env_name}=\"$(cat {file})\""));
-        }
-    }
+    // Resolve credentials for spooling manager if needed
+    args.push(format!(
+        "test -f {rw_spooling_config_file} && config-utils template {rw_spooling_config_file}",
+        rw_spooling_config_file = format!("{RW_CONFIG_DIR_NAME}/{SPOOLING_MANAGER_PROPERTIES}")
+    ));
 
     args.push("set -x".to_string());
 
diff --git a/rust/operator-binary/src/controller.rs b/rust/operator-binary/src/controller.rs
index 50e81c06..64764826 100644
--- a/rust/operator-binary/src/controller.rs
+++ b/rust/operator-binary/src/controller.rs
@@ -1200,13 +1200,7 @@ fn build_rolegroup_statefulset(
             "-c".to_string(),
         ])
         .args(vec![
-            command::container_trino_args(
-                trino_authentication_config,
-                catalogs,
-                resolved_fte_config,
-                resolved_spooling_config,
-            )
-            .join("\n"),
+            command::container_trino_args(trino_authentication_config, catalogs).join("\n"),
         ])
         .add_env_vars(env)
         .add_volume_mount("config", CONFIG_DIR_NAME)
diff --git a/rust/operator-binary/src/crd/client_protocol.rs b/rust/operator-binary/src/crd/client_protocol.rs
index 8809568b..fafc67b8 100644
--- a/rust/operator-binary/src/crd/client_protocol.rs
+++ b/rust/operator-binary/src/crd/client_protocol.rs
@@ -20,9 +20,6 @@ use crate::{
     crd::{ENV_SPOOLING_SECRET, STACKABLE_CLIENT_TLS_DIR},
 };
 
-const SPOOLING_S3_AWS_ACCESS_KEY: &str = "SPOOLING_S3_AWS_ACCESS_KEY";
-const SPOOLING_S3_AWS_SECRET_KEY: &str = "SPOOLING_S3_AWS_SECRET_KEY";
-
 #[derive(Clone, Debug, Deserialize, JsonSchema, PartialEq, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub enum ClientProtocolConfig {
@@ -88,10 +85,6 @@ pub struct ResolvedClientProtocolConfig {
     /// Volume mounts required for the configuration
     pub volume_mounts: Vec<VolumeMount>,
 
-    /// Env-Vars that should be exported from files.
-    /// You can think of it like `export <key>="$(cat <value>)"`
-    pub load_env_from_files: BTreeMap<String, String>,
-
     /// Additional commands that need to be executed before starting Trino
     /// Used to add TLS certificates to the client's trust store.
     pub init_container_extra_start_commands: Vec<String>,
@@ -110,7 +103,6 @@ impl ResolvedClientProtocolConfig {
             spooling_manager_properties: BTreeMap::new(),
             volumes: Vec::new(),
             volume_mounts: Vec::new(),
-            load_env_from_files: BTreeMap::new(),
             init_container_extra_start_commands: Vec::new(),
         };
 
@@ -194,18 +186,13 @@ impl ResolvedClientProtocolConfig {
             self.spooling_manager_properties.extend([
                 (
                     "s3.aws-access-key".to_string(),
-                    format!("${{ENV:{SPOOLING_S3_AWS_ACCESS_KEY}}}"),
+                    format!("${{file:UTF-8:{access_key_path}}}"),
                 ),
                 (
                     "s3.aws-secret-key".to_string(),
-                    format!("${{ENV:{SPOOLING_S3_AWS_SECRET_KEY}}}"),
+                    format!("${{file:UTF-8:{secret_key_path}}}"),
                 ),
             ]);
-
-            self.load_env_from_files.extend([
-                (String::from(SPOOLING_S3_AWS_ACCESS_KEY), access_key_path),
-                (String::from(SPOOLING_S3_AWS_SECRET_KEY), secret_key_path),
-            ]);
         }
 
         if let Some(tls) = s3_connection.tls.tls.as_ref() {
diff --git a/rust/operator-binary/src/crd/fault_tolerant_execution.rs b/rust/operator-binary/src/crd/fault_tolerant_execution.rs
index a58ab302..bffcde6a 100644
--- a/rust/operator-binary/src/crd/fault_tolerant_execution.rs
+++ b/rust/operator-binary/src/crd/fault_tolerant_execution.rs
@@ -226,10 +226,6 @@ pub struct ResolvedFaultTolerantExecutionConfig {
     /// Volume mounts required for the configuration
     pub volume_mounts: Vec<VolumeMount>,
 
-    /// Env-Vars that should be exported from files.
-    /// You can think of it like `export <key>="$(cat <value>)"`
-    pub load_env_from_files: BTreeMap<String, String>,
-
     /// Additional commands that need to be executed before starting Trino
     pub init_container_extra_start_commands: Vec<String>,
 }
@@ -453,7 +449,6 @@ impl ResolvedFaultTolerantExecutionConfig {
             exchange_manager_properties,
             volumes: Vec::new(),
             volume_mounts: Vec::new(),
-            load_env_from_files: BTreeMap::new(),
             init_container_extra_start_commands: Vec::new(),
         };
 
@@ -516,22 +511,14 @@ impl ResolvedFaultTolerantExecutionConfig {
         );
 
         if let Some((access_key_path, secret_key_path)) = s3_connection.credentials_mount_paths() {
-            let access_key_env = "EXCHANGE_S3_AWS_ACCESS_KEY".to_string();
-            let secret_key_env = "EXCHANGE_S3_AWS_SECRET_KEY".to_string();
-
             self.exchange_manager_properties.insert(
                 "exchange.s3.aws-access-key".to_string(),
-                format!("${{ENV:{access_key_env}}}"),
+                format!("${{file:UTF-8:{access_key_path}}}"),
             );
             self.exchange_manager_properties.insert(
                 "exchange.s3.aws-secret-key".to_string(),
-                format!("${{ENV:{secret_key_env}}}"),
+                format!("${{file:UTF-8:{secret_key_path}}}"),
             );
-
-            self.load_env_from_files
-                .insert(access_key_env, access_key_path);
-            self.load_env_from_files
-                .insert(secret_key_env, secret_key_path);
         }
 
         if let Some(tls) = s3_connection.tls.tls.as_ref() {