From 55bc5fbd59393f620108ae792dcb77768b087313 Mon Sep 17 00:00:00 2001 From: Radostin Emanuilov Date: Mon, 2 Nov 2026 09:35:20 +0000 Subject: [PATCH 1/4] feat(storage-opendal): read the s3.retry.* properties The S3 operator was always wrapped in `RetryLayer::new()`: 3 retries, a 1 s first delay, no jitter. A throttled bucket (S3 `SlowDown`) answers a burst of requests faster than that policy gives up, and the error reaches the caller after about 7 s. Define Iceberg Java's `s3.retry.num-retries`, `s3.retry.min-wait-ms` and `s3.retry.max-wait-ms` in `iceberg::io` so every S3 backend can read the same keys, and apply them to the opendal S3 operator. Setting any of them also turns on jitter; with none set the layer stays `RetryLayer::new()`. Generated-by: Claude Code (Sonnet 5.5) --- crates/iceberg/public-api.txt | 3 + crates/iceberg/src/io/storage/config/s3.rs | 7 + crates/storage/opendal/src/lib.rs | 148 ++++++++++++++++++++- 3 files changed, 153 insertions(+), 5 deletions(-) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 0ba3b6ed78..44dd2044a0 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1117,6 +1117,9 @@ pub const iceberg::io::S3_ENDPOINT: &str pub const iceberg::io::S3_MULTIPART_PART_SIZE_BYTES: &str pub const iceberg::io::S3_PATH_STYLE_ACCESS: &str pub const iceberg::io::S3_REGION: &str +pub const iceberg::io::S3_RETRY_MAX_WAIT_MS: &str +pub const iceberg::io::S3_RETRY_MIN_WAIT_MS: &str +pub const iceberg::io::S3_RETRY_NUM_RETRIES: &str pub const iceberg::io::S3_SECRET_ACCESS_KEY: &str pub const iceberg::io::S3_SESSION_TOKEN: &str pub const iceberg::io::S3_SSE_KEY: &str diff --git a/crates/iceberg/src/io/storage/config/s3.rs b/crates/iceberg/src/io/storage/config/s3.rs index 82f27e15dd..fcee3916e8 100644 --- a/crates/iceberg/src/io/storage/config/s3.rs +++ b/crates/iceberg/src/io/storage/config/s3.rs @@ -75,6 +75,13 @@ pub const S3_DISABLE_CONFIG_LOAD: &str = "s3.disable-config-load"; /// /// Only `iceberg-storage-opendal` reads this property, so [`S3Config`] has no matching field. pub const S3_MULTIPART_PART_SIZE_BYTES: &str = "s3.multipart.part-size-bytes"; +/// Number of times to retry an S3 operation, `0` to disable retrying. Retries back off exponentially, +/// with jitter, between [`S3_RETRY_MIN_WAIT_MS`] and [`S3_RETRY_MAX_WAIT_MS`]. +pub const S3_RETRY_NUM_RETRIES: &str = "s3.retry.num-retries"; +/// Minimum wait in milliseconds before retrying an S3 operation. +pub const S3_RETRY_MIN_WAIT_MS: &str = "s3.retry.min-wait-ms"; +/// Maximum wait in milliseconds between two retries of an S3 operation. +pub const S3_RETRY_MAX_WAIT_MS: &str = "s3.retry.max-wait-ms"; /// Amazon S3 storage configuration. /// diff --git a/crates/storage/opendal/src/lib.rs b/crates/storage/opendal/src/lib.rs index b1279b34b4..89b84c0dca 100644 --- a/crates/storage/opendal/src/lib.rs +++ b/crates/storage/opendal/src/lib.rs @@ -36,8 +36,8 @@ use cfg_if::cfg_if; use futures::StreamExt; use futures::stream::BoxStream; use iceberg::io::{ - FileMetadata, FileRead, FileWrite, InputFile, OutputFile, Storage, StorageConfig, - StorageFactory, + FileMetadata, FileRead, FileWrite, InputFile, OutputFile, S3_RETRY_MAX_WAIT_MS, + S3_RETRY_MIN_WAIT_MS, S3_RETRY_NUM_RETRIES, Storage, StorageConfig, StorageFactory, }; use iceberg::{Error, ErrorKind, Result}; use iceberg_property_macro::Properties; @@ -130,15 +130,35 @@ pub struct OpenDalClientConfig { #[property( key = OPENDAL_IO_TIMEOUT_MS, default = DEFAULT_IO_TIMEOUT_MS, - parse_with = parse_io_timeout_ms + parse_with = parse_positive_ms )] io_timeout_ms: NonZeroU64, + /// S3 retries. + #[property(key = S3_RETRY_NUM_RETRIES, default = None)] + s3_retry_num_retries: Option, + /// S3 delay before the first retry in milliseconds. + #[property( + key = S3_RETRY_MIN_WAIT_MS, + default = None, + parse_with = parse_positive_ms + )] + s3_retry_min_wait_ms: Option, + /// S3 upper bound of the delay between two retries in milliseconds. + #[property( + key = S3_RETRY_MAX_WAIT_MS, + default = None, + parse_with = parse_positive_ms + )] + s3_retry_max_wait_ms: Option, } impl Default for OpenDalClientConfig { fn default() -> Self { Self { io_timeout_ms: DEFAULT_IO_TIMEOUT_MS, + s3_retry_num_retries: None, + s3_retry_min_wait_ms: None, + s3_retry_max_wait_ms: None, } } } @@ -147,9 +167,30 @@ impl OpenDalClientConfig { pub(crate) fn io_timeout(&self) -> Duration { Duration::from_millis(self.io_timeout_ms.get()) } + + /// The retry policy of the `s3.retry.*` properties, with jitter; OpenDAL's default when none is set. + pub(crate) fn s3_retry_layer(&self) -> RetryLayer { + let mut layer = RetryLayer::new(); + if self.s3_retry_num_retries.is_none() + && self.s3_retry_min_wait_ms.is_none() + && self.s3_retry_max_wait_ms.is_none() + { + return layer; + } + if let Some(times) = self.s3_retry_num_retries { + layer = layer.with_max_times(times); + } + if let Some(ms) = self.s3_retry_min_wait_ms { + layer = layer.with_min_delay(Duration::from_millis(ms.get())); + } + if let Some(ms) = self.s3_retry_max_wait_ms { + layer = layer.with_max_delay(Duration::from_millis(ms.get())); + } + layer.with_jitter() + } } -fn parse_io_timeout_ms(value: &str) -> Result { +fn parse_positive_ms(value: &str) -> Result { value.parse().map_err(|error| { Error::new( ErrorKind::DataInvalid, @@ -513,10 +554,19 @@ impl OpenDalStorage { // benefits non-object-store backends. let operator = operator .layer(TimeoutLayer::new().with_io_timeout(self.client_config().io_timeout())) - .layer(RetryLayer::new()); + .layer(self.retry_layer()); Ok((operator, relative_path)) } + /// Only S3 reads the `s3.retry.*` properties; the other backends keep OpenDAL's default. + fn retry_layer(&self) -> RetryLayer { + match self { + #[cfg(feature = "opendal-s3")] + OpenDalStorage::S3 { client_config, .. } => client_config.s3_retry_layer(), + _ => RetryLayer::new(), + } + } + pub(crate) fn client_config(&self) -> &OpenDalClientConfig { match self { #[cfg(feature = "opendal-memory")] @@ -927,6 +977,94 @@ mod tests { ); } + fn retry_config(props: &[(&str, &str)]) -> Result { + OpenDalClientConfig::from_properties( + &props + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect::>(), + ) + } + + #[test] + fn test_s3_retry_defaults_match_opendal() { + // `RetryLayer` has no getters, so compare through `Debug`. + let unset = retry_config(&[]).unwrap(); + assert_eq!( + format!("{:?}", unset.s3_retry_layer()), + format!("{:?}", RetryLayer::new()) + ); + assert_ne!( + format!("{:?}", RetryLayer::new()), + format!("{:?}", RetryLayer::new().with_jitter()) + ); + } + + #[test] + fn test_s3_retry_properties_reach_retry_layer() { + let config = retry_config(&[ + (S3_RETRY_NUM_RETRIES, "10"), + (S3_RETRY_MIN_WAIT_MS, "2000"), + (S3_RETRY_MAX_WAIT_MS, "20000"), + ]) + .unwrap(); + let expected = RetryLayer::new() + .with_max_times(10) + .with_min_delay(Duration::from_secs(2)) + .with_max_delay(Duration::from_secs(20)) + .with_jitter(); + assert_eq!( + format!("{:?}", config.s3_retry_layer()), + format!("{expected:?}") + ); + + // any one property switches jitter on and leaves the others at OpenDAL's defaults + let no_retries = retry_config(&[(S3_RETRY_NUM_RETRIES, "0")]).unwrap(); + assert_eq!( + format!("{:?}", no_retries.s3_retry_layer()), + format!("{:?}", RetryLayer::new().with_max_times(0).with_jitter()) + ); + } + + #[test] + fn test_s3_retry_property_parsing_rejects_invalid_values() { + for (key, invalid) in [ + (S3_RETRY_NUM_RETRIES, "-1"), + (S3_RETRY_NUM_RETRIES, "abc"), + (S3_RETRY_NUM_RETRIES, ""), + (S3_RETRY_MIN_WAIT_MS, "0"), + (S3_RETRY_MIN_WAIT_MS, "abc"), + (S3_RETRY_MAX_WAIT_MS, "0"), + (S3_RETRY_MAX_WAIT_MS, "12.5"), + ] { + let err = retry_config(&[(key, invalid)]).unwrap_err().to_string(); + assert!(err.contains(key), "{key}={invalid}: {err}"); + } + } + + #[cfg(all(feature = "opendal-s3", feature = "opendal-memory"))] + #[test] + fn test_only_s3_applies_the_retry_properties() { + let client_config = retry_config(&[(S3_RETRY_NUM_RETRIES, "10")]).unwrap(); + let s3 = OpenDalStorage::S3 { + config: Arc::new(S3Config::default()), + customized_credential_load: None, + client_config: client_config.clone(), + }; + let memory = OpenDalStorage::Memory { + operator: default_memory_operator(), + client_config, + }; + assert_eq!( + format!("{:?}", s3.retry_layer()), + format!("{:?}", RetryLayer::new().with_max_times(10).with_jitter()) + ); + assert_eq!( + format!("{:?}", memory.retry_layer()), + format!("{:?}", RetryLayer::new()) + ); + } + #[cfg(feature = "opendal-s3")] #[test] fn test_client_config_serde_round_trip() { From e61ef6eb1be5658b79b5c9b22c89ee59a79f6c9e Mon Sep 17 00:00:00 2001 From: Radostin Emanuilov Date: Mon, 2 Nov 2026 17:40:30 +0000 Subject: [PATCH 2/4] feat(storage-opendal): add backend-independent opendal.retry.* properties `opendal.retry.num-retries`, `opendal.retry.min-wait-ms` and `opendal.retry.max-wait-ms` configure the `RetryLayer` of every backend, in the place where the `TimeoutLayer` is applied. For S3 each `s3.retry.*` key overrides the matching `opendal.retry.*` key, resolved per key; unset keeps OpenDAL's default. Any set key turns jitter on. Generated-by: Claude Code (Sonnet 5.5) --- crates/storage/opendal/public-api.txt | 3 + crates/storage/opendal/src/lib.rs | 229 +++++++++++++++++++------- 2 files changed, 171 insertions(+), 61 deletions(-) diff --git a/crates/storage/opendal/public-api.txt b/crates/storage/opendal/public-api.txt index 03e38db6ab..ef24c0f820 100644 --- a/crates/storage/opendal/public-api.txt +++ b/crates/storage/opendal/public-api.txt @@ -130,3 +130,6 @@ pub fn iceberg_storage_opendal::OpenDalResolvingStorageFactory::serialize<__S>(& impl<'de> serde_core::de::Deserialize<'de> for iceberg_storage_opendal::OpenDalResolvingStorageFactory pub fn iceberg_storage_opendal::OpenDalResolvingStorageFactory::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> pub const iceberg_storage_opendal::OPENDAL_IO_TIMEOUT_MS: &str +pub const iceberg_storage_opendal::OPENDAL_RETRY_MAX_WAIT_MS: &str +pub const iceberg_storage_opendal::OPENDAL_RETRY_MIN_WAIT_MS: &str +pub const iceberg_storage_opendal::OPENDAL_RETRY_NUM_RETRIES: &str diff --git a/crates/storage/opendal/src/lib.rs b/crates/storage/opendal/src/lib.rs index 89b84c0dca..a3e8ff60ec 100644 --- a/crates/storage/opendal/src/lib.rs +++ b/crates/storage/opendal/src/lib.rs @@ -119,6 +119,18 @@ pub use resolving::{OpenDalResolvingStorage, OpenDalResolvingStorageFactory}; /// affected and stays at its 60-second default. pub const OPENDAL_IO_TIMEOUT_MS: &str = "opendal.io-timeout-ms"; +/// Number of times to retry a failed operation on any backend, `0` to disable retrying. Unset keeps +/// OpenDAL's default. For S3, [`S3_RETRY_NUM_RETRIES`] takes precedence. +pub const OPENDAL_RETRY_NUM_RETRIES: &str = "opendal.retry.num-retries"; + +/// Minimum wait in milliseconds before retrying a failed operation on any backend. Unset keeps +/// OpenDAL's default. For S3, [`S3_RETRY_MIN_WAIT_MS`] takes precedence. +pub const OPENDAL_RETRY_MIN_WAIT_MS: &str = "opendal.retry.min-wait-ms"; + +/// Maximum wait in milliseconds between two retries on any backend. Unset keeps OpenDAL's default. +/// For S3, [`S3_RETRY_MAX_WAIT_MS`] takes precedence. +pub const OPENDAL_RETRY_MAX_WAIT_MS: &str = "opendal.retry.max-wait-ms"; + /// Matches OpenDAL's `TimeoutLayer` default. const DEFAULT_IO_TIMEOUT_MS: NonZeroU64 = NonZeroU64::new(10_000).unwrap(); @@ -133,6 +145,23 @@ pub struct OpenDalClientConfig { parse_with = parse_positive_ms )] io_timeout_ms: NonZeroU64, + /// Retries on any backend. + #[property(key = OPENDAL_RETRY_NUM_RETRIES, default = None)] + retry_num_retries: Option, + /// Delay before the first retry on any backend in milliseconds. + #[property( + key = OPENDAL_RETRY_MIN_WAIT_MS, + default = None, + parse_with = parse_positive_ms + )] + retry_min_wait_ms: Option, + /// Upper bound of the delay between two retries on any backend in milliseconds. + #[property( + key = OPENDAL_RETRY_MAX_WAIT_MS, + default = None, + parse_with = parse_positive_ms + )] + retry_max_wait_ms: Option, /// S3 retries. #[property(key = S3_RETRY_NUM_RETRIES, default = None)] s3_retry_num_retries: Option, @@ -156,6 +185,9 @@ impl Default for OpenDalClientConfig { fn default() -> Self { Self { io_timeout_ms: DEFAULT_IO_TIMEOUT_MS, + retry_num_retries: None, + retry_min_wait_ms: None, + retry_max_wait_ms: None, s3_retry_num_retries: None, s3_retry_min_wait_ms: None, s3_retry_max_wait_ms: None, @@ -168,22 +200,33 @@ impl OpenDalClientConfig { Duration::from_millis(self.io_timeout_ms.get()) } - /// The retry policy of the `s3.retry.*` properties, with jitter; OpenDAL's default when none is set. - pub(crate) fn s3_retry_layer(&self) -> RetryLayer { + /// The retry policy, with jitter; OpenDAL's default when no key is set. Each setting is resolved on its own: + /// the `s3.retry.*` key (S3 only), then the `opendal.retry.*` key, then OpenDAL's default. + pub(crate) fn retry_layer(&self, s3: bool) -> RetryLayer { + let (times, min_wait, max_wait) = if s3 { + ( + self.s3_retry_num_retries.or(self.retry_num_retries), + self.s3_retry_min_wait_ms.or(self.retry_min_wait_ms), + self.s3_retry_max_wait_ms.or(self.retry_max_wait_ms), + ) + } else { + ( + self.retry_num_retries, + self.retry_min_wait_ms, + self.retry_max_wait_ms, + ) + }; let mut layer = RetryLayer::new(); - if self.s3_retry_num_retries.is_none() - && self.s3_retry_min_wait_ms.is_none() - && self.s3_retry_max_wait_ms.is_none() - { + if times.is_none() && min_wait.is_none() && max_wait.is_none() { return layer; } - if let Some(times) = self.s3_retry_num_retries { + if let Some(times) = times { layer = layer.with_max_times(times); } - if let Some(ms) = self.s3_retry_min_wait_ms { + if let Some(ms) = min_wait { layer = layer.with_min_delay(Duration::from_millis(ms.get())); } - if let Some(ms) = self.s3_retry_max_wait_ms { + if let Some(ms) = max_wait { layer = layer.with_max_delay(Duration::from_millis(ms.get())); } layer.with_jitter() @@ -554,17 +597,16 @@ impl OpenDalStorage { // benefits non-object-store backends. let operator = operator .layer(TimeoutLayer::new().with_io_timeout(self.client_config().io_timeout())) - .layer(self.retry_layer()); + .layer(self.client_config().retry_layer(self.is_s3())); Ok((operator, relative_path)) } - /// Only S3 reads the `s3.retry.*` properties; the other backends keep OpenDAL's default. - fn retry_layer(&self) -> RetryLayer { - match self { - #[cfg(feature = "opendal-s3")] - OpenDalStorage::S3 { client_config, .. } => client_config.s3_retry_layer(), - _ => RetryLayer::new(), - } + /// Only S3 reads the `s3.retry.*` properties. + fn is_s3(&self) -> bool { + #[cfg(feature = "opendal-s3")] + return matches!(self, OpenDalStorage::S3 { .. }); + #[cfg(not(feature = "opendal-s3"))] + false } pub(crate) fn client_config(&self) -> &OpenDalClientConfig { @@ -986,66 +1028,134 @@ mod tests { ) } + /// `RetryLayer` has no getters, so layers are compared through `Debug`. + fn layer_debug(times: Option, min_ms: Option, max_ms: Option) -> String { + let mut layer = RetryLayer::new(); + if times.is_some() || min_ms.is_some() || max_ms.is_some() { + layer = layer.with_jitter(); + } + if let Some(times) = times { + layer = layer.with_max_times(times); + } + if let Some(ms) = min_ms { + layer = layer.with_min_delay(Duration::from_millis(ms)); + } + if let Some(ms) = max_ms { + layer = layer.with_max_delay(Duration::from_millis(ms)); + } + format!("{layer:?}") + } + + fn resolved(props: &[(&str, &str)], s3: bool) -> String { + format!("{:?}", retry_config(props).unwrap().retry_layer(s3)) + } + #[test] - fn test_s3_retry_defaults_match_opendal() { - // `RetryLayer` has no getters, so compare through `Debug`. - let unset = retry_config(&[]).unwrap(); - assert_eq!( - format!("{:?}", unset.s3_retry_layer()), - format!("{:?}", RetryLayer::new()) - ); + fn test_retry_defaults_match_opendal() { + for s3 in [false, true] { + assert_eq!(resolved(&[], s3), format!("{:?}", RetryLayer::new())); + } assert_ne!( format!("{:?}", RetryLayer::new()), - format!("{:?}", RetryLayer::new().with_jitter()) + layer_debug(Some(3), None, None) ); } #[test] - fn test_s3_retry_properties_reach_retry_layer() { - let config = retry_config(&[ + fn test_generic_retry_properties_apply_to_every_backend() { + let props = [ + (OPENDAL_RETRY_NUM_RETRIES, "5"), + (OPENDAL_RETRY_MIN_WAIT_MS, "100"), + (OPENDAL_RETRY_MAX_WAIT_MS, "200"), + ]; + let expected = layer_debug(Some(5), Some(100), Some(200)); + for s3 in [false, true] { + assert_eq!(resolved(&props, s3), expected); + } + // zero retries is valid, and one key alone switches the policy (with jitter) on + assert_eq!( + resolved(&[(OPENDAL_RETRY_NUM_RETRIES, "0")], false), + layer_debug(Some(0), None, None) + ); + } + + #[test] + fn test_s3_retry_properties_apply_to_s3_only() { + let props = [ (S3_RETRY_NUM_RETRIES, "10"), (S3_RETRY_MIN_WAIT_MS, "2000"), (S3_RETRY_MAX_WAIT_MS, "20000"), - ]) - .unwrap(); - let expected = RetryLayer::new() - .with_max_times(10) - .with_min_delay(Duration::from_secs(2)) - .with_max_delay(Duration::from_secs(20)) - .with_jitter(); + ]; + assert_eq!( + resolved(&props, true), + layer_debug(Some(10), Some(2000), Some(20000)) + ); + assert_eq!(resolved(&props, false), format!("{:?}", RetryLayer::new())); + } + + #[test] + fn test_s3_retry_properties_win_over_the_generic_ones_per_key() { + let both = [ + (OPENDAL_RETRY_NUM_RETRIES, "5"), + (OPENDAL_RETRY_MIN_WAIT_MS, "100"), + (OPENDAL_RETRY_MAX_WAIT_MS, "200"), + (S3_RETRY_NUM_RETRIES, "10"), + (S3_RETRY_MIN_WAIT_MS, "2000"), + (S3_RETRY_MAX_WAIT_MS, "20000"), + ]; + assert_eq!( + resolved(&both, true), + layer_debug(Some(10), Some(2000), Some(20000)) + ); + // other backends still see the generic values assert_eq!( - format!("{:?}", config.s3_retry_layer()), - format!("{expected:?}") + resolved(&both, false), + layer_debug(Some(5), Some(100), Some(200)) ); - // any one property switches jitter on and leaves the others at OpenDAL's defaults - let no_retries = retry_config(&[(S3_RETRY_NUM_RETRIES, "0")]).unwrap(); + // a key set at one layer only is taken from that layer + let mixed = [ + (S3_RETRY_NUM_RETRIES, "10"), + (OPENDAL_RETRY_MIN_WAIT_MS, "100"), + (OPENDAL_RETRY_MAX_WAIT_MS, "200"), + (S3_RETRY_MAX_WAIT_MS, "20000"), + ]; assert_eq!( - format!("{:?}", no_retries.s3_retry_layer()), - format!("{:?}", RetryLayer::new().with_max_times(0).with_jitter()) + resolved(&mixed, true), + layer_debug(Some(10), Some(100), Some(20000)) ); } #[test] - fn test_s3_retry_property_parsing_rejects_invalid_values() { - for (key, invalid) in [ - (S3_RETRY_NUM_RETRIES, "-1"), - (S3_RETRY_NUM_RETRIES, "abc"), - (S3_RETRY_NUM_RETRIES, ""), - (S3_RETRY_MIN_WAIT_MS, "0"), - (S3_RETRY_MIN_WAIT_MS, "abc"), - (S3_RETRY_MAX_WAIT_MS, "0"), - (S3_RETRY_MAX_WAIT_MS, "12.5"), + fn test_retry_property_parsing_rejects_invalid_values() { + for key in [ + OPENDAL_RETRY_NUM_RETRIES, + OPENDAL_RETRY_MIN_WAIT_MS, + OPENDAL_RETRY_MAX_WAIT_MS, + S3_RETRY_NUM_RETRIES, + S3_RETRY_MIN_WAIT_MS, + S3_RETRY_MAX_WAIT_MS, ] { - let err = retry_config(&[(key, invalid)]).unwrap_err().to_string(); - assert!(err.contains(key), "{key}={invalid}: {err}"); + let invalid: &[&str] = if key.ends_with("num-retries") { + &["-1", "abc", ""] + } else { + &["0", "abc", "12.5"] + }; + for value in invalid { + let err = retry_config(&[(key, value)]).unwrap_err().to_string(); + assert!(err.contains(key), "{key}={value}: {err}"); + } } } #[cfg(all(feature = "opendal-s3", feature = "opendal-memory"))] #[test] - fn test_only_s3_applies_the_retry_properties() { - let client_config = retry_config(&[(S3_RETRY_NUM_RETRIES, "10")]).unwrap(); + fn test_only_s3_storage_reads_the_s3_retry_properties() { + let client_config = retry_config(&[ + (OPENDAL_RETRY_NUM_RETRIES, "5"), + (S3_RETRY_NUM_RETRIES, "10"), + ]) + .unwrap(); let s3 = OpenDalStorage::S3 { config: Arc::new(S3Config::default()), customized_credential_load: None, @@ -1055,14 +1165,11 @@ mod tests { operator: default_memory_operator(), client_config, }; - assert_eq!( - format!("{:?}", s3.retry_layer()), - format!("{:?}", RetryLayer::new().with_max_times(10).with_jitter()) - ); - assert_eq!( - format!("{:?}", memory.retry_layer()), - format!("{:?}", RetryLayer::new()) - ); + let layer = |storage: &OpenDalStorage| { + format!("{:?}", storage.client_config().retry_layer(storage.is_s3())) + }; + assert_eq!(layer(&s3), layer_debug(Some(10), None, None)); + assert_eq!(layer(&memory), layer_debug(Some(5), None, None)); } #[cfg(feature = "opendal-s3")] From 96eb6fdc7d963c34d6aacfa2818f5b8867715d5a Mon Sep 17 00:00:00 2001 From: Radostin Emanuilov Date: Mon, 2 Nov 2026 17:54:56 +0000 Subject: [PATCH 3/4] refactor(storage-opendal): keep the reviewed retry tests, add only the new cases Restore the four tests of a767ae1e0 as they were (their S3-only semantics still hold) and add only the generic-keys and per-key precedence cases; `s3_retry_layer` stays as a thin wrapper of `retry_layer(true)`. Generated-by: Claude Code (Sonnet 5.5) --- crates/storage/opendal/src/lib.rs | 236 ++++++++++++++++-------------- 1 file changed, 125 insertions(+), 111 deletions(-) diff --git a/crates/storage/opendal/src/lib.rs b/crates/storage/opendal/src/lib.rs index a3e8ff60ec..3600216866 100644 --- a/crates/storage/opendal/src/lib.rs +++ b/crates/storage/opendal/src/lib.rs @@ -200,6 +200,11 @@ impl OpenDalClientConfig { Duration::from_millis(self.io_timeout_ms.get()) } + /// The retry policy of an S3 storage: see [`Self::retry_layer`]. + pub(crate) fn s3_retry_layer(&self) -> RetryLayer { + self.retry_layer(true) + } + /// The retry policy, with jitter; OpenDAL's default when no key is set. Each setting is resolved on its own: /// the `s3.retry.*` key (S3 only), then the `opendal.retry.*` key, then OpenDAL's default. pub(crate) fn retry_layer(&self, s3: bool) -> RetryLayer { @@ -597,16 +602,17 @@ impl OpenDalStorage { // benefits non-object-store backends. let operator = operator .layer(TimeoutLayer::new().with_io_timeout(self.client_config().io_timeout())) - .layer(self.client_config().retry_layer(self.is_s3())); + .layer(self.retry_layer()); Ok((operator, relative_path)) } - /// Only S3 reads the `s3.retry.*` properties. - fn is_s3(&self) -> bool { - #[cfg(feature = "opendal-s3")] - return matches!(self, OpenDalStorage::S3 { .. }); - #[cfg(not(feature = "opendal-s3"))] - false + /// Only S3 reads the `s3.retry.*` properties; the other backends read `opendal.retry.*`. + fn retry_layer(&self) -> RetryLayer { + match self { + #[cfg(feature = "opendal-s3")] + OpenDalStorage::S3 { client_config, .. } => client_config.s3_retry_layer(), + _ => self.client_config().retry_layer(false), + } } pub(crate) fn client_config(&self) -> &OpenDalClientConfig { @@ -1028,148 +1034,156 @@ mod tests { ) } - /// `RetryLayer` has no getters, so layers are compared through `Debug`. - fn layer_debug(times: Option, min_ms: Option, max_ms: Option) -> String { - let mut layer = RetryLayer::new(); - if times.is_some() || min_ms.is_some() || max_ms.is_some() { - layer = layer.with_jitter(); - } - if let Some(times) = times { - layer = layer.with_max_times(times); - } - if let Some(ms) = min_ms { - layer = layer.with_min_delay(Duration::from_millis(ms)); - } - if let Some(ms) = max_ms { - layer = layer.with_max_delay(Duration::from_millis(ms)); - } - format!("{layer:?}") + #[test] + fn test_s3_retry_defaults_match_opendal() { + // `RetryLayer` has no getters, so compare through `Debug`. + let unset = retry_config(&[]).unwrap(); + assert_eq!( + format!("{:?}", unset.s3_retry_layer()), + format!("{:?}", RetryLayer::new()) + ); + assert_ne!( + format!("{:?}", RetryLayer::new()), + format!("{:?}", RetryLayer::new().with_jitter()) + ); } - fn resolved(props: &[(&str, &str)], s3: bool) -> String { - format!("{:?}", retry_config(props).unwrap().retry_layer(s3)) + #[test] + fn test_s3_retry_properties_reach_retry_layer() { + let config = retry_config(&[ + (S3_RETRY_NUM_RETRIES, "10"), + (S3_RETRY_MIN_WAIT_MS, "2000"), + (S3_RETRY_MAX_WAIT_MS, "20000"), + ]) + .unwrap(); + let expected = RetryLayer::new() + .with_max_times(10) + .with_min_delay(Duration::from_secs(2)) + .with_max_delay(Duration::from_secs(20)) + .with_jitter(); + assert_eq!( + format!("{:?}", config.s3_retry_layer()), + format!("{expected:?}") + ); + + // any one property switches jitter on and leaves the others at OpenDAL's defaults + let no_retries = retry_config(&[(S3_RETRY_NUM_RETRIES, "0")]).unwrap(); + assert_eq!( + format!("{:?}", no_retries.s3_retry_layer()), + format!("{:?}", RetryLayer::new().with_max_times(0).with_jitter()) + ); } #[test] - fn test_retry_defaults_match_opendal() { - for s3 in [false, true] { - assert_eq!(resolved(&[], s3), format!("{:?}", RetryLayer::new())); + fn test_s3_retry_property_parsing_rejects_invalid_values() { + for (key, invalid) in [ + (S3_RETRY_NUM_RETRIES, "-1"), + (S3_RETRY_NUM_RETRIES, "abc"), + (S3_RETRY_NUM_RETRIES, ""), + (S3_RETRY_MIN_WAIT_MS, "0"), + (S3_RETRY_MIN_WAIT_MS, "abc"), + (S3_RETRY_MAX_WAIT_MS, "0"), + (S3_RETRY_MAX_WAIT_MS, "12.5"), + (OPENDAL_RETRY_NUM_RETRIES, "-1"), + (OPENDAL_RETRY_MIN_WAIT_MS, "0"), + (OPENDAL_RETRY_MAX_WAIT_MS, "abc"), + ] { + let err = retry_config(&[(key, invalid)]).unwrap_err().to_string(); + assert!(err.contains(key), "{key}={invalid}: {err}"); } - assert_ne!( - format!("{:?}", RetryLayer::new()), - layer_debug(Some(3), None, None) + } + + #[cfg(all(feature = "opendal-s3", feature = "opendal-memory"))] + #[test] + fn test_only_s3_applies_the_retry_properties() { + let client_config = retry_config(&[(S3_RETRY_NUM_RETRIES, "10")]).unwrap(); + let s3 = OpenDalStorage::S3 { + config: Arc::new(S3Config::default()), + customized_credential_load: None, + client_config: client_config.clone(), + }; + let memory = OpenDalStorage::Memory { + operator: default_memory_operator(), + client_config, + }; + assert_eq!( + format!("{:?}", s3.retry_layer()), + format!("{:?}", RetryLayer::new().with_max_times(10).with_jitter()) + ); + assert_eq!( + format!("{:?}", memory.retry_layer()), + format!("{:?}", RetryLayer::new()) ); } #[test] fn test_generic_retry_properties_apply_to_every_backend() { - let props = [ + let config = retry_config(&[ (OPENDAL_RETRY_NUM_RETRIES, "5"), (OPENDAL_RETRY_MIN_WAIT_MS, "100"), (OPENDAL_RETRY_MAX_WAIT_MS, "200"), - ]; - let expected = layer_debug(Some(5), Some(100), Some(200)); + ]) + .unwrap(); + let expected = RetryLayer::new() + .with_max_times(5) + .with_min_delay(Duration::from_millis(100)) + .with_max_delay(Duration::from_millis(200)) + .with_jitter(); for s3 in [false, true] { - assert_eq!(resolved(&props, s3), expected); + assert_eq!( + format!("{:?}", config.retry_layer(s3)), + format!("{expected:?}") + ); } - // zero retries is valid, and one key alone switches the policy (with jitter) on - assert_eq!( - resolved(&[(OPENDAL_RETRY_NUM_RETRIES, "0")], false), - layer_debug(Some(0), None, None) - ); - } - - #[test] - fn test_s3_retry_properties_apply_to_s3_only() { - let props = [ - (S3_RETRY_NUM_RETRIES, "10"), - (S3_RETRY_MIN_WAIT_MS, "2000"), - (S3_RETRY_MAX_WAIT_MS, "20000"), - ]; - assert_eq!( - resolved(&props, true), - layer_debug(Some(10), Some(2000), Some(20000)) - ); - assert_eq!(resolved(&props, false), format!("{:?}", RetryLayer::new())); } #[test] fn test_s3_retry_properties_win_over_the_generic_ones_per_key() { - let both = [ + let both = retry_config(&[ (OPENDAL_RETRY_NUM_RETRIES, "5"), (OPENDAL_RETRY_MIN_WAIT_MS, "100"), (OPENDAL_RETRY_MAX_WAIT_MS, "200"), (S3_RETRY_NUM_RETRIES, "10"), (S3_RETRY_MIN_WAIT_MS, "2000"), (S3_RETRY_MAX_WAIT_MS, "20000"), - ]; + ]) + .unwrap(); + let s3_values = RetryLayer::new() + .with_max_times(10) + .with_min_delay(Duration::from_secs(2)) + .with_max_delay(Duration::from_secs(20)) + .with_jitter(); + let generic_values = RetryLayer::new() + .with_max_times(5) + .with_min_delay(Duration::from_millis(100)) + .with_max_delay(Duration::from_millis(200)) + .with_jitter(); assert_eq!( - resolved(&both, true), - layer_debug(Some(10), Some(2000), Some(20000)) + format!("{:?}", both.retry_layer(true)), + format!("{s3_values:?}") ); - // other backends still see the generic values assert_eq!( - resolved(&both, false), - layer_debug(Some(5), Some(100), Some(200)) + format!("{:?}", both.retry_layer(false)), + format!("{generic_values:?}") ); // a key set at one layer only is taken from that layer - let mixed = [ + let mixed = retry_config(&[ (S3_RETRY_NUM_RETRIES, "10"), (OPENDAL_RETRY_MIN_WAIT_MS, "100"), (OPENDAL_RETRY_MAX_WAIT_MS, "200"), (S3_RETRY_MAX_WAIT_MS, "20000"), - ]; - assert_eq!( - resolved(&mixed, true), - layer_debug(Some(10), Some(100), Some(20000)) - ); - } - - #[test] - fn test_retry_property_parsing_rejects_invalid_values() { - for key in [ - OPENDAL_RETRY_NUM_RETRIES, - OPENDAL_RETRY_MIN_WAIT_MS, - OPENDAL_RETRY_MAX_WAIT_MS, - S3_RETRY_NUM_RETRIES, - S3_RETRY_MIN_WAIT_MS, - S3_RETRY_MAX_WAIT_MS, - ] { - let invalid: &[&str] = if key.ends_with("num-retries") { - &["-1", "abc", ""] - } else { - &["0", "abc", "12.5"] - }; - for value in invalid { - let err = retry_config(&[(key, value)]).unwrap_err().to_string(); - assert!(err.contains(key), "{key}={value}: {err}"); - } - } - } - - #[cfg(all(feature = "opendal-s3", feature = "opendal-memory"))] - #[test] - fn test_only_s3_storage_reads_the_s3_retry_properties() { - let client_config = retry_config(&[ - (OPENDAL_RETRY_NUM_RETRIES, "5"), - (S3_RETRY_NUM_RETRIES, "10"), ]) .unwrap(); - let s3 = OpenDalStorage::S3 { - config: Arc::new(S3Config::default()), - customized_credential_load: None, - client_config: client_config.clone(), - }; - let memory = OpenDalStorage::Memory { - operator: default_memory_operator(), - client_config, - }; - let layer = |storage: &OpenDalStorage| { - format!("{:?}", storage.client_config().retry_layer(storage.is_s3())) - }; - assert_eq!(layer(&s3), layer_debug(Some(10), None, None)); - assert_eq!(layer(&memory), layer_debug(Some(5), None, None)); + let expected = RetryLayer::new() + .with_max_times(10) + .with_min_delay(Duration::from_millis(100)) + .with_max_delay(Duration::from_secs(20)) + .with_jitter(); + assert_eq!( + format!("{:?}", mixed.retry_layer(true)), + format!("{expected:?}") + ); } #[cfg(feature = "opendal-s3")] From 7048c0b781f1f6dcaa3b57fd245d9413c6aae4cb Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 6 Nov 2026 14:20:04 +0000 Subject: [PATCH 4/4] refactor(storage-opendal): resolve s3.retry.* overrides when building the S3 storage Parse the s3.retry.* properties next to the other s3.* properties and overlay them, per key, onto the opendal.retry.* settings of the S3 variant's client config. OpenDalClientConfig no longer knows about S3, so retry_layer takes no backend flag and other backends neither parse nor serialize s3.retry.*. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01FhHg5U2mpkFVSMgVtmhThk --- crates/storage/opendal/src/lib.rs | 218 +++++------------------- crates/storage/opendal/src/resolving.rs | 2 +- crates/storage/opendal/src/s3.rs | 150 +++++++++++++++- 3 files changed, 186 insertions(+), 184 deletions(-) diff --git a/crates/storage/opendal/src/lib.rs b/crates/storage/opendal/src/lib.rs index 3600216866..55b651ae1f 100644 --- a/crates/storage/opendal/src/lib.rs +++ b/crates/storage/opendal/src/lib.rs @@ -36,8 +36,8 @@ use cfg_if::cfg_if; use futures::StreamExt; use futures::stream::BoxStream; use iceberg::io::{ - FileMetadata, FileRead, FileWrite, InputFile, OutputFile, S3_RETRY_MAX_WAIT_MS, - S3_RETRY_MIN_WAIT_MS, S3_RETRY_NUM_RETRIES, Storage, StorageConfig, StorageFactory, + FileMetadata, FileRead, FileWrite, InputFile, OutputFile, Storage, StorageConfig, + StorageFactory, }; use iceberg::{Error, ErrorKind, Result}; use iceberg_property_macro::Properties; @@ -120,15 +120,15 @@ pub use resolving::{OpenDalResolvingStorage, OpenDalResolvingStorageFactory}; pub const OPENDAL_IO_TIMEOUT_MS: &str = "opendal.io-timeout-ms"; /// Number of times to retry a failed operation on any backend, `0` to disable retrying. Unset keeps -/// OpenDAL's default. For S3, [`S3_RETRY_NUM_RETRIES`] takes precedence. +/// OpenDAL's default. For S3, [`S3_RETRY_NUM_RETRIES`](iceberg::io::S3_RETRY_NUM_RETRIES) takes precedence. pub const OPENDAL_RETRY_NUM_RETRIES: &str = "opendal.retry.num-retries"; /// Minimum wait in milliseconds before retrying a failed operation on any backend. Unset keeps -/// OpenDAL's default. For S3, [`S3_RETRY_MIN_WAIT_MS`] takes precedence. +/// OpenDAL's default. For S3, [`S3_RETRY_MIN_WAIT_MS`](iceberg::io::S3_RETRY_MIN_WAIT_MS) takes precedence. pub const OPENDAL_RETRY_MIN_WAIT_MS: &str = "opendal.retry.min-wait-ms"; /// Maximum wait in milliseconds between two retries on any backend. Unset keeps OpenDAL's default. -/// For S3, [`S3_RETRY_MAX_WAIT_MS`] takes precedence. +/// For S3, [`S3_RETRY_MAX_WAIT_MS`](iceberg::io::S3_RETRY_MAX_WAIT_MS) takes precedence. pub const OPENDAL_RETRY_MAX_WAIT_MS: &str = "opendal.retry.max-wait-ms"; /// Matches OpenDAL's `TimeoutLayer` default. @@ -162,23 +162,6 @@ pub struct OpenDalClientConfig { parse_with = parse_positive_ms )] retry_max_wait_ms: Option, - /// S3 retries. - #[property(key = S3_RETRY_NUM_RETRIES, default = None)] - s3_retry_num_retries: Option, - /// S3 delay before the first retry in milliseconds. - #[property( - key = S3_RETRY_MIN_WAIT_MS, - default = None, - parse_with = parse_positive_ms - )] - s3_retry_min_wait_ms: Option, - /// S3 upper bound of the delay between two retries in milliseconds. - #[property( - key = S3_RETRY_MAX_WAIT_MS, - default = None, - parse_with = parse_positive_ms - )] - s3_retry_max_wait_ms: Option, } impl Default for OpenDalClientConfig { @@ -188,9 +171,6 @@ impl Default for OpenDalClientConfig { retry_num_retries: None, retry_min_wait_ms: None, retry_max_wait_ms: None, - s3_retry_num_retries: None, - s3_retry_min_wait_ms: None, - s3_retry_max_wait_ms: None, } } } @@ -200,38 +180,22 @@ impl OpenDalClientConfig { Duration::from_millis(self.io_timeout_ms.get()) } - /// The retry policy of an S3 storage: see [`Self::retry_layer`]. - pub(crate) fn s3_retry_layer(&self) -> RetryLayer { - self.retry_layer(true) - } - - /// The retry policy, with jitter; OpenDAL's default when no key is set. Each setting is resolved on its own: - /// the `s3.retry.*` key (S3 only), then the `opendal.retry.*` key, then OpenDAL's default. - pub(crate) fn retry_layer(&self, s3: bool) -> RetryLayer { - let (times, min_wait, max_wait) = if s3 { - ( - self.s3_retry_num_retries.or(self.retry_num_retries), - self.s3_retry_min_wait_ms.or(self.retry_min_wait_ms), - self.s3_retry_max_wait_ms.or(self.retry_max_wait_ms), - ) - } else { - ( - self.retry_num_retries, - self.retry_min_wait_ms, - self.retry_max_wait_ms, - ) - }; + /// The retry policy, with jitter; OpenDAL's default when no key is set. + pub(crate) fn retry_layer(&self) -> RetryLayer { let mut layer = RetryLayer::new(); - if times.is_none() && min_wait.is_none() && max_wait.is_none() { + if self.retry_num_retries.is_none() + && self.retry_min_wait_ms.is_none() + && self.retry_max_wait_ms.is_none() + { return layer; } - if let Some(times) = times { + if let Some(times) = self.retry_num_retries { layer = layer.with_max_times(times); } - if let Some(ms) = min_wait { + if let Some(ms) = self.retry_min_wait_ms { layer = layer.with_min_delay(Duration::from_millis(ms.get())); } - if let Some(ms) = max_wait { + if let Some(ms) = self.retry_max_wait_ms { layer = layer.with_max_delay(Duration::from_millis(ms.get())); } layer.with_jitter() @@ -331,7 +295,7 @@ impl StorageFactory for OpenDalStorageFactory { config: s3_config_parse(config.props().clone())?.into(), multipart_part_size: s3_multipart_part_size_parse(config.props())?, customized_credential_load: customized_credential_load.clone(), - client_config, + client_config: s3_client_config(config.props(), client_config)?, })), #[cfg(feature = "opendal-gcs")] OpenDalStorageFactory::Gcs => Ok(Arc::new(OpenDalStorage::Gcs { @@ -602,19 +566,10 @@ impl OpenDalStorage { // benefits non-object-store backends. let operator = operator .layer(TimeoutLayer::new().with_io_timeout(self.client_config().io_timeout())) - .layer(self.retry_layer()); + .layer(self.client_config().retry_layer()); Ok((operator, relative_path)) } - /// Only S3 reads the `s3.retry.*` properties; the other backends read `opendal.retry.*`. - fn retry_layer(&self) -> RetryLayer { - match self { - #[cfg(feature = "opendal-s3")] - OpenDalStorage::S3 { client_config, .. } => client_config.s3_retry_layer(), - _ => self.client_config().retry_layer(false), - } - } - pub(crate) fn client_config(&self) -> &OpenDalClientConfig { match self { #[cfg(feature = "opendal-memory")] @@ -1034,12 +989,15 @@ mod tests { ) } + fn retry_layer_debug(props: &[(&str, &str)]) -> Result { + Ok(format!("{:?}", retry_config(props)?.retry_layer())) + } + #[test] - fn test_s3_retry_defaults_match_opendal() { + fn test_retry_defaults_match_opendal() { // `RetryLayer` has no getters, so compare through `Debug`. - let unset = retry_config(&[]).unwrap(); assert_eq!( - format!("{:?}", unset.s3_retry_layer()), + retry_layer_debug(&[]).unwrap(), format!("{:?}", RetryLayer::new()) ); assert_ne!( @@ -1049,143 +1007,45 @@ mod tests { } #[test] - fn test_s3_retry_properties_reach_retry_layer() { - let config = retry_config(&[ - (S3_RETRY_NUM_RETRIES, "10"), - (S3_RETRY_MIN_WAIT_MS, "2000"), - (S3_RETRY_MAX_WAIT_MS, "20000"), - ]) - .unwrap(); + fn test_retry_properties_reach_retry_layer() { let expected = RetryLayer::new() - .with_max_times(10) - .with_min_delay(Duration::from_secs(2)) - .with_max_delay(Duration::from_secs(20)) + .with_max_times(5) + .with_min_delay(Duration::from_millis(100)) + .with_max_delay(Duration::from_millis(200)) .with_jitter(); assert_eq!( - format!("{:?}", config.s3_retry_layer()), + retry_layer_debug(&[ + (OPENDAL_RETRY_NUM_RETRIES, "5"), + (OPENDAL_RETRY_MIN_WAIT_MS, "100"), + (OPENDAL_RETRY_MAX_WAIT_MS, "200"), + ]) + .unwrap(), format!("{expected:?}") ); // any one property switches jitter on and leaves the others at OpenDAL's defaults - let no_retries = retry_config(&[(S3_RETRY_NUM_RETRIES, "0")]).unwrap(); assert_eq!( - format!("{:?}", no_retries.s3_retry_layer()), + retry_layer_debug(&[(OPENDAL_RETRY_NUM_RETRIES, "0")]).unwrap(), format!("{:?}", RetryLayer::new().with_max_times(0).with_jitter()) ); } #[test] - fn test_s3_retry_property_parsing_rejects_invalid_values() { + fn test_retry_property_parsing_rejects_invalid_values() { for (key, invalid) in [ - (S3_RETRY_NUM_RETRIES, "-1"), - (S3_RETRY_NUM_RETRIES, "abc"), - (S3_RETRY_NUM_RETRIES, ""), - (S3_RETRY_MIN_WAIT_MS, "0"), - (S3_RETRY_MIN_WAIT_MS, "abc"), - (S3_RETRY_MAX_WAIT_MS, "0"), - (S3_RETRY_MAX_WAIT_MS, "12.5"), (OPENDAL_RETRY_NUM_RETRIES, "-1"), + (OPENDAL_RETRY_NUM_RETRIES, "abc"), + (OPENDAL_RETRY_NUM_RETRIES, ""), (OPENDAL_RETRY_MIN_WAIT_MS, "0"), - (OPENDAL_RETRY_MAX_WAIT_MS, "abc"), + (OPENDAL_RETRY_MIN_WAIT_MS, "abc"), + (OPENDAL_RETRY_MAX_WAIT_MS, "0"), + (OPENDAL_RETRY_MAX_WAIT_MS, "12.5"), ] { let err = retry_config(&[(key, invalid)]).unwrap_err().to_string(); assert!(err.contains(key), "{key}={invalid}: {err}"); } } - #[cfg(all(feature = "opendal-s3", feature = "opendal-memory"))] - #[test] - fn test_only_s3_applies_the_retry_properties() { - let client_config = retry_config(&[(S3_RETRY_NUM_RETRIES, "10")]).unwrap(); - let s3 = OpenDalStorage::S3 { - config: Arc::new(S3Config::default()), - customized_credential_load: None, - client_config: client_config.clone(), - }; - let memory = OpenDalStorage::Memory { - operator: default_memory_operator(), - client_config, - }; - assert_eq!( - format!("{:?}", s3.retry_layer()), - format!("{:?}", RetryLayer::new().with_max_times(10).with_jitter()) - ); - assert_eq!( - format!("{:?}", memory.retry_layer()), - format!("{:?}", RetryLayer::new()) - ); - } - - #[test] - fn test_generic_retry_properties_apply_to_every_backend() { - let config = retry_config(&[ - (OPENDAL_RETRY_NUM_RETRIES, "5"), - (OPENDAL_RETRY_MIN_WAIT_MS, "100"), - (OPENDAL_RETRY_MAX_WAIT_MS, "200"), - ]) - .unwrap(); - let expected = RetryLayer::new() - .with_max_times(5) - .with_min_delay(Duration::from_millis(100)) - .with_max_delay(Duration::from_millis(200)) - .with_jitter(); - for s3 in [false, true] { - assert_eq!( - format!("{:?}", config.retry_layer(s3)), - format!("{expected:?}") - ); - } - } - - #[test] - fn test_s3_retry_properties_win_over_the_generic_ones_per_key() { - let both = retry_config(&[ - (OPENDAL_RETRY_NUM_RETRIES, "5"), - (OPENDAL_RETRY_MIN_WAIT_MS, "100"), - (OPENDAL_RETRY_MAX_WAIT_MS, "200"), - (S3_RETRY_NUM_RETRIES, "10"), - (S3_RETRY_MIN_WAIT_MS, "2000"), - (S3_RETRY_MAX_WAIT_MS, "20000"), - ]) - .unwrap(); - let s3_values = RetryLayer::new() - .with_max_times(10) - .with_min_delay(Duration::from_secs(2)) - .with_max_delay(Duration::from_secs(20)) - .with_jitter(); - let generic_values = RetryLayer::new() - .with_max_times(5) - .with_min_delay(Duration::from_millis(100)) - .with_max_delay(Duration::from_millis(200)) - .with_jitter(); - assert_eq!( - format!("{:?}", both.retry_layer(true)), - format!("{s3_values:?}") - ); - assert_eq!( - format!("{:?}", both.retry_layer(false)), - format!("{generic_values:?}") - ); - - // a key set at one layer only is taken from that layer - let mixed = retry_config(&[ - (S3_RETRY_NUM_RETRIES, "10"), - (OPENDAL_RETRY_MIN_WAIT_MS, "100"), - (OPENDAL_RETRY_MAX_WAIT_MS, "200"), - (S3_RETRY_MAX_WAIT_MS, "20000"), - ]) - .unwrap(); - let expected = RetryLayer::new() - .with_max_times(10) - .with_min_delay(Duration::from_millis(100)) - .with_max_delay(Duration::from_secs(20)) - .with_jitter(); - assert_eq!( - format!("{:?}", mixed.retry_layer(true)), - format!("{expected:?}") - ); - } - #[cfg(feature = "opendal-s3")] #[test] fn test_client_config_serde_round_trip() { diff --git a/crates/storage/opendal/src/resolving.rs b/crates/storage/opendal/src/resolving.rs index 9123022de3..47459c42c8 100644 --- a/crates/storage/opendal/src/resolving.rs +++ b/crates/storage/opendal/src/resolving.rs @@ -97,7 +97,7 @@ fn build_storage_for_scheme( config: Arc::new(config), multipart_part_size: crate::s3::s3_multipart_part_size_parse(props)?, customized_credential_load: customized_credential_load.clone(), - client_config, + client_config: crate::s3::s3_client_config(props, client_config)?, }) } #[cfg(feature = "opendal-gcs")] diff --git a/crates/storage/opendal/src/s3.rs b/crates/storage/opendal/src/s3.rs index 8f1828dc46..b7e0166185 100644 --- a/crates/storage/opendal/src/s3.rs +++ b/crates/storage/opendal/src/s3.rs @@ -16,15 +16,18 @@ // under the License. use std::collections::HashMap; +use std::num::NonZeroU64; use std::sync::Arc; use iceberg::io::{ CLIENT_REGION, S3_ACCESS_KEY_ID, S3_ALLOW_ANONYMOUS, S3_ASSUME_ROLE_ARN, S3_ASSUME_ROLE_EXTERNAL_ID, S3_ASSUME_ROLE_SESSION_NAME, S3_DISABLE_CONFIG_LOAD, S3_DISABLE_EC2_METADATA, S3_ENDPOINT, S3_MULTIPART_PART_SIZE_BYTES, S3_PATH_STYLE_ACCESS, - S3_REGION, S3_SECRET_ACCESS_KEY, S3_SESSION_TOKEN, S3_SSE_KEY, S3_SSE_MD5, S3_SSE_TYPE, + S3_REGION, S3_RETRY_MAX_WAIT_MS, S3_RETRY_MIN_WAIT_MS, S3_RETRY_NUM_RETRIES, + S3_SECRET_ACCESS_KEY, S3_SESSION_TOKEN, S3_SSE_KEY, S3_SSE_MD5, S3_SSE_TYPE, }; use iceberg::{Error, ErrorKind, Result}; +use iceberg_property_macro::Properties; use opendal::services::S3Config; use opendal::{Configurator, Operator}; /// AWS credentials: access key ID, secret access key, and optional session token. @@ -35,6 +38,29 @@ use reqsign_core::{ProvideCredentialChain, ProvideCredentialDyn}; use url::Url; use crate::utils::{from_opendal_error, is_truthy}; +use crate::{OpenDalClientConfig, parse_positive_ms}; + +#[derive(Properties)] +struct S3RetryConfig { + #[property(key = S3_RETRY_NUM_RETRIES, default = None)] + num_retries: Option, + #[property(key = S3_RETRY_MIN_WAIT_MS, default = None, parse_with = parse_positive_ms)] + min_wait_ms: Option, + #[property(key = S3_RETRY_MAX_WAIT_MS, default = None, parse_with = parse_positive_ms)] + max_wait_ms: Option, +} + +/// Overlays the `s3.retry.*` properties onto the `opendal.retry.*` settings of `client_config`, per key. +pub(crate) fn s3_client_config( + props: &HashMap, + mut client_config: OpenDalClientConfig, +) -> Result { + let retry = S3RetryConfig::from_properties(props)?; + client_config.retry_num_retries = retry.num_retries.or(client_config.retry_num_retries); + client_config.retry_min_wait_ms = retry.min_wait_ms.or(client_config.retry_min_wait_ms); + client_config.retry_max_wait_ms = retry.max_wait_ms.or(client_config.retry_max_wait_ms); + Ok(client_config) +} /// S3 rejects a non-final part smaller than this. const MULTIPART_PART_SIZE_MIN: u64 = 5 * 1024 * 1024; @@ -227,12 +253,22 @@ impl CustomAwsCredentialLoader { #[cfg(test)] mod tests { use std::collections::HashMap; + use std::time::Duration; - use iceberg::io::{S3_MULTIPART_PART_SIZE_BYTES, S3_PATH_STYLE_ACCESS}; + use iceberg::Result; + use iceberg::io::{ + S3_MULTIPART_PART_SIZE_BYTES, S3_PATH_STYLE_ACCESS, S3_RETRY_MAX_WAIT_MS, + S3_RETRY_MIN_WAIT_MS, S3_RETRY_NUM_RETRIES, + }; + use opendal::layers::RetryLayer; use super::{ - MULTIPART_PART_SIZE_DEFAULT, MULTIPART_PART_SIZE_MAX, MULTIPART_PART_SIZE_MIN, Result, - s3_config_parse, s3_multipart_part_size_parse, + MULTIPART_PART_SIZE_DEFAULT, MULTIPART_PART_SIZE_MAX, MULTIPART_PART_SIZE_MIN, + s3_client_config, s3_config_parse, s3_multipart_part_size_parse, + }; + use crate::{ + OPENDAL_RETRY_MAX_WAIT_MS, OPENDAL_RETRY_MIN_WAIT_MS, OPENDAL_RETRY_NUM_RETRIES, + OpenDalClientConfig, }; fn parse_with(prop: Option<&str>) -> bool { @@ -277,4 +313,110 @@ mod tests { assert!(parse_part_size(Some(&(MULTIPART_PART_SIZE_MAX + 1).to_string())).is_err()); assert!(parse_part_size(Some("not-a-number")).is_err()); } + + fn s3_retry_layer(props: &[(&str, &str)]) -> Result { + let props = props + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect::>(); + let client_config = + s3_client_config(&props, OpenDalClientConfig::from_properties(&props)?)?; + Ok(format!("{:?}", client_config.retry_layer())) + } + + #[test] + fn test_s3_retry_properties_reach_retry_layer() { + let expected = RetryLayer::new() + .with_max_times(10) + .with_min_delay(Duration::from_secs(2)) + .with_max_delay(Duration::from_secs(20)) + .with_jitter(); + assert_eq!( + s3_retry_layer(&[ + (S3_RETRY_NUM_RETRIES, "10"), + (S3_RETRY_MIN_WAIT_MS, "2000"), + (S3_RETRY_MAX_WAIT_MS, "20000"), + ]) + .unwrap(), + format!("{expected:?}") + ); + assert_eq!( + s3_retry_layer(&[(S3_RETRY_NUM_RETRIES, "0")]).unwrap(), + format!("{:?}", RetryLayer::new().with_max_times(0).with_jitter()) + ); + assert_eq!( + s3_retry_layer(&[]).unwrap(), + format!("{:?}", RetryLayer::new()) + ); + } + + #[test] + fn test_s3_retry_properties_win_over_the_generic_ones_per_key() { + let expected = RetryLayer::new() + .with_max_times(10) + .with_min_delay(Duration::from_secs(2)) + .with_max_delay(Duration::from_secs(20)) + .with_jitter(); + assert_eq!( + s3_retry_layer(&[ + (OPENDAL_RETRY_NUM_RETRIES, "5"), + (OPENDAL_RETRY_MIN_WAIT_MS, "100"), + (OPENDAL_RETRY_MAX_WAIT_MS, "200"), + (S3_RETRY_NUM_RETRIES, "10"), + (S3_RETRY_MIN_WAIT_MS, "2000"), + (S3_RETRY_MAX_WAIT_MS, "20000"), + ]) + .unwrap(), + format!("{expected:?}") + ); + + // a key set at one layer only is taken from that layer + let mixed = RetryLayer::new() + .with_max_times(10) + .with_min_delay(Duration::from_millis(100)) + .with_max_delay(Duration::from_secs(20)) + .with_jitter(); + assert_eq!( + s3_retry_layer(&[ + (S3_RETRY_NUM_RETRIES, "10"), + (OPENDAL_RETRY_MIN_WAIT_MS, "100"), + (OPENDAL_RETRY_MAX_WAIT_MS, "200"), + (S3_RETRY_MAX_WAIT_MS, "20000"), + ]) + .unwrap(), + format!("{mixed:?}") + ); + } + + #[test] + fn test_s3_retry_property_parsing_rejects_invalid_values() { + for (key, invalid) in [ + (S3_RETRY_NUM_RETRIES, "-1"), + (S3_RETRY_NUM_RETRIES, "abc"), + (S3_RETRY_NUM_RETRIES, ""), + (S3_RETRY_MIN_WAIT_MS, "0"), + (S3_RETRY_MIN_WAIT_MS, "abc"), + (S3_RETRY_MAX_WAIT_MS, "0"), + (S3_RETRY_MAX_WAIT_MS, "12.5"), + ] { + let err = s3_retry_layer(&[(key, invalid)]).unwrap_err().to_string(); + assert!(err.contains(key), "{key}={invalid}: {err}"); + } + } + + #[cfg(feature = "opendal-memory")] + #[test] + fn test_other_backends_ignore_the_s3_retry_properties() { + use iceberg::io::{StorageConfig, StorageFactory}; + + use crate::OpenDalStorageFactory; + + let config = StorageConfig::new().with_props([ + (S3_RETRY_NUM_RETRIES.to_string(), "abc".to_string()), + (OPENDAL_RETRY_NUM_RETRIES.to_string(), "5".to_string()), + ]); + let storage = OpenDalStorageFactory::Memory.build(&config).unwrap(); + let json = serde_json::to_string(&storage).unwrap(); + assert!(json.contains(r#""retry_num_retries":5"#), "{json}"); + } }