This factory provides type-safe configuration through S3ConfigurationBuilder. Uses shared + * client with proper lifecycle management to prevent HTTP connection pool exhaustion. + */ +public class S3ClientConfigurationFactory { + + private static final Logger LOG = LoggerFactory.getLogger(S3ClientConfigurationFactory.class); + + // Shared client with reference counting for proper cleanup + private static volatile S3Client sharedClient; + private static volatile String sharedConfigHash; + private static volatile int clientRefCount = 0; + private static final Object clientLock = new Object(); + + /** Private constructor to prevent instantiation. */ + private S3ClientConfigurationFactory() {} + + /** + * Acquires a reference to the shared S3 client configured to match the given S3AFileSystem. + * Must be paired with releaseS3Client() to prevent resource leaks. + * + * @param s3aFileSystem The S3AFileSystem to match configuration for + * @return A shared S3 client with consistent configuration + */ + public static S3Client acquireS3Client(S3AFileSystem s3aFileSystem) { + try { + // Handle test scenarios where getConf() might return null + org.apache.hadoop.conf.Configuration hadoopConf = s3aFileSystem.getConf(); + if (hadoopConf == null) { + // In test environments, create a minimal configuration + hadoopConf = new org.apache.hadoop.conf.Configuration(); + LOG.debug( + "Using default configuration for test environment (S3AFileSystem.getConf() returned null)"); + } + + // Build configuration from Hadoop configuration + S3Configuration config = + S3ConfigurationBuilder.fromHadoopConfiguration(hadoopConf).build(); + + return getOrCreateSharedClient(config); + + } catch (Exception e) { + LOG.error("Failed to acquire S3 client for S3AFileSystem", e); + throw new RuntimeException("Failed to acquire S3 client: " + e.getMessage(), e); + } + } + + /** + * Acquires a reference to the shared S3 client from Hadoop configuration directly. Must be + * paired with releaseS3Client() to prevent resource leaks. + * + * @param hadoopConfig The Hadoop configuration to use + * @return A shared S3 client with consistent configuration + */ + public static S3Client acquireS3Client(org.apache.hadoop.conf.Configuration hadoopConfig) { + try { + // Build configuration from Hadoop configuration + S3Configuration config = + S3ConfigurationBuilder.fromHadoopConfiguration(hadoopConfig).build(); + + return getOrCreateSharedClient(config); + + } catch (Exception e) { + LOG.error("Failed to acquire S3 client from Hadoop configuration", e); + throw new RuntimeException("Failed to acquire S3 client: " + e.getMessage(), e); + } + } + + /** + * Releases a reference to the shared S3 client. When the last reference is released, the client + * will be closed to free up HTTP connection pools. + * + * @param client The S3 client to release (should match the client returned by acquireS3Client) + */ + public static void releaseS3Client(S3Client client) { + if (client == null) { + return; + } + + synchronized (clientLock) { + if (client == sharedClient && clientRefCount > 0) { + clientRefCount--; + LOG.debug("Released S3 client reference, remaining refs: {}", clientRefCount); + + // Close the shared client when no more references + if (clientRefCount == 0) { + LOG.debug("Closing shared S3 client - no more references"); + try { + if (sharedClient != null) { + sharedClient.close(); + sharedClient = null; + sharedConfigHash = null; + } + } catch (Exception e) { + LOG.warn("Failed to close shared S3 client", e); + } + } + } + } + } + + /** Gets or creates the shared S3 client with reference counting. */ + private static S3Client getOrCreateSharedClient(S3Configuration config) { + String configHash = config.getConfigurationHash(); + + synchronized (clientLock) { + // Check if we have a client for this configuration + if (sharedClient != null && configHash.equals(sharedConfigHash)) { + clientRefCount++; + LOG.debug("Using existing shared S3 client, refs: {}", clientRefCount); + return sharedClient; + } + + // Need to create a new client (configuration changed or first time) + if (sharedClient != null) { + LOG.debug("Configuration changed, closing previous S3 client"); + try { + sharedClient.close(); + } catch (Exception e) { + LOG.warn("Failed to close previous S3 client", e); + } + } + + // Create new shared client + LOG.debug("Creating new shared S3 client for config hash: {}", configHash); + sharedClient = createS3Client(config); + sharedConfigHash = configHash; + clientRefCount = 1; + + LOG.debug("Created shared S3 client, refs: {}", clientRefCount); + return sharedClient; + } + } + + /** + * Creates a new S3 client with the given configuration. + * + * @param config The S3 configuration + * @return A new S3 client + */ + private static S3Client createS3Client(S3Configuration config) { + software.amazon.awssdk.services.s3.S3ClientBuilder clientBuilder = + software.amazon.awssdk.services.s3.S3Client.builder(); + + // Configure region + if (config.getRegion() != null) { + clientBuilder.region(software.amazon.awssdk.regions.Region.of(config.getRegion())); + } + + // Configure endpoint if specified + if (config.getEndpoint() != null) { + clientBuilder.endpointOverride(config.getEndpoint()); + } + + // Configure path style access + clientBuilder.forcePathStyle(config.isPathStyleAccess()); + + // Configure credentials if available + if (config.getAccessKey() != null && config.getSecretKey() != null) { + software.amazon.awssdk.auth.credentials.AwsCredentials credentials; + if (config.getSessionToken() != null) { + credentials = + software.amazon.awssdk.auth.credentials.AwsSessionCredentials.create( + config.getAccessKey(), + config.getSecretKey(), + config.getSessionToken()); + } else { + credentials = + software.amazon.awssdk.auth.credentials.AwsBasicCredentials.create( + config.getAccessKey(), config.getSecretKey()); + } + clientBuilder.credentialsProvider(() -> credentials); + } else { + // Use credential provider that matches Hadoop's S3A configuration + clientBuilder.credentialsProvider(createHadoopCompatibleCredentialProvider(config)); + } + + // Use AWS SDK default HTTP client to avoid interference with Flink networking + // Custom HTTP client configuration was causing SSL/connection conflicts + + return clientBuilder.build(); + } + + /** + * Creates a credential provider that is compatible with Hadoop's S3A configuration. This + * ensures our custom S3 client uses the same credential chain as Hadoop S3A. + */ + private static software.amazon.awssdk.auth.credentials.AwsCredentialsProvider + createHadoopCompatibleCredentialProvider(S3Configuration config) { + + org.apache.hadoop.conf.Configuration hadoopConfig = config.getHadoopConfiguration(); + + // Check if Hadoop has a specific credential provider configured + if (hadoopConfig != null) { + String credentialProviders = hadoopConfig.get("fs.s3a.aws.credentials.provider"); + if (credentialProviders != null && !credentialProviders.trim().isEmpty()) { + LOG.debug("Using Hadoop-configured credential providers: {}", credentialProviders); + // For most cases, especially IAM roles, the default provider chain handles this + // well + // and is compatible with Hadoop's credential providers + } + } + + // Use AWS SDK's default credential provider chain which includes: + // 1. Environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKEN) + // 2. System properties (aws.accessKeyId, aws.secretAccessKey, aws.sessionToken) + // 3. AWS credentials file (~/.aws/credentials) + // 4. IAM instance profile credentials (for EC2 instances) + // 5. IAM ECS task role credentials (for ECS tasks) + // This is compatible with Hadoop's default credential provider chain + return software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider.create(); + } + + /** + * Releases the shared S3 client reference. Used for cleanup in tests and shutdown. This + * decrements the reference count and closes the client if no more references exist. + */ + public static void releaseS3Client() { + synchronized (clientLock) { + if (clientRefCount > 0) { + clientRefCount--; + LOG.debug("Released S3 client reference, remaining references: {}", clientRefCount); + + if (clientRefCount == 0 && sharedClient != null) { + try { + sharedClient.close(); + LOG.debug("Closed shared S3 client"); + } catch (Exception e) { + LOG.warn("Error closing S3 client", e); + } finally { + sharedClient = null; + sharedConfigHash = null; + } + } + } + } + } +} diff --git a/flink-filesystems/flink-s3-fs-hadoop/src/main/java/org/apache/flink/fs/s3hadoop/S3Configuration.java b/flink-filesystems/flink-s3-fs-hadoop/src/main/java/org/apache/flink/fs/s3hadoop/S3Configuration.java new file mode 100644 index 0000000000000..9387b962986e7 --- /dev/null +++ b/flink-filesystems/flink-s3-fs-hadoop/src/main/java/org/apache/flink/fs/s3hadoop/S3Configuration.java @@ -0,0 +1,297 @@ +/* + * 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.s3hadoop; + +import org.apache.flink.annotation.Internal; + +import javax.annotation.Nullable; + +import java.net.URI; +import java.time.Duration; +import java.util.Objects; + +/** + * Immutable configuration object for S3 client settings. This centralizes all S3 configuration and + * provides a clean, type-safe API for accessing configuration values. + */ +@Internal +public final class S3Configuration { + + // Connection settings + private final Duration connectionTimeout; + private final Duration socketTimeout; + private final int maxConnections; + + // Retry settings + private final int maxRetries; + private final Duration retryInterval; + + // Timeouts + private final Duration apiCallTimeout; + private final Duration apiCallAttemptTimeout; + + // Credentials + private final String accessKey; + private final String secretKey; + private final String sessionToken; + + // Regional settings + private final String region; + private final URI endpoint; + + // Service settings + private final boolean pathStyleAccess; + private final boolean checksumValidation; + + // SSL settings + private final boolean sslEnabled; + private final boolean verifySslCertificates; + private final String trustStorePath; + private final String trustStorePassword; + + // Buffer settings + private final int bufferSize; + + // Hadoop configuration for credential provider access + private final org.apache.hadoop.conf.Configuration hadoopConfiguration; + + S3Configuration(S3ConfigurationBuilder builder) { + this.connectionTimeout = builder.getConnectionTimeout(); + this.socketTimeout = builder.getSocketTimeout(); + this.maxConnections = builder.getMaxConnections(); + this.maxRetries = builder.getMaxRetries(); + this.retryInterval = builder.getRetryInterval(); + this.apiCallTimeout = builder.getApiCallTimeout(); + this.apiCallAttemptTimeout = builder.getApiCallAttemptTimeout(); + this.accessKey = builder.getAccessKey(); + this.secretKey = builder.getSecretKey(); + this.sessionToken = builder.getSessionToken(); + this.region = builder.getRegion(); + this.endpoint = builder.getEndpoint(); + this.pathStyleAccess = builder.isPathStyleAccess(); + this.checksumValidation = builder.isChecksumValidation(); + this.sslEnabled = builder.isSslEnabled(); + this.verifySslCertificates = builder.isVerifySslCertificates(); + this.trustStorePath = builder.getTrustStorePath(); + this.trustStorePassword = builder.getTrustStorePassword(); + this.bufferSize = builder.getBufferSize(); + this.hadoopConfiguration = builder.getHadoopConfiguration(); + } + + // Getters + public Duration getConnectionTimeout() { + return connectionTimeout; + } + + public Duration getSocketTimeout() { + return socketTimeout; + } + + public int getMaxConnections() { + return maxConnections; + } + + public int getMaxRetries() { + return maxRetries; + } + + public Duration getRetryInterval() { + return retryInterval; + } + + public Duration getApiCallTimeout() { + return apiCallTimeout; + } + + public Duration getApiCallAttemptTimeout() { + return apiCallAttemptTimeout; + } + + @Nullable + public String getAccessKey() { + return accessKey; + } + + @Nullable + public String getSecretKey() { + return secretKey; + } + + @Nullable + public String getSessionToken() { + return sessionToken; + } + + public String getRegion() { + return region; + } + + @Nullable + public URI getEndpoint() { + return endpoint; + } + + public boolean isPathStyleAccess() { + return pathStyleAccess; + } + + public boolean isChecksumValidation() { + return checksumValidation; + } + + public boolean isSslEnabled() { + return sslEnabled; + } + + public boolean isVerifySslCertificates() { + return verifySslCertificates; + } + + @Nullable + public String getTrustStorePath() { + return trustStorePath; + } + + @Nullable + public String getTrustStorePassword() { + return trustStorePassword; + } + + public int getBufferSize() { + return bufferSize; + } + + @Nullable + public org.apache.hadoop.conf.Configuration getHadoopConfiguration() { + return hadoopConfiguration; + } + + /** + * Generates a configuration hash for caching purposes. This should be consistent across + * different instances with the same configuration. + */ + public String getConfigurationHash() { + return Integer.toHexString( + Objects.hash( + connectionTimeout, + socketTimeout, + maxConnections, + maxRetries, + retryInterval, + apiCallTimeout, + apiCallAttemptTimeout, + accessKey, + secretKey, + sessionToken, + region, + endpoint, + pathStyleAccess, + checksumValidation, + sslEnabled, + verifySslCertificates, + trustStorePath, + trustStorePassword, + bufferSize)); + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } + if (obj == null || getClass() != obj.getClass()) { + return false; + } + S3Configuration that = (S3Configuration) obj; + return maxConnections == that.maxConnections + && maxRetries == that.maxRetries + && pathStyleAccess == that.pathStyleAccess + && checksumValidation == that.checksumValidation + && sslEnabled == that.sslEnabled + && verifySslCertificates == that.verifySslCertificates + && bufferSize == that.bufferSize + && Objects.equals(connectionTimeout, that.connectionTimeout) + && Objects.equals(socketTimeout, that.socketTimeout) + && Objects.equals(retryInterval, that.retryInterval) + && Objects.equals(apiCallTimeout, that.apiCallTimeout) + && Objects.equals(apiCallAttemptTimeout, that.apiCallAttemptTimeout) + && Objects.equals(accessKey, that.accessKey) + && Objects.equals(secretKey, that.secretKey) + && Objects.equals(sessionToken, that.sessionToken) + && Objects.equals(region, that.region) + && Objects.equals(endpoint, that.endpoint) + && Objects.equals(trustStorePath, that.trustStorePath) + && Objects.equals(trustStorePassword, that.trustStorePassword); + } + + @Override + public int hashCode() { + return Objects.hash( + connectionTimeout, + socketTimeout, + maxConnections, + maxRetries, + retryInterval, + apiCallTimeout, + apiCallAttemptTimeout, + accessKey, + secretKey, + sessionToken, + region, + endpoint, + pathStyleAccess, + checksumValidation, + sslEnabled, + verifySslCertificates, + trustStorePath, + trustStorePassword, + bufferSize); + } + + @Override + public String toString() { + return "S3Configuration{" + + "connectionTimeout=" + + connectionTimeout + + ", socketTimeout=" + + socketTimeout + + ", maxConnections=" + + maxConnections + + ", maxRetries=" + + maxRetries + + ", retryInterval=" + + retryInterval + + ", region='" + + region + + '\'' + + ", endpoint=" + + endpoint + + ", pathStyleAccess=" + + pathStyleAccess + + ", checksumValidation=" + + checksumValidation + + ", sslEnabled=" + + sslEnabled + + ", bufferSize=" + + bufferSize + + + // Note: Don't include credentials in toString for security + '}'; + } +} diff --git a/flink-filesystems/flink-s3-fs-hadoop/src/main/java/org/apache/flink/fs/s3hadoop/S3ConfigurationBuilder.java b/flink-filesystems/flink-s3-fs-hadoop/src/main/java/org/apache/flink/fs/s3hadoop/S3ConfigurationBuilder.java new file mode 100644 index 0000000000000..0fc9834900090 --- /dev/null +++ b/flink-filesystems/flink-s3-fs-hadoop/src/main/java/org/apache/flink/fs/s3hadoop/S3ConfigurationBuilder.java @@ -0,0 +1,290 @@ +/* + * 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.s3hadoop; + +import org.apache.flink.annotation.Internal; + +import org.apache.hadoop.conf.Configuration; + +import javax.annotation.Nullable; + +import java.net.URI; +import java.time.Duration; + +/** + * Builder for S3 client configuration that provides a fluent API and comprehensive validation. This + * centralizes all configuration logic and makes it easier to test and maintain. + */ +@Internal +public class S3ConfigurationBuilder { + + // Connection settings + private Duration connectionTimeout = Duration.ofMillis(200000); + private Duration socketTimeout = Duration.ofMillis(200000); + private int maxConnections = 96; + + // Retry settings + private int maxRetries = 10; + private Duration retryInterval = Duration.ofMillis(500); + + // Timeouts + private Duration apiCallTimeout = Duration.ofMinutes(5); + private Duration apiCallAttemptTimeout = Duration.ofMinutes(1); + + // Credentials + private String accessKey; + private String secretKey; + private String sessionToken; + + // Regional settings + private String region = "us-east-1"; + private URI endpoint; + + // Service settings + private boolean pathStyleAccess = false; + private boolean checksumValidation = true; + + // SSL settings + private boolean sslEnabled = true; + private boolean verifySslCertificates = true; + private String trustStorePath; + private String trustStorePassword; + + // Buffer settings + private int bufferSize = 32 * 1024; // 32KB default + + // Hadoop configuration for credential provider access + private org.apache.hadoop.conf.Configuration hadoopConfiguration; + + public static S3ConfigurationBuilder fromHadoopConfiguration(Configuration hadoopConfig) { + S3ConfigurationBuilder builder = new S3ConfigurationBuilder(); + return builder.loadFromHadoop(hadoopConfig); + } + + private S3ConfigurationBuilder loadFromHadoop(Configuration hadoopConfig) { + // Load all configuration with proper defaults and validation + this.connectionTimeout = + parseDuration(hadoopConfig, "fs.s3a.connection.timeout", connectionTimeout); + this.socketTimeout = parseDuration(hadoopConfig, "fs.s3a.socket.timeout", socketTimeout); + this.maxConnections = + parseInteger(hadoopConfig, "fs.s3a.connection.maximum", maxConnections, 1, 1000); + + this.maxRetries = parseInteger(hadoopConfig, "fs.s3a.retry.limit", maxRetries, 0, 50); + this.retryInterval = parseDuration(hadoopConfig, "fs.s3a.retry.interval", retryInterval); + + this.apiCallTimeout = parseDuration(hadoopConfig, "fs.s3a.api.timeout", apiCallTimeout); + this.apiCallAttemptTimeout = + parseDuration(hadoopConfig, "fs.s3a.api.attempt.timeout", apiCallAttemptTimeout); + + // Credentials (secure) + this.accessKey = hadoopConfig.get("fs.s3a.access.key"); + this.secretKey = hadoopConfig.get("fs.s3a.secret.key"); + this.sessionToken = hadoopConfig.get("fs.s3a.session.token"); + + // Store the Hadoop configuration for credential provider access + this.hadoopConfiguration = hadoopConfig; + + // Regional settings + this.region = hadoopConfig.get("fs.s3a.endpoint.region", region); + String endpointStr = hadoopConfig.get("fs.s3a.endpoint"); + if (endpointStr != null && !endpointStr.trim().isEmpty()) { + // Ensure backward compatibility: add https:// if no scheme is provided + String trimmedEndpoint = endpointStr.trim(); + if (!trimmedEndpoint.contains("://")) { + trimmedEndpoint = "https://" + trimmedEndpoint; + } + this.endpoint = URI.create(trimmedEndpoint); + } + + // Service settings + this.pathStyleAccess = hadoopConfig.getBoolean("fs.s3a.path.style.access", pathStyleAccess); + this.checksumValidation = + hadoopConfig.getBoolean("fs.s3a.checksum.validation", checksumValidation); + + // SSL settings + this.sslEnabled = hadoopConfig.getBoolean("fs.s3a.connection.ssl.enabled", sslEnabled); + // Note: fs.s3a.ssl.channel.mode is not a boolean - it's an enum ("default_jsse", "openssl", + // etc.) + // Use the correct boolean config for SSL certificate verification + this.verifySslCertificates = + hadoopConfig.getBoolean("fs.s3a.connection.ssl.cert.verify", verifySslCertificates); + this.trustStorePath = hadoopConfig.get("fs.s3a.ssl.truststore.path"); + this.trustStorePassword = hadoopConfig.get("fs.s3a.ssl.truststore.password"); + + // Buffer settings + this.bufferSize = + parseInteger(hadoopConfig, "fs.s3a.block.size", bufferSize, 4096, 1024 * 1024); + + return this; + } + + private Duration parseDuration(Configuration config, String key, Duration defaultValue) { + try { + String value = config.get(key); + if (value == null) { + return defaultValue; + } + + // Handle common suffixes + if (value.endsWith("s")) { + return Duration.ofSeconds(Long.parseLong(value.substring(0, value.length() - 1))); + } else if (value.endsWith("ms")) { + return Duration.ofMillis(Long.parseLong(value.substring(0, value.length() - 2))); + } else { + return Duration.ofMillis(Long.parseLong(value)); + } + } catch (Exception e) { + return defaultValue; + } + } + + private int parseInteger(Configuration config, String key, int defaultValue, int min, int max) { + try { + int value = config.getInt(key, defaultValue); + return Math.max(min, Math.min(max, value)); + } catch (Exception e) { + return defaultValue; + } + } + + public S3Configuration build() { + validate(); + return new S3Configuration(this); + } + + private void validate() { + if (connectionTimeout.toMillis() < 1000 || connectionTimeout.toMillis() > 600000) { + throw new IllegalArgumentException("Connection timeout must be between 1s and 10min"); + } + if (socketTimeout.toMillis() < 1000 || socketTimeout.toMillis() > 600000) { + throw new IllegalArgumentException("Socket timeout must be between 1s and 10min"); + } + if (maxConnections < 1 || maxConnections > 1000) { + throw new IllegalArgumentException("Max connections must be between 1 and 1000"); + } + if (maxRetries < 0 || maxRetries > 50) { + throw new IllegalArgumentException("Max retries must be between 0 and 50"); + } + if (region != null && !isValidAwsRegion(region)) { + throw new IllegalArgumentException("Invalid AWS region: " + region); + } + if (endpoint != null && !isValidEndpointUrl(endpoint)) { + throw new IllegalArgumentException("Invalid endpoint URL: " + endpoint); + } + } + + private boolean isValidAwsRegion(String region) { + return region.matches("^[a-z0-9-]+$") && region.length() >= 3 && region.length() <= 20; + } + + private boolean isValidEndpointUrl(URI endpoint) { + String scheme = endpoint.getScheme(); + return (scheme != null && (scheme.equals("http") || scheme.equals("https"))) + && endpoint.getHost() != null + && !endpoint.getHost().trim().isEmpty(); + } + + // Getters for the S3Configuration class + Duration getConnectionTimeout() { + return connectionTimeout; + } + + Duration getSocketTimeout() { + return socketTimeout; + } + + int getMaxConnections() { + return maxConnections; + } + + int getMaxRetries() { + return maxRetries; + } + + Duration getRetryInterval() { + return retryInterval; + } + + Duration getApiCallTimeout() { + return apiCallTimeout; + } + + Duration getApiCallAttemptTimeout() { + return apiCallAttemptTimeout; + } + + @Nullable + String getAccessKey() { + return accessKey; + } + + @Nullable + String getSecretKey() { + return secretKey; + } + + @Nullable + String getSessionToken() { + return sessionToken; + } + + String getRegion() { + return region; + } + + @Nullable + URI getEndpoint() { + return endpoint; + } + + boolean isPathStyleAccess() { + return pathStyleAccess; + } + + boolean isChecksumValidation() { + return checksumValidation; + } + + boolean isSslEnabled() { + return sslEnabled; + } + + boolean isVerifySslCertificates() { + return verifySslCertificates; + } + + @Nullable + String getTrustStorePath() { + return trustStorePath; + } + + @Nullable + String getTrustStorePassword() { + return trustStorePassword; + } + + int getBufferSize() { + return bufferSize; + } + + @Nullable + org.apache.hadoop.conf.Configuration getHadoopConfiguration() { + return hadoopConfiguration; + } +} diff --git a/flink-filesystems/flink-s3-fs-hadoop/src/main/java/org/apache/flink/fs/s3hadoop/S3FileSystemFactory.java b/flink-filesystems/flink-s3-fs-hadoop/src/main/java/org/apache/flink/fs/s3hadoop/S3FileSystemFactory.java index cab089361eaf4..7b8a90590dc84 100644 --- a/flink-filesystems/flink-s3-fs-hadoop/src/main/java/org/apache/flink/fs/s3hadoop/S3FileSystemFactory.java +++ b/flink-filesystems/flink-s3-fs-hadoop/src/main/java/org/apache/flink/fs/s3hadoop/S3FileSystemFactory.java @@ -37,17 +37,40 @@ public class S3FileSystemFactory extends AbstractS3FileSystemFactory { private static final Logger LOG = LoggerFactory.getLogger(S3FileSystemFactory.class); + + static { + // Static initializer - this will log when the class is first loaded + String timestamp = java.time.Instant.now().toString(); + String message = String.format("=== CUSTOM S3FileSystemFactory LOADED [%s] - Hadoop 3.4.2 + Conditional Writes ===", timestamp); + System.err.println(message); + LOG.error(message); + + // Also log the JAR location to help with debugging + try { + String jarLocation = S3FileSystemFactory.class.getProtectionDomain().getCodeSource().getLocation().toString(); + LOG.error("=== JAR Location: {} ===", jarLocation); + System.err.println("=== JAR Location: " + jarLocation + " ==="); + } catch (Exception e) { + LOG.error("=== Could not determine JAR location: {} ===", e.getMessage()); + } + } private static final String[] FLINK_CONFIG_PREFIXES = {"s3.", "s3a.", "fs.s3a."}; private static final String[][] MIRRORED_CONFIG_KEYS = { {"fs.s3a.access-key", "fs.s3a.access.key"}, {"fs.s3a.secret-key", "fs.s3a.secret.key"}, - {"fs.s3a.path-style-access", "fs.s3a.path.style.access"} + {"fs.s3a.path-style-access", "fs.s3a.path.style.access"}, + {"fs.s3a.requester-pays-enabled", "fs.s3a.requester.pays.enabled"}, + {"fs.s3a.create-conditional-enabled", "fs.s3a.create.conditional.enabled"}, + {"s3a.create-conditional-enabled", "fs.s3a.create.conditional.enabled"} }; public S3FileSystemFactory() { super("Hadoop s3a file system", createHadoopConfigLoader()); + + // Distinctive log message to confirm our custom JAR is loaded + LOG.error("=== CUSTOM S3 FILESYSTEM JAR LOADED - Hadoop 3.4.2 + Conditional Writes ==="); } @Override diff --git a/flink-filesystems/flink-s3-fs-hadoop/src/main/resources/META-INF/NOTICE b/flink-filesystems/flink-s3-fs-hadoop/src/main/resources/META-INF/NOTICE index 74d3a9f470461..f20b03a4d1b0e 100644 --- a/flink-filesystems/flink-s3-fs-hadoop/src/main/resources/META-INF/NOTICE +++ b/flink-filesystems/flink-s3-fs-hadoop/src/main/resources/META-INF/NOTICE @@ -28,12 +28,12 @@ This project bundles the following dependencies under the Apache Software Licens - org.apache.commons:commons-configuration2:2.1.1 - org.apache.commons:commons-lang3:3.12.0 - org.apache.commons:commons-text:1.10.0 -- org.apache.hadoop.thirdparty:hadoop-shaded-guava:1.1.1 -- org.apache.hadoop.thirdparty:hadoop-shaded-protobuf_3_7:1.1.1 -- org.apache.hadoop:hadoop-annotations:3.3.4 -- org.apache.hadoop:hadoop-auth:3.3.4 -- org.apache.hadoop:hadoop-aws:3.3.4 -- org.apache.hadoop:hadoop-common:3.3.4 +- org.apache.hadoop.thirdparty:hadoop-shaded-guava:1.4.0 +- org.apache.hadoop.thirdparty:hadoop-shaded-protobuf_3_25:1.4.0 +- org.apache.hadoop:hadoop-annotations:3.4.2 +- org.apache.hadoop:hadoop-auth:3.4.2 +- org.apache.hadoop:hadoop-aws:3.4.2 +- org.apache.hadoop:hadoop-common:3.4.2 - org.apache.httpcomponents:httpclient:4.5.13 - org.apache.httpcomponents:httpcore:4.4.14 - org.apache.kerby:kerb-core:1.0.1 diff --git a/flink-filesystems/flink-s3-fs-hadoop/src/test/java/org/apache/flink/fs/s3hadoop/HadoopS3AccessHelperTest.java b/flink-filesystems/flink-s3-fs-hadoop/src/test/java/org/apache/flink/fs/s3hadoop/HadoopS3AccessHelperTest.java new file mode 100644 index 0000000000000..3646dec5588c9 --- /dev/null +++ b/flink-filesystems/flink-s3-fs-hadoop/src/test/java/org/apache/flink/fs/s3hadoop/HadoopS3AccessHelperTest.java @@ -0,0 +1,655 @@ +/* + * 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.s3hadoop; + +import com.amazonaws.services.s3.model.CompleteMultipartUploadResult; +import com.amazonaws.services.s3.model.PutObjectResult; +import com.amazonaws.services.s3.model.UploadPartResult; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.s3a.S3AFileSystem; +import org.apache.hadoop.fs.s3a.WriteOperationHelper; +import org.apache.hadoop.fs.s3a.impl.PutObjectOptions; +import org.junit.Test; +import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadResponse; +import software.amazon.awssdk.services.s3.model.PutObjectResponse; +import software.amazon.awssdk.services.s3.model.RequestCharged; +import software.amazon.awssdk.services.s3.model.UploadPartResponse; + +import java.lang.reflect.Method; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +/** + * Unit tests for {@link HadoopS3AccessHelper} to verify correct AWS SDK v2 to v1 conversions and + * Hadoop 3.4.2 API compatibility. + */ +public class HadoopS3AccessHelperTest { + + @Test + public void testHadoop342ApiCompatibility() throws Exception { + // Verify that we're using Hadoop 3.4.2 compatible APIs + + // Check that PutObjectOptions.keepingDirs() method exists (Hadoop 3.4.2 API) + Method keepingDirsMethod = PutObjectOptions.class.getMethod("keepingDirs"); + assertNotNull(keepingDirsMethod); + + // Verify the method returns the expected type + PutObjectOptions options = PutObjectOptions.keepingDirs(); + assertNotNull(options); + } + + @Test + public void testCallbacksImplementS3Operations() throws Exception { + // Get the callbacks to test that they actually implement S3 operations + WriteOperationHelper.WriteOperationHelperCallbacks callbacks = getDefaultCallbacks(); + + // Test that uploadPart callback attempts to perform actual S3 operations + software.amazon.awssdk.services.s3.model.UploadPartRequest request = + software.amazon.awssdk.services.s3.model.UploadPartRequest.builder() + .bucket("test-bucket") + .key("test-key") + .uploadId("test-upload-id") + .partNumber(1) + .build(); + software.amazon.awssdk.core.sync.RequestBody body = + software.amazon.awssdk.core.sync.RequestBody.empty(); + + try { + callbacks.uploadPart(request, body, null); + // If we get here, the operation succeeded (unlikely in test environment) + assertTrue("S3 operation succeeded", true); + } catch (RuntimeException e) { + // Expected in test environment due to missing AWS credentials or network + assertTrue( + "Should get RuntimeException for S3 operations in test environment", + e.getMessage().contains("Failed to upload S3 part") + || e.getMessage().contains("Failed to create S3 client")); + } + } + + @Test + public void testCompleteMultipartUploadCallback() throws Exception { + // Get the callbacks to test that they actually implement S3 operations + WriteOperationHelper.WriteOperationHelperCallbacks callbacks = getDefaultCallbacks(); + + // Test that completeMultipartUpload callback attempts to perform actual S3 operations + software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest request = + software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest.builder() + .bucket("test-bucket") + .key("test-key") + .uploadId("test-upload-id") + .build(); + + try { + callbacks.completeMultipartUpload(request); + // If we get here, the operation succeeded (unlikely in test environment) + assertTrue("S3 complete operation succeeded", true); + } catch (RuntimeException e) { + // Expected in test environment due to missing AWS credentials or network + assertTrue( + "Should get RuntimeException for S3 operations in test environment", + e.getMessage().contains("Failed to complete S3 multipart upload") + || e.getMessage().contains("Failed to create S3 client")); + } + } + + @Test + public void testUploadPartResponseConversion() { + // Create an UploadPartResponse with available properties + UploadPartResponse response = + UploadPartResponse.builder() + .eTag("test-etag-123") + .requestCharged(RequestCharged.REQUESTER) + .sseCustomerAlgorithm("AES256") + .build(); + + UploadPartResult result = convertUploadPartResponse(response, 5); + + // Verify all properties are correctly copied + assertEquals("test-etag-123", result.getETag()); + assertEquals(5, result.getPartNumber()); + assertTrue(result.isRequesterCharged()); + + // Verify SSE algorithm property + assertEquals("AES256", result.getSSECustomerAlgorithm()); + } + + @Test + public void testUploadPartResponseConversionWithoutRequestCharged() { + // Create an UploadPartResponse without requestCharged + UploadPartResponse response = UploadPartResponse.builder().eTag("test-etag-456").build(); + + UploadPartResult result = convertUploadPartResponse(response, 3); + + // Verify properties are correctly set + assertEquals("test-etag-456", result.getETag()); + assertEquals(3, result.getPartNumber()); + assertFalse(result.isRequesterCharged()); // Should default to false + } + + @Test + public void testPutObjectResponseConversion() { + // Create a PutObjectResponse with all properties + PutObjectResponse response = + PutObjectResponse.builder() + .eTag("put-object-etag") + .requestCharged(RequestCharged.REQUESTER) + .build(); + + PutObjectResult result = convertPutObjectResponse(response); + + // Verify all properties are correctly copied + assertEquals("put-object-etag", result.getETag()); + assertTrue(result.isRequesterCharged()); + } + + @Test + public void testPutObjectResponseConversionWithoutRequestCharged() { + // Create a PutObjectResponse without requestCharged + PutObjectResponse response = PutObjectResponse.builder().eTag("put-object-etag-2").build(); + + PutObjectResult result = convertPutObjectResponse(response); + + // Verify properties are correctly set + assertEquals("put-object-etag-2", result.getETag()); + assertFalse(result.isRequesterCharged()); // Should default to false + } + + @Test + public void testCompleteMultipartUploadResponseConversion() { + // Create a CompleteMultipartUploadResponse with all properties + CompleteMultipartUploadResponse response = + CompleteMultipartUploadResponse.builder() + .eTag("complete-etag") + .bucket("test-bucket") + .key("test/key") + .requestCharged(RequestCharged.REQUESTER) + .build(); + + CompleteMultipartUploadResult result = convertCompleteMultipartUploadResponse(response); + + // Verify all properties are correctly copied + assertEquals("complete-etag", result.getETag()); + assertEquals("test-bucket", result.getBucketName()); + assertEquals("test/key", result.getKey()); + assertTrue(result.isRequesterCharged()); + } + + @Test + public void testCompleteMultipartUploadResponseConversionWithoutRequestCharged() { + // Create a CompleteMultipartUploadResponse without requestCharged + CompleteMultipartUploadResponse response = + CompleteMultipartUploadResponse.builder() + .eTag("complete-etag-2") + .bucket("test-bucket-2") + .key("test/key2") + .build(); + + CompleteMultipartUploadResult result = convertCompleteMultipartUploadResponse(response); + + // Verify properties are correctly set + assertEquals("complete-etag-2", result.getETag()); + assertEquals("test-bucket-2", result.getBucketName()); + assertEquals("test/key2", result.getKey()); + assertFalse(result.isRequesterCharged()); // Should default to false + } + + @Test + public void testRequestChargedConversionLogic() { + // Test various RequestCharged values + assertTrue(isRequesterCharged(RequestCharged.REQUESTER)); + assertFalse(isRequesterCharged(RequestCharged.UNKNOWN_TO_SDK_VERSION)); + assertFalse(isRequesterCharged(null)); + } + + // Helper methods for testing internal conversion logic + + private WriteOperationHelper.WriteOperationHelperCallbacks getDefaultCallbacks() + throws Exception { + // Since createCallbacks is now non-static, we need a mock instance + try { + // For test purposes, we'll create a mock instance to access the callbacks + S3AFileSystem mockS3a = new S3AFileSystem(); + Configuration conf = new Configuration(); + HadoopS3AccessHelper helper = new HadoopS3AccessHelper(mockS3a, conf); + + Method createCallbacksMethod = + HadoopS3AccessHelper.class.getDeclaredMethod("createCallbacks"); + createCallbacksMethod.setAccessible(true); + return (WriteOperationHelper.WriteOperationHelperCallbacks) + createCallbacksMethod.invoke(helper); + } catch (Exception e) { + // If we can't create S3A helper in test, skip the callback tests + throw new org.junit.AssumptionViolatedException( + "Cannot test callbacks without S3A setup", e); + } + } + + private UploadPartResult convertUploadPartResponse( + UploadPartResponse response, int partNumber) { + // Simulate the conversion logic from uploadPart method + UploadPartResult result = new UploadPartResult(); + result.setETag(response.eTag()); + result.setPartNumber(partNumber); + if (response.requestCharged() != null) { + result.setRequesterCharged(response.requestCharged().toString().equals("requester")); + } + + // Copy server-side encryption algorithm if available + if (response.sseCustomerAlgorithm() != null) { + result.setSSECustomerAlgorithm(response.sseCustomerAlgorithm()); + } + + return result; + } + + private PutObjectResult convertPutObjectResponse(PutObjectResponse response) { + // Simulate the conversion logic from putObject method + PutObjectResult result = new PutObjectResult(); + result.setETag(response.eTag()); + if (response.requestCharged() != null) { + result.setRequesterCharged(response.requestCharged().toString().equals("requester")); + } + + // Copy server-side encryption algorithm if available + if (response.sseCustomerAlgorithm() != null) { + result.setSSECustomerAlgorithm(response.sseCustomerAlgorithm()); + } + + return result; + } + + private CompleteMultipartUploadResult convertCompleteMultipartUploadResponse( + CompleteMultipartUploadResponse response) { + // Simulate the conversion logic from commitMultiPartUpload method + CompleteMultipartUploadResult result = new CompleteMultipartUploadResult(); + result.setETag(response.eTag()); + result.setBucketName(response.bucket()); + result.setKey(response.key()); + if (response.requestCharged() != null) { + result.setRequesterCharged(response.requestCharged().toString().equals("requester")); + } + + // CompleteMultipartUploadResponse typically only has basic properties + // SSE properties are not commonly available on this response type + + return result; + } + + private boolean isRequesterCharged(RequestCharged requestCharged) { + // Simulate the requester charged conversion logic + return requestCharged != null && requestCharged.toString().equals("requester"); + } + + // ========== Hadoop 3.4.2 API Verification Tests ========== + + @Test + public void testHadoop342VersionInUse() throws Exception { + // Verify we're using Hadoop 3.4.2 by checking the version info + String hadoopVersion = org.apache.hadoop.util.VersionInfo.getVersion(); + + assertNotNull("Hadoop version should not be null", hadoopVersion); + assertTrue( + "Expected Hadoop version 3.4.2, but got: " + hadoopVersion, + hadoopVersion.startsWith("3.4.2")); + } + + @Test + public void testHadoop342S3AFileSystemPackageExists() throws Exception { + // Test that core S3A package classes exist in Hadoop 3.4.2 + try { + // These should be available in Hadoop 3.4.2 + Class.forName("org.apache.hadoop.fs.s3a.S3AFileSystem"); + Class.forName("org.apache.hadoop.fs.s3a.WriteOperationHelper"); + + // Basic test passes if we can load these core classes + assertTrue("Core S3A classes found in Hadoop 3.4.2", true); + + } catch (ClassNotFoundException e) { + fail("Core S3A classes not found - may not be using Hadoop 3.4.2: " + e.getMessage()); + } + } + + @Test + public void testHadoop342S3AFileSystemClassExists() throws Exception { + // Verify that S3AFileSystem class exists and is accessible + try { + Class s3aFileSystemClass = Class.forName("org.apache.hadoop.fs.s3a.S3AFileSystem"); + assertNotNull("S3AFileSystem class should be accessible", s3aFileSystemClass); + + // Check for some methods that should exist in Hadoop 3.4.2 + s3aFileSystemClass.getMethod("getCanonicalServiceName"); + s3aFileSystemClass.getMethod("getBucket"); + + } catch (ClassNotFoundException e) { + fail("S3AFileSystem class not found: " + e.getMessage()); + } catch (NoSuchMethodException e) { + fail( + "Expected S3AFileSystem methods not found - may not be using correct Hadoop version: " + + e.getMessage()); + } + } + + @Test + public void testHadoop342WriteOperationHelperExists() throws Exception { + // Verify WriteOperationHelper class exists (methods may vary by version) + try { + Class writeOpHelperClass = + Class.forName("org.apache.hadoop.fs.s3a.WriteOperationHelper"); + assertNotNull("WriteOperationHelper class should be accessible", writeOpHelperClass); + + // Just verify the class exists - method signatures may vary in different versions + assertTrue("WriteOperationHelper class found in Hadoop 3.4.2", true); + + } catch (ClassNotFoundException e) { + fail("WriteOperationHelper class not found: " + e.getMessage()); + } + } + + @Test + public void testS3ClientConsistencyInMultipartUploadLifecycle() throws Exception { + // This test verifies the fix for the NoSuchUploadException issue where different S3 clients + // were used for upload initiation vs. callbacks, causing upload IDs to be invalid across + // clients. + // + // The issue was: + // 1. startMultiPartUpload() used s3accessHelper (S3AFileSystem's client) + // 2. Callbacks used createS3Client() (created NEW client) + // 3. Upload ID from one client didn't work with another client! + // + // The fix ensures callbacks use consistent S3 client configuration via + // getS3ClientFromFileSystem() + + // Get the default callbacks to test that they are properly implemented + WriteOperationHelper.WriteOperationHelperCallbacks callbacks = getDefaultCallbacks(); + + // Test uploadPart callback - should attempt real S3 operation, not throw + // UnsupportedOperationException + software.amazon.awssdk.services.s3.model.UploadPartRequest uploadRequest = + software.amazon.awssdk.services.s3.model.UploadPartRequest.builder() + .bucket("test-bucket") + .key("test-key") + .uploadId("test-upload-id") + .partNumber(1) + .build(); + software.amazon.awssdk.core.sync.RequestBody requestBody = + software.amazon.awssdk.core.sync.RequestBody.empty(); + + try { + callbacks.uploadPart(uploadRequest, requestBody, null); + fail("Should throw exception due to missing AWS credentials/connectivity, not succeed"); + } catch (RuntimeException e) { + // Expected: Should be AWS connectivity/credential error, NOT + // UnsupportedOperationException + assertFalse( + "Should not throw UnsupportedOperationException - callbacks are implemented", + e instanceof UnsupportedOperationException); + assertTrue( + "Should be AWS-related error indicating real S3 operation was attempted", + e.getMessage().toLowerCase().contains("s3") + || e.getMessage().toLowerCase().contains("aws") + || e.getMessage().toLowerCase().contains("credentials") + || e.getMessage().toLowerCase().contains("unable to execute")); + } + + // Test completeMultipartUpload callback - should attempt real S3 operation + software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest completeRequest = + software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest.builder() + .bucket("test-bucket") + .key("test-key") + .uploadId("test-upload-id") + .multipartUpload( + m -> + m.parts( + software.amazon.awssdk.services.s3.model + .CompletedPart.builder() + .partNumber(1) + .eTag("test-etag") + .build())) + .build(); + + try { + callbacks.completeMultipartUpload(completeRequest); + fail("Should throw exception due to missing AWS credentials/connectivity, not succeed"); + } catch (RuntimeException e) { + // Expected: Should be AWS connectivity/credential error, NOT + // UnsupportedOperationException + assertFalse( + "Should not throw UnsupportedOperationException - callbacks are implemented", + e instanceof UnsupportedOperationException); + assertTrue( + "Should be AWS-related error indicating real S3 operation was attempted", + e.getMessage().toLowerCase().contains("s3") + || e.getMessage().toLowerCase().contains("aws") + || e.getMessage().toLowerCase().contains("credentials") + || e.getMessage().toLowerCase().contains("unable to execute")); + } + } + + @Test + public void testHadoop342SpecificClasses() throws Exception { + // Test for classes that we know should exist in Hadoop 3.4.2 + String[] expectedClasses = { + "org.apache.hadoop.fs.s3a.S3AFileSystem", + "org.apache.hadoop.fs.s3a.WriteOperationHelper", + "org.apache.hadoop.util.VersionInfo", + "org.apache.hadoop.fs.statistics.DurationTrackerFactory" + }; + + for (String className : expectedClasses) { + try { + Class.forName(className); + // Class exists - good! + } catch (ClassNotFoundException e) { + fail( + "Expected Hadoop 3.4.2 class not found: " + + className + + " - " + + e.getMessage()); + } + } + } + + @Test + public void testHadoop342AWSSDKIntegration() throws Exception { + // Verify that we can access AWS SDK v2 classes that are used in Hadoop 3.4.2 + try { + // These should be available in our shaded JAR + Class.forName("software.amazon.awssdk.services.s3.model.UploadPartRequest"); + Class.forName("software.amazon.awssdk.services.s3.model.UploadPartResponse"); + Class.forName( + "software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest"); + Class.forName( + "software.amazon.awssdk.services.s3.model.CompleteMultipartUploadResponse"); + Class.forName("software.amazon.awssdk.services.s3.model.RequestCharged"); + + } catch (ClassNotFoundException e) { + fail( + "Expected AWS SDK v2 classes not found - may indicate Hadoop 3.4.2 integration issue: " + + e.getMessage()); + } + } + + @Test + public void testHadoop342BuildInformation() throws Exception { + // Additional verification of Hadoop build information + String version = org.apache.hadoop.util.VersionInfo.getVersion(); + String revision = org.apache.hadoop.util.VersionInfo.getRevision(); + + assertNotNull("Hadoop version should not be null", version); + assertNotNull("Hadoop revision should not be null", revision); + + // Print build info for debugging + System.out.println("Hadoop Version: " + version); + System.out.println("Hadoop Revision: " + revision); + + // Verify this is indeed 3.4.2 + assertTrue("Expected Hadoop 3.4.2, got: " + version, version.startsWith("3.4.2")); + } + + /** + * Test for S3 client consistency throughout multipart upload lifecycle. This test verifies the + * fix for the client inconsistency issue that caused NoSuchUploadException. + * + *
Background: The issue occurred because different S3 client instances were used: - + * startMultiPartUpload() used S3AFileSystem's internal client - uploadPart() callback created a + * NEW S3Client instance - completeMultipartUpload() callback created ANOTHER NEW S3Client + * instance + * + *
This caused upload IDs from one client to be invalid for another, leading to: "The + * specified multipart upload does not exist. The upload ID may be invalid, or the upload may + * have been aborted or completed." + */ + @Test + public void testS3ClientConsistencyThroughoutMultipartUploadLifecycle() throws Exception { + // Test that verifies the S3 client caching mechanism that prevents NoSuchUploadException + + // The core issue was that different S3 client instances were used: + // 1. startMultiPartUpload() used S3AFileSystem's internal client + // 2. uploadPart() callback created a NEW S3Client instance + // 3. completeMultipartUpload() callback created ANOTHER NEW S3Client instance + + // This test verifies that the caching mechanism ensures consistency + + // Since HadoopS3AccessHelper requires S3AFileSystem, and mocking is complex, + // we test that the fix exists by verifying the cached S3 client field is present + + // Verify that the factory-based approach is used instead of instance caching + try { + HadoopS3AccessHelper.class.getDeclaredField("cachedS3Client"); + fail("HadoopS3AccessHelper should NOT have cachedS3Client field with factory approach"); + } catch (NoSuchFieldException expected) { + // This is expected - the factory approach doesn't use instance caching + } + + // Verify the S3ClientConfigurationFactory class exists for the shared approach + try { + Class factoryClass = + Class.forName("org.apache.flink.fs.s3hadoop.S3ClientConfigurationFactory"); + assertTrue("S3ClientConfigurationFactory should exist", factoryClass != null); + + // Verify the factory has the acquireS3Client method + Method acquireS3ClientMethod = + factoryClass.getDeclaredMethod( + "acquireS3Client", org.apache.hadoop.fs.s3a.S3AFileSystem.class); + assertTrue("Factory should have acquireS3Client method", acquireS3ClientMethod != null); + assertTrue( + "acquireS3Client should be static", + java.lang.reflect.Modifier.isStatic(acquireS3ClientMethod.getModifiers())); + + } catch (ClassNotFoundException | NoSuchMethodException e) { + throw new AssertionError( + "S3ClientConfigurationFactory should exist with acquireS3Client method for shared approach", + e); + } + + // Verify the getS3ClientFromFileSystem method exists + try { + Method getS3ClientMethod = + HadoopS3AccessHelper.class.getDeclaredMethod("getS3ClientFromFileSystem"); + assertTrue("getS3ClientFromFileSystem method should exist", getS3ClientMethod != null); + + } catch (NoSuchMethodException e) { + throw new AssertionError( + "HadoopS3AccessHelper should have getS3ClientFromFileSystem method for consistency", + e); + } + + // Verify the close method exists for resource cleanup + try { + Method closeMethod = HadoopS3AccessHelper.class.getDeclaredMethod("close"); + assertTrue("close method should exist for resource cleanup", closeMethod != null); + + } catch (NoSuchMethodException e) { + throw new AssertionError( + "HadoopS3AccessHelper should have close method for resource cleanup", e); + } + } + + /** + * Test specifically for the NoSuchUploadException prevention. This test verifies that the fix + * mechanism is in place. + */ + @Test + public void testNoSuchUploadExceptionPrevention() throws Exception { + // Verify that the factory-based approach replaces the old createS3Client method + // The shared factory ensures S3 clients have consistent configuration + + try { + HadoopS3AccessHelper.class.getDeclaredMethod("createS3Client"); + fail( + "HadoopS3AccessHelper should NOT have createS3Client method with factory approach"); + } catch (NoSuchMethodException expected) { + // This is expected - the factory approach centralizes client creation + } + + // Verify the factory approach is used instead + try { + Class factoryClass = + Class.forName("org.apache.flink.fs.s3hadoop.S3ClientConfigurationFactory"); + + // Verify the factory has static client acquisition capabilities + Method acquireS3ClientMethod = + factoryClass.getDeclaredMethod( + "acquireS3Client", org.apache.hadoop.fs.s3a.S3AFileSystem.class); + assertTrue("Factory should have acquireS3Client method", acquireS3ClientMethod != null); + assertTrue( + "acquireS3Client should be static", + java.lang.reflect.Modifier.isStatic(acquireS3ClientMethod.getModifiers())); + + // Verify no global caching methods exist (they were removed to fix resource leaks) + try { + factoryClass.getDeclaredMethod("cleanup"); + fail( + "Factory should not have cleanup method - no global caching to avoid resource leaks"); + } catch (NoSuchMethodException expected) { + // This is expected - global caching was removed + } + + } catch (ClassNotFoundException | NoSuchMethodException e) { + throw new AssertionError( + "S3ClientConfigurationFactory should exist with client acquisition support", e); + } + + // Verify the reflection-based access method exists + // This provides a fallback when direct S3AFileSystem access fails + + try { + java.lang.reflect.Method getS3ClientFromFileSystemMethod = + HadoopS3AccessHelper.class.getDeclaredMethod("getS3ClientFromFileSystem"); + + // Verify it's private (internal implementation) + assertTrue( + "getS3ClientFromFileSystem should be private", + java.lang.reflect.Modifier.isPrivate( + getS3ClientFromFileSystemMethod.getModifiers())); + + } catch (NoSuchMethodException e) { + throw new AssertionError( + "HadoopS3AccessHelper should have getS3ClientFromFileSystem method", e); + } + + // This test ensures that the S3 client consistency fix is properly implemented + // The actual functionality is tested in integration tests where real S3 operations occur + } +} diff --git a/flink-filesystems/flink-s3-fs-hadoop/src/test/java/org/apache/flink/fs/s3hadoop/S3CallbackImplementationTest.java b/flink-filesystems/flink-s3-fs-hadoop/src/test/java/org/apache/flink/fs/s3hadoop/S3CallbackImplementationTest.java new file mode 100644 index 0000000000000..70cc399b62c42 --- /dev/null +++ b/flink-filesystems/flink-s3-fs-hadoop/src/test/java/org/apache/flink/fs/s3hadoop/S3CallbackImplementationTest.java @@ -0,0 +1,270 @@ +/* + * 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.s3hadoop; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.s3a.S3AFileSystem; +import org.apache.hadoop.fs.s3a.WriteOperationHelper; +import org.junit.Test; + +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +/** + * Unit tests specifically for the S3 callback implementation in HadoopS3AccessHelper. These tests + * verify that the callbacks actually implement S3 operations using AWS SDK v2 instead of throwing + * UnsupportedOperationException. + */ +public class S3CallbackImplementationTest { + + @Test + public void testUploadPartCallbackImplementation() throws Exception { + // Test that uploadPart callback attempts real S3 operations + WriteOperationHelper.WriteOperationHelperCallbacks callbacks = createTestCallbacks(); + + // Create a well-formed upload part request + software.amazon.awssdk.services.s3.model.UploadPartRequest uploadRequest = + software.amazon.awssdk.services.s3.model.UploadPartRequest.builder() + .bucket("test-bucket") + .key("test-key") + .uploadId("test-upload-id") + .partNumber(1) + .contentLength(100L) + .build(); + + software.amazon.awssdk.core.sync.RequestBody body = + software.amazon.awssdk.core.sync.RequestBody.fromString("test content"); + + try { + callbacks.uploadPart(uploadRequest, body, null); + // If we get here, the S3 operation succeeded (unlikely in test environment) + assertTrue("S3 upload operation succeeded", true); + } catch (RuntimeException e) { + // Expected in test environment due to no AWS credentials/connectivity + assertTrue( + "Should attempt S3 operation and fail with network/auth error, got: " + + e.getMessage(), + e.getMessage().contains("Failed to upload S3 part") + || e.getMessage().contains("Failed to create S3 client") + || e.getMessage().contains("Unable to load AWS") + || e.getMessage().contains("credentials")); + } + } + + @Test + public void testCompleteMultipartUploadCallbackImplementation() throws Exception { + // Test that completeMultipartUpload callback attempts real S3 operations + WriteOperationHelper.WriteOperationHelperCallbacks callbacks = createTestCallbacks(); + + // Create a well-formed complete multipart upload request + software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest completeRequest = + software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest.builder() + .bucket("test-bucket") + .key("test-key") + .uploadId("test-upload-id") + .multipartUpload( + m -> + m.parts( + software.amazon.awssdk.services.s3.model + .CompletedPart.builder() + .partNumber(1) + .eTag("test-etag") + .build())) + .build(); + + try { + callbacks.completeMultipartUpload(completeRequest); + // If we get here, the S3 operation succeeded (unlikely in test environment) + assertTrue("S3 complete operation succeeded", true); + } catch (RuntimeException e) { + // Expected in test environment due to no AWS credentials/connectivity + assertTrue( + "Should attempt S3 operation and fail with network/auth error, got: " + + e.getMessage(), + e.getMessage().contains("Failed to complete S3 multipart upload") + || e.getMessage().contains("Failed to create S3 client") + || e.getMessage().contains("Unable to load AWS") + || e.getMessage().contains("credentials")); + } + } + + @Test + public void testCallbackErrorHandlingWithInvalidRequests() throws Exception { + // Test error handling with malformed requests + WriteOperationHelper.WriteOperationHelperCallbacks callbacks = createTestCallbacks(); + + // Test upload part with missing required fields + software.amazon.awssdk.services.s3.model.UploadPartRequest invalidUploadRequest = + software.amazon.awssdk.services.s3.model.UploadPartRequest.builder() + // Missing bucket, key, uploadId - should cause error + .partNumber(1) + .build(); + + software.amazon.awssdk.core.sync.RequestBody body = + software.amazon.awssdk.core.sync.RequestBody.empty(); + + try { + callbacks.uploadPart(invalidUploadRequest, body, null); + fail("Should have thrown exception for invalid upload request"); + } catch (RuntimeException e) { + assertTrue( + "Should get error for invalid request", + e.getMessage().contains("Failed to upload S3 part") + || e.getMessage().contains("Failed to create S3 client")); + } + + // Test complete multipart upload with missing required fields + software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest + invalidCompleteRequest = + software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest + .builder() + // Missing bucket, key, uploadId - should cause error + .build(); + + try { + callbacks.completeMultipartUpload(invalidCompleteRequest); + fail("Should have thrown exception for invalid complete request"); + } catch (RuntimeException e) { + assertTrue( + "Should get error for invalid request", + e.getMessage().contains("Failed to complete S3 multipart upload") + || e.getMessage().contains("Failed to create S3 client")); + } + } + + @Test + public void testAWSSDKv2TypesUsed() { + // Verify we're using the correct AWS SDK v2 types + + // Test UploadPartRequest + software.amazon.awssdk.services.s3.model.UploadPartRequest uploadRequest = + software.amazon.awssdk.services.s3.model.UploadPartRequest.builder() + .bucket("test") + .key("test") + .uploadId("test") + .partNumber(1) + .contentLength(100L) + .build(); + + assertNotNull("UploadPartRequest should be creatable", uploadRequest); + assertTrue( + "Should be AWS SDK v2 type", + uploadRequest.getClass().getName().contains("software.amazon.awssdk")); + + // Test CompleteMultipartUploadRequest + software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest completeRequest = + software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest.builder() + .bucket("test") + .key("test") + .uploadId("test") + .build(); + + assertNotNull("CompleteMultipartUploadRequest should be creatable", completeRequest); + assertTrue( + "Should be AWS SDK v2 type", + completeRequest.getClass().getName().contains("software.amazon.awssdk")); + + // Test RequestBody + software.amazon.awssdk.core.sync.RequestBody body = + software.amazon.awssdk.core.sync.RequestBody.fromString("test"); + + assertNotNull("RequestBody should be creatable", body); + assertTrue( + "Should be AWS SDK v2 type", + body.getClass().getName().contains("software.amazon.awssdk")); + } + + @Test + public void testCallbacksDoNotThrowUnsupportedOperationException() throws Exception { + // Verify callbacks no longer throw UnsupportedOperationException + WriteOperationHelper.WriteOperationHelperCallbacks callbacks = createTestCallbacks(); + + // Create valid requests + software.amazon.awssdk.services.s3.model.UploadPartRequest uploadRequest = + software.amazon.awssdk.services.s3.model.UploadPartRequest.builder() + .bucket("test-bucket") + .key("test-key") + .uploadId("test-upload-id") + .partNumber(1) + .build(); + + software.amazon.awssdk.core.sync.RequestBody body = + software.amazon.awssdk.core.sync.RequestBody.empty(); + + software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest completeRequest = + software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest.builder() + .bucket("test-bucket") + .key("test-key") + .uploadId("test-upload-id") + .build(); + + // Both methods should attempt S3 operations, not throw UnsupportedOperationException + try { + callbacks.uploadPart(uploadRequest, body, null); + } catch (UnsupportedOperationException e) { + fail( + "uploadPart should not throw UnsupportedOperationException, it should attempt S3 operations"); + } catch (RuntimeException e) { + // Expected - should be network/auth errors, not UnsupportedOperationException + assertTrue( + "Should not be UnsupportedOperationException", + !e.getClass().equals(UnsupportedOperationException.class)); + } + + try { + callbacks.completeMultipartUpload(completeRequest); + } catch (UnsupportedOperationException e) { + fail( + "completeMultipartUpload should not throw UnsupportedOperationException, it should attempt S3 operations"); + } catch (RuntimeException e) { + // Expected - should be network/auth errors, not UnsupportedOperationException + assertTrue( + "Should not be UnsupportedOperationException", + !e.getClass().equals(UnsupportedOperationException.class)); + } + } + + /** + * Helper method to create test callbacks. We can't easily create a full S3AFileSystem in tests, + * but we can test the callback behavior. + */ + private WriteOperationHelper.WriteOperationHelperCallbacks createTestCallbacks() + throws Exception { + try { + // Try to create a minimal S3AFileSystem for testing + S3AFileSystem s3a = new S3AFileSystem(); + Configuration conf = new Configuration(); + + // Create HadoopS3AccessHelper to get the callbacks + HadoopS3AccessHelper helper = new HadoopS3AccessHelper(s3a, conf); + + // Use reflection to access the createCallbacks method for testing + java.lang.reflect.Method method = + HadoopS3AccessHelper.class.getDeclaredMethod("createCallbacks"); + method.setAccessible(true); + return (WriteOperationHelper.WriteOperationHelperCallbacks) method.invoke(helper); + + } catch (Exception e) { + // If we can't create the helper, we'll assume the test environment doesn't support it + throw new org.junit.AssumptionViolatedException( + "Cannot create S3 helper in test environment", e); + } + } +} diff --git a/flink-filesystems/flink-s3-fs-hadoop/src/test/java/org/apache/flink/fs/s3hadoop/S3ClientCredentialProviderTest.java b/flink-filesystems/flink-s3-fs-hadoop/src/test/java/org/apache/flink/fs/s3hadoop/S3ClientCredentialProviderTest.java new file mode 100644 index 0000000000000..def7c0a759cf3 --- /dev/null +++ b/flink-filesystems/flink-s3-fs-hadoop/src/test/java/org/apache/flink/fs/s3hadoop/S3ClientCredentialProviderTest.java @@ -0,0 +1,237 @@ +/* + * 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.s3hadoop; + +import org.apache.hadoop.conf.Configuration; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.lang.reflect.Method; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests for credential provider compatibility between Hadoop S3A and our custom S3 client. + * + *
This test verifies that {@link S3ClientConfigurationFactory} properly handles Hadoop's
+ * credential provider configuration to ensure consistent authentication between Hadoop's S3A
+ * filesystem and our custom S3 client used in multipart upload callbacks.
+ */
+class S3ClientCredentialProviderTest {
+
+ private Configuration hadoopConfig;
+
+ @BeforeEach
+ void setUp() {
+ hadoopConfig = new Configuration();
+ }
+
+ @AfterEach
+ void tearDown() {
+ // Clean up any S3 clients created during tests
+ try {
+ Method releaseMethod =
+ S3ClientConfigurationFactory.class.getDeclaredMethod("releaseS3Client");
+ releaseMethod.setAccessible(true);
+ releaseMethod.invoke(null);
+ } catch (Exception e) {
+ // Ignore cleanup errors in tests
+ }
+ }
+
+ @Test
+ void testCredentialProviderWithExplicitCredentials() throws Exception {
+ // Test case: Explicit access key and secret key provided
+ hadoopConfig.set("fs.s3a.access.key", "test-access-key");
+ hadoopConfig.set("fs.s3a.secret.key", "test-secret-key");
+
+ S3Configuration s3Config =
+ S3ConfigurationBuilder.fromHadoopConfiguration(hadoopConfig).build();
+
+ // Verify explicit credentials are stored
+ assertThat(s3Config.getAccessKey()).isEqualTo("test-access-key");
+ assertThat(s3Config.getSecretKey()).isEqualTo("test-secret-key");
+ assertThat(s3Config.getSessionToken()).isNull();
+ assertThat(s3Config.getHadoopConfiguration()).isEqualTo(hadoopConfig);
+ }
+
+ @Test
+ void testCredentialProviderWithSessionToken() throws Exception {
+ // Test case: Temporary credentials with session token
+ hadoopConfig.set("fs.s3a.access.key", "temp-access-key");
+ hadoopConfig.set("fs.s3a.secret.key", "temp-secret-key");
+ hadoopConfig.set("fs.s3a.session.token", "temp-session-token");
+
+ S3Configuration s3Config =
+ S3ConfigurationBuilder.fromHadoopConfiguration(hadoopConfig).build();
+
+ // Verify temporary credentials are stored
+ assertThat(s3Config.getAccessKey()).isEqualTo("temp-access-key");
+ assertThat(s3Config.getSecretKey()).isEqualTo("temp-secret-key");
+ assertThat(s3Config.getSessionToken()).isEqualTo("temp-session-token");
+ assertThat(s3Config.getHadoopConfiguration()).isEqualTo(hadoopConfig);
+ }
+
+ @Test
+ void testCredentialProviderWithHadoopProviderChain() throws Exception {
+ // Test case: No explicit credentials - should use Hadoop's credential provider chain
+ hadoopConfig.set(
+ "fs.s3a.aws.credentials.provider",
+ "org.apache.hadoop.fs.s3a.auth.IAMInstanceCredentialsProvider");
+
+ S3Configuration s3Config =
+ S3ConfigurationBuilder.fromHadoopConfiguration(hadoopConfig).build();
+
+ // Verify no explicit credentials but Hadoop config is preserved
+ assertThat(s3Config.getAccessKey()).isNull();
+ assertThat(s3Config.getSecretKey()).isNull();
+ assertThat(s3Config.getSessionToken()).isNull();
+ assertThat(s3Config.getHadoopConfiguration()).isNotNull();
+ assertThat(s3Config.getHadoopConfiguration().get("fs.s3a.aws.credentials.provider"))
+ .isEqualTo("org.apache.hadoop.fs.s3a.auth.IAMInstanceCredentialsProvider");
+ }
+
+ @Test
+ void testCredentialProviderWithMultipleProviders() throws Exception {
+ // Test case: Multiple credential providers in chain (typical production setup)
+ String providerChain =
+ "org.apache.hadoop.fs.s3a.auth.IAMInstanceCredentialsProvider,"
+ + "com.amazonaws.auth.EnvironmentVariableCredentialsProvider,"
+ + "com.amazonaws.auth.SystemPropertiesCredentialsProvider";
+
+ hadoopConfig.set("fs.s3a.aws.credentials.provider", providerChain);
+
+ S3Configuration s3Config =
+ S3ConfigurationBuilder.fromHadoopConfiguration(hadoopConfig).build();
+
+ // Verify the provider chain is preserved
+ assertThat(s3Config.getHadoopConfiguration().get("fs.s3a.aws.credentials.provider"))
+ .isEqualTo(providerChain);
+ }
+
+ @Test
+ void testCredentialProviderCompatibilityMethod() throws Exception {
+ // Test the createHadoopCompatibleCredentialProvider method via reflection
+ hadoopConfig.set(
+ "fs.s3a.aws.credentials.provider",
+ "org.apache.hadoop.fs.s3a.auth.IAMInstanceCredentialsProvider");
+
+ S3Configuration s3Config =
+ S3ConfigurationBuilder.fromHadoopConfiguration(hadoopConfig).build();
+
+ // Access the private method via reflection
+ Method createProviderMethod =
+ S3ClientConfigurationFactory.class.getDeclaredMethod(
+ "createHadoopCompatibleCredentialProvider", S3Configuration.class);
+ createProviderMethod.setAccessible(true);
+
+ // Invoke the method
+ software.amazon.awssdk.auth.credentials.AwsCredentialsProvider provider =
+ (software.amazon.awssdk.auth.credentials.AwsCredentialsProvider)
+ createProviderMethod.invoke(null, s3Config);
+
+ // Verify that a credential provider is returned
+ assertThat(provider).isNotNull();
+ assertThat(provider)
+ .isInstanceOf(
+ software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider.class);
+ }
+
+ @Test
+ void testCredentialProviderWithEmptyConfiguration() throws Exception {
+ // Test case: Empty configuration - should still work with defaults
+ S3Configuration s3Config =
+ S3ConfigurationBuilder.fromHadoopConfiguration(hadoopConfig).build();
+
+ // Verify defaults are handled correctly
+ assertThat(s3Config.getAccessKey()).isNull();
+ assertThat(s3Config.getSecretKey()).isNull();
+ assertThat(s3Config.getSessionToken()).isNull();
+ assertThat(s3Config.getHadoopConfiguration()).isEqualTo(hadoopConfig);
+ }
+
+ @Test
+ void testS3ClientCreationWithHadoopCredentials() throws Exception {
+ // Test that S3 client can be created with Hadoop credential configuration
+ hadoopConfig.set(
+ "fs.s3a.aws.credentials.provider",
+ "software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider");
+ hadoopConfig.set("fs.s3a.endpoint.region", "us-west-2");
+
+ try {
+ S3Configuration s3Config =
+ S3ConfigurationBuilder.fromHadoopConfiguration(hadoopConfig).build();
+
+ // This should not throw an exception even without real AWS credentials in test
+ // environment
+ software.amazon.awssdk.services.s3.S3Client client =
+ S3ClientConfigurationFactory.acquireS3Client(hadoopConfig);
+
+ assertThat(client).isNotNull();
+
+ // Clean up
+ S3ClientConfigurationFactory.releaseS3Client();
+
+ } catch (Exception e) {
+ // In test environment, we might get credential-related exceptions, which is expected
+ // The important thing is that the configuration parsing doesn't fail
+ assertThat(e.getMessage()).contains("credentials");
+ }
+ }
+
+ @Test
+ void testConfigurationHashIncludesCredentialInfo() throws Exception {
+ // Test that configuration hash accounts for credential settings
+ hadoopConfig.set("fs.s3a.access.key", "test-key");
+ S3Configuration config1 =
+ S3ConfigurationBuilder.fromHadoopConfiguration(hadoopConfig).build();
+
+ hadoopConfig.set("fs.s3a.access.key", "different-key");
+ S3Configuration config2 =
+ S3ConfigurationBuilder.fromHadoopConfiguration(hadoopConfig).build();
+
+ // Different credentials should produce different hashes
+ assertThat(config1.getConfigurationHash()).isNotEqualTo(config2.getConfigurationHash());
+ }
+
+ @Test
+ void testCredentialProviderLogging() throws Exception {
+ // Test that credential provider configuration is logged appropriately
+ hadoopConfig.set(
+ "fs.s3a.aws.credentials.provider",
+ "org.apache.hadoop.fs.s3a.auth.IAMInstanceCredentialsProvider");
+
+ S3Configuration s3Config =
+ S3ConfigurationBuilder.fromHadoopConfiguration(hadoopConfig).build();
+
+ // Access the private method to verify it logs the credential provider
+ Method createProviderMethod =
+ S3ClientConfigurationFactory.class.getDeclaredMethod(
+ "createHadoopCompatibleCredentialProvider", S3Configuration.class);
+ createProviderMethod.setAccessible(true);
+
+ // This should not throw and should handle the logging internally
+ software.amazon.awssdk.auth.credentials.AwsCredentialsProvider provider =
+ (software.amazon.awssdk.auth.credentials.AwsCredentialsProvider)
+ createProviderMethod.invoke(null, s3Config);
+
+ assertThat(provider).isNotNull();
+ }
+}
diff --git a/flink-filesystems/pom.xml b/flink-filesystems/pom.xml
index 22cbc0ce836d6..2a668cd7bc0eb 100644
--- a/flink-filesystems/pom.xml
+++ b/flink-filesystems/pom.xml
@@ -34,7 +34,7 @@ under the License.