[FLINK-39113][s3] Fix s3.sse.kms.encryption-context config in native s3 connector by gaborgsomogyi · Pull Request #28070 · apache/flink

gaborgsomogyi · 2026-04-29T15:07:07Z

What is the purpose of the change

s3.sse.kms.encryption-context was listed in the README as a supported configuration option for SSE-KMS, but the corresponding ConfigOption never existed in NativeS3FileSystemFactory. As a result, the encryption context was silently ignored regardless of what users configured, and the sseKms(keyId, context) code path in S3EncryptionConfig was dead code unreachable from production.

This PR closes the gap by adding the missing config option and wiring it end-to-end.

Brief change log

Fix: Add SSE_KMS_ENCRYPTION_CONTEXT (s3.sse.kms.encryption-context) ConfigOption to NativeS3FileSystemFactory and wire it through S3EncryptionConfig.fromConfig() — the feature was documented but never implemented
Fix: Update docs (README, s3.md EN/ZH) with correct aws:s3:arn encryption context example reflecting real AWS usage
Tests: Add unit tests for S3EncryptionConfig, S3ExceptionUtils, S3FileStatus, S3BlockLocation

Verifying this change

Existing and new unit tests.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): no
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
The S3 file system connector: yes

Documentation

Does this pull request introduce a new feature? no
If yes, how is the feature documented? docs

Was generative AI tooling used to co-author this PR?

Yes (please specify the tool below)

Generated-by: Claude code

gaborgsomogyi · 2026-04-29T15:07:47Z

cc @Samrat002

flinkbot · 2026-04-29T15:16:34Z

CI report:

d00a143 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

Samrat002

NativeS3OutputStream.uploadToS3() is a separate write path for small files. It builds a PutObjectRequest but currently doesn't call ssekmsEncryptionContext(). The context wired through NativeS3ObjectOperations.applyEncryption() won't apply to writes that go through NativeS3OutputStream. Could you extend that code path as well?

Samrat002

Left few comments.

Cheers,
Samrat

Samrat002 · 2026-05-01T05:34:29Z

    }

    @VisibleForTesting
+    @Nullable


Suggested change

@Nullable

Keep the field non-nullable. It must fail fast.

This alone is not enough because there are nullable marker scattered around and we should make that consistent... We need to check if we remove all then what will happen

Samrat002 · 2026-05-01T05:35:50Z

+     * Creates a config for SSE-KMS encryption with the default KMS key and an encryption context.
     *
+     * @param encryptionContext The encryption context key-value pairs
+     */
+    public static S3EncryptionConfig sseKms(Map<String, String> encryptionContext) {
+        return new S3EncryptionConfig(EncryptionType.SSE_KMS, null, encryptionContext);
+    }


Suggested change

* Creates a config for SSE-KMS encryption with the default KMS key and an encryption context.

*

* @param encryptionContext The encryption context key-value pairs

*/

public static S3EncryptionConfig sseKms(Map<String, String> encryptionContext) {

return new S3EncryptionConfig(EncryptionType.SSE_KMS, null, encryptionContext);

}

Samrat002 · 2026-05-01T05:37:15Z

+                return kmsKeyId != null && !kmsKeyId.isEmpty()
+                        ? sseKms(kmsKeyId, encryptionContext)
+                        : sseKms(encryptionContext);


Suggested change

return kmsKeyId != null && !kmsKeyId.isEmpty()

? sseKms(kmsKeyId, encryptionContext)

: sseKms(encryptionContext);

return sseKms(kmsKeyId, encryptionContext);

This can be abstracted into a single method. There is no requirement for an overloaded method for sseKms

public static S3EncryptionConfig sseKms( String kmsKeyId, Map<String, String> encryptionContext) { return new S3EncryptionConfig(EncryptionType.SSE_KMS, kmsKeyId, encryptionContext); if (kmsKeyId != null && !kmsKeyId.isEmpty()) { return new S3EncryptionConfig(EncryptionType.SSE_KMS, kmsKeyId, encryptionContext); } else { return new S3EncryptionConfig(EncryptionType.SSE_KMS, null, encryptionContext); }

Samrat002 · 2026-05-01T05:40:06Z

+    @MethodSource
+    void sseKms_contextOnlyFactory_absentContext_hasEncryptionContextFalse(
+            Map<String, String> context) {
+        S3EncryptionConfig c = S3EncryptionConfig.sseKms(context);


Suggested change

S3EncryptionConfig c = S3EncryptionConfig.sseKms(context);

S3EncryptionConfig c = S3EncryptionConfig.sseKms(null, context);

Samrat002 · 2026-05-01T05:40:54Z

+    @Test
+    void sseKms_contextOnlyFactory_contextMutatedAfterCreation_contextUnchanged() {
+        Map<String, String> ctx = new HashMap<>(Map.of("dept", "finance"));
+        S3EncryptionConfig c = S3EncryptionConfig.sseKms(ctx);


Suggested change

S3EncryptionConfig c = S3EncryptionConfig.sseKms(ctx);

S3EncryptionConfig c = S3EncryptionConfig.sseKms(null, ctx);

Samrat002 · 2026-05-01T05:43:22Z

+    @MethodSource
+    void serializeEncryptionContext_jsonSpecialChars_escapedCorrectly(
+            String key, String value, String expectedFragment) {
+        S3EncryptionConfig c = S3EncryptionConfig.sseKms(Map.of(key, value));


Suggested change

S3EncryptionConfig c = S3EncryptionConfig.sseKms(Map.of(key, value));

S3EncryptionConfig c = S3EncryptionConfig.sseKms(null, Map.of(key, value));

Samrat002 · 2026-05-01T05:45:46Z

+    private static S3Exception s3ExceptionStatusOnly(int statusCode) {
+        S3Exception.Builder b = S3Exception.builder();
+        b.statusCode(statusCode);
+        return (S3Exception) b.build();
+    }


Suggested change

private static S3Exception s3ExceptionStatusOnly(int statusCode) {

S3Exception.Builder b = S3Exception.builder();

b.statusCode(statusCode);

return (S3Exception) b.build();

}

private static AwsServiceException s3ExceptionStatusOnly(int statusCode) {

return S3Exception.builder().statusCode(statusCode).build();

}

Samrat002 · 2026-05-01T05:46:20Z

+    private static S3Exception s3ExceptionWithMessageAndDetails(
+            int statusCode, String message, AwsErrorDetails details) {
+        S3Exception.Builder b = S3Exception.builder();
+        b.statusCode(statusCode);
+        b.message(message);
+        b.awsErrorDetails(details);
+        return (S3Exception) b.build();
+    }


Suggested change

private static S3Exception s3ExceptionWithMessageAndDetails(

int statusCode, String message, AwsErrorDetails details) {

S3Exception.Builder b = S3Exception.builder();

b.statusCode(statusCode);

b.message(message);

b.awsErrorDetails(details);

return (S3Exception) b.build();

}

private static AwsServiceException s3ExceptionWithMessageAndDetails(

int statusCode, String message, AwsErrorDetails details) {

return S3Exception.builder()

.statusCode(statusCode)

.message(message)

.awsErrorDetails(details)

.build();

}

Samrat002 · 2026-05-01T05:46:55Z

+    private static S3Exception s3Exception(int statusCode, String message) {
+        S3Exception.Builder b = S3Exception.builder();
+        b.statusCode(statusCode);
+        b.message(message);
+        return (S3Exception) b.build();
+    }


Suggested change

private static S3Exception s3Exception(int statusCode, String message) {

S3Exception.Builder b = S3Exception.builder();

b.statusCode(statusCode);

b.message(message);

return (S3Exception) b.build();

}

private static AwsServiceException s3Exception(int statusCode, String message) {

return S3Exception.builder().statusCode(statusCode).message(message).build();

}

Samrat002 · 2026-05-01T05:47:21Z

+    private static S3Exception s3Exception(int statusCode, AwsErrorDetails details) {
+        S3Exception.Builder b = S3Exception.builder();
+        b.statusCode(statusCode);
+        b.awsErrorDetails(details);
+        return (S3Exception) b.build();
+    }


Suggested change

private static S3Exception s3Exception(int statusCode, AwsErrorDetails details) {

S3Exception.Builder b = S3Exception.builder();

b.statusCode(statusCode);

b.awsErrorDetails(details);

return (S3Exception) b.build();

}

private static AwsServiceException s3Exception(int statusCode, AwsErrorDetails details) {

return S3Exception.builder().statusCode(statusCode).awsErrorDetails(details).build();

}

…s3 connector

…ryption context - Switch fromConfig() normalization to toLowerCase(Locale.ROOT), removing the SSE_KMS placeholder case - Add sseKms(Map) factory overload so encryption context is preserved when using the default AWS-managed key - Revert Map.copyOf() to null-tolerant unmodifiableMap(new HashMap<>())

…eam write path Move serializeEncryptionContext to S3EncryptionConfig and apply it in NativeS3OutputStream.uploadToS3(), which previously dropped the KMS encryption context for small-file (non-multipart) writes.

… change

gaborgsomogyi force-pushed the FLINK-39579 branch from d335870 to b3f1760 Compare April 29, 2026 15:34

gaborgsomogyi changed the title ~~[FLINK-39579][s3] Fix s3.sse.kms.encryption-context config in native s3 connector~~ [FLINK-39113][s3] Fix s3.sse.kms.encryption-context config in native s3 connector Apr 29, 2026

gaborgsomogyi force-pushed the FLINK-39579 branch from b3f1760 to 8e16a10 Compare April 30, 2026 07:37

Samrat002 reviewed Apr 30, 2026

View reviewed changes

gaborgsomogyi force-pushed the FLINK-39579 branch 2 times, most recently from 69ec88c to cdd8670 Compare April 30, 2026 16:02

github-actions Bot added the community-reviewed PR has been reviewed by the community. label Apr 30, 2026

gaborgsomogyi requested a review from Samrat002 April 30, 2026 16:44

Samrat002 reviewed May 1, 2026

View reviewed changes

gaborgsomogyi added 5 commits May 1, 2026 10:07

[FLINK-39113][s3] Fix s3.sse.kms.encryption-context config in native …

0c9bff4

…s3 connector

[FLINK-39113][s3] Fix missing encryption context in NativeS3OutputStr…

3d8b5dd

…eam write path Move serializeEncryptionContext to S3EncryptionConfig and apply it in NativeS3OutputStream.uploadToS3(), which previously dropped the KMS encryption context for small-file (non-multipart) writes.

[FLINK-39113][s3] Fix stale Javadoc on fromConfig after normalization…

a1874e3

… change

Test simplification and restructuring

d00a143

gaborgsomogyi force-pushed the FLINK-39579 branch from cdd8670 to d00a143 Compare May 1, 2026 08:08

	S3EncryptionConfig c = S3EncryptionConfig.sseKms(context);
	S3EncryptionConfig c = S3EncryptionConfig.sseKms(null, context);

Conversation

gaborgsomogyi commented Apr 29, 2026

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Was generative AI tooling used to co-author this PR?

Uh oh!

gaborgsomogyi commented Apr 29, 2026

Uh oh!

flinkbot commented Apr 29, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

Samrat002 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Samrat002 left a comment

Choose a reason for hiding this comment

Uh oh!

Samrat002 May 1, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

flinkbot commented Apr 29, 2026 •

edited

Loading

Samrat002 May 1, 2026 •

edited

Loading