AWS: Support similar S3 Sync Client configurations for S3 Async Clients

aws-bundle/build.gradle

@@ -27,6 +27,7 @@ project(":iceberg-aws-bundle") {
     implementation platform(libs.awssdk.bom)
     implementation libs.awssdk.s3accessgrants
     implementation "software.amazon.awssdk:apache-client"
+    implementation "software.amazon.awssdk:netty-nio-client"
     implementation "software.amazon.awssdk:auth"
     implementation "software.amazon.awssdk:crt-core"
     implementation "software.amazon.awssdk:http-auth-aws-crt"

aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogCommitFailure.java

@@ -18,6 +18,7 @@
  */
 package org.apache.iceberg.aws.glue;
 
+import static org.apache.iceberg.aws.s3.S3TestUtil.skipIfAnalyticsAcceleratorEnabled;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
@@ -28,6 +29,7 @@
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableMetadata;
 import org.apache.iceberg.aws.AwsIntegTestUtil;
+import org.apache.iceberg.aws.s3.S3FileIOProperties;
 import org.apache.iceberg.aws.s3.S3TestUtil;
 import org.apache.iceberg.aws.util.RetryDetector;
 import org.apache.iceberg.catalog.TableIdentifier;
@@ -39,6 +41,8 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
 import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariables;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
 import org.mockito.Mockito;
 import software.amazon.awssdk.core.metrics.CoreMetric;
 import software.amazon.awssdk.metrics.MetricCollector;
@@ -156,10 +160,14 @@ public void testCheckCommitStatusAfterRetries() {
         .isEqualTo(2);
   }
 
-  @Test
-  public void testNoRetryAwarenessCorruptsTable() {
+  @ParameterizedTest
+  @MethodSource("org.apache.iceberg.aws.s3.S3TestUtil#analyticsAcceleratorLibraryProperties")
+  public void testNoRetryAwarenessCorruptsTable(Map<String, String> aalProperties) {
     // This test exists to replicate the issue the prior test validates the fix for
     // See https://github.com/apache/iceberg/issues/7151
+    skipIfAnalyticsAcceleratorEnabled(
+        new S3FileIOProperties(aalProperties),
+        "Analytics Accelerator Library does not support custom Iceberg exception: NotFoundException");
     String namespace = createNamespace();
     String tableName = createTable(namespace);
     TableIdentifier tableId = TableIdentifier.of(namespace, tableName);

aws/src/integration/java/org/apache/iceberg/aws/s3/MinioUtil.java

@@ -26,6 +26,8 @@
 import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
 import software.amazon.awssdk.regions.Region;
 import software.amazon.awssdk.services.s3.LegacyMd5Plugin;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.S3AsyncClientBuilder;
 import software.amazon.awssdk.services.s3.S3Client;
 import software.amazon.awssdk.services.s3.S3ClientBuilder;
 
@@ -73,4 +75,16 @@ public static S3Client createS3Client(MinIOContainer container, boolean legacyMd
     builder.forcePathStyle(true); // OSX won't resolve subdomains
     return builder.build();
   }
+
+  public static S3AsyncClient createS3AsyncClient(MinIOContainer container) {
+    URI uri = URI.create(container.getS3URL());
+    S3AsyncClientBuilder builder = S3AsyncClient.builder();
+    builder.credentialsProvider(
+        StaticCredentialsProvider.create(
+            AwsBasicCredentials.create(container.getUserName(), container.getPassword())));
+    builder.applyMutation(mutator -> mutator.endpointOverride(uri));
+    builder.region(Region.US_EAST_1);
+    builder.forcePathStyle(true); // OSX won't resolve subdomains
+    return builder.build();
+  }
 }

aws/src/integration/java/org/apache/iceberg/aws/s3/S3TestUtil.java

@@ -18,8 +18,20 @@
  */
 package org.apache.iceberg.aws.s3;
 
+import static org.assertj.core.api.Assumptions.assumeThat;
+
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Stream;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.params.provider.Arguments;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 public class S3TestUtil {
 
+  private static final Logger LOG = LoggerFactory.getLogger(S3TestUtil.class);
+
   private S3TestUtil() {}
 
   public static String getBucketFromUri(String s3Uri) {
@@ -29,4 +41,38 @@ public static String getBucketFromUri(String s3Uri) {
   public static String getKeyFromUri(String s3Uri) {
     return new S3URI(s3Uri).key();
   }
+
+  /**
+   * Skip a test if the Analytics Accelerator Library for Amazon S3 is enabled.
+   *
+   * @param properties properties to probe
+   */
+  public static void skipIfAnalyticsAcceleratorEnabled(
+      S3FileIOProperties properties, String message) {
+    boolean isAcceleratorEnabled = properties.isS3AnalyticsAcceleratorEnabled();
+    if (isAcceleratorEnabled) {
+      LOG.warn(message);
+    }
+    assumeThat(!isAcceleratorEnabled).describedAs(message).isTrue();
+  }
+
+  public static Stream<Arguments> analyticsAcceleratorLibraryProperties() {
+    return listAnalyticsAcceleratorLibraryProperties().stream().map(Arguments::of);
+  }
+
+  public static List<Map<String, String>> listAnalyticsAcceleratorLibraryProperties() {
+    return List.of(
+        ImmutableMap.of(
+            S3FileIOProperties.S3_ANALYTICS_ACCELERATOR_ENABLED, Boolean.toString(true)),
+        ImmutableMap.of(
+            S3FileIOProperties.S3_ANALYTICS_ACCELERATOR_ENABLED, Boolean.toString(false)));
+  }
+
+  public static Map<String, String> mergeProperties(
+      Map<String, String> aalProperties, Map<String, String> testProperties) {
+    return ImmutableMap.<String, String>builder()
+        .putAll(aalProperties)
+        .putAll(testProperties)
+        .build();
+  }
 }