[emanuilov]

Which issue does this PR close?

• Closes feat(storage): write pipeline robustness and edge-case handling for object_store backend #3166.

What changes are included in this PR?

Problem. Every opendal operator is wrapped in RetryLayer::new(): 3 retries starting at 1 s, factor 2, no jitter. Nothing exposes it (#3263 made the per-operation timeout configurable, not the retry policy). High-throughput writers hit this. One example is a streaming writer into Amazon S3 Tables: a backfill flushing hundreds of partitions at once, or dozens of writers creating files in new tables in one table bucket. S3 answers 503 SlowDown while it scales the prefix up, which is expected behaviour, and AWS's guidance is to retry with exponential backoff and jitter. With the fixed policy:

• the error reaches the caller after about 7 s, usually before S3 has scaled, so the whole commit cycle fails and is redone, adding to the load that caused the throttling;
• without jitter, every upload throttled in the same burst retries at the same moments and is throttled again together.

The only workaround today is a custom StorageFactory that re-implements the opendal storage to change one layer.

Change. Two layers of retry keys, resolved per key:

• opendal.retry.num-retries, opendal.retry.min-wait-ms and opendal.retry.max-wait-ms live on OpenDalClientConfig, next to opendal.io-timeout-ms, and apply to every opendal backend.
• s3.retry.num-retries, s3.retry.min-wait-ms and s3.retry.max-wait-ms have the names and meaning of Iceberg Java's S3FileIO, so a catalog configuration written for Java works unchanged. They are defined in iceberg::io next to the other S3_* constants, parsed in the opendal s3 module, and only the S3 storage reads them.
• When the S3 storage is built (by OpenDalStorageFactory::S3 or the resolving factory), each s3.retry.* key that is set overrides the matching opendal.retry.* value. The order is s3.retry.*, then opendal.retry.*, then OpenDAL's default. Other backends don't parse or validate s3.retry.*.
• The values map onto opendal's RetryLayer (with_max_times, with_min_delay, with_max_delay). Setting any key also enables jitter: Java has no jitter key, and the AWS SDK always jitters.
• With no key set, the layer stays RetryLayer::new(), so existing users see no change.

Example, retrying S3 requests for up to about 2.5 minutes:

s3.retry.num-retries = 10
s3.retry.min-wait-ms = 2000
s3.retry.max-wait-ms = 20000

Are these changes tested?

Unit tests in iceberg-storage-opendal:

• no key set gives the same layer as RetryLayer::new();
• opendal.retry.* and s3.retry.* values (and num-retries = 0) reach the layer, with jitter on;
• for S3, s3.retry.* wins per key, and a key set at only one layer is taken from that layer;
• zero, negative, fractional and non-numeric values are rejected at both layers, with the key in the error;
• a non-S3 backend ignores s3.retry.*, even an invalid value, and still applies opendal.retry.*.

cargo fmt, clippy -D warnings and the tests of iceberg (io) and iceberg-storage-opendal pass. An earlier revision ran end to end against real S3 Tables through FileIO. Not tested against a real throttled store.

AI Disclosure

Drafted with Claude Code; I reviewed the change and ran fmt, clippy and the tests of the two crates. Points worth a look:

• waits must be positive, while num-retries may be 0;
• setting only some keys leaves the rest at OpenDAL's defaults (1 s / 60 s), not Java's (2 s / 20 s);
• min-wait-ms greater than max-wait-ms is not rejected (OpenDAL's backoff just caps at the maximum);
• the test that a non-S3 backend ignores s3.retry.* checks the serialized storage for the resolved value, since Arc<dyn Storage> can't be downcast;
• parse_io_timeout_ms is renamed parse_positive_ms, because it now also validates the waits.

[xanderbailey]

Onboard with adding more configuration options here but mild concerns about their placement. Let me know what you think!

[xanderbailey]

Is there a reason we would put these in the s3 namespace and not in the new opendal namespace? It seems like these could apply to all opendal backends?

[emanuilov]

I put them under s3. on purpose: they’re the same keys Iceberg Java’s S3FileIO uses (s3.retry.num-retries, s3.retry.min-wait-ms, s3.retry.max-wait-ms), so a catalog configuration written for Java works unchanged with iceberg-rust. They aren’t meant to be opendal-only: the object_store S3 backend (#3165) should read the same keys. To make that explicit, I can move the parsing into a small backend-neutral retry config in iceberg::io, which the opendal and object_store backends each turn into their own retry setup. Other schemes (GCS, Azure) could feed the same struct from their own keys later. Want me to do that in this PR?

[xanderbailey]

That makes sense, wanting java configured tables to work out of the box makes sense but I wonder if we could layer these configs better. i.e have both namespaces available with the more specific one taking preference? Does something like that make sense?

[emanuilov]

Just to check I understand: do you mean a generic set of retry keys in the opendal namespace that every OpenDAL backend uses, with s3.retry.* taking precedence for S3?

If so, here’s what I’d do:

• Add opendal.retry.num-retries, opendal.retry.min-wait-ms and opendal.retry.max-wait-ms to OpenDalClientConfig, next to opendal.io-timeout-ms, and build every backend’s RetryLayer from them.
• For S3, any s3.retry.* key that is set overrides the matching field, so the order is s3.retry., then opendal.retry., then OpenDAL’s defaults. Each key is resolved separately, so you can set one at either layer.
• Java-configured tables work unchanged. Anyone who wants the same retries on every backend sets the opendal.retry.* keys once.
• Tests for generic-only, S3-only, both set (S3 wins), a key set at only one layer, and a non-S3 backend ignoring s3.retry.*.

Is that what you had in mind? If so, I’ll add it to this PR.

[xanderbailey]

Yes! I think this makes sense to me!

[emanuilov]

Pushed the change with the layered config: opendal.retry.* for every backend, with s3.retry.* overriding per key for S3.

Tested locally:

• The four existing retry tests are unchanged. New tests cover generic keys applying to every backend, S3 keys winning per key over the generic ones, a key set at only one layer, and the new keys rejecting invalid values.
• Downstream: I built our Kafka → Iceberg (S3 Tables) sink against this branch (iceberg, iceberg-catalog-s3tables, iceberg-storage-opendal patched to it) and ran its end-to-end tests on a local kind cluster against real S3 Tables:
• with s3.retry.num-retries=10, s3.retry.min-wait-ms=2000, s3.retry.max-wait-ms=20000 and opendal.retry.num-retries=3 all set, the run committed every row exactly once;
• with an invalid opendal.retry.min-wait-ms=0, and separately s3.retry.min-wait-ms=0, the sink fails at startup with a property error naming the key, so both layers are read through the real FileIO path.

I didn’t reproduce real S3 throttling here, so which layer wins at runtime is covered by the unit tests rather than the e2e.

[xanderbailey]

These are S3-specific so I don't think they belong on OpenDalClientConfig

[xanderbailey]

If the S3 overrides are resolved into the client config when the S3 storage is built, this can drop the s3: bool flag and just build the layer from opendal.retry.* fields, which by then already hold the right values for each backend.

[xanderbailey]

With the overrides resolved at build time this per-variant match isn't needed; create_operator can call self.client_config().retry_layer() the same way it does for io_timeout().

[xanderbailey]

Thanks for picking up the layering so quickly! The precedence behaviour is what I was after, but I think I explained the placement badly: I didn't mean for the s3.retry.* keys to live on OpenDalClientConfig alongside the opendal.* ones. OpenDalClientConfig is documented as the backend-independent settings shared by every variant, and with the S3 fields on it every backend now parses (and validates) s3.retry.*, and they get serialized into client_config for GCS/Azure/memory etc. too.

What I had in mind:

• OpenDalClientConfig only knows about opendal.retry.* (plus opendal.io-timeout-ms).
• The s3.retry.* keys are parsed in s3.rs, next to the rest of the s3.* properties that s3_config_parse already handles.
• The S3 arm of OpenDalStorageFactory::build overlays those S3 values onto the client config per key, so the S3 variant's client_config holds the resolved policy.

Then retry_layer doesn't need to know which backend it's building for, the OpenDalStorage::retry_layer dispatch and s3_retry_layer go away, and adding GCS/Azure-specific overrides later follows the same pattern without growing the shared struct.

Also, the title and description still say only S3 applies retry settings and the other backends keep OpenDAL's default, which isn't true anymore with opendal.retry.*. Could you update them?

[emanuilov]

Thanks for picking up the layering so quickly! The precedence behaviour is what I was after, but I think I explained the placement badly: I didn't mean for the s3.retry.* keys to live on OpenDalClientConfig alongside the opendal.* ones. OpenDalClientConfig is documented as the backend-independent settings shared by every variant, and with the S3 fields on it every backend now parses (and validates) s3.retry.*, and they get serialized into client_config for GCS/Azure/memory etc. too.

What I had in mind:

• OpenDalClientConfig only knows about opendal.retry.* (plus opendal.io-timeout-ms).
• The s3.retry.* keys are parsed in s3.rs, next to the rest of the s3.* properties that s3_config_parse already handles.
• The S3 arm of OpenDalStorageFactory::build overlays those S3 values onto the client config per key, so the S3 variant's client_config holds the resolved policy.

Then retry_layer doesn't need to know which backend it's building for, the OpenDalStorage::retry_layer dispatch and s3_retry_layer go away, and adding GCS/Azure-specific overrides later follows the same pattern without growing the shared struct.

Also, the title and description still say only S3 applies retry settings and the other backends keep OpenDAL's default, which isn't true anymore with opendal.retry.*. Could you update them?

Thanks, that's clearer. Reworked it that way: OpenDalClientConfig now only has opendal.io-timeout-ms and opendal.retry.*. s3.retry.* is parsed in s3.rs, and the S3 arm of OpenDalStorageFactory::build writes those values over the client config per key, so the S3 variant's client_config holds the resolved policy. The resolving factory's S3 path does the same, otherwise it would silently ignore s3.retry.*. Other backends no longer parse, validate or serialize s3.retry.*; there's a test that an invalid s3.retry.* value doesn't break a memory storage. I also rebased onto main and updated the title and description.

Tested locally beyond the unit tests:

• file_io_s3_test passes against the docker-compose object store.
• End to end through FileIO, with a small proxy in front of the object store that answers 503 SlowDown a set number of times and records when each request arrives:
  • with no keys set, the write gives up after OpenDAL's default 3 retries (gaps of 1 s, 2 s, 4 s);
  • s3.retry.* alone and opendal.retry.* alone each let the write ride out 5 throttled attempts, and the measured gaps match the configured waits with jitter;
  • with both set, s3.retry.* wins per key: s3.retry.num-retries=2 beats opendal.retry.num-retries=10, and a retry count from s3.retry.* combines with waits from opendal.retry.*;
  • s3.retry.num-retries=0 makes one attempt;
  • the resolving factory (s3:// paths) reads both layers the same way;
  • invalid values at either layer fail with the key in the error, and a memory backend ignores an invalid s3.retry.* value.
• As a sanity check, dropping the S3 overlay from both factories makes every s3.retry.* case fail while the opendal.retry.* ones still pass.