fix: Native S3 scan on EKS/IRSA turns a transient STS throttle into a hard 403 storm (#6025) (#6323)

* fix: Native S3 scan on EKS/IRSA turns a transient STS throttle into a hard 403 storm

* simplify and cleanup

* prettier

* address comments

* address review comments

* rework provider to use sdk's resolved config

* address more review comments

* address more comments

* address more review comments.

* address more comments

* resolve more issues

* more comments addressed

(cherry picked from commit 65a0cda)

Co-authored-by: Parth Chandra <parthc@apache.org>

Author: andygrove

docs/source/contributor-guide/s3-credential-provider-design.md

@@ -90,6 +90,21 @@ This is consistent with [Why no Comet-side cache](#why-no-comet-side-cache): the
 
 The dispatcher returns `null` for a provider that does not implement the interface without calling it, and Comet builds the same plain store as before. The Iceberg path does not use locations. Operations other than reads route by path without the retry, since Comet only reads through these stores.
 
+## The IRSA web-identity provider is the exception that does cache
+
+The "no Comet-side cache" rule above is about the vendor _bridge_. There is one Comet-owned credential provider that deliberately does cache: the EKS/IRSA web-identity provider in `native/core/src/cloud/s3/web_identity.rs`. It is not a vendor path -- there is no JVM SPI involved -- so the reasoning above does not apply.
+
+It exists because the implicit IRSA credential path on the **Iceberg scan** mishandles STS throttling. `AssumeRoleWithWebIdentity` is resolved per reader thread; a concurrent startup burst throttles STS; opendal's default reqsign chain does not retry the throttle and falls through to the EKS node instance role, which lacks bucket access, turning a transient throttle into a hard `403` (this is the reported failure). The raw-Parquet path is intentionally left alone: its AWS SDK default chain already retries and stops on a provider error rather than downgrading, and — because `NativeConfig.extractObjectStoreOptions` forwards Hadoop's `core-default.xml` default for `fs.s3a.aws.credentials.provider` — a Parquet take-over gated on "no provider configured" could never engage anyway. So the take-over is Iceberg-only, and covers both reads and writes (both build their `FileIO` through `load_file_io`): when both `AWS_WEB_IDENTITY_TOKEN_FILE` and `AWS_ROLE_ARN` are set, a region is set, and the catalog configures no explicit credentials, `iceberg_common.rs::build_s3_credential_loader` installs `WebIdentityCredentialProvider` (via a `CustomAwsCredentialLoader`) instead of returning `Ok(None)`. It:
+
+- builds its STS client from the AWS SDK's fully-resolved `SdkConfig` (`aws_config::defaults(...).load()`) with a raised `RetryConfig`, and calls `AssumeRoleWithWebIdentity` on it; because the client comes from the resolved config rather than a hand-assembled one, it honors region, `use_fips`, `use_dual_stack` and any profile/custom STS endpoint with no per-setting copying to keep in sync,
+- reports the credential's real STS expiry (not an early deadline). reqsign's signer refuses to sign within ~10s of the reported expiry and reloads within 120s of it, so an early deadline would open a signing dead zone before every refresh; instead our own cache margin (`min_ttl`, floored to 120s) refreshes at or before the signer's reload point, and a still-valid credential keeps being served if a refresh is throttled,
+- only ever calls `AssumeRoleWithWebIdentity` -- no credential chain, no IMDS/instance-role fallback -- so a throttle that outlasts the retries errors instead of downgrading, and
+- caches one credential per process keyed by identity `(role_arn, token_file, region)` plus the resolved settings (so a catalog's own knobs are honored regardless of init order) and single-flights refreshes, so a startup burst makes one STS call per executor rather than one per reader thread; a failed refresh is briefly remembered so a throttled burst also costs one call rather than one per reader.
+
+Unlike the bridge, this provider owns the whole credential lifecycle (there is no vendor to delegate to), which is why caching lives here. Config keys are catalog properties under the `s3.` prefix (`s3.comet.credential.webIdentity.*`), the same route as `s3.comet.credential.provider.class`; see the user guide "EKS / IRSA" section. It stands aside for any credential source opendal's default chain ranks ahead of web-identity -- a bridge class, catalog static keys / `client.assume-role.arn`, static `AWS_ACCESS_KEY_ID` / `AWS_SECRET_ACCESS_KEY`, or a configured profile (`AWS_PROFILE`, or a shared credentials / config file) -- and when no region is set (a web-identity STS client with no region fails opaquely, while the default chain has a no-region global fallback). Deferring on a present profile/config file also covers a profile-configured custom STS `endpoint_url`. So it only changes the otherwise-default behavior.
+
+STS endpoint selection (`configure_sts_endpoint`) overrides the SDK's own endpoint resolution only in the one narrow case where doing so is safe and matches reqsign, and defers to the SDK for everything else. It overrides to the global `sts.amazonaws.com` (signed `us-east-1`) only when the region is in the commercial partition, FIPS is off, dual-stack is off, no custom/profile STS `endpoint_url` is set, and `AWS_STS_REGIONAL_ENDPOINTS` is `legacy` or unset. That is the case the old reqsign chain hit, so an executor whose egress only reaches the global endpoint keeps working after the take-over. Every other combination keeps the SDK's regional resolution: FIPS (`AWS_USE_FIPS_ENDPOINT` / profile) uses the regional FIPS endpoint because there is no global FIPS STS endpoint (an accompanying `AWS_STS_REGIONAL_ENDPOINTS=legacy` is ignored with a one-time warning); dual-stack (`AWS_USE_DUALSTACK_ENDPOINT`) uses the dual-stack regional endpoint, because the SDK rejects a custom endpoint combined with dual-stack before it sends anything; a custom/VPC/profile `endpoint_url` is honored rather than silently replaced; `AWS_STS_REGIONAL_ENDPOINTS=regional` stays regional; and the China, GovCloud, ISO and EUSC partitions (and any region whose prefix Comet does not recognize) keep their own regional hosts (none of them has a `sts.amazonaws.com` global endpoint). The commercial-partition check is an allowlist of known commercial region prefixes, so an unrecognized prefix from a future partition defers to the SDK rather than being sent to the global endpoint.
+
 ## Path-specific behavior
 
 `object_store::CredentialProvider` and `reqsign_core::ProvideCredential` differ in what they consume:

docs/source/user-guide/latest/s3-credential-providers.md

@@ -100,6 +100,40 @@ Without the config set, no credential-related log lines appear at startup; nativ
 
 **Credentials silently going stale during long-running jobs.** When a vendor returns `expirationEpochMillis=0`, the bridge substitutes a 5-minute expiry before handing the credential to `opendal`, so `opendal`'s cache cannot hold a stale credential indefinitely. Returning a real expiry is preferred; the 5-minute fallback is a safety net, not a knob.
 
+## EKS / IRSA: STS throttling protection (native Iceberg reads and writes)
+
+This is automatic; there is nothing to configure to get the protection, and it does not involve a bridge class. It applies to native Iceberg **reads and writes** (both build their `FileIO` through the same path). The raw-Parquet path is unaffected: it uses the AWS SDK default chain, which already retries and stops on a provider error rather than downgrading.
+
+On EKS with [IAM Roles for Service Accounts (IRSA)](https://docs.aws.amazon.com/eks/latest/userguide/iam-roles-for-service-accounts.html), executors assume the application role by calling STS `AssumeRoleWithWebIdentity`. Under a large concurrent startup burst (many executors times many cores, all fetching credentials at once), STS can throttle that call. On the Iceberg path, opendal's default credential chain does not retry the throttle and falls through to the EKS node instance role, which usually lacks bucket access, so every native read then fails with a hard `403 AccessDenied` even though the throttle was transient.
+
+When Comet detects IRSA (both `AWS_WEB_IDENTITY_TOKEN_FILE` and `AWS_ROLE_ARN` are set), a region is set, and the catalog configures no explicit credentials, the native Iceberg reader resolves web-identity credentials itself instead of using opendal's default chain. It builds the STS client from the AWS SDK's fully-resolved config, so region, `AWS_USE_FIPS_ENDPOINT`, `AWS_USE_DUALSTACK_ENDPOINT`, and any profile/custom STS endpoint are honored. The provider:
+
+- retries the throttled `AssumeRoleWithWebIdentity` call with the AWS SDK's exponential backoff and jitter (a throttle that outlasts the retries becomes an error that Spark's task retry then picks up),
+- never falls back to the node instance role, and
+- caches one assumed-role credential per executor process, shared across all reader threads and scans, so a startup burst makes one STS call per executor rather than one per thread. If a refresh is throttled while the current credential is still valid, it keeps serving that credential.
+
+**STS endpoint selection** uses the global `sts.amazonaws.com` endpoint only in the plain commercial case, and leaves everything else to the AWS SDK. The global endpoint is used when your region is in the standard commercial partition, FIPS and dual-stack are both off, no custom STS endpoint is set, and `AWS_STS_REGIONAL_ENDPOINTS` is `legacy` or unset. That matches the previous behavior, so a network that only reaches the global endpoint keeps working. In every other case the SDK's own regional endpoint is used: FIPS (`AWS_USE_FIPS_ENDPOINT`) always uses the regional FIPS endpoint since there is no global FIPS STS endpoint (an accompanying `AWS_STS_REGIONAL_ENDPOINTS=legacy` is ignored and a warning is logged), dual-stack (`AWS_USE_DUALSTACK_ENDPOINT`) uses the dual-stack regional endpoint, a custom STS endpoint (`AWS_ENDPOINT_URL_STS` or a profile) is honored, `AWS_STS_REGIONAL_ENDPOINTS=regional` stays regional, and the China, GovCloud, ISO and EUSC regions (and any region Comet does not recognize as commercial) use their own regional endpoints.
+
+It stands aside whenever a higher-precedence credential source is configured -- a Comet bridge class or catalog static keys / `client.assume-role.arn`, static credentials in the environment (`AWS_ACCESS_KEY_ID` + `AWS_SECRET_ACCESS_KEY`), or a configured profile (`AWS_PROFILE`, or a shared credentials / config file such as `~/.aws/credentials` or `~/.aws/config`) -- and when no region is set (`AWS_REGION` / `AWS_DEFAULT_REGION`). These all rank ahead of web-identity in opendal's default chain or need its no-region fallback, so the take-over only changes the otherwise-default behavior and never switches away from an identity -- or a profile-configured STS endpoint -- you set explicitly. On an EKS/IRSA pod none of these apply, so the take-over still engages there.
+
+Tuning is rarely needed. Set these as catalog properties under the `s3.` prefix (the same namespace as the credential-provider SPI key):
+
+| Property (per Iceberg catalog)                         | Default | Meaning                                                                               |
+| ------------------------------------------------------ | ------- | ------------------------------------------------------------------------------------- |
+| `s3.comet.credential.webIdentity.enabled`              | `true`  | Set `false` to opt out and use opendal's default chain.                               |
+| `s3.comet.credential.webIdentity.maxAttempts`          | `5`     | STS attempts before the assume-role call is treated as failed.                        |
+| `s3.comet.credential.webIdentity.minTtlSeconds`        | `300`   | Refresh this many seconds before expiry (floored to 120s).                            |
+| `s3.comet.credential.webIdentity.refreshJitterSeconds` | `60`    | Random slack added on top of `minTtlSeconds` so executors do not all refresh at once. |
+
+For example, to opt out of the take-over or raise the retry count for one catalog:
+
+```
+spark.sql.catalog.<catalog>.s3.comet.credential.webIdentity.enabled=false
+spark.sql.catalog.<catalog>.s3.comet.credential.webIdentity.maxAttempts=8
+```
+
+Use the `s3.` prefix. Comet forwards the unfiltered catalog property bag, so a bare, unprefixed key does technically reach the native reader, but the settings are looked up under `s3.` (matching the credential-provider SPI key), so only the prefixed spelling takes effect.
+
 ## Iceberg: explicit S3 region required
 
 `iceberg-storage-opendal` does not auto-detect a bucket's region, with or without the bridge. When neither the catalog (`s3.region` or `client.region`) nor the executor environment (`AWS_REGION` / `AWS_DEFAULT_REGION`) supplies a region, Comet uses `us-east-1`. That suits most non-AWS S3-compatible services but fails for AWS buckets in other regions, so set the region (and the endpoint for non-AWS) explicitly on the Spark catalog:

native/Cargo.lock

@@ -64,6 +64,9 @@ object_store = { version = "0.13.2", features = ["gcp", "azure", "aws", "http"]
 url = "2.2"
 aws-config = "1.8.18"
 aws-credential-types = "1.2.13"
+# default-features off: the STS client is built from a fully-resolved SdkConfig, so it needs none
+# of aws-sdk-sts's own transport/crypto defaults, and enabling them would pull extra crates.
+aws-sdk-sts = { version = "1.114.0", default-features = false }
 # Pinned to a revision rather than a release because no published iceberg-rust
 # version is on our arrow: 0.10.1 requires arrow/parquet ^58. Move to a plain
 # version requirement once a release ships on arrow 59. Bump policy: #5645.

native/Cargo.toml

@@ -68,6 +68,9 @@ object_store = { workspace = true }
 url = { workspace = true }
 aws-config = { workspace = true }
 aws-credential-types = { workspace = true }
+aws-sdk-sts = { workspace = true }
+# Only for the `SharedHttpClient` type in `build_provider`'s test seam; already in the graph.
+aws-smithy-runtime-api = "1.16.0"
 parking_lot = "0.12.5"
 # Optional Delta Lake contrib (enabled by the `contrib-delta` feature). Source lives
 # under `contrib/delta/native/` so non-Delta committers can ignore it.
@@ -102,6 +105,12 @@ lazy_static = "1.4"
 assertables = "10"
 hex = "0.4.3"
 datafusion-functions-nested = { version = "55.1.0" }
+# Test-only helpers for a hand-rolled in-memory HTTP client that exercises the web-identity STS
+# retry path without a socket, plus reqsign's SigV4 signer for the signing round-trip test. Already
+# in the dependency graph (transitive via aws-config and iceberg-storage-opendal).
+aws-smithy-types = "1.6.3"
+http = "1.5.0"
+reqsign-aws-v4 = "3"
 
 [features]
 backtrace = ["datafusion/backtrace"]

native/core/Cargo.toml

@@ -42,8 +42,9 @@ use std::sync::Arc;
 use std::time::Duration;
 
 /// Cap on opendal's credential cache when the provider does not report an expiry. Prevents the
-/// executor from holding a stale credential for the entire job lifetime.
-const DEFAULT_EXPIRY_WHEN_UNKNOWN: Duration = Duration::from_secs(300);
+/// executor from holding a stale credential for the entire job lifetime. Shared with the IRSA
+/// web-identity provider (`super::web_identity`).
+pub(crate) const DEFAULT_EXPIRY_WHEN_UNKNOWN: Duration = Duration::from_secs(300);
 
 /// Once-per-process latch for the "missing expiry" warning. Bridges are per-scan, so a per-bridge
 /// latch would re-log on every scan.

native/core/src/cloud/s3/credential_bridge.rs

@@ -16,3 +16,4 @@
 // under the License.
 
 pub mod credential_bridge;
+pub mod web_identity;