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 7a675c9565 [filesystems][oss] support customer CMK (BYOK) SSE-KMS via
fs.oss.server-side-encryption-key-id (#8331)
7a675c9565 is described below
commit 7a675c9565897e5a12bc879aa8ca8100e7f8c42e
Author: Jiajia Li <[email protected]>
AuthorDate: Tue Jun 23 21:48:34 2026 +0800
[filesystems][oss] support customer CMK (BYOK) SSE-KMS via
fs.oss.server-side-encryption-key-id (#8331)
hadoop-aliyun exposes only the SSE algorithm
(fs.oss.server-side-encryption-algorithm),
never a custom CMK key-id, even though the OSS SDK supports it.
Following the pattern of
PR #6413 (reflect into the underlying OSSClient), when
fs.oss.server-side-encryption-key-id
is set we swap the client's object/multipart operations for subclasses
that stamp the
OSS-native SSE-KMS headers (x-oss-server-side-encryption=KMS +
x-oss-server-side-encryption-key-id=<CMK>) onto the write paths.
PutObject, InitiateMultipartUpload, and CopyObject (server-side copy
used by rename/commit)
are overridden so every write Paimon performs carries the CMK; multipart
parts inherit it
from the initiation. UploadPart/HeadObject/GetObject are left untouched
(stamping SSE headers
on those makes OSS reject the request with 400).
For PutObject/InitiateMultipartUpload the CMK is set on the request's
ObjectMetadata; for
CopyObject it is set on the CopyObjectRequest's own SSE fields (not
newObjectMetadata), so the
copy is encrypted without forcing a metadata-directive REPLACE that
would drop the source
object's Content-Type / user metadata.
The key-id is validated at filesystem init (rejecting blank or
whitespace-containing values so
config errors fail fast), and the operation swap is fail-closed: if a
future SDK change makes
the swap a no-op, init throws rather than silently writing unencrypted
objects. Adds unit tests
pinning the reflection contract (field names + getters) against the
bundled aliyun-oss-sdk and
covering the new ReflectionUtils.setPrivateFieldValue.
---
docs/docs/maintenance/filesystems.mdx | 10 ++
.../org/apache/paimon/utils/ReflectionUtils.java | 16 +++
.../apache/paimon/utils/ReflectionUtilsTest.java | 60 ++++++++
.../main/java/org/apache/paimon/oss/OSSFileIO.java | 151 ++++++++++++++++++++-
.../java/org/apache/paimon/oss/OSSFileIOTest.java | 46 +++++++
5 files changed, 279 insertions(+), 4 deletions(-)
diff --git a/docs/docs/maintenance/filesystems.mdx
b/docs/docs/maintenance/filesystems.mdx
index 98a90ec45e..8f1cc844a5 100644
--- a/docs/docs/maintenance/filesystems.mdx
+++ b/docs/docs/maintenance/filesystems.mdx
@@ -332,6 +332,16 @@ Download
[paimon-jindo-@@VERSION@@.jar](https://repository.apache.org/snapshots/
</Unstable>
+### Server-Side Encryption with Customer CMK (SSE-KMS)
+
+To encrypt OSS writes with a customer-managed CMK (BYOK) via SSE-KMS,
configure the CMK key id:
+
+```yaml
+fs.oss.server-side-encryption-key-id: your-cmk-key-id
+```
+
+When set, Paimon stamps the given CMK on the OSS write paths it performs:
`PutObject`, server-side `CopyObject` (used by rename/commit), and
multipart-upload initiation (parts inherit the CMK).
+
## S3
<Stable>
diff --git
a/paimon-common/src/main/java/org/apache/paimon/utils/ReflectionUtils.java
b/paimon-common/src/main/java/org/apache/paimon/utils/ReflectionUtils.java
index 3d02271959..8d2da7da71 100644
--- a/paimon-common/src/main/java/org/apache/paimon/utils/ReflectionUtils.java
+++ b/paimon-common/src/main/java/org/apache/paimon/utils/ReflectionUtils.java
@@ -81,4 +81,20 @@ public class ReflectionUtils {
}
throw new NoSuchFieldException(fieldName);
}
+
+ public static void setPrivateFieldValue(Object obj, String fieldName,
Object value)
+ throws NoSuchFieldException, IllegalAccessException {
+ Class<?> clazz = obj.getClass();
+ while (clazz != null) {
+ try {
+ Field field = clazz.getDeclaredField(fieldName);
+ field.setAccessible(true);
+ field.set(obj, value);
+ return;
+ } catch (NoSuchFieldException e) {
+ clazz = clazz.getSuperclass();
+ }
+ }
+ throw new NoSuchFieldException(fieldName);
+ }
}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/utils/ReflectionUtilsTest.java
b/paimon-common/src/test/java/org/apache/paimon/utils/ReflectionUtilsTest.java
new file mode 100644
index 0000000000..32030502af
--- /dev/null
+++
b/paimon-common/src/test/java/org/apache/paimon/utils/ReflectionUtilsTest.java
@@ -0,0 +1,60 @@
+/*
+ * 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.utils;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link ReflectionUtils}. */
+public class ReflectionUtilsTest {
+
+ @Test
+ public void testSetPrivateFieldValueOnDeclaringClass() throws Exception {
+ Child child = new Child();
+ ReflectionUtils.setPrivateFieldValue(child, "ownField", "new-value");
+ assertThat((String) ReflectionUtils.getPrivateFieldValue(child,
"ownField"))
+ .isEqualTo("new-value");
+ }
+
+ @Test
+ public void testSetPrivateFieldValueWalksSuperclass() throws Exception {
+ Child child = new Child();
+ ReflectionUtils.setPrivateFieldValue(child, "parentField",
"from-parent");
+ assertThat((String) ReflectionUtils.getPrivateFieldValue(child,
"parentField"))
+ .isEqualTo("from-parent");
+ }
+
+ @Test
+ public void testSetPrivateFieldValueUnknownFieldThrows() {
+ assertThatThrownBy(() -> ReflectionUtils.setPrivateFieldValue(new
Child(), "missing", "x"))
+ .isInstanceOf(NoSuchFieldException.class);
+ }
+
+ private static class Parent {
+ @SuppressWarnings("unused")
+ private String parentField = "parent";
+ }
+
+ private static class Child extends Parent {
+ @SuppressWarnings("unused")
+ private String ownField = "child";
+ }
+}
diff --git
a/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java
b/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java
index b9c18bf994..e126277288 100644
---
a/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java
+++
b/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java
@@ -30,8 +30,17 @@ import org.apache.paimon.utils.StringUtils;
import com.aliyun.oss.OSSClient;
import com.aliyun.oss.OSSException;
+import com.aliyun.oss.common.auth.CredentialsProvider;
import com.aliyun.oss.common.comm.ServiceClient;
+import com.aliyun.oss.internal.OSSMultipartOperation;
+import com.aliyun.oss.internal.OSSObjectOperation;
+import com.aliyun.oss.model.CopyObjectRequest;
+import com.aliyun.oss.model.CopyObjectResult;
+import com.aliyun.oss.model.InitiateMultipartUploadRequest;
+import com.aliyun.oss.model.InitiateMultipartUploadResult;
import com.aliyun.oss.model.ObjectMetadata;
+import com.aliyun.oss.model.PutObjectRequest;
+import com.aliyun.oss.model.PutObjectResult;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem;
@@ -69,6 +78,10 @@ public class OSSFileIO extends HadoopCompliantFileIO
implements HadoopOptionsPro
private static final String OSS_ACCESS_KEY_SECRET =
"fs.oss.accessKeySecret";
private static final String OSS_SECURITY_TOKEN = "fs.oss.securityToken";
private static final String OSS_SECOND_LEVEL_DOMAIN_ENABLED =
"fs.oss.sld.enabled";
+ private static final String OSS_SSE_KMS_KEY_ID =
"fs.oss.server-side-encryption-key-id";
+
+ /** OSS-native value for the {@code x-oss-server-side-encryption} header
when using a CMK. */
+ private static final String SSE_KMS_ALGORITHM = "KMS";
private static final Map<String, String> CASE_SENSITIVE_KEYS =
new HashMap<String, String>() {
@@ -174,6 +187,22 @@ public class OSSFileIO extends HadoopCompliantFileIO
implements HadoopOptionsPro
enableSecondLevelDomain(fs);
}
+ String sseKmsKeyId = hadoopOptions.get(OSS_SSE_KMS_KEY_ID);
+ if (sseKmsKeyId != null) {
+ String trimmed = sseKmsKeyId.trim();
+ // Reject what can never be a key id, so config errors
fail fast at
+ // init instead of mid-job.
+ if (trimmed.isEmpty()
+ ||
trimmed.chars().anyMatch(Character::isWhitespace)) {
+ throw new IllegalArgumentException(
+ "Invalid value for '"
+ + OSS_SSE_KMS_KEY_ID
+ + "': the CMK key id must not be
blank or contain"
+ + " whitespace/newlines.");
+ }
+ enableSseKms(fs, trimmed);
+ }
+
return fs;
};
@@ -203,9 +232,8 @@ public class OSSFileIO extends HadoopCompliantFileIO
implements HadoopOptionsPro
}
AliyunOSSFileSystem fs = (AliyunOSSFileSystem)
getFileSystem(path(path));
- AliyunOSSFileSystemStore store = fs.getStore();
try {
- OSSClient ossClient = ReflectionUtils.getPrivateFieldValue(store,
"ossClient");
+ OSSClient ossClient = getOssClient(fs);
ossClient.putObject(bucket, objectKey, new
ByteArrayInputStream(bytes), metadata);
return true;
} catch (OSSException e) {
@@ -228,9 +256,8 @@ public class OSSFileIO extends HadoopCompliantFileIO
implements HadoopOptionsPro
}
public void enableSecondLevelDomain(AliyunOSSFileSystem fs) {
- AliyunOSSFileSystemStore store = fs.getStore();
try {
- OSSClient ossClient = ReflectionUtils.getPrivateFieldValue(store,
"ossClient");
+ OSSClient ossClient = getOssClient(fs);
ServiceClient serviceClient =
ReflectionUtils.getPrivateFieldValue(ossClient,
"serviceClient");
serviceClient.getClientConfiguration().setSLDEnabled(true);
@@ -240,6 +267,122 @@ public class OSSFileIO extends HadoopCompliantFileIO
implements HadoopOptionsPro
}
}
+ /** Reflectively extract the underlying {@link OSSClient} that
hadoop-aliyun keeps private. */
+ private static OSSClient getOssClient(AliyunOSSFileSystem fs) throws
Exception {
+ AliyunOSSFileSystemStore store = fs.getStore();
+ return ReflectionUtils.getPrivateFieldValue(store, "ossClient");
+ }
+
+ /**
+ * Enable SSE-KMS with a customer CMK. hadoop-aliyun exposes no CMK key-id
option, so swap the
+ * OSS client's object/multipart operations for subclasses that stamp the
OSS-native SSE-KMS
+ * headers on the write paths: PutObject, server-side CopyObject
(rename/commit) and
+ * InitiateMultipartUpload. UploadPart/HeadObject/GetObject are left
untouched (stamping SSE
+ * headers on those makes OSS reject the request with 400).
+ */
+ private void enableSseKms(AliyunOSSFileSystem fs, String kmsKeyId) {
+ try {
+ swapSseKmsOperations(getOssClient(fs), kmsKeyId);
+ } catch (Exception e) {
+ LOG.error("Failed to enable SSE-KMS BYOK.", e);
+ throw new RuntimeException("Failed to enable SSE-KMS BYOK.", e);
+ }
+ }
+
+ /**
+ * Swap the OSS client's object/multipart operations for the CMK-stamping
subclasses. Package
+ * private so a unit test can pin the reflected field names and getters
against the bundled
+ * aliyun-oss-sdk version.
+ */
+ static void swapSseKmsOperations(OSSClient ossClient, String kmsKeyId)
throws Exception {
+ ServiceClient serviceClient =
+ ReflectionUtils.getPrivateFieldValue(ossClient,
"serviceClient");
+ CredentialsProvider credsProvider =
+ ReflectionUtils.getPrivateFieldValue(ossClient,
"credsProvider");
+ // Replacement operations must inherit the endpoint or requests NPE.
+ URI endpoint = ossClient.getEndpoint();
+ SseKmsObjectOperation objectOperation =
+ new SseKmsObjectOperation(serviceClient, credsProvider,
kmsKeyId);
+ objectOperation.setEndpoint(endpoint);
+ SseKmsMultipartOperation multipartOperation =
+ new SseKmsMultipartOperation(serviceClient, credsProvider,
kmsKeyId);
+ multipartOperation.setEndpoint(endpoint);
+ ReflectionUtils.setPrivateFieldValue(ossClient, "objectOperation",
objectOperation);
+ ReflectionUtils.setPrivateFieldValue(ossClient, "multipartOperation",
multipartOperation);
+ // Fail closed: if a future SDK change makes the swap a no-op, fail
loudly
+ // rather than silently writing objects without the customer CMK.
+ if (ossClient.getObjectOperation() != objectOperation
+ || ossClient.getMultipartOperation() != multipartOperation) {
+ throw new IllegalStateException(
+ "SSE-KMS BYOK operation swap did not take effect; refusing
to write "
+ + "without server-side encryption. The
aliyun-oss-sdk internals "
+ + "may have changed.");
+ }
+ }
+
+ /** Stamps the customer CMK on the given metadata, allocating one if
absent. */
+ private static ObjectMetadata applySseKms(ObjectMetadata metadata, String
kmsKeyId) {
+ if (metadata == null) {
+ metadata = new ObjectMetadata();
+ }
+ String existing = metadata.getServerSideEncryption();
+ if (existing != null && !SSE_KMS_ALGORITHM.equals(existing)) {
+ LOG.warn(
+ "Customer CMK SSE-KMS is overriding the previously-set
server-side "
+ + "encryption algorithm '{}' with 'KMS'.",
+ existing);
+ }
+ metadata.setServerSideEncryption(SSE_KMS_ALGORITHM);
+ metadata.setServerSideEncryptionKeyId(kmsKeyId);
+ return metadata;
+ }
+
+ /** Stamps the customer CMK on every simple PutObject. */
+ static class SseKmsObjectOperation extends OSSObjectOperation {
+ private final String kmsKeyId;
+
+ SseKmsObjectOperation(
+ ServiceClient serviceClient, CredentialsProvider
credsProvider, String kmsKeyId) {
+ super(serviceClient, credsProvider);
+ this.kmsKeyId = kmsKeyId;
+ }
+
+ @Override
+ public PutObjectResult putObject(PutObjectRequest request) {
+ request.setMetadata(applySseKms(request.getMetadata(), kmsKeyId));
+ return super.putObject(request);
+ }
+
+ @Override
+ public CopyObjectResult copyObject(CopyObjectRequest request) {
+ // Rename/commit uses server-side copy. populateCopyObjectHeaders
reads the SSE
+ // headers straight from these request fields, so setting them
here stamps the CMK
+ // without touching newObjectMetadata. Setting newObjectMetadata
would force
+ // metadata-directive REPLACE and drop the source object's
Content-Type/user metadata.
+ request.setServerSideEncryption(SSE_KMS_ALGORITHM);
+ request.setServerSideEncryptionKeyId(kmsKeyId);
+ return super.copyObject(request);
+ }
+ }
+
+ /** Stamps the customer CMK on the multipart-upload init; parts inherit
it. */
+ static class SseKmsMultipartOperation extends OSSMultipartOperation {
+ private final String kmsKeyId;
+
+ SseKmsMultipartOperation(
+ ServiceClient serviceClient, CredentialsProvider
credsProvider, String kmsKeyId) {
+ super(serviceClient, credsProvider);
+ this.kmsKeyId = kmsKeyId;
+ }
+
+ @Override
+ public InitiateMultipartUploadResult initiateMultipartUpload(
+ InitiateMultipartUploadRequest request) {
+ request.setObjectMetadata(applySseKms(request.getObjectMetadata(),
kmsKeyId));
+ return super.initiateMultipartUpload(request);
+ }
+ }
+
private static class CacheKey {
private final Options options;
diff --git
a/paimon-filesystems/paimon-oss-impl/src/test/java/org/apache/paimon/oss/OSSFileIOTest.java
b/paimon-filesystems/paimon-oss-impl/src/test/java/org/apache/paimon/oss/OSSFileIOTest.java
new file mode 100644
index 0000000000..e5100ce3ef
--- /dev/null
+++
b/paimon-filesystems/paimon-oss-impl/src/test/java/org/apache/paimon/oss/OSSFileIOTest.java
@@ -0,0 +1,46 @@
+/*
+ * 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.oss;
+
+import com.aliyun.oss.OSSClient;
+import com.aliyun.oss.OSSClientBuilder;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link OSSFileIO}. */
+public class OSSFileIOTest {
+
+ /** The swap must replace both operations with the CMK-stamping
subclasses. */
+ @Test
+ public void testSseKmsOperationSwapTakesEffect() throws Exception {
+ OSSClient ossClient =
+ (OSSClient) new
OSSClientBuilder().build("http://oss.example.com";, "ak", "sk");
+ try {
+ OSSFileIO.swapSseKmsOperations(ossClient, "my-cmk-key-id");
+
+ assertThat(ossClient.getObjectOperation())
+ .isInstanceOf(OSSFileIO.SseKmsObjectOperation.class);
+ assertThat(ossClient.getMultipartOperation())
+ .isInstanceOf(OSSFileIO.SseKmsMultipartOperation.class);
+ } finally {
+ ossClient.shutdown();
+ }
+ }
+}
