Small fixups

Author: michaelkoepf

fluss-filesystems/fluss-fs-s3/src/main/java/org/apache/fluss/fs/s3/S3ConfigOptions.java

@@ -19,5 +19,7 @@
 
 /** S3 configuration options that are explicitly read by name. */
 public final class S3ConfigOptions {
-    public static final String PROVIDER_CONFIG_NAME = "fs.s3a.aws.credentials.provider";
+    public static final String CREDENTIALS_PROVIDER_CONFIG_NAME = "fs.s3a.aws.credentials.provider";
+    public static final String ASSUMED_ROLE_CREDENTIALS_PROVIDER_CONFIG_NAME =
+            "fs.s3a.assumed.role.credentials.provider";
 }

fluss-filesystems/fluss-fs-s3/src/main/java/org/apache/fluss/fs/s3/S3FileSystemPlugin.java

@@ -28,6 +28,7 @@
 import org.apache.fluss.fs.s3.token.S3ADelegationTokenReceiver;
 import org.apache.fluss.fs.s3.token.S3DelegationTokenProvider;
 import org.apache.fluss.fs.s3.token.S3DelegationTokenReceiver;
+
 import org.apache.hadoop.fs.s3a.S3AFileSystem;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -73,7 +74,8 @@ public class S3FileSystemPlugin implements FileSystemPlugin {
                             "access.key",
                             "secret-key",
                             "secret.key",
-                            "aws.credentials.provider"));
+                            "aws.credentials.provider",
+                            "assumed.role.credentials.provider"));
 
     private static final List<String> TOKEN_DELEGATION_DEFAULT_CREDENTIAL_PROVIDER =
             Collections.singletonList("org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider");
@@ -143,7 +145,8 @@ org.apache.hadoop.conf.Configuration getHadoopConfiguration(Configuration flussC
         org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();
 
         // Credential providers must be explicitly set by the user
-        conf.set(S3ConfigOptions.PROVIDER_CONFIG_NAME, "");
+        conf.set(S3ConfigOptions.CREDENTIALS_PROVIDER_CONFIG_NAME, "");
+        conf.set(S3ConfigOptions.ASSUMED_ROLE_CREDENTIALS_PROVIDER_CONFIG_NAME, "");
 
         if (flussConfig == null) {
             return conf;

fluss-filesystems/fluss-fs-s3/src/main/java/org/apache/fluss/fs/s3/token/S3DelegationTokenReceiver.java

@@ -49,7 +49,7 @@ public static void updateHadoopConfigCredentialProviders(
             org.apache.hadoop.conf.Configuration hadoopConfig, List<String> credentialProvider) {
         LOG.info("Updating credential providers in Hadoop configuration");
 
-        String providers = hadoopConfig.get(S3ConfigOptions.PROVIDER_CONFIG_NAME, "");
+        String providers = hadoopConfig.get(S3ConfigOptions.CREDENTIALS_PROVIDER_CONFIG_NAME, "");
         List<String> credentialProviderPrependOrder = new ArrayList<>(credentialProvider);
         Collections.reverse(credentialProviderPrependOrder);
 
@@ -62,7 +62,7 @@ public static void updateHadoopConfigCredentialProviders(
                     providers = credentialProviderName + "," + providers;
                     LOG.debug("Prepending provider, new providers value: {}", providers);
                 }
-                hadoopConfig.set(S3ConfigOptions.PROVIDER_CONFIG_NAME, providers);
+                hadoopConfig.set(S3ConfigOptions.CREDENTIALS_PROVIDER_CONFIG_NAME, providers);
             } else {
                 LOG.debug("Provider {} already exists in chain", credentialProviderName);
             }