[fs] Support IRSA with AssumeRole delegation tokens

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));

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),

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

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

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: <your-delegation-role-arn>
+```
+
+EC2 instance profiles are covered by the default chain and need no change.
 
 ### Active Segment Retention Rollout
 

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