feature(s3/manager): add option to control default checksums

As announced in #2960,
AWS SDK for Go v2 service/s3 v1.73.0 shipped a change
(#2808) that adopted new
default integrity protections, automatically calculating CRC32 checksums
for operations like PutObject and UploadPart.

While it is possible to revert to the previous behavior by setting
`AWS_REQUEST_CHECKSUM_CALCULATION=when_required`, this config setting
did not apply to the S3 Manager multipart uploader, which always enabled
CRC32 checksums by default regardless of the global setting.

This commit adds a new `RequestChecksumCalculation` field to the Uploader
struct that allows users to control checksum behavior:

- `RequestChecksumCalculationWhenSupported` (default): Always calculates
  CRC32 checksums for multipart uploads
- `RequestChecksumCalculationWhenRequired`: Only calculates checksums
  when explicitly set by the user, preserving backwards compatibility

For example:

```go
uploader := manager.NewUploader(client, func(u *manager.Uploader) {
    u.RequestChecksumCalculation = aws.RequestChecksumCalculationWhenRequired
})
```

S3 Express One Zone buckets always require CRC32 checksums regardless of
this setting, as mandated by the S3 Express service requirements.  The
uploader automatically detects S3 Express buckets (names ending with
`--x-s3`) and applies CRC32 checksums unconditionally.

Fixes #3007

Author: stanhu

feature/s3/manager/upload.go

@@ -8,6 +8,7 @@ import (
 	"io"
 	"net/http"
 	"sort"
+	"strings"
 	"sync"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
@@ -245,6 +246,21 @@ type Uploader struct {
 	// Defines the buffer strategy used when uploading a part
 	BufferProvider ReadSeekerWriteToProvider
 
+	// RequestChecksumCalculation determines when request checksum calculation is performed
+	// for multipart uploads.
+	//
+	// There are two possible values for this setting:
+	//
+	// 1. RequestChecksumCalculationWhenSupported (default): The checksum is always calculated
+	//    if the operation supports it, regardless of whether the user sets an algorithm in the request.
+	//
+	// 2. RequestChecksumCalculationWhenRequired: The checksum is only calculated if the user
+	//    explicitly sets a checksum algorithm in the request. This preserves backwards compatibility
+	//    for applications that don't want automatic checksum calculation.
+	//
+	// Note: S3 Express buckets always require CRC32 checksums regardless of this setting.
+	RequestChecksumCalculation aws.RequestChecksumCalculation
+
 	// partPool allows for the re-usage of streaming payload part buffers between upload calls
 	partPool byteSlicePool
 }
@@ -274,12 +290,13 @@ type Uploader struct {
 //	})
 func NewUploader(client UploadAPIClient, options ...func(*Uploader)) *Uploader {
 	u := &Uploader{
-		S3:                client,
-		PartSize:          DefaultUploadPartSize,
-		Concurrency:       DefaultUploadConcurrency,
-		LeavePartsOnError: false,
-		MaxUploadParts:    MaxUploadParts,
-		BufferProvider:    defaultUploadBufferProvider(),
+		S3:                         client,
+		PartSize:                   DefaultUploadPartSize,
+		Concurrency:                DefaultUploadConcurrency,
+		LeavePartsOnError:          false,
+		MaxUploadParts:             MaxUploadParts,
+		RequestChecksumCalculation: aws.RequestChecksumCalculationWhenSupported,
+		BufferProvider:             defaultUploadBufferProvider(),
 	}
 
 	for _, option := range options {
@@ -776,10 +793,23 @@ func (u *multiuploader) initChecksumAlgorithm() {
 	case u.in.ChecksumSHA256 != nil:
 		u.in.ChecksumAlgorithm = types.ChecksumAlgorithmSha256
 	default:
-		u.in.ChecksumAlgorithm = types.ChecksumAlgorithmCrc32
+		if u.isS3ExpressBucket() {
+			u.in.ChecksumAlgorithm = types.ChecksumAlgorithmCrc32
+			return
+		}
+
+		if u.cfg.RequestChecksumCalculation != aws.RequestChecksumCalculationWhenRequired {
+			u.in.ChecksumAlgorithm = types.ChecksumAlgorithmCrc32
+		}
 	}
 }
 
+// isS3ExpressBucket returns true if the bucket has the S3 Express suffix.
+func (u *multiuploader) isS3ExpressBucket() bool {
+	bucketName := aws.ToString(u.in.Bucket)
+	return strings.HasSuffix(bucketName, "--x-s3")
+}
+
 // geterr is a thread-safe getter for the error object
 func (u *multiuploader) geterr() error {
 	u.m.Lock()

feature/s3/manager/upload_test.go

@@ -1013,6 +1013,155 @@ func TestUploaderValidARN(t *testing.T) {
 	}
 }
 
+// TestUploadRequestChecksumCalculation tests that checksum behavior respects
+// the RequestChecksumCalculation setting for backwards compatibility.
+func TestUploadRequestChecksumCalculation(t *testing.T) {
+	testCases := []struct {
+		name                       string
+		requestChecksumCalculation aws.RequestChecksumCalculation
+		inputChecksumAlgorithm     types.ChecksumAlgorithm
+		expectedChecksumAlgorithm  types.ChecksumAlgorithm
+		description                string
+	}{
+		{
+			name:                       "WhenRequired_NoDefault",
+			requestChecksumCalculation: aws.RequestChecksumCalculationWhenRequired,
+			inputChecksumAlgorithm:     "",
+			expectedChecksumAlgorithm:  "",
+			description:                "When RequestChecksumCalculationWhenRequired is set, no default checksum should be applied (backwards compatibility)",
+		},
+		{
+			name:                       "WhenSupported_HasDefault",
+			requestChecksumCalculation: aws.RequestChecksumCalculationWhenSupported,
+			inputChecksumAlgorithm:     "",
+			expectedChecksumAlgorithm:  types.ChecksumAlgorithmCrc32,
+			description:                "When RequestChecksumCalculationWhenSupported is set, default CRC32 checksum should be applied",
+		},
+		{
+			name:                       "WhenRequired_ExplicitPreserved",
+			requestChecksumCalculation: aws.RequestChecksumCalculationWhenRequired,
+			inputChecksumAlgorithm:     types.ChecksumAlgorithmSha256,
+			expectedChecksumAlgorithm:  types.ChecksumAlgorithmSha256,
+			description:                "Explicit checksums should always be preserved regardless of RequestChecksumCalculation setting",
+		},
+		{
+			name:                       "WhenSupported_ExplicitPreserved",
+			requestChecksumCalculation: aws.RequestChecksumCalculationWhenSupported,
+			inputChecksumAlgorithm:     types.ChecksumAlgorithmSha1,
+			expectedChecksumAlgorithm:  types.ChecksumAlgorithmSha1,
+			description:                "Explicit checksums should override defaults even with WhenSupported",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			c, invocations, args := s3testing.NewUploadLoggingClient(nil)
+
+			// Configure uploader with RequestChecksumCalculation directly on the uploader
+			mgr := manager.NewUploader(c, func(u *manager.Uploader) {
+				u.RequestChecksumCalculation = tc.requestChecksumCalculation
+			})
+
+			input := &s3.PutObjectInput{
+				Bucket: aws.String("Bucket"),
+				Key:    aws.String("Key"),
+				Body:   bytes.NewReader(buf12MB), // Large enough to trigger multipart upload
+			}
+			if tc.inputChecksumAlgorithm != "" {
+				input.ChecksumAlgorithm = tc.inputChecksumAlgorithm
+			}
+
+			resp, err := mgr.Upload(context.Background(), input)
+			if err != nil {
+				t.Errorf("Expected no error but received %v", err)
+			}
+
+			expectedOps := []string{"CreateMultipartUpload", "UploadPart", "UploadPart", "UploadPart", "CompleteMultipartUpload"}
+			if diff := cmpDiff(expectedOps, *invocations); len(diff) > 0 {
+				t.Error(diff)
+			}
+
+			if resp.UploadID != "UPLOAD-ID" {
+				t.Errorf("expect %q, got %q", "UPLOAD-ID", resp.UploadID)
+			}
+
+			cmu := (*args)[0].(*s3.CreateMultipartUploadInput)
+			if cmu.ChecksumAlgorithm != tc.expectedChecksumAlgorithm {
+				t.Errorf("%s: Expected checksum algorithm %v in CreateMultipartUpload, but got %v",
+					tc.description, tc.expectedChecksumAlgorithm, cmu.ChecksumAlgorithm)
+			}
+
+			for i := 1; i <= 3; i++ {
+				uploadPart := (*args)[i].(*s3.UploadPartInput)
+				if uploadPart.ChecksumAlgorithm != tc.expectedChecksumAlgorithm {
+					t.Errorf("%s: Expected checksum algorithm %v in UploadPart %d, but got %v",
+						tc.description, tc.expectedChecksumAlgorithm, i, uploadPart.ChecksumAlgorithm)
+				}
+			}
+		})
+	}
+}
+
+// TestUploadS3ExpressAlwaysRequiresChecksum tests that S3 Express buckets
+// always get CRC32 checksums regardless of RequestChecksumCalculation setting.
+func TestUploadS3ExpressAlwaysRequiresChecksum(t *testing.T) {
+	testCases := []struct {
+		name                       string
+		requestChecksumCalculation aws.RequestChecksumCalculation
+		description                string
+	}{
+		{
+			name:                       "S3Express_WhenRequired_StillGetsCRC32",
+			requestChecksumCalculation: aws.RequestChecksumCalculationWhenRequired,
+			description:                "S3 Express buckets should get CRC32 even with RequestChecksumCalculationWhenRequired",
+		},
+		{
+			name:                       "S3Express_WhenSupported_GetsCRC32",
+			requestChecksumCalculation: aws.RequestChecksumCalculationWhenSupported,
+			description:                "S3 Express buckets should get CRC32 with RequestChecksumCalculationWhenSupported",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			c, invocations, args := s3testing.NewUploadLoggingClient(nil)
+
+			mgr := manager.NewUploader(c, func(u *manager.Uploader) {
+				u.RequestChecksumCalculation = tc.requestChecksumCalculation
+			})
+
+			_, err := mgr.Upload(context.Background(), &s3.PutObjectInput{
+				Bucket: aws.String("my-express-bucket--x-s3"),
+				Key:    aws.String("Key"),
+				Body:   bytes.NewReader(buf12MB),
+			})
+
+			if err != nil {
+				t.Errorf("Expected no error but received %v", err)
+			}
+
+			expectedOps := []string{"CreateMultipartUpload", "UploadPart", "UploadPart", "UploadPart", "CompleteMultipartUpload"}
+			if diff := cmpDiff(expectedOps, *invocations); len(diff) > 0 {
+				t.Error(diff)
+			}
+
+			cmu := (*args)[0].(*s3.CreateMultipartUploadInput)
+			if cmu.ChecksumAlgorithm != types.ChecksumAlgorithmCrc32 {
+				t.Errorf("%s: Expected CRC32 checksum for S3 Express bucket, but got %v",
+					tc.description, cmu.ChecksumAlgorithm)
+			}
+
+			for i := 1; i <= 3; i++ {
+				uploadPart := (*args)[i].(*s3.UploadPartInput)
+				if uploadPart.ChecksumAlgorithm != types.ChecksumAlgorithmCrc32 {
+					t.Errorf("%s: Expected CRC32 checksum in UploadPart %d for S3 Express bucket, but got %v",
+						tc.description, i, uploadPart.ChecksumAlgorithm)
+				}
+			}
+		})
+	}
+}
+
 type mockS3UploadServer struct {
 	*http.ServeMux