AWS: remove deprecated API for 1.4.0 in iceberg-aws

aws/src/integration/java/org/apache/iceberg/aws/TestAssumeRoleAwsClientFactory.java

@@ -82,7 +82,8 @@ public void before() {
     assumeRoleProperties = Maps.newHashMap();
     assumeRoleProperties.put(
         AwsProperties.CLIENT_FACTORY, AssumeRoleAwsClientFactory.class.getName());
-    assumeRoleProperties.put(AwsProperties.HTTP_CLIENT_TYPE, AwsProperties.HTTP_CLIENT_TYPE_APACHE);
+    assumeRoleProperties.put(
+        HttpClientProperties.CLIENT_TYPE, HttpClientProperties.CLIENT_TYPE_APACHE);
     assumeRoleProperties.put(
         AwsProperties.CLIENT_ASSUME_ROLE_REGION, AwsIntegTestUtil.testRegion());
     assumeRoleProperties.put(AwsProperties.CLIENT_ASSUME_ROLE_ARN, response.role().arn());

aws/src/integration/java/org/apache/iceberg/aws/TestDefaultAwsClientFactory.java

@@ -20,6 +20,7 @@
 
 import java.util.Map;
 import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.aws.s3.S3FileIOProperties;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.junit.Test;
 import software.amazon.awssdk.core.exception.SdkClientException;
@@ -48,7 +49,7 @@ public void testGlueEndpointOverride() {
   @Test
   public void testS3FileIoEndpointOverride() {
     Map<String, String> properties = Maps.newHashMap();
-    properties.put(AwsProperties.S3FILEIO_ENDPOINT, "https://unknown:1234");
+    properties.put(S3FileIOProperties.ENDPOINT, "https://unknown:1234");
     AwsClientFactory factory = AwsClientFactories.from(properties);
     S3Client s3Client = factory.s3();
     AssertHelpers.assertThrowsCause(
@@ -61,8 +62,8 @@ public void testS3FileIoEndpointOverride() {
   @Test
   public void testS3FileIoCredentialsOverride() {
     Map<String, String> properties = Maps.newHashMap();
-    properties.put(AwsProperties.S3FILEIO_ACCESS_KEY_ID, "unknown");
-    properties.put(AwsProperties.S3FILEIO_SECRET_ACCESS_KEY, "unknown");
+    properties.put(S3FileIOProperties.ACCESS_KEY_ID, "unknown");
+    properties.put(S3FileIOProperties.SECRET_ACCESS_KEY, "unknown");
     AwsClientFactory factory = AwsClientFactories.from(properties);
     S3Client s3Client = factory.s3();
     AssertHelpers.assertThrows(

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

@@ -28,6 +28,7 @@
 import org.apache.iceberg.aws.AwsClientFactory;
 import org.apache.iceberg.aws.AwsIntegTestUtil;
 import org.apache.iceberg.aws.AwsProperties;
+import org.apache.iceberg.aws.s3.S3FileIOProperties;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
@@ -77,15 +78,29 @@ public class GlueTestBase {
   @BeforeClass
   public static void beforeClass() {
     glueCatalog = new GlueCatalog();
-    AwsProperties properties = new AwsProperties();
-    properties.setS3FileIoDeleteBatchSize(10);
-    glueCatalog.initialize(catalogName, testBucketPath, properties, glue, null, ImmutableMap.of());
+    AwsProperties awsProperties = new AwsProperties();
+    S3FileIOProperties s3FileIOProperties = new S3FileIOProperties();
+    s3FileIOProperties.setDeleteBatchSize(10);
+    glueCatalog.initialize(
+        catalogName,
+        testBucketPath,
+        awsProperties,
+        s3FileIOProperties,
+        glue,
+        null,
+        ImmutableMap.of());
 
     glueCatalogWithSkipNameValidation = new GlueCatalog();
     AwsProperties propertiesSkipNameValidation = new AwsProperties();
     propertiesSkipNameValidation.setGlueCatalogSkipNameValidation(true);
     glueCatalogWithSkipNameValidation.initialize(
-        catalogName, testBucketPath, propertiesSkipNameValidation, glue, null, ImmutableMap.of());
+        catalogName,
+        testBucketPath,
+        propertiesSkipNameValidation,
+        new S3FileIOProperties(),
+        glue,
+        null,
+        ImmutableMap.of());
   }
 
   @AfterClass

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

@@ -33,6 +33,7 @@
 import org.apache.iceberg.Table;
 import org.apache.iceberg.aws.AwsProperties;
 import org.apache.iceberg.aws.dynamodb.DynamoDbLockManager;
+import org.apache.iceberg.aws.s3.S3FileIOProperties;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.util.concurrent.MoreExecutors;
@@ -56,11 +57,13 @@ public static void beforeClass() {
     lockTableName = getRandomName();
     glueCatalog = new GlueCatalog();
     AwsProperties awsProperties = new AwsProperties();
+    S3FileIOProperties s3FileIOProperties = new S3FileIOProperties();
     dynamo = clientFactory.dynamo();
     glueCatalog.initialize(
         catalogName,
         testBucketPath,
         awsProperties,
+        s3FileIOProperties,
         glue,
         new DynamoDbLockManager(dynamo, lockTableName),
         ImmutableMap.of());

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

@@ -38,6 +38,7 @@
 import org.apache.iceberg.TableOperations;
 import org.apache.iceberg.Transaction;
 import org.apache.iceberg.aws.AwsProperties;
+import org.apache.iceberg.aws.s3.S3FileIOProperties;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.AlreadyExistsException;
@@ -139,6 +140,7 @@ public void testCreateAndLoadTableWithoutWarehouseLocation() {
         catalogName,
         null,
         new AwsProperties(),
+        new S3FileIOProperties(),
         glue,
         LockManagers.defaultLockManager(),
         ImmutableMap.of());
@@ -380,6 +382,7 @@ public void testCommitTableSkipArchive() {
         catalogName,
         testBucketPath,
         properties,
+        new S3FileIOProperties(),
         glue,
         LockManagers.defaultLockManager(),
         ImmutableMap.of());
@@ -596,11 +599,17 @@ public void testTableLevelS3Tags() {
     String testBucketPath = "s3://" + testBucketName + "/" + testPathPrefix;
     Map<String, String> properties =
         ImmutableMap.of(
-            AwsProperties.S3_WRITE_TABLE_TAG_ENABLED,
+            S3FileIOProperties.WRITE_TABLE_TAG_ENABLED,
             "true",
-            AwsProperties.S3_WRITE_NAMESPACE_TAG_ENABLED,
+            S3FileIOProperties.WRITE_NAMESPACE_TAG_ENABLED,
             "true");
-    glueCatalog.initialize(catalogName, testBucketPath, new AwsProperties(properties), glue, null);
+    glueCatalog.initialize(
+        catalogName,
+        testBucketPath,
+        new AwsProperties(properties),
+        new S3FileIOProperties(properties),
+        glue,
+        null);
     String namespace = createNamespace();
     String tableName = getRandomName();
     createTable(namespace, tableName);
@@ -617,9 +626,9 @@ public void testTableLevelS3Tags() {
             .tagSet();
     Map<String, String> tagMap = tags.stream().collect(Collectors.toMap(Tag::key, Tag::value));
 
-    Assert.assertTrue(tagMap.containsKey(AwsProperties.S3_TAG_ICEBERG_TABLE));
-    Assert.assertEquals(tableName, tagMap.get(AwsProperties.S3_TAG_ICEBERG_TABLE));
-    Assert.assertTrue(tagMap.containsKey(AwsProperties.S3_TAG_ICEBERG_NAMESPACE));
-    Assert.assertEquals(namespace, tagMap.get(AwsProperties.S3_TAG_ICEBERG_NAMESPACE));
+    Assert.assertTrue(tagMap.containsKey(S3FileIOProperties.S3_TAG_ICEBERG_TABLE));
+    Assert.assertEquals(tableName, tagMap.get(S3FileIOProperties.S3_TAG_ICEBERG_TABLE));
+    Assert.assertTrue(tagMap.containsKey(S3FileIOProperties.S3_TAG_ICEBERG_NAMESPACE));
+    Assert.assertEquals(namespace, tagMap.get(S3FileIOProperties.S3_TAG_ICEBERG_NAMESPACE));
   }
 }

aws/src/integration/java/org/apache/iceberg/aws/lakeformation/LakeFormationTestBase.java

@@ -30,6 +30,7 @@
 import org.apache.iceberg.aws.AssumeRoleAwsClientFactory;
 import org.apache.iceberg.aws.AwsIntegTestUtil;
 import org.apache.iceberg.aws.AwsProperties;
+import org.apache.iceberg.aws.HttpClientProperties;
 import org.apache.iceberg.aws.glue.GlueCatalog;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
@@ -227,7 +228,8 @@ public static void beforeClass() throws Exception {
         AwsProperties.CLIENT_ASSUME_ROLE_REGION, AwsIntegTestUtil.testRegion());
     assumeRoleProperties.put(AwsProperties.GLUE_LAKEFORMATION_ENABLED, "true");
     assumeRoleProperties.put(AwsProperties.GLUE_ACCOUNT_ID, AwsIntegTestUtil.testAccountId());
-    assumeRoleProperties.put(AwsProperties.HTTP_CLIENT_TYPE, AwsProperties.HTTP_CLIENT_TYPE_APACHE);
+    assumeRoleProperties.put(
+        HttpClientProperties.CLIENT_TYPE, HttpClientProperties.CLIENT_TYPE_APACHE);
     assumeRoleProperties.put(AwsProperties.CLIENT_ASSUME_ROLE_ARN, lfPrivilegedRoleArn);
     assumeRoleProperties.put(
         AwsProperties.CLIENT_ASSUME_ROLE_TAGS_PREFIX

aws/src/integration/java/org/apache/iceberg/aws/lakeformation/TestLakeFormationAwsClientFactory.java

@@ -22,6 +22,7 @@
 import java.util.UUID;
 import org.apache.iceberg.aws.AwsIntegTestUtil;
 import org.apache.iceberg.aws.AwsProperties;
+import org.apache.iceberg.aws.HttpClientProperties;
 import org.apache.iceberg.aws.glue.GlueCatalog;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
@@ -82,7 +83,8 @@ public void before() {
     assumeRoleProperties = Maps.newHashMap();
     assumeRoleProperties.put(AwsProperties.CLIENT_ASSUME_ROLE_REGION, "us-east-1");
     assumeRoleProperties.put(AwsProperties.GLUE_LAKEFORMATION_ENABLED, "true");
-    assumeRoleProperties.put(AwsProperties.HTTP_CLIENT_TYPE, AwsProperties.HTTP_CLIENT_TYPE_APACHE);
+    assumeRoleProperties.put(
+        HttpClientProperties.CLIENT_TYPE, HttpClientProperties.CLIENT_TYPE_APACHE);
     assumeRoleProperties.put(AwsProperties.CLIENT_ASSUME_ROLE_ARN, response.role().arn());
     assumeRoleProperties.put(
         AwsProperties.CLIENT_ASSUME_ROLE_TAGS_PREFIX

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

@@ -37,7 +37,6 @@
 import org.apache.iceberg.aws.AwsClientFactories;
 import org.apache.iceberg.aws.AwsClientFactory;
 import org.apache.iceberg.aws.AwsIntegTestUtil;
-import org.apache.iceberg.aws.AwsProperties;
 import org.apache.iceberg.io.InputFile;
 import org.apache.iceberg.io.OutputFile;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
@@ -377,7 +376,7 @@ public void testDeleteFilesMultipleBatchesWithAccessPoints() throws Exception {
 
   @Test
   public void testDeleteFilesMultipleBatchesWithCrossRegionAccessPoints() throws Exception {
-    clientFactory.initialize(ImmutableMap.of(AwsProperties.S3_USE_ARN_REGION_ENABLED, "true"));
+    clientFactory.initialize(ImmutableMap.of(S3FileIOProperties.USE_ARN_REGION_ENABLED, "true"));
     S3FileIO s3FileIO = new S3FileIO(clientFactory::s3, getDeletionTestProperties());
     s3FileIO.initialize(
         ImmutableMap.of(

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

@@ -47,33 +47,31 @@ public <T extends AwsSyncClientBuilder> void configureHttpClientBuilder(T awsCli
   private void initialize(Map<String, String> httpClientProperties) {
     this.connectionTimeoutMs =
         PropertyUtil.propertyAsNullableLong(
-            httpClientProperties, AwsProperties.HTTP_CLIENT_APACHE_CONNECTION_TIMEOUT_MS);
+            httpClientProperties, HttpClientProperties.APACHE_CONNECTION_TIMEOUT_MS);
     this.socketTimeoutMs =
         PropertyUtil.propertyAsNullableLong(
-            httpClientProperties, AwsProperties.HTTP_CLIENT_APACHE_SOCKET_TIMEOUT_MS);
+            httpClientProperties, HttpClientProperties.APACHE_SOCKET_TIMEOUT_MS);
     this.acquisitionTimeoutMs =
         PropertyUtil.propertyAsNullableLong(
-            httpClientProperties,
-            AwsProperties.HTTP_CLIENT_APACHE_CONNECTION_ACQUISITION_TIMEOUT_MS);
+            httpClientProperties, HttpClientProperties.APACHE_CONNECTION_ACQUISITION_TIMEOUT_MS);
     this.connectionMaxIdleTimeMs =
         PropertyUtil.propertyAsNullableLong(
-            httpClientProperties, AwsProperties.HTTP_CLIENT_APACHE_CONNECTION_MAX_IDLE_TIME_MS);
+            httpClientProperties, HttpClientProperties.APACHE_CONNECTION_MAX_IDLE_TIME_MS);
     this.connectionTimeToLiveMs =
         PropertyUtil.propertyAsNullableLong(
-            httpClientProperties, AwsProperties.HTTP_CLIENT_APACHE_CONNECTION_TIME_TO_LIVE_MS);
+            httpClientProperties, HttpClientProperties.APACHE_CONNECTION_TIME_TO_LIVE_MS);
     this.expectContinueEnabled =
         PropertyUtil.propertyAsNullableBoolean(
-            httpClientProperties, AwsProperties.HTTP_CLIENT_APACHE_EXPECT_CONTINUE_ENABLED);
+            httpClientProperties, HttpClientProperties.APACHE_EXPECT_CONTINUE_ENABLED);
     this.maxConnections =
         PropertyUtil.propertyAsNullableInt(
-            httpClientProperties, AwsProperties.HTTP_CLIENT_APACHE_MAX_CONNECTIONS);
+            httpClientProperties, HttpClientProperties.APACHE_MAX_CONNECTIONS);
     this.tcpKeepAliveEnabled =
         PropertyUtil.propertyAsNullableBoolean(
-            httpClientProperties, AwsProperties.HTTP_CLIENT_APACHE_TCP_KEEP_ALIVE_ENABLED);
+            httpClientProperties, HttpClientProperties.APACHE_TCP_KEEP_ALIVE_ENABLED);
     this.useIdleConnectionReaperEnabled =
         PropertyUtil.propertyAsNullableBoolean(
-            httpClientProperties,
-            AwsProperties.HTTP_CLIENT_APACHE_USE_IDLE_CONNECTION_REAPER_ENABLED);
+            httpClientProperties, HttpClientProperties.APACHE_USE_IDLE_CONNECTION_REAPER_ENABLED);
   }
 
   @VisibleForTesting

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

@@ -20,6 +20,7 @@
 
 import java.util.Map;
 import java.util.UUID;
+import org.apache.iceberg.aws.s3.S3FileIOProperties;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import software.amazon.awssdk.awscore.client.builder.AwsClientBuilder;
 import software.amazon.awssdk.awscore.client.builder.AwsSyncClientBuilder;
@@ -34,47 +35,51 @@
 
 public class AssumeRoleAwsClientFactory implements AwsClientFactory {
   private AwsProperties awsProperties;
+  private HttpClientProperties httpClientProperties;
+  private S3FileIOProperties s3FileIOProperties;
   private String roleSessionName;
 
   @Override
   public S3Client s3() {
     return S3Client.builder()
         .applyMutation(this::applyAssumeRoleConfigurations)
-        .applyMutation(awsProperties::applyHttpClientConfigurations)
-        .applyMutation(awsProperties::applyS3EndpointConfigurations)
-        .applyMutation(awsProperties::applyS3ServiceConfigurations)
-        .applyMutation(awsProperties::applyS3SignerConfiguration)
+        .applyMutation(httpClientProperties::applyHttpClientConfigurations)
+        .applyMutation(s3FileIOProperties::applyEndpointConfigurations)
+        .applyMutation(s3FileIOProperties::applyServiceConfigurations)
+        .applyMutation(s3FileIOProperties::applySignerConfiguration)
         .build();
   }
 
   @Override
   public GlueClient glue() {
     return GlueClient.builder()
         .applyMutation(this::applyAssumeRoleConfigurations)
-        .applyMutation(awsProperties::applyHttpClientConfigurations)
+        .applyMutation(httpClientProperties::applyHttpClientConfigurations)
         .build();
   }
 
   @Override
   public KmsClient kms() {
     return KmsClient.builder()
         .applyMutation(this::applyAssumeRoleConfigurations)
-        .applyMutation(awsProperties::applyHttpClientConfigurations)
+        .applyMutation(httpClientProperties::applyHttpClientConfigurations)
         .build();
   }
 
   @Override
   public DynamoDbClient dynamo() {
     return DynamoDbClient.builder()
         .applyMutation(this::applyAssumeRoleConfigurations)
-        .applyMutation(awsProperties::applyHttpClientConfigurations)
+        .applyMutation(httpClientProperties::applyHttpClientConfigurations)
         .applyMutation(awsProperties::applyDynamoDbEndpointConfigurations)
         .build();
   }
 
   @Override
   public void initialize(Map<String, String> properties) {
     this.awsProperties = new AwsProperties(properties);
+    this.s3FileIOProperties = new S3FileIOProperties(properties);
+    this.httpClientProperties = new HttpClientProperties(properties);
     this.roleSessionName = genSessionName();
     Preconditions.checkNotNull(
         awsProperties.clientAssumeRoleArn(),
@@ -112,8 +117,18 @@ protected AwsProperties awsProperties() {
     return awsProperties;
   }
 
+  protected HttpClientProperties httpClientProperties() {
+    return httpClientProperties;
+  }
+
+  protected S3FileIOProperties s3FileIOProperties() {
+    return s3FileIOProperties;
+  }
+
   private StsClient sts() {
-    return StsClient.builder().applyMutation(awsProperties::applyHttpClientConfigurations).build();
+    return StsClient.builder()
+        .applyMutation(httpClientProperties::applyHttpClientConfigurations)
+        .build();
   }
 
   private String genSessionName() {