From ccd2beeef2603a153ca59486cfe1d8ccfc054e78 Mon Sep 17 00:00:00 2001 From: Anton Borisov Date: Fri, 18 Sep 2026 16:05:07 +0100 Subject: [PATCH] [fs] Support IRSA with AssumeRole delegation tokens --- .../fluss/fs/s3/S3FileSystemPlugin.java | 4 --- .../s3/token/S3DelegationTokenProvider.java | 10 +++--- .../fluss/fs/s3/S3FileSystemPluginTest.java | 15 ++++---- .../token/S3DelegationTokenProviderTest.java | 36 +++++++++++++++---- .../operations/upgrade-notes-1.0.md | 11 +++++- .../tiered-storage/filesystems/s3.md | 12 +++++-- 6 files changed, 64 insertions(+), 24 deletions(-) diff --git a/fluss-filesystems/fluss-fs-s3/src/main/java/org/apache/fluss/fs/s3/S3FileSystemPlugin.java b/fluss-filesystems/fluss-fs-s3/src/main/java/org/apache/fluss/fs/s3/S3FileSystemPlugin.java index ea4785c3506..df0e6ea0689 100644 --- a/fluss-filesystems/fluss-fs-s3/src/main/java/org/apache/fluss/fs/s3/S3FileSystemPlugin.java +++ b/fluss-filesystems/fluss-fs-s3/src/main/java/org/apache/fluss/fs/s3/S3FileSystemPlugin.java @@ -177,10 +177,6 @@ private void setCredentialProvider(org.apache.hadoop.conf.Configuration hadoopCo boolean hasRoleArn = hadoopConfig.get(ROLE_ARN_KEY) != null; if (hasCredentialProvider) { - if (hasRoleArn) { - throw new IllegalArgumentException( - "AssumeRole and a custom AWS credentials provider cannot be configured together."); - } LOG.info( "Using configured AWS credential provider(s) for server-side S3 access: {}", hadoopConfig.get(PROVIDER_CONFIG_NAME)); diff --git a/fluss-filesystems/fluss-fs-s3/src/main/java/org/apache/fluss/fs/s3/token/S3DelegationTokenProvider.java b/fluss-filesystems/fluss-fs-s3/src/main/java/org/apache/fluss/fs/s3/token/S3DelegationTokenProvider.java index 9525ca0e244..c8c065d604b 100644 --- a/fluss-filesystems/fluss-fs-s3/src/main/java/org/apache/fluss/fs/s3/token/S3DelegationTokenProvider.java +++ b/fluss-filesystems/fluss-fs-s3/src/main/java/org/apache/fluss/fs/s3/token/S3DelegationTokenProvider.java @@ -94,10 +94,6 @@ public S3DelegationTokenProvider(String scheme, Configuration conf) throws IOExc checkArgument( (accessKey == null) == (secretKey == null), "S3 access key and secret key must both be set or both be unset."); - if (hasCredentialProvider && roleArn != null) { - throw new IllegalArgumentException( - "AssumeRole and a custom AWS credentials provider cannot be configured together."); - } if (hasCredentialProvider) { checkArgument( !Arrays.asList(conf.getTrimmedStrings(AWS_CREDENTIALS_PROVIDER)) @@ -182,6 +178,12 @@ private StsClient buildStsClient() { @Nullable AwsCredentialsProvider createStsCredentialsProvider() { if (credentialProviderList != null) { + if (roleArn != null) { + // Not the list itself: StsClient.close() would close it, breaking + // every later token refresh. + LOG.info("Using configured AWS credentials provider as the AssumeRole caller."); + return () -> credentialProviderList.resolveCredentials(); + } AwsCredentials credentials = credentialProviderList.resolveCredentials(); checkArgument( !(credentials instanceof AwsSessionCredentials), diff --git a/fluss-filesystems/fluss-fs-s3/src/test/java/org/apache/fluss/fs/s3/S3FileSystemPluginTest.java b/fluss-filesystems/fluss-fs-s3/src/test/java/org/apache/fluss/fs/s3/S3FileSystemPluginTest.java index fbbe66f80c8..3d48d4aa863 100644 --- a/fluss-filesystems/fluss-fs-s3/src/test/java/org/apache/fluss/fs/s3/S3FileSystemPluginTest.java +++ b/fluss-filesystems/fluss-fs-s3/src/test/java/org/apache/fluss/fs/s3/S3FileSystemPluginTest.java @@ -32,7 +32,6 @@ import java.util.Collections; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Tests for server/client detection in {@link S3FileSystemPlugin}. */ class S3FileSystemPluginTest { @@ -111,7 +110,7 @@ void testServerModeWithConfiguredCredentialProviderForS3A() { } @Test - void testConfiguredCredentialProviderWithRoleArnThrows() { + void testConfiguredCredentialProviderWithRoleArnIsAccepted() { Configuration flussConfig = new Configuration(); flussConfig.setString( PROVIDER_CONFIG, @@ -121,10 +120,14 @@ void testConfiguredCredentialProviderWithRoleArnThrows() { S3FileSystemPlugin plugin = new S3FileSystemPlugin(); - assertThatThrownBy(() -> plugin.buildHadoopConfiguration(flussConfig)) - .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining("AssumeRole") - .hasMessageContaining("custom AWS credentials provider"); + org.apache.hadoop.conf.Configuration hadoopConfig = + plugin.buildHadoopConfiguration(flussConfig); + assertThat(hadoopConfig.get(PROVIDER_CONFIG)) + .isEqualTo( + S3DelegationTokenProviderTest.RefreshableCredentialsProvider.class + .getName()); + assertThat(hadoopConfig.get("fs.s3a.assumed.role.arn")) + .isEqualTo("arn:aws:iam::123456789012:role/test-role"); } @Test diff --git a/fluss-filesystems/fluss-fs-s3/src/test/java/org/apache/fluss/fs/s3/token/S3DelegationTokenProviderTest.java b/fluss-filesystems/fluss-fs-s3/src/test/java/org/apache/fluss/fs/s3/token/S3DelegationTokenProviderTest.java index b19d309ad32..34988126a6e 100644 --- a/fluss-filesystems/fluss-fs-s3/src/test/java/org/apache/fluss/fs/s3/token/S3DelegationTokenProviderTest.java +++ b/fluss-filesystems/fluss-fs-s3/src/test/java/org/apache/fluss/fs/s3/token/S3DelegationTokenProviderTest.java @@ -23,6 +23,8 @@ import software.amazon.awssdk.auth.credentials.AwsCredentials; import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; import software.amazon.awssdk.auth.credentials.AwsSessionCredentials; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.sts.StsClient; import java.io.IOException; import java.net.URI; @@ -98,16 +100,38 @@ void testConfiguredProviderRequiresRegion() { } @Test - void testConfiguredProviderWithRoleArnThrows() { + void testConfiguredProviderWithRoleArnCallsAssumeRoleAsTheProvider() throws IOException { Configuration conf = new Configuration(); conf.set("fs.s3a.region", "us-east-1"); - setConfiguredProvider(conf, RefreshableCredentialsProvider.class); + setConfiguredProvider(conf, SessionCredentialsProvider.class); conf.set("fs.s3a.assumed.role.arn", "arn:aws:iam::123456789012:role/test-role"); - assertThatThrownBy(() -> new S3DelegationTokenProvider("s3", conf)) - .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining("AssumeRole") - .hasMessageContaining("custom AWS credentials provider"); + S3DelegationTokenProvider provider = new S3DelegationTokenProvider("s3", conf); + + assertThat(provider.createStsCredentialsProvider().resolveCredentials()) + .isInstanceOf(AwsSessionCredentials.class); + } + + @Test + void testStsClientCloseDoesNotInvalidateTheConfiguredProvider() throws IOException { + Configuration conf = new Configuration(); + conf.set("fs.s3a.region", "us-east-1"); + setConfiguredProvider(conf, SessionCredentialsProvider.class); + conf.set("fs.s3a.assumed.role.arn", "arn:aws:iam::123456789012:role/test-role"); + + S3DelegationTokenProvider provider = new S3DelegationTokenProvider("s3", conf); + AwsCredentialsProvider stsCaller = provider.createStsCredentialsProvider(); + + try (StsClient client = + StsClient.builder() + .region(Region.of("us-east-1")) + .credentialsProvider(stsCaller) + .build()) { + assertThat(client).isNotNull(); + } + + assertThat(provider.createStsCredentialsProvider().resolveCredentials()) + .isInstanceOf(AwsSessionCredentials.class); } @Test diff --git a/website/docs/maintenance/operations/upgrade-notes-1.0.md b/website/docs/maintenance/operations/upgrade-notes-1.0.md index 9076b7e7ace..af64b529285 100644 --- a/website/docs/maintenance/operations/upgrade-notes-1.0.md +++ b/website/docs/maintenance/operations/upgrade-notes-1.0.md @@ -32,7 +32,16 @@ If `s3.aws.credentials.provider`, `s3a.aws.credentials.provider`, or `fs.s3a.aws Custom credentials providers must now implement `software.amazon.awssdk.auth.credentials.AwsCredentialsProvider` instead of `com.amazonaws.auth.AWSCredentialsProvider`. Implementations should provide credentials through `resolveCredentials()` rather than the SDK v1 `getCredentials()` and `refresh()` methods. -Deployments using static access keys or the default AWS credentials provider chain do not require configuration changes. +Deployments using static access keys do not require configuration changes. + +Deployments on EKS that rely on IRSA do. Hadoop S3A's default provider chain contains no web-identity provider, so the server's own S3 access falls through to the EC2 instance profile instead. Set the provider explicitly, alongside the delegation role: + +```yaml +s3.aws.credentials.provider: software.amazon.awssdk.auth.credentials.WebIdentityTokenFileCredentialsProvider +s3.assumed.role.arn: +``` + +EC2 instance profiles are covered by the default chain and need no change. ### Active Segment Retention Rollout diff --git a/website/docs/maintenance/tiered-storage/filesystems/s3.md b/website/docs/maintenance/tiered-storage/filesystems/s3.md index 2ffeca3cba0..a272d74e17d 100644 --- a/website/docs/maintenance/tiered-storage/filesystems/s3.md +++ b/website/docs/maintenance/tiered-storage/filesystems/s3.md @@ -98,7 +98,13 @@ Without `s3.assumed.role.arn`, Fluss falls back to `GetSessionToken` (the defaul When running Fluss on Kubernetes with [IAM Roles for Service Accounts (IRSA)](https://docs.aws.amazon.com/eks/latest/userguide/iam-roles-for-service-accounts.html) or on EC2 with instance profiles, you can omit `s3.access-key` and `s3.secret-key`. The server will authenticate using the default AWS credential chain. :::note -IRSA and instance profiles return session credentials, so they must use this default chain — not the explicit [Custom AWS Credentials Provider](#custom-aws-credentials-provider) mode, which requires long-term credentials. +Hadoop S3A's default provider chain contains no web-identity provider, so it does **not** pick up IRSA for the server's own S3 access — it falls through to the EC2 instance profile. On EKS, set the provider explicitly: + +```yaml +s3.aws.credentials.provider: software.amazon.awssdk.auth.credentials.WebIdentityTokenFileCredentialsProvider +``` + +EC2 instance profiles are covered by the default chain and need no provider. ::: In this mode, `s3.assumed.role.arn` is required — the server uses `AssumeRole` to generate temporary credentials for clients (Flink/Spark connectors) that read tiered data from S3. @@ -129,8 +135,8 @@ s3.aws.credentials.provider: software.amazon.awssdk.auth.credentials.ProfileCred | Configuration | Description | |---|---| -| `s3.aws.credentials.provider` | Fully qualified class name of an AWS SDK V2 / Hadoop S3A credentials provider implementing `AwsCredentialsProvider`. It is used both for the server's own S3 access and for resolving the credentials used to mint temporary client credentials via `GetSessionToken`. Takes precedence over `s3.access-key`/`s3.secret-key` when both are configured. | +| `s3.aws.credentials.provider` | Fully qualified class name of an AWS SDK V2 / Hadoop S3A credentials provider implementing `AwsCredentialsProvider`. It supplies the server's own S3 access, and it is also the identity Fluss uses to mint temporary client credentials: with `s3.assumed.role.arn` set it is the `AssumeRole` caller, and without it the credentials passed to `GetSessionToken`. Takes precedence over `s3.access-key`/`s3.secret-key` when both are configured. | :::note -This mode only supports providers that return **long-term** access-key credentials, and it cannot be combined with `s3.assumed.role.arn`. Providers that return session credentials — including IRSA and EC2 instance profiles — must use the [default AWS credential chain](#default-aws-credential-chain-irsa-instance-profiles) instead. +`GetSessionToken` cannot be called with temporary credentials, so **without** `s3.assumed.role.arn` this mode requires a provider that returns long-term access keys. Providers that return session credentials, such as IRSA, must set `s3.assumed.role.arn` as well: `AssumeRole` accepts a caller holding temporary credentials, so the provider serves the data path while the role mints client credentials. :::