[AWS2] Bump AWS v2 SDK (closes #26323) (#26380)

buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy

@@ -520,7 +520,7 @@ class BeamModulePlugin implements Plugin<Project> {
     def autovalue_version = "1.9"
     def autoservice_version = "1.0.1"
     def aws_java_sdk_version = "1.12.135"
-    def aws_java_sdk2_version = "2.17.127"
+    def aws_java_sdk2_version = "2.20.47"
     def cassandra_driver_version = "3.10.2"
     def cdap_version = "6.5.1"
     def checkerframework_version = "3.27.0"

sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/options/AwsModule.java

@@ -49,7 +49,6 @@
 import org.apache.beam.sdk.annotations.Experimental.Kind;
 import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableSet;
 import org.checkerframework.checker.nullness.qual.NonNull;
-import org.slf4j.LoggerFactory;
 import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider;
 import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
 import software.amazon.awssdk.auth.credentials.AwsCredentials;
@@ -253,17 +252,6 @@ public void serializeWithType(
         if (profileName != null && !profileName.equals(envProfileName)) {
           jsonGenerator.writeStringField(PROFILE_NAME, profileName);
         }
-        try {
-          Exception exception = (Exception) readField(credentialsProvider, "loadException");
-          if (exception != null) {
-            LoggerFactory.getLogger(AwsModule.class)
-                .warn("Serialized ProfileCredentialsProvider in faulty state.", exception);
-          }
-        } catch (RuntimeException e) {
-          LoggerFactory.getLogger(AwsModule.class)
-              .warn("Failed to check ProfileCredentialsProvider for loadException.", e);
-        }
-
       } else if (providerClass.equals(StsAssumeRoleCredentialsProvider.class)) {
         Supplier<AssumeRoleRequest> reqSupplier =
             (Supplier<AssumeRoleRequest>)

sdks/java/io/amazon-web-services2/src/test/java/org/apache/beam/sdk/io/aws2/options/AwsModuleTest.java

@@ -20,6 +20,7 @@
 import static org.apache.beam.repackaged.core.org.apache.commons.lang3.reflect.FieldUtils.readField;
 import static org.apache.beam.sdk.io.aws2.options.SerializationTestUtil.serialize;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.hamcrest.Matchers.hasItem;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.junit.Assert.assertEquals;
@@ -61,6 +62,7 @@
 import software.amazon.awssdk.auth.credentials.ProfileCredentialsProvider;
 import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
 import software.amazon.awssdk.auth.credentials.SystemPropertyCredentialsProvider;
+import software.amazon.awssdk.core.exception.SdkClientException;
 import software.amazon.awssdk.http.apache.ProxyConfiguration;
 import software.amazon.awssdk.regions.Region;
 import software.amazon.awssdk.services.sts.StsClient;
@@ -147,11 +149,13 @@ public void testProfileCredentialsProviderSerDeWithDefaultProfile() throws Excep
           assertThat(serializedProvider).isEqualTo("{\"@type\":\"ProfileCredentialsProvider\"}");
 
           AwsCredentialsProvider actual = deserialize(serializedProvider);
-          assertThat(actual.resolveCredentials())
-              .isEqualToComparingFieldByField(DEFAULT_CREDENTIALS);
-          return assertThat(actual)
+          // compare providers before loading credentials (triggering lazy initialization)
+          assertThat(actual)
               .isExactlyInstanceOf(ProfileCredentialsProvider.class)
               .isEqualToComparingFieldByFieldRecursively(provider);
+
+          return assertThat(actual.resolveCredentials())
+              .isEqualToComparingFieldByField(DEFAULT_CREDENTIALS);
         });
   }
 
@@ -167,10 +171,13 @@ public void testProfileCredentialsProviderSerDeWithCustomProfile() throws Except
               .isEqualTo("{\"@type\":\"ProfileCredentialsProvider\",\"profileName\":\"other\"}");
 
           AwsCredentialsProvider actual = deserialize(serializedProvider);
-          assertThat(actual.resolveCredentials()).isEqualToComparingFieldByField(OTHER_CREDENTIALS);
-          return assertThat(actual)
+          // compare providers before loading credentials (triggering lazy initialization)
+          assertThat(actual)
               .isExactlyInstanceOf(ProfileCredentialsProvider.class)
               .isEqualToComparingFieldByFieldRecursively(provider);
+
+          return assertThat(actual.resolveCredentials())
+              .isEqualToComparingFieldByField(OTHER_CREDENTIALS);
         });
   }
 
@@ -185,11 +192,13 @@ public void testProfileCredentialsProviderSerDeWithCustomDefaultProfile() throws
           assertThat(serializedProvider).isEqualTo("{\"@type\":\"ProfileCredentialsProvider\"}");
 
           AwsCredentialsProvider actual = deserialize(serializedProvider);
-          assertThat(actual.resolveCredentials())
-              .isEqualToComparingFieldByFieldRecursively(OTHER_CREDENTIALS);
-          return assertThat(actual)
+          // compare providers before loading credentials (triggering lazy initialization)
+          assertThat(actual)
               .isExactlyInstanceOf(ProfileCredentialsProvider.class)
               .isEqualToComparingFieldByFieldRecursively(provider);
+
+          return assertThat(actual.resolveCredentials())
+              .isEqualToComparingFieldByFieldRecursively(OTHER_CREDENTIALS);
         });
   }
 
@@ -201,16 +210,20 @@ public void testProfileCredentialsProviderSerDeWithUnknownProfile() throws Excep
           AwsCredentialsProvider provider = ProfileCredentialsProvider.create("unknown");
           String serializedProvider = serialize(provider);
 
-          // ProfileCredentialsProvider SILENTLY drops unknown profiles
-          assertThat(serializedProvider).isEqualTo("{\"@type\":\"ProfileCredentialsProvider\"}");
+          assertThat(serializedProvider)
+              .isEqualTo("{\"@type\":\"ProfileCredentialsProvider\",\"profileName\":\"unknown\"}");
 
           AwsCredentialsProvider actual = deserialize(serializedProvider);
-          // NOTE: This documents the unexpected behavior in case a faulty provider is serialized
-          return assertThat(actual.resolveCredentials())
-              .isEqualToComparingFieldByField(DEFAULT_CREDENTIALS);
-        });
+          // compare providers before loading credentials (triggering lazy initialization)
+          assertThat(actual)
+              .isExactlyInstanceOf(ProfileCredentialsProvider.class)
+              .isEqualToComparingFieldByFieldRecursively(provider);
 
-    logs.verifyWarn("Serialized ProfileCredentialsProvider in faulty state.");
+          // Exceptions for invalid profiles are thrown lazily on resolve credentials
+          return assertThatThrownBy(() -> actual.resolveCredentials())
+              .isInstanceOf(SdkClientException.class)
+              .hasMessageContaining("Profile file contained no credentials for profile 'unknown'");
+        });
   }
 
   @Test