feat(storage-opendal): configurable retry policy via opendal.retry.* and s3.retry.*

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

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.
 ///

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<Self, <__D as serde_core::de::Deserializer>::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

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`](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`](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`](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.
 const DEFAULT_IO_TIMEOUT_MS: NonZeroU64 = NonZeroU64::new(10_000).unwrap();
 
@@ -130,15 +142,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,
+    /// Retries on any backend.
+    #[property(key = OPENDAL_RETRY_NUM_RETRIES, default = None)]
+    retry_num_retries: Option<usize>,
+    /// 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<NonZeroU64>,
+    /// 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<NonZeroU64>,
 }
 
 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,
         }
     }
 }
@@ -147,9 +179,30 @@ impl OpenDalClientConfig {
     pub(crate) fn io_timeout(&self) -> Duration {
         Duration::from_millis(self.io_timeout_ms.get())
     }
+
+    /// 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 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) = self.retry_num_retries {
+            layer = layer.with_max_times(times);
+        }
+        if let Some(ms) = self.retry_min_wait_ms {
+            layer = layer.with_min_delay(Duration::from_millis(ms.get()));
+        }
+        if let Some(ms) = self.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<NonZeroU64> {
+fn parse_positive_ms(value: &str) -> Result<NonZeroU64> {
     value.parse().map_err(|error| {
         Error::new(
             ErrorKind::DataInvalid,
@@ -242,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 {
@@ -513,7 +566,7 @@ 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.client_config().retry_layer());
         Ok((operator, relative_path))
     }
 
@@ -927,6 +980,72 @@ mod tests {
         );
     }
 
+    fn retry_config(props: &[(&str, &str)]) -> Result<OpenDalClientConfig> {
+        OpenDalClientConfig::from_properties(
+            &props
+                .iter()
+                .map(|(k, v)| (k.to_string(), v.to_string()))
+                .collect::<HashMap<_, _>>(),
+        )
+    }
+
+    fn retry_layer_debug(props: &[(&str, &str)]) -> Result<String> {
+        Ok(format!("{:?}", retry_config(props)?.retry_layer()))
+    }
+
+    #[test]
+    fn test_retry_defaults_match_opendal() {
+        // `RetryLayer` has no getters, so compare through `Debug`.
+        assert_eq!(
+            retry_layer_debug(&[]).unwrap(),
+            format!("{:?}", RetryLayer::new())
+        );
+        assert_ne!(
+            format!("{:?}", RetryLayer::new()),
+            format!("{:?}", RetryLayer::new().with_jitter())
+        );
+    }
+
+    #[test]
+    fn test_retry_properties_reach_retry_layer() {
+        let expected = RetryLayer::new()
+            .with_max_times(5)
+            .with_min_delay(Duration::from_millis(100))
+            .with_max_delay(Duration::from_millis(200))
+            .with_jitter();
+        assert_eq!(
+            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
+        assert_eq!(
+            retry_layer_debug(&[(OPENDAL_RETRY_NUM_RETRIES, "0")]).unwrap(),
+            format!("{:?}", RetryLayer::new().with_max_times(0).with_jitter())
+        );
+    }
+
+    #[test]
+    fn test_retry_property_parsing_rejects_invalid_values() {
+        for (key, invalid) in [
+            (OPENDAL_RETRY_NUM_RETRIES, "-1"),
+            (OPENDAL_RETRY_NUM_RETRIES, "abc"),
+            (OPENDAL_RETRY_NUM_RETRIES, ""),
+            (OPENDAL_RETRY_MIN_WAIT_MS, "0"),
+            (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(feature = "opendal-s3")]
     #[test]
     fn test_client_config_serde_round_trip() {

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")]