Aws: Add Iceberg version to UserAgent in S3 requests

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

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

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

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

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

@@ -24,6 +24,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
+import org.apache.iceberg.EnvironmentContext;
 import org.apache.iceberg.aws.AwsClientProperties;
 import org.apache.iceberg.aws.glue.GlueCatalog;
 import org.apache.iceberg.aws.s3.signer.S3V4RestSignerClient;
@@ -375,6 +376,14 @@ public class S3FileIOProperties implements Serializable {
 
   public static final boolean PRELOAD_CLIENT_ENABLED_DEFAULT = false;
 
+  /**
+   * User Agent Prefix set by the S3 client.
+   *
+   * <p>This allows developers to monitor which version of Iceberg they have deployed to a cluster
+   * (for example, through the S3 Access Logs, which contain the user agent field).
+   */
+  private static final String S3_FILE_IO_USER_AGENT = "s3fileio/" + EnvironmentContext.get();
+
   private String sseType;
   private String sseKey;
   private String sseMd5;
@@ -819,6 +828,11 @@ public <T extends S3ClientBuilder> void applyS3AccessGrantsConfigurations(T buil
     }
   }
 
+  public <T extends S3ClientBuilder> void applyUserAgentConfigurations(T builder) {
+    builder.overrideConfiguration(
+        c -> c.putAdvancedOption(SdkAdvancedClientOption.USER_AGENT_PREFIX, S3_FILE_IO_USER_AGENT));
+  }
+
   /**
    * Dynamically load the http client builder to avoid runtime deps requirements of any optional SDK
    * Plugins

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

@@ -478,4 +478,14 @@ public void testApplyEndpointConfiguration() {
     s3FileIOProperties.applyEndpointConfigurations(mockS3ClientBuilder);
     Mockito.verify(mockS3ClientBuilder).endpointOverride(Mockito.any(URI.class));
   }
+
+  @Test
+  public void testApplyUserAgentConfigurations() {
+    Map<String, String> properties = Maps.newHashMap();
+    S3FileIOProperties s3FileIOProperties = new S3FileIOProperties(properties);
+    S3ClientBuilder mockS3ClientBuilder = Mockito.mock(S3ClientBuilder.class);
+    s3FileIOProperties.applyUserAgentConfigurations(mockS3ClientBuilder);
+
+    Mockito.verify(mockS3ClientBuilder).overrideConfiguration(Mockito.any(Consumer.class));
+  }
 }