AWS: Add S3 Access Grants Integration

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

@@ -112,6 +112,7 @@ public S3Client s3() {
           .applyMutation(
               b -> s3FileIOProperties.applyCredentialConfigurations(awsClientProperties, b))
           .applyMutation(s3FileIOProperties::applySignerConfiguration)
+          .applyMutation(s3FileIOProperties::applyS3AccessGrantsConfigurations)
           .build();
     }
 

aws/src/main/java/org/apache/iceberg/aws/s3/DefaultS3FileIOAwsClientFactory.java

@@ -53,6 +53,7 @@ public S3Client s3() {
                 s3FileIOProperties.applyCredentialConfigurations(
                     awsClientProperties, s3ClientBuilder))
         .applyMutation(s3FileIOProperties::applySignerConfiguration)
+        .applyMutation(s3FileIOProperties::applyS3AccessGrantsConfigurations)
         .build();
   }
 }

aws/src/main/java/org/apache/iceberg/aws/s3/S3AccessGrantsPluginConfigurations.java

@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.aws.s3;
+
+import java.util.Map;
+import org.apache.iceberg.util.PropertyUtil;
+import software.amazon.awssdk.s3accessgrants.plugin.S3AccessGrantsPlugin;
+import software.amazon.awssdk.services.s3.S3ClientBuilder;
+
+class S3AccessGrantsPluginConfigurations {
+  private boolean isS3AccessGrantsFallbackToIamEnabled;
+
+  private S3AccessGrantsPluginConfigurations() {}
+
+  public <T extends S3ClientBuilder> void configureS3ClientBuilder(T builder) {
+    S3AccessGrantsPlugin s3AccessGrantsPlugin =
+        S3AccessGrantsPlugin.builder().enableFallback(isS3AccessGrantsFallbackToIamEnabled).build();
+    builder.addPlugin(s3AccessGrantsPlugin);
+  }
+
+  private void initialize(Map<String, String> properties) {
+    this.isS3AccessGrantsFallbackToIamEnabled =
+        PropertyUtil.propertyAsBoolean(
+            properties,
+            S3FileIOProperties.S3_ACCESS_GRANTS_FALLBACK_TO_IAM_ENABLED,
+            S3FileIOProperties.S3_ACCESS_GRANTS_FALLBACK_TO_IAM_ENABLED_DEFAULT);
+  }
+
+  public static S3AccessGrantsPluginConfigurations create(Map<String, String> properties) {
+    S3AccessGrantsPluginConfigurations configurations = new S3AccessGrantsPluginConfigurations();
+    configurations.initialize(properties);
+    return configurations;
+  }
+}

aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIOProperties.java

@@ -27,6 +27,7 @@
 import org.apache.iceberg.aws.AwsClientProperties;
 import org.apache.iceberg.aws.glue.GlueCatalog;
 import org.apache.iceberg.aws.s3.signer.S3V4RestSignerClient;
+import org.apache.iceberg.common.DynMethods;
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
@@ -50,6 +51,28 @@ public class S3FileIOProperties implements Serializable {
    */
   public static final String CLIENT_FACTORY = "s3.client-factory-impl";
 
+  /**
+   * This property is used to enable using the S3 Access Grants product to control authorization to
+   * S3 data. More information regarding this feature can be found at:
+   * https://aws.amazon.com/s3/features/access-grants/.
+   */
+  public static final String S3_ACCESS_GRANTS_ENABLED = "s3.access-grants.enabled";
+
+  public static final boolean S3_ACCESS_GRANTS_ENABLED_DEFAULT = false;
+
+  /**
+   * The fallback-to-iam property allows users to customize whether or not they would like their
+   * jobs fall back to the Job Execution IAM role in case they get an Access Denied from the S3
+   * Access Grants call. Further documentation regarding this flag can be found in the S3 Access
+   * Grants Plugin GitHub:
+   *
+   * <p>For more details, see: https://github.com/aws/aws-s3-accessgrants-plugin-java-v2
+   */
+  public static final String S3_ACCESS_GRANTS_FALLBACK_TO_IAM_ENABLED =
+      "s3.access-grants.fallback-to-iam";
+
+  public static final boolean S3_ACCESS_GRANTS_FALLBACK_TO_IAM_ENABLED_DEFAULT = false;
+
   /**
    * Type of S3 Server side encryption used, default to {@link S3FileIOProperties#SSE_TYPE_NONE}.
    *
@@ -358,6 +381,8 @@ public class S3FileIOProperties implements Serializable {
   private String accessKeyId;
   private String secretAccessKey;
   private String sessionToken;
+  private boolean isS3AccessGrantsEnabled;
+  private boolean isS3AccessGrantsFallbackToIamEnabled;
   private int multipartUploadThreads;
   private int multiPartSize;
   private int deleteBatchSize;
@@ -410,6 +435,8 @@ public S3FileIOProperties() {
     this.isUseArnRegionEnabled = USE_ARN_REGION_ENABLED_DEFAULT;
     this.isAccelerationEnabled = ACCELERATION_ENABLED_DEFAULT;
     this.isRemoteSigningEnabled = REMOTE_SIGNING_ENABLED_DEFAULT;
+    this.isS3AccessGrantsEnabled = S3_ACCESS_GRANTS_ENABLED_DEFAULT;
+    this.isS3AccessGrantsFallbackToIamEnabled = S3_ACCESS_GRANTS_FALLBACK_TO_IAM_ENABLED_DEFAULT;
     this.allProperties = Maps.newHashMap();
 
     ValidationException.check(
@@ -500,6 +527,14 @@ public S3FileIOProperties(Map<String, String> properties) {
             properties, REMOTE_SIGNING_ENABLED, REMOTE_SIGNING_ENABLED_DEFAULT);
     this.writeStorageClass = properties.get(WRITE_STORAGE_CLASS);
     this.allProperties = SerializableMap.copyOf(properties);
+    this.isS3AccessGrantsEnabled =
+        PropertyUtil.propertyAsBoolean(
+            properties, S3_ACCESS_GRANTS_ENABLED, S3_ACCESS_GRANTS_ENABLED_DEFAULT);
+    this.isS3AccessGrantsFallbackToIamEnabled =
+        PropertyUtil.propertyAsBoolean(
+            properties,
+            S3_ACCESS_GRANTS_FALLBACK_TO_IAM_ENABLED,
+            S3_ACCESS_GRANTS_FALLBACK_TO_IAM_ENABLED_DEFAULT);
 
     ValidationException.check(
         keyIdAccessKeyBothConfigured(),
@@ -684,6 +719,22 @@ private Set<Tag> toS3Tags(Map<String, String> properties, String prefix) {
         .collect(Collectors.toSet());
   }
 
+  public boolean isS3AccessGrantsEnabled() {
+    return isS3AccessGrantsEnabled;
+  }
+
+  public void setS3AccessGrantsEnabled(boolean s3AccessGrantsEnabled) {
+    this.isS3AccessGrantsEnabled = s3AccessGrantsEnabled;
+  }
+
+  public boolean isS3AccessGrantsFallbackToIamEnabled() {
+    return isS3AccessGrantsFallbackToIamEnabled;
+  }
+
+  public void setS3AccessGrantsFallbackToIamEnabled(boolean s3AccessGrantsFallbackToIamEnabled) {
+    this.isS3AccessGrantsFallbackToIamEnabled = s3AccessGrantsFallbackToIamEnabled;
+  }
+
   private boolean keyIdAccessKeyBothConfigured() {
     return (accessKeyId == null) == (secretAccessKey == null);
   }
@@ -749,4 +800,43 @@ public <T extends S3ClientBuilder> void applyEndpointConfigurations(T builder) {
       builder.endpointOverride(URI.create(endpoint));
     }
   }
+
+  /**
+   * Add the S3 Access Grants Plugin for an S3 client.
+   *
+   * <p>Sample usage:
+   *
+   * <pre>
+   *     S3Client.builder().applyMutation(s3FileIOProperties::applyS3AccessGrantsConfigurations)
+   * </pre>
+   */
+  public <T extends S3ClientBuilder> void applyS3AccessGrantsConfigurations(T builder) {
+    if (isS3AccessGrantsEnabled) {
+      S3AccessGrantsPluginConfigurations s3AccessGrantsPluginConfigurations =
+          loadSdkPluginConfigurations(
+              S3AccessGrantsPluginConfigurations.class.getName(), allProperties);
+      s3AccessGrantsPluginConfigurations.configureS3ClientBuilder(builder);
+    }
+  }
+
+  /**
+   * Dynamically load the http client builder to avoid runtime deps requirements of any optional SDK
+   * Plugins
+   */
+  private <T> T loadSdkPluginConfigurations(String impl, Map<String, String> properties) {
+    Object sdkPluginConfigurations;
+    try {
+      sdkPluginConfigurations =
+          DynMethods.builder("create")
+              .hiddenImpl(impl, Map.class)
+              .buildStaticChecked()
+              .invoke(properties);
+      return (T) sdkPluginConfigurations;
+    } catch (NoSuchMethodException e) {
+      throw new IllegalArgumentException(
+          String.format(
+              "Cannot create %s to generate and configure the client SDK Plugin builder", impl),
+          e);
+    }
+  }
 }

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

@@ -249,4 +249,36 @@ public void testS3RemoteSigningDisabled() {
         builder.overrideConfiguration().advancedOption(SdkAdvancedClientOption.SIGNER);
     Assertions.assertThat(signer).isNotPresent();
   }
+
+  @Test
+  public void testS3AccessGrantsEnabled() {
+    // Explicitly true
+    Map<String, String> properties =
+        ImmutableMap.of(S3FileIOProperties.S3_ACCESS_GRANTS_ENABLED, "true");
+    S3FileIOProperties s3Properties = new S3FileIOProperties(properties);
+    S3ClientBuilder builder = S3Client.builder();
+
+    s3Properties.applyS3AccessGrantsConfigurations(builder);
+    Assertions.assertThat(builder.plugins().size()).isEqualTo(1);
+  }
+
+  @Test
+  public void testS3AccessGrantsDisabled() {
+    // Explicitly false
+    Map<String, String> properties =
+        ImmutableMap.of(S3FileIOProperties.S3_ACCESS_GRANTS_ENABLED, "false");
+    S3FileIOProperties s3Properties = new S3FileIOProperties(properties);
+    S3ClientBuilder builder = S3Client.builder();
+
+    s3Properties.applyS3AccessGrantsConfigurations(builder);
+    Assertions.assertThat(builder.plugins().size()).isEqualTo(0);
+
+    // Implicitly false
+    properties = ImmutableMap.of();
+    s3Properties = new S3FileIOProperties(properties);
+    builder = S3Client.builder();
+
+    s3Properties.applyS3AccessGrantsConfigurations(builder);
+    Assertions.assertThat(builder.plugins().size()).isEqualTo(0);
+  }
 }

build.gradle

@@ -465,6 +465,7 @@ project(':iceberg-aws') {
     implementation "com.fasterxml.jackson.core:jackson-core"
 
     compileOnly(platform(libs.awssdk.bom))
+    compileOnly(libs.awssdk.s3accessgrants)
     compileOnly("software.amazon.awssdk:url-connection-client")
     compileOnly("software.amazon.awssdk:apache-client")
     compileOnly("software.amazon.awssdk:auth")
@@ -487,6 +488,7 @@ project(':iceberg-aws') {
     testImplementation(platform(libs.awssdk.bom))
     testImplementation("software.amazon.awssdk:iam")
     testImplementation("software.amazon.awssdk:s3control")
+    testImplementation("software.amazon.s3.accessgrants:aws-s3-accessgrants-java-plugin")
     testImplementation project(path: ':iceberg-api', configuration: 'testArtifacts')
     testImplementation(libs.s3mock.junit5) {
       exclude module: "spring-boot-starter-logging"

gradle/libs.versions.toml

@@ -29,6 +29,7 @@ assertj-core = "3.24.2"
 awaitility = "4.2.0"
 awssdk-bom = "2.21.42"
 azuresdk-bom = "1.2.18"
+awssdk-s3accessgrants = "1.0.1"
 caffeine = "2.9.3"
 calcite = "1.10.0"
 delta-standalone = "0.6.0"
@@ -96,6 +97,7 @@ arrow-memory-netty = { module = "org.apache.arrow:arrow-memory-netty", version.r
 arrow-vector = { module = "org.apache.arrow:arrow-vector", version.ref = "arrow" }
 avro-avro = { module = "org.apache.avro:avro", version.ref = "avro" }
 awssdk-bom = { module = "software.amazon.awssdk:bom", version.ref = "awssdk-bom" }
+awssdk-s3accessgrants = { module = "software.amazon.s3.accessgrants:aws-s3-accessgrants-java-plugin", version.ref = "awssdk-s3accessgrants" }
 azuresdk-bom = { module = "com.azure:azure-sdk-bom", version.ref = "azuresdk-bom" }
 caffeine = { module = "com.github.ben-manes.caffeine:caffeine", version.ref = "caffeine" }
 calcite-core = { module = "org.apache.calcite:calcite-core", version.ref = "calcite" }