gaborgsomogyi commented on code in PR #27788: URL: https://github.com/apache/flink/pull/27788#discussion_r3116980072
########## flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/BucketConfigProvider.java: ########## @@ -0,0 +1,178 @@ +/* + * 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.apache.flink.annotation.Internal; +import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.IllegalConfigurationException; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.util.Collections; +import java.util.Comparator; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.function.BiConsumer; +import java.util.stream.Collectors; + +/** + * Parses bucket-specific S3 configuration using format {@code s3.bucket.<bucket-name>.<property>}. + * + * <p>Enables per-bucket overrides for endpoints, credentials, encryption, and IAM roles. Bucket + * names containing dots are supported; properties are matched by longest suffix first. + * + * <p>Immutable and thread-safe after construction. + */ +@Internal +final class BucketConfigProvider { + + private static final Logger LOG = LoggerFactory.getLogger(BucketConfigProvider.class); + static final String BUCKET_CONFIG_PREFIX = "s3.bucket."; + static final Map<String, BiConsumer<S3BucketConfig.Builder, String>> PROPERTY_APPLICATORS; + static final List<String> KNOWN_PROPERTIES_BY_LENGTH; + + static { + final Map<String, BiConsumer<S3BucketConfig.Builder, String>> applicators = + new LinkedHashMap<>(); + applicators.put("access-key", (b, v) -> b.accessKey(v)); + applicators.put("assume-role.arn", (b, v) -> b.assumeRoleArn(v)); + applicators.put("assume-role.external-id", (b, v) -> b.assumeRoleExternalId(v)); + applicators.put( + "assume-role.session-duration", + (b, v) -> { + try { + b.assumeRoleSessionDurationSeconds(Integer.parseInt(v)); + } catch (NumberFormatException e) { + throw new IllegalConfigurationException( + String.format( + "Invalid assume-role.session-duration '%s' for bucket '%s'. " + + "Must be a valid integer (e.g., 3600)", + v, b.getBucketName()), + e); + } + }); + applicators.put("assume-role.session-name", (b, v) -> b.assumeRoleSessionName(v)); + applicators.put("credentials.provider", (b, v) -> b.credentialsProvider(v)); + applicators.put("endpoint", (b, v) -> b.endpoint(v)); + applicators.put("path-style-access", (b, v) -> b.pathStyleAccess(Boolean.parseBoolean(v))); + applicators.put("region", (b, v) -> b.region(v)); + applicators.put("sse.kms-key-id", (b, v) -> b.sseKmsKeyId(v)); + applicators.put("sse.type", (b, v) -> b.sseType(v)); + applicators.put("secret-key", (b, v) -> b.secretKey(v)); + PROPERTY_APPLICATORS = Collections.unmodifiableMap(applicators); + + KNOWN_PROPERTIES_BY_LENGTH = + applicators.keySet().stream() + .sorted(Comparator.comparingInt(String::length).reversed()) + .collect(Collectors.toList()); + } + + private final Map<String, S3BucketConfig> bucketConfigs; + + BucketConfigProvider(Configuration flinkConfig) { + this.bucketConfigs = Collections.unmodifiableMap(parseBucketConfigs(flinkConfig)); + } + + @Nullable + S3BucketConfig getBucketConfig(String bucketName) { + return bucketConfigs.get(bucketName); + } + + @VisibleForTesting + boolean hasBucketConfig(String bucketName) { + return bucketConfigs.containsKey(bucketName); + } + + @VisibleForTesting + int size() { + return bucketConfigs.size(); + } + + private static Map<String, S3BucketConfig> parseBucketConfigs(Configuration flinkConfig) { + final Map<String, Map<String, String>> rawConfigs = new HashMap<>(); + + for (final String key : flinkConfig.keySet()) { + if (!key.startsWith(BUCKET_CONFIG_PREFIX)) { + continue; + } + final String suffix = key.substring(BUCKET_CONFIG_PREFIX.length()); Review Comment: We can skip is whitespace only ########## flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java: ########## @@ -31,16 +31,32 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import javax.annotation.Nullable; + import java.io.IOException; import java.net.URI; import java.time.Duration; +import java.util.concurrent.TimeUnit; /** - * Factory for creating Native S3 FileSystem instances. + * Factory for creating Native S3 FileSystem instances using AWS SDK v2. + * + * <h3>Bucket-Level Configuration</h3> * - * <p>This factory creates {@link NativeS3FileSystem} instances for accessing Amazon S3 buckets - * using AWS SDK v2. The Native S3 FileSystem provides a drop-in replacement for Presto and Hadoop - * S3 implementations with minimal external dependencies. + * <p>Supports per-bucket configuration overrides using the format {@code + * s3.bucket.<bucket-name>.<property>}. Bucket-level settings override global settings; unset + * properties inherit global values. Supported properties: + * + * <ul> + * <li>{@code region} - AWS region for this bucket + * <li>{@code endpoint} - custom S3 endpoint + * <li>{@code path-style-access} - use path-style S3 access + * <li>{@code access-key} / {@code secret-key} - bucket-specific credentials + * <li>{@code sse.type} / {@code sse.kms-key-id} - server-side encryption + * <li>{@code assume-role.arn} / {@code assume-role.external-id} / {@code + * assume-role.session-name} / {@code assume-role.session-duration} - IAM assume role + * <li>{@code credentials.provider} - custom credentials provider class + * </ul> Review Comment: I would not keep supported configs in comment, later it will slip from implementation ########## flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java: ########## @@ -285,13 +301,54 @@ public FileSystem create(URI fsUri) throws IOException { String region = config.get(REGION); String endpoint = config.get(ENDPOINT); boolean pathStyleAccess = config.get(PATH_STYLE_ACCESS); - + String sseType = config.get(SSE_TYPE); + String sseKmsKeyId = config.get(SSE_KMS_KEY_ID); + String assumeRoleArn = config.get(ASSUME_ROLE_ARN); + String assumeRoleExternalId = config.get(ASSUME_ROLE_EXTERNAL_ID); + String assumeRoleSessionName = config.get(ASSUME_ROLE_SESSION_NAME); + int assumeRoleSessionDuration = config.get(ASSUME_ROLE_SESSION_DURATION_SECONDS); + String credentialsProviderClasses = config.get(AWS_CREDENTIALS_PROVIDER); + + // Apply bucket-specific overrides + String bucketName = fsUri.getHost(); Review Comment: Is this code virtual-hosted path compliant? ########## flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java: ########## @@ -271,6 +286,7 @@ public int getPriority() { @Override public void configure(Configuration config) { this.flinkConfig = config; + this.bucketConfigProvider = new BucketConfigProvider(config); Review Comment: It's nullable but `Configuration` is not and we set both at the same time. Which one is true then? ########## flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java: ########## @@ -359,35 +413,51 @@ public FileSystem create(URI fsUri) throws IOException { .socketTimeout(config.get(SOCKET_TIMEOUT)) .connectionMaxIdleTime(config.get(CONNECTION_MAX_IDLE_TIME)) .clientCloseTimeout(config.get(CLIENT_CLOSE_TIMEOUT)) - .assumeRoleArn(config.get(ASSUME_ROLE_ARN)) - .assumeRoleExternalId(config.get(ASSUME_ROLE_EXTERNAL_ID)) - .assumeRoleSessionName(config.get(ASSUME_ROLE_SESSION_NAME)) - .assumeRoleSessionDurationSeconds( - config.get(ASSUME_ROLE_SESSION_DURATION_SECONDS)) + .assumeRoleArn(assumeRoleArn) + .assumeRoleExternalId(assumeRoleExternalId) + .assumeRoleSessionName(assumeRoleSessionName) + .assumeRoleSessionDurationSeconds(assumeRoleSessionDuration) .maxRetries(config.get(MAX_RETRIES)) - .credentialsProviderClasses(config.get(AWS_CREDENTIALS_PROVIDER)) + .credentialsProviderClasses(credentialsProviderClasses) .encryptionConfig(encryptionConfig) .build(); - NativeS3BulkCopyHelper bulkCopyHelper = null; Review Comment: I don't get why these changes needed at all ########## flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java: ########## @@ -191,7 +206,7 @@ public class NativeS3FileSystemFactory implements FileSystemFactory { public static final ConfigOption<Integer> ASSUME_ROLE_SESSION_DURATION_SECONDS = ConfigOptions.key("s3.assume-role.session-duration") .intType() - .defaultValue(3600) // 1 hour default + .defaultValue(3600) Review Comment: Presume these are unrelated and no value here, right? ########## flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java: ########## @@ -255,14 +270,14 @@ public class NativeS3FileSystemFactory implements FileSystemFactory { + "When not set, the default chain is used: delegation tokens -> " + "static credentials (if configured) -> DefaultCredentialsProvider."); - private Configuration flinkConfig; + private volatile Configuration flinkConfig; + @Nullable private volatile BucketConfigProvider bucketConfigProvider; @Override public String getScheme() { return "s3"; } - // setting to least priority so that it is not used by default Review Comment: Unrelated ########## flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java: ########## @@ -285,13 +301,54 @@ public FileSystem create(URI fsUri) throws IOException { String region = config.get(REGION); String endpoint = config.get(ENDPOINT); boolean pathStyleAccess = config.get(PATH_STYLE_ACCESS); - + String sseType = config.get(SSE_TYPE); + String sseKmsKeyId = config.get(SSE_KMS_KEY_ID); + String assumeRoleArn = config.get(ASSUME_ROLE_ARN); + String assumeRoleExternalId = config.get(ASSUME_ROLE_EXTERNAL_ID); + String assumeRoleSessionName = config.get(ASSUME_ROLE_SESSION_NAME); + int assumeRoleSessionDuration = config.get(ASSUME_ROLE_SESSION_DURATION_SECONDS); + String credentialsProviderClasses = config.get(AWS_CREDENTIALS_PROVIDER); + + // Apply bucket-specific overrides + String bucketName = fsUri.getHost(); + if (bucketName == null || bucketName.isEmpty()) { + throw new IOException("Invalid S3 URI: missing or empty bucket name in URI: " + fsUri); + } + if (bucketConfigProvider != null) { + S3BucketConfig overrides = bucketConfigProvider.getBucketConfig(bucketName); + if (overrides != null) { + LOG.debug( + "Applying bucket-specific configuration for bucket '{}': {}", + bucketName, + overrides); + accessKey = firstNonNull(overrides.getAccessKey(), accessKey); + secretKey = firstNonNull(overrides.getSecretKey(), secretKey); + region = firstNonNull(overrides.getRegion(), region); + endpoint = firstNonNull(overrides.getEndpoint(), endpoint); + sseType = firstNonNull(overrides.getSseType(), sseType); + sseKmsKeyId = firstNonNull(overrides.getSseKmsKeyId(), sseKmsKeyId); + assumeRoleArn = firstNonNull(overrides.getAssumeRoleArn(), assumeRoleArn); + assumeRoleExternalId = + firstNonNull(overrides.getAssumeRoleExternalId(), assumeRoleExternalId); + assumeRoleSessionName = + firstNonNull(overrides.getAssumeRoleSessionName(), assumeRoleSessionName); + credentialsProviderClasses = + firstNonNull( + overrides.getCredentialsProvider(), credentialsProviderClasses); + if (overrides.getPathStyleAccess() != null) { + pathStyleAccess = overrides.getPathStyleAccess(); + } + if (overrides.getAssumeRoleSessionDurationSeconds() != null) { + assumeRoleSessionDuration = overrides.getAssumeRoleSessionDurationSeconds(); + } Review Comment: Why have these special treatment? ########## flink-filesystems/flink-s3-fs-native/README.md: ########## @@ -89,6 +89,29 @@ input.sinkTo(FileSink.forRowFormat(new Path("s3://my-bucket/output"), | s3.assume-role.session-name | flink-s3-session | Session name for the assumed role | | s3.assume-role.session-duration | 3600 | Session duration in seconds (900-43200) | +## Bucket-Level Configuration + +The Native S3 FileSystem supports per-bucket configuration overrides, allowing different S3 buckets to use different connection settings within the same Flink cluster. This enables scenarios like: + +- **Checkpointing to one bucket** with specific credentials +- **Savepoints to another bucket** with different region/endpoint +- **Data sinks to third-party buckets** with cross-account IAM roles + +### Format + +Bucket-level configuration uses the format: `s3.bucket.<bucket-name>.<property>` + +Bucket names containing dots (e.g., `my.company.data`) are fully supported through longest-suffix matching. Review Comment: AWS statement for this feature. I think we should mention this in a compacted way. ``` Avoid using periods (.) in bucket names For best compatibility, we recommend that you avoid using periods (.) in bucket names, except for buckets that are used only for static website hosting. If you include periods in a bucket's name, you can't use virtual-host-style addressing over HTTPS, unless you perform your own certificate validation. The security certificates used for virtual hosting of buckets don't work for buckets with periods in their names. ``` ########## flink-filesystems/flink-s3-fs-native/README.md: ########## @@ -89,6 +89,29 @@ input.sinkTo(FileSink.forRowFormat(new Path("s3://my-bucket/output"), | s3.assume-role.session-name | flink-s3-session | Session name for the assumed role | | s3.assume-role.session-duration | 3600 | Session duration in seconds (900-43200) | +## Bucket-Level Configuration + +The Native S3 FileSystem supports per-bucket configuration overrides, allowing different S3 buckets to use different connection settings within the same Flink cluster. This enables scenarios like: + +- **Checkpointing to one bucket** with specific credentials +- **Savepoints to another bucket** with different region/endpoint Review Comment: I think savepoints and checkpoint can phisically go into a single bucket -- 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]
