S3: Disable strong integrity checksums

Author: ebyhr

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

@@ -275,6 +275,7 @@ private List<String> deleteBatch(String bucket, Collection<String> keysToDelete)
         DeleteObjectsRequest.builder()
             .bucket(bucket)
             .delete(Delete.builder().objects(objectIds).build())
+            .overrideConfiguration(S3RequestUtil.disableStrongIntegrityChecksums())
             .build();
     List<String> failures = Lists.newArrayList();
     try {

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

@@ -424,7 +424,10 @@ private void completeUploads() {
                       .orElseGet(() -> new ByteArrayInputStream(new byte[0])));
 
       PutObjectRequest.Builder requestBuilder =
-          PutObjectRequest.builder().bucket(location.bucket()).key(location.key());
+          PutObjectRequest.builder()
+              .bucket(location.bucket())
+              .key(location.key())
+              .overrideConfiguration(S3RequestUtil.disableStrongIntegrityChecksums());
 
       if (writeTags != null && !writeTags.isEmpty()) {
         requestBuilder.tagging(Tagging.builder().tagSet(writeTags).build());

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

@@ -20,6 +20,8 @@
 
 import java.util.Locale;
 import java.util.function.Function;
+import software.amazon.awssdk.auth.signer.AwsS3V4Signer;
+import software.amazon.awssdk.awscore.AwsRequestOverrideConfiguration;
 import software.amazon.awssdk.services.s3.model.CreateMultipartUploadRequest;
 import software.amazon.awssdk.services.s3.model.GetObjectRequest;
 import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
@@ -149,4 +151,10 @@ static void configurePermission(
       Function<ObjectCannedACL, S3Request.Builder> aclSetter) {
     aclSetter.apply(s3FileIOProperties.acl());
   }
+
+  // TODO Remove me once all of the S3-compatible storage support strong integrity checks
+  @SuppressWarnings("deprecation")
+  static AwsRequestOverrideConfiguration disableStrongIntegrityChecksums() {
+    return AwsRequestOverrideConfiguration.builder().signer(AwsS3V4Signer.create()).build();
+  }
 }

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

@@ -29,15 +29,18 @@
 import software.amazon.awssdk.services.s3.S3ClientBuilder;
 
 public class MinioUtil {
+  public static final String LATEST_TAG = "latest";
+  // This version doesn't support strong integrity checks
+  static final String LEGACY_TAG = "RELEASE.2024-12-18T13-15-44Z";
 
   private MinioUtil() {}
 
   public static MinIOContainer createContainer() {
-    return createContainer(null);
+    return createContainer(LATEST_TAG, null);
   }
 
-  public static MinIOContainer createContainer(AwsCredentials credentials) {
-    var container = new MinIOContainer(DockerImageName.parse("minio/minio:latest"));
+  public static MinIOContainer createContainer(String tag, AwsCredentials credentials) {
+    var container = new MinIOContainer(DockerImageName.parse("minio/minio").withTag(tag));
 
     // this enables virtual-host-style requests. see
     // https://github.com/minio/minio/tree/master/docs/config#domain

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

@@ -105,9 +105,9 @@
 
 @Testcontainers
 public class TestS3FileIO {
-  @Container private static final MinIOContainer MINIO = MinioUtil.createContainer();
+  @Container private final MinIOContainer minio = createMinIOContainer();
 
-  private final SerializableSupplier<S3Client> s3 = () -> MinioUtil.createS3Client(MINIO);
+  private final SerializableSupplier<S3Client> s3 = () -> MinioUtil.createS3Client(minio);
   private final S3Client s3mock = mock(S3Client.class, delegatesTo(s3.get()));
   private final Random random = new Random(1);
   private final int numBucketsForBatchDeletion = 3;
@@ -125,6 +125,12 @@ public class TestS3FileIO {
           "s3.delete.batch-size",
           Integer.toString(batchDeletionSize));
 
+  protected MinIOContainer createMinIOContainer() {
+    MinIOContainer container = MinioUtil.createContainer();
+    container.start();
+    return container;
+  }
+
   @BeforeEach
   public void before() {
     s3FileIO = new S3FileIO(() -> s3mock);

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

@@ -0,0 +1,32 @@
+/*
+ * 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 org.testcontainers.containers.MinIOContainer;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+@Testcontainers
+public class TestS3FileIOWithLegacyMinIO extends TestS3FileIO {
+  @Override
+  protected MinIOContainer createMinIOContainer() {
+    MinIOContainer container = MinioUtil.createContainer(MinioUtil.LEGACY_TAG, null);
+    container.start();
+    return container;
+  }
+}

aws/src/test/java/org/apache/iceberg/aws/s3/signer/TestS3RestSigner.java

@@ -83,7 +83,7 @@ public class TestS3RestSigner {
 
   @Container
   private static final MinIOContainer MINIO_CONTAINER =
-      MinioUtil.createContainer(CREDENTIALS_PROVIDER.resolveCredentials());
+      MinioUtil.createContainer(MinioUtil.LATEST_TAG, CREDENTIALS_PROVIDER.resolveCredentials());
 
   private static Server httpServer;
   private static ValidatingSigner validatingSigner;