feat: Handle s3 region

Author: NickLarsenNZ

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

@@ -182,7 +182,7 @@ impl S3LogDir {
     pub fn history_server_spark_config(&self) -> Result<BTreeMap<String, String>, Error> {
         let connection = &self.bucket.connection;
 
-        Ok(BTreeMap::from([
+        let mut config = BTreeMap::from([
             ("spark.history.fs.logDirectory".to_string(), self.url()),
             (
                 "spark.hadoop.fs.s3a.endpoint".to_string(),
@@ -192,7 +192,16 @@ impl S3LogDir {
                 "spark.hadoop.fs.s3a.path.style.access".to_string(),
                 (connection.access_style == S3AccessStyle::Path).to_string(),
             ),
-        ]))
+        ]);
+
+        if let Some(region_name) = connection.region.name() {
+            config.insert(
+                "spark.hadoop.fs.s3a.endpoint.region".to_string(),
+                region_name.to_string(),
+            );
+        };
+
+        Ok(config)
     }
 
     pub fn application_spark_config(&self) -> Result<BTreeMap<String, String>, Error> {
@@ -211,6 +220,12 @@ impl S3LogDir {
             format!("spark.hadoop.fs.s3a.bucket.{bucket_name}.path.style.access"),
             (connection.access_style == S3AccessStyle::Path).to_string(),
         );
+        if let Some(region_name) = connection.region.name() {
+            result.insert(
+                format!("spark.hadoop.fs.s3a.bucket.{bucket_name}.region"),
+                region_name.to_string(),
+            );
+        };
         if let Some(secret_dir) = self.credentials_mount_path() {
             // We don't use the credentials at all here but assume they are available
             result.insert(

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

@@ -565,6 +565,11 @@ impl v1alpha1::SparkApplication {
                 "--conf spark.hadoop.fs.s3a.path.style.access={}",
                 s3conn.access_style == S3AccessStyle::Path
             ));
+            if let Some(region_name) = s3conn.region.name() {
+                submit_cmd.push(format!(
+                    "--conf spark.hadoop.fs.s3a.endpoint.region=\"{region_name}\""
+                ));
+            };
             if let Some(credentials) = &s3conn.credentials {
                 let secret_class_name = credentials.secret_class.clone();
                 let secret_dir = format!("{S3_SECRET_DIR_NAME}/{secret_class_name}");