Samrat002 commented on code in PR #28070:
URL: https://github.com/apache/flink/pull/28070#discussion_r3172225222
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3EncryptionConfig.java:
##########
@@ -141,29 +151,32 @@ public static S3EncryptionConfig sseKms(
/**
* Creates an encryption config from configuration strings.
*
- * @param encryptionTypeStr The encryption type: "none", "sse-s3",
"sse-kms", or "SSE_S3",
- * "SSE_KMS"
+ * @param encryptionTypeStr The encryption type: "none", "sse-s3",
"sse-kms", "aws:kms",
+ * "aes256" (case-insensitive)
* @param kmsKeyId The KMS key ID (required for SSE-KMS, ignored for other
types)
* @return The encryption configuration
* @throws IllegalArgumentException if the encryption type is invalid
*/
public static S3EncryptionConfig fromConfig(
- @Nullable String encryptionTypeStr, @Nullable String kmsKeyId) {
- if (encryptionTypeStr == null
- || encryptionTypeStr.isEmpty()
+ @Nullable String encryptionTypeStr,
+ @Nullable String kmsKeyId,
+ Map<String, String> encryptionContext) {
+ if (StringUtils.isNullOrWhitespaceOnly(encryptionTypeStr)
|| "none".equalsIgnoreCase(encryptionTypeStr)) {
return none();
}
- String normalizedType = encryptionTypeStr.toUpperCase().replace("-",
"_").replace(":", "_");
+ String normalizedType = encryptionTypeStr.toLowerCase(Locale.ROOT);
switch (normalizedType) {
- case "SSE_S3":
- case "AES256":
+ case "sse-s3":
+ case "aes256":
return sseS3();
- case "SSE_KMS":
- case "AWS_KMS":
- return kmsKeyId != null && !kmsKeyId.isEmpty() ?
sseKms(kmsKeyId) : sseKms();
+ case "sse-kms":
+ case "aws:kms":
+ return kmsKeyId != null && !kmsKeyId.isEmpty()
+ ? sseKms(kmsKeyId, encryptionContext)
+ : sseKms(encryptionContext);
Review Comment:
```suggestion
return sseKms(kmsKeyId, encryptionContext);
```
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3ClientProvider.java:
##########
@@ -135,6 +135,7 @@ public S3EncryptionConfig getEncryptionConfig() {
}
@VisibleForTesting
+ @Nullable
Review Comment:
```suggestion
```
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/S3EncryptionConfigTest.java:
##########
@@ -0,0 +1,239 @@
+/*
+ * 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.flink.fs.s3native;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import software.amazon.awssdk.services.s3.model.ServerSideEncryption;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Stream;
+
+import static
org.apache.flink.fs.s3native.S3EncryptionConfig.EncryptionType.NONE;
+import static
org.apache.flink.fs.s3native.S3EncryptionConfig.EncryptionType.SSE_KMS;
+import static
org.apache.flink.fs.s3native.S3EncryptionConfig.EncryptionType.SSE_S3;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link S3EncryptionConfig}. */
+class S3EncryptionConfigTest {
+
+ @Test
+ void sseKms_withContext_contextStoredDefensively() {
+ Map<String, String> ctx = new HashMap<>(Map.of("dept", "finance"));
+ S3EncryptionConfig c = S3EncryptionConfig.sseKms("key-id", ctx);
+ ctx.put("extra", "value");
+
+ assertThat(c.getEncryptionContext()).isEqualTo(Map.of("dept",
"finance"));
+ assertThat(c.hasEncryptionContext()).isTrue();
+ }
+
+ @Test
+ void sseKms_nullContext_contextIsEmpty() {
+ S3EncryptionConfig c = S3EncryptionConfig.sseKms("key-id", null);
+
+ assertThat(c.getEncryptionContext()).isEmpty();
+ assertThat(c.hasEncryptionContext()).isFalse();
+ }
+
+ @ParameterizedTest
+ @MethodSource
+ void sseKms_contextOnlyFactory_absentContext_hasEncryptionContextFalse(
+ Map<String, String> context) {
+ S3EncryptionConfig c = S3EncryptionConfig.sseKms(context);
+
+ assertThat(c.getEncryptionContext()).isEmpty();
+ assertThat(c.hasEncryptionContext()).isFalse();
+ }
+
+ static Stream<Arguments>
sseKms_contextOnlyFactory_absentContext_hasEncryptionContextFalse() {
+ return Stream.of(Arguments.of(Collections.emptyMap()),
Arguments.of((Object) null));
+ }
+
+ @Test
+ void
sseKms_contextOnlyFactory_contextMutatedAfterCreation_contextUnchanged() {
+ Map<String, String> ctx = new HashMap<>(Map.of("dept", "finance"));
+ S3EncryptionConfig c = S3EncryptionConfig.sseKms(ctx);
+ ctx.put("extra", "value");
+
+ assertThat(c.getEncryptionContext()).isEqualTo(Map.of("dept",
"finance"));
+ }
+
+ @ParameterizedTest
+ @MethodSource
+ void getServerSideEncryption_allTypes_returnsCorrectSseValue(
+ String configType, ServerSideEncryption expected) {
+ S3EncryptionConfig c =
+ S3EncryptionConfig.fromConfig(configType, null,
Collections.emptyMap());
+
+ assertThat(c.getServerSideEncryption()).isEqualTo(expected);
+ }
+
+ static Stream<Arguments>
getServerSideEncryption_allTypes_returnsCorrectSseValue() {
+ return Stream.of(
+ Arguments.of(null, null),
+ Arguments.of("sse-s3", ServerSideEncryption.AES256),
+ Arguments.of("sse-kms", ServerSideEncryption.AWS_KMS));
+ }
+
+ @ParameterizedTest
+ @MethodSource
+ void fromConfig_typeVariants_returnExpectedType(
+ String input, S3EncryptionConfig.EncryptionType expected) {
+ assertThat(
+ S3EncryptionConfig.fromConfig(input, null,
Collections.emptyMap())
+ .getEncryptionType())
+ .isEqualTo(expected);
+ }
+
+ static Stream<Arguments> fromConfig_typeVariants_returnExpectedType() {
+ return Stream.of(
+ Arguments.of(null, NONE),
+ Arguments.of("", NONE),
+ Arguments.of("none", NONE),
+ Arguments.of("NONE", NONE),
+ Arguments.of(" ", NONE),
+ Arguments.of("sse-s3", SSE_S3),
+ Arguments.of("AES256", SSE_S3),
+ Arguments.of("sse-kms", SSE_KMS),
+ Arguments.of("aws:kms", SSE_KMS));
+ }
+
+ @Test
+ void fromConfig_sseKmsWithKeyAndContext_keyAndContextPreserved() {
+ S3EncryptionConfig result =
+ S3EncryptionConfig.fromConfig("sse-kms", "my-key",
Map.of("env", "prod"));
+
+ assertThat(result.getKmsKeyId()).isEqualTo("my-key");
+ assertThat(result.getEncryptionContext()).isEqualTo(Map.of("env",
"prod"));
+ }
+
+ @ParameterizedTest
+ @MethodSource
+ void fromConfig_sseKmsWithNoKeyId_keyIdIsNull(String kmsKeyId) {
+ assertThat(
+ S3EncryptionConfig.fromConfig("sse-kms", kmsKeyId,
Collections.emptyMap())
+ .getKmsKeyId())
+ .isNull();
+ }
+
+ static Stream<Arguments> fromConfig_sseKmsWithNoKeyId_keyIdIsNull() {
+ return Stream.of(Arguments.of((Object) null), Arguments.of(""));
+ }
+
+ @Test
+ void fromConfig_sseKmsDefaultKeyWithContext_contextPreserved() {
+ assertThat(
+ S3EncryptionConfig.fromConfig("sse-kms", null,
Map.of("env", "prod"))
+ .hasEncryptionContext())
+ .isTrue();
+ }
+
+ @Test
+ void fromConfig_sseS3WithContext_contextIgnored() {
+ S3EncryptionConfig c =
+ S3EncryptionConfig.fromConfig("sse-s3", null, Map.of("dept",
"finance"));
+
+ assertThat(c.getEncryptionType()).isEqualTo(SSE_S3);
+ assertThat(c.getEncryptionContext()).isEmpty();
+ }
+
+ @Test
+ void fromConfig_unknownType_throwsIllegalArgument() {
+ assertThatThrownBy(
+ () ->
+ S3EncryptionConfig.fromConfig(
+ "invalid-type", null,
Collections.emptyMap()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("invalid-type");
+ }
+
+ @ParameterizedTest
+ @MethodSource
+ void serializeEncryptionContext_exactOutput_correctBase64Json(
+ Map<String, String> context, String expectedDecoded) {
+ S3EncryptionConfig c = S3EncryptionConfig.sseKms(context);
+ String decoded = new
String(Base64.getDecoder().decode(c.serializeEncryptionContext()));
+
+ assertThat(decoded).isEqualTo(expectedDecoded);
+ }
+
+ static Stream<Arguments>
serializeEncryptionContext_exactOutput_correctBase64Json() {
+ return Stream.of(
+ Arguments.of(Collections.emptyMap(), "{}"),
+ Arguments.of(Map.of("k", "v"), "{\"k\":\"v\"}"));
+ }
+
+ @Test
+ void serializeEncryptionContext_multipleEntries_allEntriesPresent() {
+ S3EncryptionConfig c = S3EncryptionConfig.sseKms(Map.of("k1", "v1",
"k2", "v2"));
+ String decoded = new
String(Base64.getDecoder().decode(c.serializeEncryptionContext()));
+
+ assertThat(decoded).contains("\"k1\":\"v1\"", "\"k2\":\"v2\"");
+ }
+
+ @ParameterizedTest
+ @MethodSource
+ void serializeEncryptionContext_jsonSpecialChars_escapedCorrectly(
+ String key, String value, String expectedFragment) {
+ S3EncryptionConfig c = S3EncryptionConfig.sseKms(Map.of(key, value));
Review Comment:
```suggestion
S3EncryptionConfig c = S3EncryptionConfig.sseKms(null, Map.of(key,
value));
```
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3EncryptionConfig.java:
##########
@@ -105,8 +108,15 @@ public static S3EncryptionConfig sseKms() {
}
/**
- * Creates a config for SSE-KMS encryption with a specific KMS key.
+ * 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);
+ }
Review Comment:
```suggestion
```
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/S3EncryptionConfigTest.java:
##########
@@ -0,0 +1,239 @@
+/*
+ * 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.flink.fs.s3native;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import software.amazon.awssdk.services.s3.model.ServerSideEncryption;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Stream;
+
+import static
org.apache.flink.fs.s3native.S3EncryptionConfig.EncryptionType.NONE;
+import static
org.apache.flink.fs.s3native.S3EncryptionConfig.EncryptionType.SSE_KMS;
+import static
org.apache.flink.fs.s3native.S3EncryptionConfig.EncryptionType.SSE_S3;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link S3EncryptionConfig}. */
+class S3EncryptionConfigTest {
+
+ @Test
+ void sseKms_withContext_contextStoredDefensively() {
+ Map<String, String> ctx = new HashMap<>(Map.of("dept", "finance"));
+ S3EncryptionConfig c = S3EncryptionConfig.sseKms("key-id", ctx);
+ ctx.put("extra", "value");
+
+ assertThat(c.getEncryptionContext()).isEqualTo(Map.of("dept",
"finance"));
+ assertThat(c.hasEncryptionContext()).isTrue();
+ }
+
+ @Test
+ void sseKms_nullContext_contextIsEmpty() {
+ S3EncryptionConfig c = S3EncryptionConfig.sseKms("key-id", null);
+
+ assertThat(c.getEncryptionContext()).isEmpty();
+ assertThat(c.hasEncryptionContext()).isFalse();
+ }
+
+ @ParameterizedTest
+ @MethodSource
+ void sseKms_contextOnlyFactory_absentContext_hasEncryptionContextFalse(
+ Map<String, String> context) {
+ S3EncryptionConfig c = S3EncryptionConfig.sseKms(context);
Review Comment:
```suggestion
S3EncryptionConfig c = S3EncryptionConfig.sseKms(null, context);
```
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/S3ExceptionUtilsTest.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.flink.fs.s3native;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import software.amazon.awssdk.awscore.exception.AwsErrorDetails;
+import software.amazon.awssdk.services.s3.model.S3Exception;
+
+import java.io.IOException;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link S3ExceptionUtils}. */
+class S3ExceptionUtilsTest {
+
+ static Stream<Arguments> toIoExceptionCases() {
+ return Stream.of(
+ Arguments.of(
+ s3Exception(
+ 404,
+ AwsErrorDetails.builder().errorMessage("key
not found").build()),
+ "delete object",
+ "delete object (HTTP 404: key not found)"),
+ Arguments.of(
+ s3Exception(500, "internal error"),
+ "put object",
+ "put object (HTTP 500: internal error)"));
+ }
+
+ @ParameterizedTest
+ @MethodSource("toIoExceptionCases")
+ void toIOException_contextAndStatusCode_formatsMessageAndPreservesCause(
+ S3Exception cause, String context, String expectedMessage) {
+ IOException result = S3ExceptionUtils.toIOException(context, cause);
+
+ assertThat(result.getMessage()).contains(expectedMessage);
+ assertThat(result.getCause()).isSameAs(cause);
+ }
+
+ static Stream<Arguments> errorMessageExactCases() {
+ return Stream.of(
+ Arguments.of(
+ s3Exception(
+ 404,
+ AwsErrorDetails.builder()
+ .errorMessage("The specified key does
not exist.")
+ .build()),
+ "The specified key does not exist."),
+ Arguments.of(s3ExceptionStatusOnly(500), "Unknown S3 error"));
+ }
+
+ @ParameterizedTest
+ @MethodSource("errorMessageExactCases")
+ void errorMessage_exactReturn_matchesExpected(S3Exception e, String
expected) {
+ assertThat(S3ExceptionUtils.errorMessage(e)).isEqualTo(expected);
+ }
+
+ static Stream<Arguments> errorMessageFallbackCases() {
+ return Stream.of(
+ Arguments.of(s3Exception(500, "connection timeout"),
"connection timeout"),
+ Arguments.of(
+ s3ExceptionWithMessageAndDetails(
+ 500,
+ "fallback message",
+
AwsErrorDetails.builder().errorCode("InternalError").build()),
+ "fallback message"));
+ }
+
+ @ParameterizedTest
+ @MethodSource("errorMessageFallbackCases")
+ void errorMessage_fallbackToExceptionMessage_containsExpected(
+ S3Exception e, String expectedMessage) {
+ assertThat(S3ExceptionUtils.errorMessage(e)).contains(expectedMessage);
+ }
+
+ static Stream<Arguments> errorCodeAllCases() {
+ return Stream.of(
+ Arguments.of(
+ s3Exception(404,
AwsErrorDetails.builder().errorCode("NoSuchKey").build()),
+ "NoSuchKey"),
+ Arguments.of(s3Exception(500, "some error"), "Unknown"),
+ Arguments.of(
+ s3Exception(
+ 500,
+ AwsErrorDetails.builder()
+ .errorMessage("Something went wrong")
+ .build()),
+ "Unknown"));
+ }
+
+ @ParameterizedTest
+ @MethodSource("errorCodeAllCases")
+ void errorCode_allCases_returnsExpected(S3Exception e, String expected) {
+ assertThat(S3ExceptionUtils.errorCode(e)).isEqualTo(expected);
+ }
+
+ 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 S3Exception s3Exception(int statusCode, String message) {
+ S3Exception.Builder b = S3Exception.builder();
+ b.statusCode(statusCode);
+ b.message(message);
+ return (S3Exception) b.build();
+ }
+
+ 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 S3Exception s3ExceptionStatusOnly(int statusCode) {
+ S3Exception.Builder b = S3Exception.builder();
+ b.statusCode(statusCode);
+ return (S3Exception) b.build();
+ }
Review Comment:
```suggestion
private static AwsServiceException s3ExceptionStatusOnly(int statusCode)
{
return S3Exception.builder().statusCode(statusCode).build();
}
```
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/S3EncryptionConfigTest.java:
##########
@@ -0,0 +1,239 @@
+/*
+ * 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.flink.fs.s3native;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import software.amazon.awssdk.services.s3.model.ServerSideEncryption;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Stream;
+
+import static
org.apache.flink.fs.s3native.S3EncryptionConfig.EncryptionType.NONE;
+import static
org.apache.flink.fs.s3native.S3EncryptionConfig.EncryptionType.SSE_KMS;
+import static
org.apache.flink.fs.s3native.S3EncryptionConfig.EncryptionType.SSE_S3;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link S3EncryptionConfig}. */
+class S3EncryptionConfigTest {
+
+ @Test
+ void sseKms_withContext_contextStoredDefensively() {
+ Map<String, String> ctx = new HashMap<>(Map.of("dept", "finance"));
+ S3EncryptionConfig c = S3EncryptionConfig.sseKms("key-id", ctx);
+ ctx.put("extra", "value");
+
+ assertThat(c.getEncryptionContext()).isEqualTo(Map.of("dept",
"finance"));
+ assertThat(c.hasEncryptionContext()).isTrue();
+ }
+
+ @Test
+ void sseKms_nullContext_contextIsEmpty() {
+ S3EncryptionConfig c = S3EncryptionConfig.sseKms("key-id", null);
+
+ assertThat(c.getEncryptionContext()).isEmpty();
+ assertThat(c.hasEncryptionContext()).isFalse();
+ }
+
+ @ParameterizedTest
+ @MethodSource
+ void sseKms_contextOnlyFactory_absentContext_hasEncryptionContextFalse(
+ Map<String, String> context) {
+ S3EncryptionConfig c = S3EncryptionConfig.sseKms(context);
+
+ assertThat(c.getEncryptionContext()).isEmpty();
+ assertThat(c.hasEncryptionContext()).isFalse();
+ }
+
+ static Stream<Arguments>
sseKms_contextOnlyFactory_absentContext_hasEncryptionContextFalse() {
+ return Stream.of(Arguments.of(Collections.emptyMap()),
Arguments.of((Object) null));
+ }
+
+ @Test
+ void
sseKms_contextOnlyFactory_contextMutatedAfterCreation_contextUnchanged() {
+ Map<String, String> ctx = new HashMap<>(Map.of("dept", "finance"));
+ S3EncryptionConfig c = S3EncryptionConfig.sseKms(ctx);
Review Comment:
```suggestion
S3EncryptionConfig c = S3EncryptionConfig.sseKms(null, ctx);
```
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/S3EncryptionConfigTest.java:
##########
@@ -0,0 +1,239 @@
+/*
+ * 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.flink.fs.s3native;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import software.amazon.awssdk.services.s3.model.ServerSideEncryption;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Stream;
+
+import static
org.apache.flink.fs.s3native.S3EncryptionConfig.EncryptionType.NONE;
+import static
org.apache.flink.fs.s3native.S3EncryptionConfig.EncryptionType.SSE_KMS;
+import static
org.apache.flink.fs.s3native.S3EncryptionConfig.EncryptionType.SSE_S3;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link S3EncryptionConfig}. */
+class S3EncryptionConfigTest {
+
+ @Test
+ void sseKms_withContext_contextStoredDefensively() {
+ Map<String, String> ctx = new HashMap<>(Map.of("dept", "finance"));
+ S3EncryptionConfig c = S3EncryptionConfig.sseKms("key-id", ctx);
+ ctx.put("extra", "value");
+
+ assertThat(c.getEncryptionContext()).isEqualTo(Map.of("dept",
"finance"));
+ assertThat(c.hasEncryptionContext()).isTrue();
+ }
+
+ @Test
+ void sseKms_nullContext_contextIsEmpty() {
+ S3EncryptionConfig c = S3EncryptionConfig.sseKms("key-id", null);
+
+ assertThat(c.getEncryptionContext()).isEmpty();
+ assertThat(c.hasEncryptionContext()).isFalse();
+ }
+
+ @ParameterizedTest
+ @MethodSource
+ void sseKms_contextOnlyFactory_absentContext_hasEncryptionContextFalse(
+ Map<String, String> context) {
+ S3EncryptionConfig c = S3EncryptionConfig.sseKms(context);
+
+ assertThat(c.getEncryptionContext()).isEmpty();
+ assertThat(c.hasEncryptionContext()).isFalse();
+ }
+
+ static Stream<Arguments>
sseKms_contextOnlyFactory_absentContext_hasEncryptionContextFalse() {
+ return Stream.of(Arguments.of(Collections.emptyMap()),
Arguments.of((Object) null));
+ }
+
+ @Test
+ void
sseKms_contextOnlyFactory_contextMutatedAfterCreation_contextUnchanged() {
+ Map<String, String> ctx = new HashMap<>(Map.of("dept", "finance"));
+ S3EncryptionConfig c = S3EncryptionConfig.sseKms(ctx);
+ ctx.put("extra", "value");
+
+ assertThat(c.getEncryptionContext()).isEqualTo(Map.of("dept",
"finance"));
+ }
+
+ @ParameterizedTest
+ @MethodSource
+ void getServerSideEncryption_allTypes_returnsCorrectSseValue(
+ String configType, ServerSideEncryption expected) {
+ S3EncryptionConfig c =
+ S3EncryptionConfig.fromConfig(configType, null,
Collections.emptyMap());
+
+ assertThat(c.getServerSideEncryption()).isEqualTo(expected);
+ }
+
+ static Stream<Arguments>
getServerSideEncryption_allTypes_returnsCorrectSseValue() {
+ return Stream.of(
+ Arguments.of(null, null),
+ Arguments.of("sse-s3", ServerSideEncryption.AES256),
+ Arguments.of("sse-kms", ServerSideEncryption.AWS_KMS));
+ }
+
+ @ParameterizedTest
+ @MethodSource
+ void fromConfig_typeVariants_returnExpectedType(
+ String input, S3EncryptionConfig.EncryptionType expected) {
+ assertThat(
+ S3EncryptionConfig.fromConfig(input, null,
Collections.emptyMap())
+ .getEncryptionType())
+ .isEqualTo(expected);
+ }
+
+ static Stream<Arguments> fromConfig_typeVariants_returnExpectedType() {
+ return Stream.of(
+ Arguments.of(null, NONE),
+ Arguments.of("", NONE),
+ Arguments.of("none", NONE),
+ Arguments.of("NONE", NONE),
+ Arguments.of(" ", NONE),
+ Arguments.of("sse-s3", SSE_S3),
+ Arguments.of("AES256", SSE_S3),
+ Arguments.of("sse-kms", SSE_KMS),
+ Arguments.of("aws:kms", SSE_KMS));
+ }
+
+ @Test
+ void fromConfig_sseKmsWithKeyAndContext_keyAndContextPreserved() {
+ S3EncryptionConfig result =
+ S3EncryptionConfig.fromConfig("sse-kms", "my-key",
Map.of("env", "prod"));
+
+ assertThat(result.getKmsKeyId()).isEqualTo("my-key");
+ assertThat(result.getEncryptionContext()).isEqualTo(Map.of("env",
"prod"));
+ }
+
+ @ParameterizedTest
+ @MethodSource
+ void fromConfig_sseKmsWithNoKeyId_keyIdIsNull(String kmsKeyId) {
+ assertThat(
+ S3EncryptionConfig.fromConfig("sse-kms", kmsKeyId,
Collections.emptyMap())
+ .getKmsKeyId())
+ .isNull();
+ }
+
+ static Stream<Arguments> fromConfig_sseKmsWithNoKeyId_keyIdIsNull() {
+ return Stream.of(Arguments.of((Object) null), Arguments.of(""));
+ }
+
+ @Test
+ void fromConfig_sseKmsDefaultKeyWithContext_contextPreserved() {
+ assertThat(
+ S3EncryptionConfig.fromConfig("sse-kms", null,
Map.of("env", "prod"))
+ .hasEncryptionContext())
+ .isTrue();
+ }
+
+ @Test
+ void fromConfig_sseS3WithContext_contextIgnored() {
+ S3EncryptionConfig c =
+ S3EncryptionConfig.fromConfig("sse-s3", null, Map.of("dept",
"finance"));
+
+ assertThat(c.getEncryptionType()).isEqualTo(SSE_S3);
+ assertThat(c.getEncryptionContext()).isEmpty();
+ }
+
+ @Test
+ void fromConfig_unknownType_throwsIllegalArgument() {
+ assertThatThrownBy(
+ () ->
+ S3EncryptionConfig.fromConfig(
+ "invalid-type", null,
Collections.emptyMap()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("invalid-type");
+ }
+
+ @ParameterizedTest
+ @MethodSource
+ void serializeEncryptionContext_exactOutput_correctBase64Json(
+ Map<String, String> context, String expectedDecoded) {
+ S3EncryptionConfig c = S3EncryptionConfig.sseKms(context);
+ String decoded = new
String(Base64.getDecoder().decode(c.serializeEncryptionContext()));
+
+ assertThat(decoded).isEqualTo(expectedDecoded);
+ }
+
+ static Stream<Arguments>
serializeEncryptionContext_exactOutput_correctBase64Json() {
+ return Stream.of(
+ Arguments.of(Collections.emptyMap(), "{}"),
+ Arguments.of(Map.of("k", "v"), "{\"k\":\"v\"}"));
+ }
+
+ @Test
+ void serializeEncryptionContext_multipleEntries_allEntriesPresent() {
+ S3EncryptionConfig c = S3EncryptionConfig.sseKms(Map.of("k1", "v1",
"k2", "v2"));
Review Comment:
```suggestion
S3EncryptionConfig c = S3EncryptionConfig.sseKms(null, Map.of("k1",
"v1", "k2", "v2"));
```
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/S3ExceptionUtilsTest.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.flink.fs.s3native;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import software.amazon.awssdk.awscore.exception.AwsErrorDetails;
+import software.amazon.awssdk.services.s3.model.S3Exception;
+
+import java.io.IOException;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link S3ExceptionUtils}. */
+class S3ExceptionUtilsTest {
+
+ static Stream<Arguments> toIoExceptionCases() {
+ return Stream.of(
+ Arguments.of(
+ s3Exception(
+ 404,
+ AwsErrorDetails.builder().errorMessage("key
not found").build()),
+ "delete object",
+ "delete object (HTTP 404: key not found)"),
+ Arguments.of(
+ s3Exception(500, "internal error"),
+ "put object",
+ "put object (HTTP 500: internal error)"));
+ }
+
+ @ParameterizedTest
+ @MethodSource("toIoExceptionCases")
+ void toIOException_contextAndStatusCode_formatsMessageAndPreservesCause(
+ S3Exception cause, String context, String expectedMessage) {
+ IOException result = S3ExceptionUtils.toIOException(context, cause);
+
+ assertThat(result.getMessage()).contains(expectedMessage);
+ assertThat(result.getCause()).isSameAs(cause);
+ }
+
+ static Stream<Arguments> errorMessageExactCases() {
+ return Stream.of(
+ Arguments.of(
+ s3Exception(
+ 404,
+ AwsErrorDetails.builder()
+ .errorMessage("The specified key does
not exist.")
+ .build()),
+ "The specified key does not exist."),
+ Arguments.of(s3ExceptionStatusOnly(500), "Unknown S3 error"));
+ }
+
+ @ParameterizedTest
+ @MethodSource("errorMessageExactCases")
+ void errorMessage_exactReturn_matchesExpected(S3Exception e, String
expected) {
+ assertThat(S3ExceptionUtils.errorMessage(e)).isEqualTo(expected);
+ }
+
+ static Stream<Arguments> errorMessageFallbackCases() {
+ return Stream.of(
+ Arguments.of(s3Exception(500, "connection timeout"),
"connection timeout"),
+ Arguments.of(
+ s3ExceptionWithMessageAndDetails(
+ 500,
+ "fallback message",
+
AwsErrorDetails.builder().errorCode("InternalError").build()),
+ "fallback message"));
+ }
+
+ @ParameterizedTest
+ @MethodSource("errorMessageFallbackCases")
+ void errorMessage_fallbackToExceptionMessage_containsExpected(
+ S3Exception e, String expectedMessage) {
+ assertThat(S3ExceptionUtils.errorMessage(e)).contains(expectedMessage);
+ }
+
+ static Stream<Arguments> errorCodeAllCases() {
+ return Stream.of(
+ Arguments.of(
+ s3Exception(404,
AwsErrorDetails.builder().errorCode("NoSuchKey").build()),
+ "NoSuchKey"),
+ Arguments.of(s3Exception(500, "some error"), "Unknown"),
+ Arguments.of(
+ s3Exception(
+ 500,
+ AwsErrorDetails.builder()
+ .errorMessage("Something went wrong")
+ .build()),
+ "Unknown"));
+ }
+
+ @ParameterizedTest
+ @MethodSource("errorCodeAllCases")
+ void errorCode_allCases_returnsExpected(S3Exception e, String expected) {
+ assertThat(S3ExceptionUtils.errorCode(e)).isEqualTo(expected);
+ }
+
+ private static S3Exception s3Exception(int statusCode, AwsErrorDetails
details) {
+ S3Exception.Builder b = S3Exception.builder();
+ b.statusCode(statusCode);
+ b.awsErrorDetails(details);
+ return (S3Exception) b.build();
+ }
Review Comment:
```suggestion
private static AwsServiceException s3Exception(int statusCode,
AwsErrorDetails details) {
return
S3Exception.builder().statusCode(statusCode).awsErrorDetails(details).build();
}
```
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/S3ExceptionUtilsTest.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.flink.fs.s3native;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import software.amazon.awssdk.awscore.exception.AwsErrorDetails;
+import software.amazon.awssdk.services.s3.model.S3Exception;
+
+import java.io.IOException;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link S3ExceptionUtils}. */
+class S3ExceptionUtilsTest {
+
+ static Stream<Arguments> toIoExceptionCases() {
+ return Stream.of(
+ Arguments.of(
+ s3Exception(
+ 404,
+ AwsErrorDetails.builder().errorMessage("key
not found").build()),
+ "delete object",
+ "delete object (HTTP 404: key not found)"),
+ Arguments.of(
+ s3Exception(500, "internal error"),
+ "put object",
+ "put object (HTTP 500: internal error)"));
+ }
+
+ @ParameterizedTest
+ @MethodSource("toIoExceptionCases")
+ void toIOException_contextAndStatusCode_formatsMessageAndPreservesCause(
+ S3Exception cause, String context, String expectedMessage) {
+ IOException result = S3ExceptionUtils.toIOException(context, cause);
+
+ assertThat(result.getMessage()).contains(expectedMessage);
+ assertThat(result.getCause()).isSameAs(cause);
+ }
+
+ static Stream<Arguments> errorMessageExactCases() {
+ return Stream.of(
+ Arguments.of(
+ s3Exception(
+ 404,
+ AwsErrorDetails.builder()
+ .errorMessage("The specified key does
not exist.")
+ .build()),
+ "The specified key does not exist."),
+ Arguments.of(s3ExceptionStatusOnly(500), "Unknown S3 error"));
+ }
+
+ @ParameterizedTest
+ @MethodSource("errorMessageExactCases")
+ void errorMessage_exactReturn_matchesExpected(S3Exception e, String
expected) {
+ assertThat(S3ExceptionUtils.errorMessage(e)).isEqualTo(expected);
+ }
+
+ static Stream<Arguments> errorMessageFallbackCases() {
+ return Stream.of(
+ Arguments.of(s3Exception(500, "connection timeout"),
"connection timeout"),
+ Arguments.of(
+ s3ExceptionWithMessageAndDetails(
+ 500,
+ "fallback message",
+
AwsErrorDetails.builder().errorCode("InternalError").build()),
+ "fallback message"));
+ }
+
+ @ParameterizedTest
+ @MethodSource("errorMessageFallbackCases")
+ void errorMessage_fallbackToExceptionMessage_containsExpected(
+ S3Exception e, String expectedMessage) {
+ assertThat(S3ExceptionUtils.errorMessage(e)).contains(expectedMessage);
+ }
+
+ static Stream<Arguments> errorCodeAllCases() {
+ return Stream.of(
+ Arguments.of(
+ s3Exception(404,
AwsErrorDetails.builder().errorCode("NoSuchKey").build()),
+ "NoSuchKey"),
+ Arguments.of(s3Exception(500, "some error"), "Unknown"),
+ Arguments.of(
+ s3Exception(
+ 500,
+ AwsErrorDetails.builder()
+ .errorMessage("Something went wrong")
+ .build()),
+ "Unknown"));
+ }
+
+ @ParameterizedTest
+ @MethodSource("errorCodeAllCases")
+ void errorCode_allCases_returnsExpected(S3Exception e, String expected) {
+ assertThat(S3ExceptionUtils.errorCode(e)).isEqualTo(expected);
+ }
+
+ 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 S3Exception s3Exception(int statusCode, String message) {
+ S3Exception.Builder b = S3Exception.builder();
+ b.statusCode(statusCode);
+ b.message(message);
+ return (S3Exception) b.build();
+ }
Review Comment:
```suggestion
private static AwsServiceException s3Exception(int statusCode, String
message) {
return
S3Exception.builder().statusCode(statusCode).message(message).build();
}
```
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/S3EncryptionConfigTest.java:
##########
@@ -0,0 +1,239 @@
+/*
+ * 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.flink.fs.s3native;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import software.amazon.awssdk.services.s3.model.ServerSideEncryption;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Stream;
+
+import static
org.apache.flink.fs.s3native.S3EncryptionConfig.EncryptionType.NONE;
+import static
org.apache.flink.fs.s3native.S3EncryptionConfig.EncryptionType.SSE_KMS;
+import static
org.apache.flink.fs.s3native.S3EncryptionConfig.EncryptionType.SSE_S3;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link S3EncryptionConfig}. */
+class S3EncryptionConfigTest {
+
+ @Test
+ void sseKms_withContext_contextStoredDefensively() {
+ Map<String, String> ctx = new HashMap<>(Map.of("dept", "finance"));
+ S3EncryptionConfig c = S3EncryptionConfig.sseKms("key-id", ctx);
+ ctx.put("extra", "value");
+
+ assertThat(c.getEncryptionContext()).isEqualTo(Map.of("dept",
"finance"));
+ assertThat(c.hasEncryptionContext()).isTrue();
+ }
+
+ @Test
+ void sseKms_nullContext_contextIsEmpty() {
+ S3EncryptionConfig c = S3EncryptionConfig.sseKms("key-id", null);
+
+ assertThat(c.getEncryptionContext()).isEmpty();
+ assertThat(c.hasEncryptionContext()).isFalse();
+ }
+
+ @ParameterizedTest
+ @MethodSource
+ void sseKms_contextOnlyFactory_absentContext_hasEncryptionContextFalse(
+ Map<String, String> context) {
+ S3EncryptionConfig c = S3EncryptionConfig.sseKms(context);
+
+ assertThat(c.getEncryptionContext()).isEmpty();
+ assertThat(c.hasEncryptionContext()).isFalse();
+ }
+
+ static Stream<Arguments>
sseKms_contextOnlyFactory_absentContext_hasEncryptionContextFalse() {
+ return Stream.of(Arguments.of(Collections.emptyMap()),
Arguments.of((Object) null));
+ }
+
+ @Test
+ void
sseKms_contextOnlyFactory_contextMutatedAfterCreation_contextUnchanged() {
+ Map<String, String> ctx = new HashMap<>(Map.of("dept", "finance"));
+ S3EncryptionConfig c = S3EncryptionConfig.sseKms(ctx);
+ ctx.put("extra", "value");
+
+ assertThat(c.getEncryptionContext()).isEqualTo(Map.of("dept",
"finance"));
+ }
+
+ @ParameterizedTest
+ @MethodSource
+ void getServerSideEncryption_allTypes_returnsCorrectSseValue(
+ String configType, ServerSideEncryption expected) {
+ S3EncryptionConfig c =
+ S3EncryptionConfig.fromConfig(configType, null,
Collections.emptyMap());
+
+ assertThat(c.getServerSideEncryption()).isEqualTo(expected);
+ }
+
+ static Stream<Arguments>
getServerSideEncryption_allTypes_returnsCorrectSseValue() {
+ return Stream.of(
+ Arguments.of(null, null),
+ Arguments.of("sse-s3", ServerSideEncryption.AES256),
+ Arguments.of("sse-kms", ServerSideEncryption.AWS_KMS));
+ }
+
+ @ParameterizedTest
+ @MethodSource
+ void fromConfig_typeVariants_returnExpectedType(
+ String input, S3EncryptionConfig.EncryptionType expected) {
+ assertThat(
+ S3EncryptionConfig.fromConfig(input, null,
Collections.emptyMap())
+ .getEncryptionType())
+ .isEqualTo(expected);
+ }
+
+ static Stream<Arguments> fromConfig_typeVariants_returnExpectedType() {
+ return Stream.of(
+ Arguments.of(null, NONE),
+ Arguments.of("", NONE),
+ Arguments.of("none", NONE),
+ Arguments.of("NONE", NONE),
+ Arguments.of(" ", NONE),
+ Arguments.of("sse-s3", SSE_S3),
+ Arguments.of("AES256", SSE_S3),
+ Arguments.of("sse-kms", SSE_KMS),
+ Arguments.of("aws:kms", SSE_KMS));
+ }
+
+ @Test
+ void fromConfig_sseKmsWithKeyAndContext_keyAndContextPreserved() {
+ S3EncryptionConfig result =
+ S3EncryptionConfig.fromConfig("sse-kms", "my-key",
Map.of("env", "prod"));
+
+ assertThat(result.getKmsKeyId()).isEqualTo("my-key");
+ assertThat(result.getEncryptionContext()).isEqualTo(Map.of("env",
"prod"));
+ }
+
+ @ParameterizedTest
+ @MethodSource
+ void fromConfig_sseKmsWithNoKeyId_keyIdIsNull(String kmsKeyId) {
+ assertThat(
+ S3EncryptionConfig.fromConfig("sse-kms", kmsKeyId,
Collections.emptyMap())
+ .getKmsKeyId())
+ .isNull();
+ }
+
+ static Stream<Arguments> fromConfig_sseKmsWithNoKeyId_keyIdIsNull() {
+ return Stream.of(Arguments.of((Object) null), Arguments.of(""));
+ }
+
+ @Test
+ void fromConfig_sseKmsDefaultKeyWithContext_contextPreserved() {
+ assertThat(
+ S3EncryptionConfig.fromConfig("sse-kms", null,
Map.of("env", "prod"))
+ .hasEncryptionContext())
+ .isTrue();
+ }
+
+ @Test
+ void fromConfig_sseS3WithContext_contextIgnored() {
+ S3EncryptionConfig c =
+ S3EncryptionConfig.fromConfig("sse-s3", null, Map.of("dept",
"finance"));
+
+ assertThat(c.getEncryptionType()).isEqualTo(SSE_S3);
+ assertThat(c.getEncryptionContext()).isEmpty();
+ }
+
+ @Test
+ void fromConfig_unknownType_throwsIllegalArgument() {
+ assertThatThrownBy(
+ () ->
+ S3EncryptionConfig.fromConfig(
+ "invalid-type", null,
Collections.emptyMap()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("invalid-type");
+ }
+
+ @ParameterizedTest
+ @MethodSource
+ void serializeEncryptionContext_exactOutput_correctBase64Json(
+ Map<String, String> context, String expectedDecoded) {
+ S3EncryptionConfig c = S3EncryptionConfig.sseKms(context);
Review Comment:
```suggestion
S3EncryptionConfig c = S3EncryptionConfig.sseKms(null, context);
```
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/S3ExceptionUtilsTest.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.flink.fs.s3native;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import software.amazon.awssdk.awscore.exception.AwsErrorDetails;
+import software.amazon.awssdk.services.s3.model.S3Exception;
+
+import java.io.IOException;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link S3ExceptionUtils}. */
+class S3ExceptionUtilsTest {
+
+ static Stream<Arguments> toIoExceptionCases() {
+ return Stream.of(
+ Arguments.of(
+ s3Exception(
+ 404,
+ AwsErrorDetails.builder().errorMessage("key
not found").build()),
+ "delete object",
+ "delete object (HTTP 404: key not found)"),
+ Arguments.of(
+ s3Exception(500, "internal error"),
+ "put object",
+ "put object (HTTP 500: internal error)"));
+ }
+
+ @ParameterizedTest
+ @MethodSource("toIoExceptionCases")
+ void toIOException_contextAndStatusCode_formatsMessageAndPreservesCause(
+ S3Exception cause, String context, String expectedMessage) {
+ IOException result = S3ExceptionUtils.toIOException(context, cause);
+
+ assertThat(result.getMessage()).contains(expectedMessage);
+ assertThat(result.getCause()).isSameAs(cause);
+ }
+
+ static Stream<Arguments> errorMessageExactCases() {
+ return Stream.of(
+ Arguments.of(
+ s3Exception(
+ 404,
+ AwsErrorDetails.builder()
+ .errorMessage("The specified key does
not exist.")
+ .build()),
+ "The specified key does not exist."),
+ Arguments.of(s3ExceptionStatusOnly(500), "Unknown S3 error"));
+ }
+
+ @ParameterizedTest
+ @MethodSource("errorMessageExactCases")
+ void errorMessage_exactReturn_matchesExpected(S3Exception e, String
expected) {
+ assertThat(S3ExceptionUtils.errorMessage(e)).isEqualTo(expected);
+ }
+
+ static Stream<Arguments> errorMessageFallbackCases() {
+ return Stream.of(
+ Arguments.of(s3Exception(500, "connection timeout"),
"connection timeout"),
+ Arguments.of(
+ s3ExceptionWithMessageAndDetails(
+ 500,
+ "fallback message",
+
AwsErrorDetails.builder().errorCode("InternalError").build()),
+ "fallback message"));
+ }
+
+ @ParameterizedTest
+ @MethodSource("errorMessageFallbackCases")
+ void errorMessage_fallbackToExceptionMessage_containsExpected(
+ S3Exception e, String expectedMessage) {
+ assertThat(S3ExceptionUtils.errorMessage(e)).contains(expectedMessage);
+ }
+
+ static Stream<Arguments> errorCodeAllCases() {
+ return Stream.of(
+ Arguments.of(
+ s3Exception(404,
AwsErrorDetails.builder().errorCode("NoSuchKey").build()),
+ "NoSuchKey"),
+ Arguments.of(s3Exception(500, "some error"), "Unknown"),
+ Arguments.of(
+ s3Exception(
+ 500,
+ AwsErrorDetails.builder()
+ .errorMessage("Something went wrong")
+ .build()),
+ "Unknown"));
+ }
+
+ @ParameterizedTest
+ @MethodSource("errorCodeAllCases")
+ void errorCode_allCases_returnsExpected(S3Exception e, String expected) {
+ assertThat(S3ExceptionUtils.errorCode(e)).isEqualTo(expected);
+ }
+
+ 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 S3Exception s3Exception(int statusCode, String message) {
+ S3Exception.Builder b = S3Exception.builder();
+ b.statusCode(statusCode);
+ b.message(message);
+ return (S3Exception) b.build();
+ }
+
+ 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();
+ }
Review Comment:
```suggestion
private static AwsServiceException s3ExceptionWithMessageAndDetails(
int statusCode, String message, AwsErrorDetails details) {
return S3Exception.builder()
.statusCode(statusCode)
.message(message)
.awsErrorDetails(details)
.build();
}
```
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3EncryptionConfig.java:
##########
@@ -141,29 +151,32 @@ public static S3EncryptionConfig sseKms(
/**
* Creates an encryption config from configuration strings.
*
- * @param encryptionTypeStr The encryption type: "none", "sse-s3",
"sse-kms", or "SSE_S3",
- * "SSE_KMS"
+ * @param encryptionTypeStr The encryption type: "none", "sse-s3",
"sse-kms", "aws:kms",
+ * "aes256" (case-insensitive)
* @param kmsKeyId The KMS key ID (required for SSE-KMS, ignored for other
types)
* @return The encryption configuration
* @throws IllegalArgumentException if the encryption type is invalid
*/
public static S3EncryptionConfig fromConfig(
- @Nullable String encryptionTypeStr, @Nullable String kmsKeyId) {
- if (encryptionTypeStr == null
- || encryptionTypeStr.isEmpty()
+ @Nullable String encryptionTypeStr,
+ @Nullable String kmsKeyId,
+ Map<String, String> encryptionContext) {
+ if (StringUtils.isNullOrWhitespaceOnly(encryptionTypeStr)
|| "none".equalsIgnoreCase(encryptionTypeStr)) {
return none();
}
- String normalizedType = encryptionTypeStr.toUpperCase().replace("-",
"_").replace(":", "_");
+ String normalizedType = encryptionTypeStr.toLowerCase(Locale.ROOT);
switch (normalizedType) {
- case "SSE_S3":
- case "AES256":
+ case "sse-s3":
+ case "aes256":
return sseS3();
- case "SSE_KMS":
- case "AWS_KMS":
- return kmsKeyId != null && !kmsKeyId.isEmpty() ?
sseKms(kmsKeyId) : sseKms();
+ case "sse-kms":
+ case "aws:kms":
+ return kmsKeyId != null && !kmsKeyId.isEmpty()
+ ? sseKms(kmsKeyId, encryptionContext)
+ : sseKms(encryptionContext);
Review Comment:
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);
}
```
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3ClientProvider.java:
##########
@@ -135,6 +135,7 @@ public S3EncryptionConfig getEncryptionConfig() {
}
@VisibleForTesting
+ @Nullable
Review Comment:
```suggestion
```
Keep the field non-nullable. It must fail fast.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
