sarutak commented on code in PR #57998: URL: https://github.com/apache/spark/pull/57998#discussion_r3780116322
########## connector/credential-aws/src/main/java/org/apache/spark/security/aws/SparkOidcAwsCredentialsProvider.java: ########## @@ -0,0 +1,112 @@ +/* + * 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.spark.security.aws; + +import java.util.Map; + +import software.amazon.awssdk.auth.credentials.AwsCredentials; +import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; +import software.amazon.awssdk.auth.credentials.AwsSessionCredentials; + +import org.apache.spark.SparkEnv; +import org.apache.spark.VersionedCredentials; +import org.apache.spark.annotation.DeveloperApi; +import org.apache.spark.deploy.security.UserCredentialManager; +import org.apache.spark.security.ServiceCredential; +import org.apache.spark.security.UserCredentials; + +/** + * :: DeveloperApi :: Review Comment: Ditto. ########## connector/credential-aws/src/main/java/org/apache/spark/security/aws/SparkOidcAwsCredentialsProvider.java: ########## @@ -0,0 +1,112 @@ +/* + * 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.spark.security.aws; + +import java.util.Map; + +import software.amazon.awssdk.auth.credentials.AwsCredentials; +import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; +import software.amazon.awssdk.auth.credentials.AwsSessionCredentials; + +import org.apache.spark.SparkEnv; +import org.apache.spark.VersionedCredentials; +import org.apache.spark.annotation.DeveloperApi; +import org.apache.spark.deploy.security.UserCredentialManager; +import org.apache.spark.security.ServiceCredential; +import org.apache.spark.security.UserCredentials; + +/** + * :: DeveloperApi :: + * A dynamic AWS credentials provider for executor-side S3A access that reads from + * Spark's credential store on every {@code resolveCredentials()} call. + * <p> + * This provider never caches credentials internally. Each call reads the latest + * {@link ServiceCredential} from the executor's {@code SparkEnv.userCredentials} store, + * ensuring that credential refreshes (delivered via {@code UpdateUserCredentials} RPC or + * {@code TaskDescription}) are immediately visible to S3A without requiring FileSystem + * cache invalidation. + * <p> + * Configure via: + * {@code fs.s3a.aws.credentials.provider=org.apache.spark.security.aws.SparkOidcAwsCredentialsProvider} + * <p> + * When {@code spark.security.oidc.enabled=true} and the user has not explicitly set + * {@code fs.s3a.aws.credentials.provider}, this provider is auto-configured. + * + * @since 5.0.0 + */ +@DeveloperApi +public class SparkOidcAwsCredentialsProvider implements AwsCredentialsProvider { Review Comment: AWS SDK requires `AwsCredentialsProvider` implementations to be thread-safe (shared across S3A signing threads). The current implementation IS thread-safe (stateless reads from AtomicReference), but this contract is not documented. Please add to the class Javadoc: ```java * <p>This implementation is thread-safe. Multiple threads may call * {@code resolveCredentials()} concurrently without external synchronization. * Each invocation independently reads from the atomic credential store. ``` ########## connector/credential-aws/src/main/java/org/apache/spark/security/aws/SparkOidcAwsCredentialsProvider.java: ########## @@ -0,0 +1,112 @@ +/* + * 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.spark.security.aws; + +import java.util.Map; + +import software.amazon.awssdk.auth.credentials.AwsCredentials; +import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; +import software.amazon.awssdk.auth.credentials.AwsSessionCredentials; + +import org.apache.spark.SparkEnv; +import org.apache.spark.VersionedCredentials; +import org.apache.spark.annotation.DeveloperApi; +import org.apache.spark.deploy.security.UserCredentialManager; +import org.apache.spark.security.ServiceCredential; +import org.apache.spark.security.UserCredentials; + +/** + * :: DeveloperApi :: + * A dynamic AWS credentials provider for executor-side S3A access that reads from + * Spark's credential store on every {@code resolveCredentials()} call. + * <p> + * This provider never caches credentials internally. Each call reads the latest + * {@link ServiceCredential} from the executor's {@code SparkEnv.userCredentials} store, + * ensuring that credential refreshes (delivered via {@code UpdateUserCredentials} RPC or + * {@code TaskDescription}) are immediately visible to S3A without requiring FileSystem + * cache invalidation. + * <p> + * Configure via: + * {@code fs.s3a.aws.credentials.provider=org.apache.spark.security.aws.SparkOidcAwsCredentialsProvider} + * <p> + * When {@code spark.security.oidc.enabled=true} and the user has not explicitly set + * {@code fs.s3a.aws.credentials.provider}, this provider is auto-configured. + * + * @since 5.0.0 Review Comment: `@since 4.4.0` ########## connector/credential-aws/src/main/java/org/apache/spark/security/aws/SparkOidcAwsCredentialsProvider.java: ########## @@ -0,0 +1,112 @@ +/* + * 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.spark.security.aws; + +import java.util.Map; + +import software.amazon.awssdk.auth.credentials.AwsCredentials; +import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; +import software.amazon.awssdk.auth.credentials.AwsSessionCredentials; + +import org.apache.spark.SparkEnv; +import org.apache.spark.VersionedCredentials; +import org.apache.spark.annotation.DeveloperApi; +import org.apache.spark.deploy.security.UserCredentialManager; +import org.apache.spark.security.ServiceCredential; +import org.apache.spark.security.UserCredentials; + +/** + * :: DeveloperApi :: + * A dynamic AWS credentials provider for executor-side S3A access that reads from + * Spark's credential store on every {@code resolveCredentials()} call. + * <p> + * This provider never caches credentials internally. Each call reads the latest + * {@link ServiceCredential} from the executor's {@code SparkEnv.userCredentials} store, + * ensuring that credential refreshes (delivered via {@code UpdateUserCredentials} RPC or + * {@code TaskDescription}) are immediately visible to S3A without requiring FileSystem + * cache invalidation. + * <p> + * Configure via: + * {@code fs.s3a.aws.credentials.provider=org.apache.spark.security.aws.SparkOidcAwsCredentialsProvider} + * <p> + * When {@code spark.security.oidc.enabled=true} and the user has not explicitly set + * {@code fs.s3a.aws.credentials.provider}, this provider is auto-configured. + * + * @since 5.0.0 + */ +@DeveloperApi Review Comment: This is not a developer API but an actual implementation so please remove this annotation. ########## core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala: ########## @@ -1089,6 +1089,15 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp // is processed by DriverEndpoint. VersionedCredentials.updateIfNewer( SparkEnv.get.userCredentials, version, initialCredentials) + + // Auto-configure the executor-side S3A credentials provider if the user has not + // explicitly set one. This ensures S3A reads dynamic credentials from the executor + // store without requiring manual Hadoop configuration. + val s3aProviderKey = "spark.hadoop.fs.s3a.aws.credentials.provider" Review Comment: I traced the propagation path and confirmed this works correctly for executors: - `sparkProperties` (used in `SparkAppConfig` reply) is a `lazy val` on `conf.getAll`, evaluated after `setupUserCredentialManager` completes -> OK - Executor's `SparkHadoopUtil.newConfiguration(conf)` picks up `spark.hadoop.*` from the received conf -> OK However, two questions: **(a) Driver-side S3A:** `SparkContext._hadoopConfiguration` is created (in `SparkContext` constructor, via `SparkHadoopUtil.get.newConfiguration(_conf)`) before `TaskSchedulerImpl.start()` triggers `setupUserCredentialManager`. This means the driver's own Hadoop Configuration does not pick up this late mutation. Is this intentional? - If yes (executor-only by design): please add a comment explaining that this setting only takes effect on executors because the driver's Hadoop Configuration is already materialized. - If no (driver should also use this provider): the auto-config needs to be moved earlier, or the driver needs to refresh its Hadoop Configuration after setup. **(b) Classpath guard:** The `credential-aws` module is activated by an optional Maven/SBT profile (`-Pcredential-aws`). If a user enables OIDC but doesn't include the profile, this sets a class name that doesn't exist on the classpath. The failure won't surface until the first S3A access (a `ClassNotFoundException` inside Hadoop's reflection-based provider instantiation). Please consider adding: ```scala val providerClass = "org.apache.spark.security.aws.SparkOidcAwsCredentialsProvider" if (!conf.contains(s3aProviderKey)) { try { Utils.classForName(providerClass) conf.set(s3aProviderKey, providerClass) } catch { case _: ClassNotFoundException => logWarning(log"credential-aws module not on classpath; " + log"skipping S3A provider auto-configuration") } } ``` ########## connector/credential-aws/src/main/java/org/apache/spark/security/aws/SparkOidcAwsCredentialsProvider.java: ########## @@ -0,0 +1,112 @@ +/* + * 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.spark.security.aws; + +import java.util.Map; + +import software.amazon.awssdk.auth.credentials.AwsCredentials; +import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; +import software.amazon.awssdk.auth.credentials.AwsSessionCredentials; + +import org.apache.spark.SparkEnv; +import org.apache.spark.VersionedCredentials; +import org.apache.spark.annotation.DeveloperApi; +import org.apache.spark.deploy.security.UserCredentialManager; +import org.apache.spark.security.ServiceCredential; +import org.apache.spark.security.UserCredentials; + +/** + * :: DeveloperApi :: + * A dynamic AWS credentials provider for executor-side S3A access that reads from + * Spark's credential store on every {@code resolveCredentials()} call. + * <p> + * This provider never caches credentials internally. Each call reads the latest + * {@link ServiceCredential} from the executor's {@code SparkEnv.userCredentials} store, + * ensuring that credential refreshes (delivered via {@code UpdateUserCredentials} RPC or + * {@code TaskDescription}) are immediately visible to S3A without requiring FileSystem + * cache invalidation. + * <p> + * Configure via: + * {@code fs.s3a.aws.credentials.provider=org.apache.spark.security.aws.SparkOidcAwsCredentialsProvider} + * <p> + * When {@code spark.security.oidc.enabled=true} and the user has not explicitly set + * {@code fs.s3a.aws.credentials.provider}, this provider is auto-configured. + * + * @since 5.0.0 + */ +@DeveloperApi +public class SparkOidcAwsCredentialsProvider implements AwsCredentialsProvider { + + /** S3A credential property keys (same as produced by AwsStsCredentialProvider). */ + private static final String ACCESS_KEY = "fs.s3a.access.key"; + private static final String SECRET_KEY = "fs.s3a.secret.key"; + private static final String SESSION_TOKEN = "fs.s3a.session.token"; + + /** The S3A scheme used to look up credentials in the UserCredentials bundle. */ + private static final String S3A_SCHEME = "s3a"; + + @Override + public AwsCredentials resolveCredentials() { Review Comment: S3A calls `resolveCredentials()` for every AWS API call (every GET, PUT, HEAD, multipart upload part). The current implementation deserializes the full `UserCredentials` via `ObjectInputStream` on each invocation. For I/O-heavy workloads (large shuffles to S3, Parquet column reads), this means thousands of deserialization operations per task. I understand the JIRA description says "never cache credentials internally." The intent of that requirement is to guarantee that stale credentials are never returned after a refresh, and a version-keyed memoization preserves that guarantee. The `VersionedCredentials.version()` is monotonically increasing and only changes when the driver pushes new credentials. If the version hasn't changed, the credential bytes are identical, so returning a cached deserialized result is semantically equivalent to re-deserializing. The credential store version only changes on driver-initiated renewal (minutes-scale), so the cache hit rate would be >99.99%. Could you add version-based caching like as follows? ```java private record CachedResult(long version, AwsSessionCredentials credentials) {} private volatile CachedResult cached; @Override public AwsCredentials resolveCredentials() { SparkEnv env = SparkEnv.get(); if (env == null) { ... } VersionedCredentials versioned = env.userCredentials().get(); if (versioned == null) { ... } CachedResult current = cached; if (current != null && current.version() == versioned.version()) { return current.credentials(); } UserCredentials credentials; try { credentials = UserCredentialManager.deserializeUserCredentials(versioned.bytes()); } catch (...) { ... } // ... extract s3a cred ... AwsSessionCredentials result = AwsSessionCredentials.create(accessKey, secretKey, sessionToken); cached = new CachedResult(versioned.version(), result); return result; } ``` ########## connector/credential-aws/src/main/java/org/apache/spark/security/aws/SparkOidcAwsCredentialsProvider.java: ########## @@ -0,0 +1,112 @@ +/* + * 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.spark.security.aws; + +import java.util.Map; + +import software.amazon.awssdk.auth.credentials.AwsCredentials; +import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; +import software.amazon.awssdk.auth.credentials.AwsSessionCredentials; + +import org.apache.spark.SparkEnv; +import org.apache.spark.VersionedCredentials; +import org.apache.spark.annotation.DeveloperApi; +import org.apache.spark.deploy.security.UserCredentialManager; +import org.apache.spark.security.ServiceCredential; +import org.apache.spark.security.UserCredentials; + +/** + * :: DeveloperApi :: + * A dynamic AWS credentials provider for executor-side S3A access that reads from + * Spark's credential store on every {@code resolveCredentials()} call. + * <p> + * This provider never caches credentials internally. Each call reads the latest + * {@link ServiceCredential} from the executor's {@code SparkEnv.userCredentials} store, + * ensuring that credential refreshes (delivered via {@code UpdateUserCredentials} RPC or + * {@code TaskDescription}) are immediately visible to S3A without requiring FileSystem + * cache invalidation. + * <p> + * Configure via: + * {@code fs.s3a.aws.credentials.provider=org.apache.spark.security.aws.SparkOidcAwsCredentialsProvider} + * <p> + * When {@code spark.security.oidc.enabled=true} and the user has not explicitly set + * {@code fs.s3a.aws.credentials.provider}, this provider is auto-configured. + * + * @since 5.0.0 + */ +@DeveloperApi +public class SparkOidcAwsCredentialsProvider implements AwsCredentialsProvider { + + /** S3A credential property keys (same as produced by AwsStsCredentialProvider). */ + private static final String ACCESS_KEY = "fs.s3a.access.key"; + private static final String SECRET_KEY = "fs.s3a.secret.key"; + private static final String SESSION_TOKEN = "fs.s3a.session.token"; + + /** The S3A scheme used to look up credentials in the UserCredentials bundle. */ + private static final String S3A_SCHEME = "s3a"; + + @Override + public AwsCredentials resolveCredentials() { + SparkEnv env = SparkEnv.get(); + if (env == null) { + throw new IllegalStateException( + "SparkEnv is not available. SparkOidcAwsCredentialsProvider can only be used " + + "within an active Spark executor."); + } + + VersionedCredentials versioned = env.userCredentials().get(); + if (versioned == null) { + throw new IllegalStateException( + "No credentials available in the executor credential store. " + + "Ensure spark.security.oidc.enabled=true and the driver has acquired " + + "credentials before executor tasks run."); + } + + UserCredentials credentials; + try { + credentials = UserCredentialManager.deserializeUserCredentials(versioned.bytes()); + } catch (Exception e) { + throw new IllegalStateException( + "Failed to deserialize credentials from executor store (version=" + + versioned.version() + "). The credential bytes may be corrupted or " + + "incompatible with this Spark version.", e); + } + + ServiceCredential s3aCred = credentials.forScheme(S3A_SCHEME).orElse(null); + if (s3aCred == null) { + throw new IllegalStateException( + "No credential found for scheme '" + S3A_SCHEME + "' in the executor " + + "credential store. Ensure an S3A-compatible CredentialProvider " + + "(e.g., AwsStsCredentialProvider) is configured on the driver."); + } + + Map<String, String> props = s3aCred.getProperties(); + String accessKey = props.get(ACCESS_KEY); + String secretKey = props.get(SECRET_KEY); + String sessionToken = props.get(SESSION_TOKEN); + + if (accessKey == null || secretKey == null || sessionToken == null) { Review Comment: An empty string `""` would pass this check but produce an invalid `AwsSessionCredentials`. The failure would appear as a cryptic AWS SDK auth error (signature mismatch or 403) at request-signing time rather than a clear Spark-level error. Please consider: ```java if (accessKey == null || accessKey.isEmpty() || secretKey == null || secretKey.isEmpty() || sessionToken == null || sessionToken.isEmpty()) { ``` -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
