This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 99559570fe [filesystem][s3] Fix multipart part upload missing SSE-C
encryption parameters (#8969)
99559570fe is described below
commit 99559570fedb11a626c6e1e726713b1cbe182e32
Author: Eunbin Son <[email protected]>
AuthorDate: Mon Aug 3 20:04:25 2026 +0900
[filesystem][s3] Fix multipart part upload missing SSE-C encryption
parameters (#8969)
---
.../org/apache/paimon/s3/S3MultiPartUpload.java | 26 +++--
.../apache/paimon/s3/S3MultiPartUploadTest.java | 105 +++++++++++++++++++++
2 files changed, 124 insertions(+), 7 deletions(-)
diff --git
a/paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3/S3MultiPartUpload.java
b/paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3/S3MultiPartUpload.java
index ecd4686787..c995dd0881 100644
---
a/paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3/S3MultiPartUpload.java
+++
b/paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3/S3MultiPartUpload.java
@@ -18,6 +18,7 @@
package org.apache.paimon.s3;
+import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.fs.MultiPartUploadStore;
import org.apache.hadoop.conf.Configuration;
@@ -89,18 +90,29 @@ public class S3MultiPartUpload
String objectName, String uploadId, int partNumber, File file, int
byteLength)
throws IOException {
UploadPartRequest request =
- UploadPartRequest.builder()
- .bucket(s3a.getBucket())
- .key(objectName)
- .uploadId(uploadId)
- .partNumber(partNumber)
- .contentLength((long) byteLength)
- .build();
+ newUploadPartRequest(objectName, uploadId, partNumber,
byteLength);
RequestBody body =
RequestBody.fromBytes(Files.readAllBytes(file.toPath()));
UploadPartResponse response = s3accessHelper.uploadPart(request, body,
null);
return
CompletedPart.builder().partNumber(partNumber).eTag(response.eTag()).build();
}
+ /**
+ * Builds the part request through the S3A request factory, the same way
the multipart upload is
+ * initiated. Hand-assembling the request would drop the SSE-C encryption
parameters, which S3
+ * requires on every part when the upload was initiated with a
customer-provided key.
+ *
+ * <p>{@code isLastPart} is always {@code false}: the caller does not know
in advance which part
+ * is the last one, and the factory only uses the flag to set {@code
sdkPartType}, which the
+ * hand-assembled request never set either.
+ */
+ @VisibleForTesting
+ UploadPartRequest newUploadPartRequest(
+ String objectName, String uploadId, int partNumber, int
byteLength) throws IOException {
+ return s3accessHelper
+ .newUploadPartRequestBuilder(objectName, uploadId, partNumber,
false, byteLength)
+ .build();
+ }
+
@Override
public void abortMultipartUpload(String destKey, String uploadId) throws
IOException {
s3accessHelper.abortMultipartUpload(destKey, uploadId, false, null);
diff --git
a/paimon-filesystems/paimon-s3-impl/src/test/java/org/apache/paimon/s3/S3MultiPartUploadTest.java
b/paimon-filesystems/paimon-s3-impl/src/test/java/org/apache/paimon/s3/S3MultiPartUploadTest.java
new file mode 100644
index 0000000000..03702ff29c
--- /dev/null
+++
b/paimon-filesystems/paimon-s3-impl/src/test/java/org/apache/paimon/s3/S3MultiPartUploadTest.java
@@ -0,0 +1,105 @@
+/*
+ * 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.paimon.s3;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.s3a.S3AFileSystem;
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.services.s3.model.UploadPartRequest;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.Base64;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for the {@link UploadPartRequest} built by {@link S3MultiPartUpload}.
No S3 backend is
+ * contacted: {@code fs.s3a.bucket.probe=0} makes {@link
S3AFileSystem#initialize} skip the bucket
+ * existence check, so the request factory can be exercised offline.
+ */
+class S3MultiPartUploadTest {
+
+ private static final String BUCKET = "paimon-test-bucket";
+ private static final String OBJECT_NAME = "path/to/object";
+ private static final String UPLOAD_ID = "test-upload-id";
+
+ @Test
+ void testUploadPartRequestCarriesSseCustomerParameters() throws Exception {
+ Configuration conf = baseConfiguration();
+ conf.set("fs.s3a.encryption.algorithm", "SSE-C");
+ conf.set("fs.s3a.encryption.key", sseCustomerKey());
+
+ UploadPartRequest request = newUploadPartRequest(conf);
+
+ assertThat(request.sseCustomerAlgorithm()).isEqualTo("AES256");
+ assertThat(request.sseCustomerKey()).isNotNull();
+ assertThat(request.sseCustomerKeyMD5()).isNotNull();
+ }
+
+ @Test
+ void testUploadPartRequestWithoutEncryption() throws Exception {
+ UploadPartRequest request = newUploadPartRequest(baseConfiguration());
+
+ assertThat(request.sseCustomerAlgorithm()).isNull();
+ assertThat(request.sseCustomerKey()).isNull();
+ assertThat(request.sseCustomerKeyMD5()).isNull();
+ }
+
+ @Test
+ void testUploadPartRequestKeepsPartCoordinates() throws Exception {
+ UploadPartRequest request = newUploadPartRequest(baseConfiguration());
+
+ assertThat(request.bucket()).isEqualTo(BUCKET);
+ assertThat(request.key()).isEqualTo(OBJECT_NAME);
+ assertThat(request.uploadId()).isEqualTo(UPLOAD_ID);
+ assertThat(request.partNumber()).isEqualTo(3);
+ assertThat(request.contentLength()).isEqualTo(1024L);
+ assertThat(request.sdkPartType()).isNull();
+ }
+
+ private static UploadPartRequest newUploadPartRequest(Configuration conf)
throws IOException {
+ try (S3AFileSystem fs = new S3AFileSystem()) {
+ fs.initialize(URI.create("s3a://" + BUCKET + "/"), conf);
+ S3MultiPartUpload upload = new S3MultiPartUpload(fs, conf);
+ return upload.newUploadPartRequest(OBJECT_NAME, UPLOAD_ID, 3,
1024);
+ }
+ }
+
+ private static Configuration baseConfiguration() {
+ Configuration conf = new Configuration(false);
+ // Never reach the network: no bucket probe, dummy credentials and an
unroutable endpoint.
+ conf.setInt("fs.s3a.bucket.probe", 0);
+ conf.set("fs.s3a.endpoint", "http://localhost:1";);
+ conf.set("fs.s3a.endpoint.region", "us-east-1");
+ conf.set("fs.s3a.path.style.access", "true");
+ conf.set("fs.s3a.access.key", "dummy-access-key");
+ conf.set("fs.s3a.secret.key", "dummy-secret-key");
+ return conf;
+ }
+
+ /** A 256-bit key, base64 encoded, as required by {@code
fs.s3a.encryption.key} for SSE-C. */
+ private static String sseCustomerKey() {
+ byte[] key = new byte[32];
+ for (int i = 0; i < key.length; i++) {
+ key[i] = (byte) i;
+ }
+ return Base64.getEncoder().encodeToString(key);
+ }
+}
