AWS: avoid static global credentials provider which doesn't play well…

… with lifecycle management (#8677) (#8843)

Co-authored-by: Kristin Cowalcijk <[email protected]>

aws/src/main/java/org/apache/iceberg/aws/AwsClientFactories.java

@@ -226,7 +226,8 @@ static AwsCredentialsProvider credentialsProvider(
             AwsSessionCredentials.create(accessKeyId, secretAccessKey, sessionToken));
       }
     } else {
-      return DefaultCredentialsProvider.create();
+      // Create a new credential provider for each client
+      return DefaultCredentialsProvider.builder().build();
     }
   }
 }

aws/src/main/java/org/apache/iceberg/aws/AwsClientProperties.java

@@ -150,7 +150,8 @@ public AwsCredentialsProvider credentialsProvider(
       return credentialsProvider(this.clientCredentialsProvider);
     }
 
-    return DefaultCredentialsProvider.create();
+    // Create a new credential provider for each client
+    return DefaultCredentialsProvider.builder().build();
   }
 
   private AwsCredentialsProvider credentialsProvider(String credentialsProviderClass) {

aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java

@@ -485,7 +485,8 @@ private AwsCredentialsProvider credentialsProvider(
       return credentialsProvider(this.clientCredentialsProvider);
     }
 
-    return DefaultCredentialsProvider.create();
+    // Create a new credential provider for each client
+    return DefaultCredentialsProvider.builder().build();
   }
 
   private AwsCredentialsProvider credentialsProvider(String credentialsProviderClass) {

aws/src/test/java/org/apache/iceberg/aws/AwsClientPropertiesTest.java

@@ -61,6 +61,19 @@ public void testDefaultCredentialsConfiguration() {
         .isTrue();
   }
 
+  @Test
+  public void testCreatesNewInstanceOfDefaultCredentialsConfiguration() {
+    AwsClientProperties awsClientProperties = new AwsClientProperties();
+    AwsCredentialsProvider credentialsProvider =
+        awsClientProperties.credentialsProvider(null, null, null);
+    AwsCredentialsProvider credentialsProvider2 =
+        awsClientProperties.credentialsProvider(null, null, null);
+
+    Assertions.assertThat(credentialsProvider)
+        .withFailMessage("Should create a new instance in each call")
+        .isNotSameAs(credentialsProvider2);
+  }
+
   @Test
   public void testBasicCredentialsConfiguration() {
     AwsClientProperties awsClientProperties = new AwsClientProperties();