Add config option for disabling S3 SDK stalled stream protection (#6339)

Author: guilload

quickwit/quickwit-config/src/storage_config.rs

@@ -331,6 +331,10 @@ pub struct S3StorageConfig {
     pub disable_multipart_upload: bool,
     #[serde(default)]
     pub disable_checksums: bool,
+    #[serde(default)]
+    pub disable_stalled_stream_protection_upload: bool,
+    #[serde(default)]
+    pub disable_stalled_stream_protection_download: bool,
 }
 
 impl S3StorageConfig {
@@ -399,6 +403,16 @@ impl fmt::Debug for S3StorageConfig {
                 "disable_multi_object_delete",
                 &self.disable_multi_object_delete,
             )
+            .field("disable_multipart_upload", &self.disable_multipart_upload)
+            .field("disable_checksums", &self.disable_checksums)
+            .field(
+                "disable_stalled_stream_protection_upload",
+                &self.disable_stalled_stream_protection_upload,
+            )
+            .field(
+                "disable_stalled_stream_protection_download",
+                &self.disable_stalled_stream_protection_download,
+            )
             .finish()
     }
 }
@@ -634,6 +648,8 @@ mod tests {
                 disable_multi_object_delete_requests: true
                 disable_multipart_upload: true
                 disable_checksums: true
+                disable_stalled_stream_protection_upload: true
+                disable_stalled_stream_protection_download: true
             "#;
             let s3_storage_config: S3StorageConfig =
                 serde_yaml::from_str(s3_storage_config_yaml).unwrap();
@@ -645,6 +661,8 @@ mod tests {
                 disable_multi_object_delete: true,
                 disable_multipart_upload: true,
                 disable_checksums: true,
+                disable_stalled_stream_protection_upload: true,
+                disable_stalled_stream_protection_download: true,
                 ..Default::default()
             };
             assert_eq!(s3_storage_config, expected_s3_config);

quickwit/quickwit-storage/src/object_storage/s3_compatible_storage.rs

@@ -22,6 +22,7 @@ use std::{fmt, io};
 
 use anyhow::{Context as AnyhhowContext, anyhow};
 use async_trait::async_trait;
+use aws_config::stalled_stream_protection::StalledStreamProtectionConfig;
 use aws_credential_types::provider::SharedCredentialsProvider;
 use aws_sdk_s3::Client as S3Client;
 use aws_sdk_s3::config::{
@@ -145,7 +146,11 @@ pub async fn create_s3_client(s3_storage_config: &S3StorageConfig) -> S3Client {
     s3_config.set_http_client(aws_config.http_client());
     s3_config.set_retry_config(aws_config.retry_config().cloned());
     s3_config.set_sleep_impl(aws_config.sleep_impl());
-    s3_config.set_stalled_stream_protection(aws_config.stalled_stream_protection());
+    let stalled_stream_protection = StalledStreamProtectionConfig::enabled()
+        .upload_enabled(!s3_storage_config.disable_stalled_stream_protection_upload)
+        .download_enabled(!s3_storage_config.disable_stalled_stream_protection_download)
+        .build();
+    s3_config.set_stalled_stream_protection(Some(stalled_stream_protection));
     s3_config.set_timeout_config(aws_config.timeout_config().cloned());
 
     if s3_storage_config.disable_checksums {