andygrove commented on code in PR #6023: URL: https://github.com/apache/datafusion-comet/pull/6023#discussion_r4096120368
########## spark/src/main/spark-4.x/org/apache/comet/cloud/s3/AwsSdkCredentialProviderAdapter.java: ########## @@ -0,0 +1,109 @@ +/* + * 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.comet.cloud.s3; + +import java.net.URI; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.s3a.S3AUtils; + +import org.apache.comet.annotation.Public; +import org.apache.comet.util.ClassLoaders; + +import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; + +/** + * Wraps a raw AWS SDK v2 {@link AwsCredentialsProvider} named via {@code + * fs.s3a.comet.credential.adapter.class}, for a provider not registered through S3A. This is the + * spark-4.x (SDK v2) body. Prefer {@link HadoopS3ACredentialProviderAdapter} unless the provider is + * a plain SDK class not wired through Hadoop. + * + * <pre> + * spark.hadoop.fs.s3a.comet.credential.provider.class=org.apache.comet.cloud.s3.AwsSdkCredentialProviderAdapter + * spark.hadoop.fs.s3a.comet.credential.adapter.class=<FQCN of an AwsCredentialsProvider> + * </pre> + */ +@Public +public class AwsSdkCredentialProviderAdapter implements CometS3CredentialProvider { + + static final String DELEGATE_CLASS_PROPERTY = "comet.credential.adapter.class"; + + private Map<String, String> properties; + private final ConcurrentHashMap<String, AwsCredentialsProvider> delegates = + new ConcurrentHashMap<>(); + + @Override + public void initialize(Map<String, String> catalogProperties) { + this.properties = catalogProperties; + } + + @Override + public CometS3Credentials getCredentialsForPath(CometS3CredentialContext context) + throws Exception { + AwsCredentialsProvider provider = ensureDelegate(context.getBucket()); + return SdkCredentialExtraction.toCometCredentials(provider.resolveCredentials()); + } + + private AwsCredentialsProvider ensureDelegate(String bucket) throws Exception { + AwsCredentialsProvider existing = delegates.get(bucket); + if (existing != null) { + return existing; + } + synchronized (this) { + AwsCredentialsProvider delegate = delegates.get(bucket); + if (delegate == null) { + delegate = instantiate(bucket); + delegates.put(bucket, delegate); + } + return delegate; + } + } + + private AwsCredentialsProvider instantiate(String bucket) throws Exception { + String className = AdapterSupport.lookup(properties, bucket, DELEGATE_CLASS_PROPERTY); + if (className == null) { + throw new IllegalStateException( + "AwsSdkCredentialProviderAdapter requires fs.s3a." + + DELEGATE_CLASS_PROPERTY + + " (or the per-bucket variant) to name an AwsCredentialsProvider"); + } + Class<?> clazz = ClassLoaders.loadClass(className); Review Comment: This lookup runs on whichever thread makes the first credential fetch for the bucket. On a stage whose only leaf is a native scan, `jni_api.rs` spawns the stream onto Tokio. When I measured those workers for #5282, they had a null context class loader. `ClassLoaders.loadClass` then falls back to Comet's own loader, which can't see `--jars`. The dispatcher loads the adapter during planning on the task thread, so the adapter loads fine and then its delegate doesn't. The user guide's troubleshooting section points vendors at `--jars`, so that's where I'd expect a raw SDK provider to live. I reproduced it with a delegate that only exists on a child loader. I called `ensureInitialized` on a thread whose context loader is the child, then made the first `getCredentialsForPath` call on a thread with no context loader. That call throws `ClassNotFoundException`. The same handle then works on a thread that has the child loader, and after that it works everywhere from the cache. So whether a query succeeds depends on which plan shape happened to touch the bucket first in that executor. Could `initialize()` capture `ClassLoaders.contextOrDefault(...)`, since the dispatcher calls it during planning on the task thread, and use that loader for the delegate? The spark-3.x body has the same line. ########## spark/src/test/scala/org/apache/comet/cloud/s3/HadoopS3ACredentialProviderAdapterBridgeSuite.scala: ########## @@ -0,0 +1,130 @@ +/* + * 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.comet.cloud.s3 + +import scala.collection.mutable +import scala.util.Try + +import org.apache.spark.SparkConf +import org.apache.spark.sql.SaveMode +import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper +import org.apache.spark.sql.functions.{col, sum} + +import org.apache.comet.CometS3TestBase + +/** + * End-to-end MinIO tests for [[HadoopS3ACredentialProviderAdapter]] on the native Parquet path, + * using a delegate the native Rust list deliberately rejects. A successful read proves the + * adapter routed credential resolution through Hadoop S3A rather than the native reader failing + * with `Unsupported credential provider`. Together the two cases exercise the full round trip: + * the `fs.s3a.*` map crossing JNI and the adapter rebuilding a `Configuration` from it. + */ +class HadoopS3ACredentialProviderAdapterBridgeSuite + extends CometS3TestBase + with AdaptiveSparkPlanHelper { + + override protected val testBucketName = "hadoop-adapter-bucket" + private val staticKeyBucket = "hadoop-adapter-static-bucket" + + // The AWS default-chain FQCN must match what the active Hadoop-aws line's provider factory + // accepts, not merely which SDK jar is on the test classpath: the v2 SDK is present on the + // Spark 3.x test classpath too (Iceberg's S3 test deps), but Hadoop 3.3.4's factory only accepts + // the v1 interface. CredentialProviderListFactory exists only in Hadoop 3.4+ (the v2 line), so + // its presence is the reliable per-profile signal. Neither class is in Comet's native list. + private val defaultChainClass: String = + if (Try( + Class.forName("org.apache.hadoop.fs.s3a.auth.CredentialProviderListFactory")).isSuccess) { + "software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider" + } else { + "com.amazonaws.auth.DefaultAWSCredentialsProviderChain" + } + + private val savedProps = mutable.Map[String, String]() + private def setProp(key: String, value: String): Unit = { + savedProps(key) = System.getProperty(key) + System.setProperty(key, value) + } + private def restoreProps(): Unit = { + savedProps.foreach { + case (key, null) => System.clearProperty(key) + case (key, value) => System.setProperty(key, value) + } + savedProps.clear() + } + + override protected def sparkConf: SparkConf = { + val conf = super.sparkConf + conf.set( + "spark.hadoop.fs.s3a.comet.credential.provider.class", + classOf[HadoopS3ACredentialProviderAdapter].getName) + // Default bucket: delegate to the AWS default chain (credentials come from JVM system + // properties, set within the test that uses it). + conf.set( + s"spark.hadoop.fs.s3a.bucket.$testBucketName.aws.credentials.provider", + defaultChainClass) + // Static-key bucket: a chain whose first entry reads the static keys and whose fallback is the + // (native-rejected) default chain. With no system properties set, the fallback resolves + // nothing, so the read succeeds only if the static keys were forwarded and the first entry won. + conf.set( + s"spark.hadoop.fs.s3a.bucket.$staticKeyBucket.aws.credentials.provider", + s"org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider,$defaultChainClass") + conf + } + + test( + "native Parquet read via HadoopS3ACredentialProviderAdapter (AWS default chain delegate)") { + // Both the v1 (aws.secretKey) and v2 (aws.secretAccessKey) secret property names are set so the + // default chain resolves regardless of which SDK is on the classpath. + setProp("aws.accessKeyId", userName) + setProp("aws.secretKey", password) + setProp("aws.secretAccessKey", password) + try { + val path = s"s3a://$testBucketName/data/adapter.parquet" + val rowCount = 1000L + spark.range(0, rowCount).write.format("parquet").mode(SaveMode.Overwrite).save(path) + val expectedSum = (0L until rowCount).sum + + val df = spark.read.format("parquet").load(path).agg(sum(col("id"))) + val plan = df.queryExecution.executedPlan + assert(cometScans(plan).nonEmpty, s"Expected a Comet Parquet scan in plan:\n$plan") + // Success is only reachable if the adapter resolved credentials; otherwise the native reader + // throws "Unsupported credential provider: $defaultChainClass". + assert(df.first().getLong(0) == expectedSum) + } finally { + restoreProps() + } + } + + test("native Parquet read forwards static keys end to end through the adapter") { + createBucketIfNotExists(staticKeyBucket) + val path = s"s3a://$staticKeyBucket/data/static.parquet" + val rowCount = 500L + spark.range(0, rowCount).write.format("parquet").mode(SaveMode.Overwrite).save(path) + val expectedSum = (0L until rowCount).sum + + // No system properties are set here, so the chain's fallback cannot resolve. The read succeeds Review Comment: I don't think this case can fail anymore if the forwarding breaks. `AdapterSupport.toConfiguration` now seeds from `SparkEnv`'s `spark.hadoop.*`, and `CometS3TestBase` sets the MinIO keys in exactly that spelling. So the adapter finds `fs.s3a.access.key` in the seed whether or not it crossed JNI. I checked directly. With a local `SparkContext` carrying those keys and this bucket's `SimpleAWSCredentialsProvider,<default chain>` list, `initialize(Collections.emptyMap())` followed by `getCredentialsForPath` resolves the `SparkConf` key. The first case is in the same position, because its per-bucket provider is also set through `spark.hadoop.*`. Could this case give the real keys only to a path the seed can't see, such as `withSQLConf` around the write and the read, or read options? Then set the `spark.hadoop.*` spelling for this bucket to a wrong value, so a forward that drops the keys reads with the wrong one and fails. The scaladoc at line 36 and the class comment on the v2 `HadoopS3ACredentialProviderAdapterTest` say this suite covers the forwarding, so they should match whatever it ends up testing. ########## spark/pom.xml: ########## @@ -261,6 +285,25 @@ under the License. <profile> <id>spark-3.4</id> <dependencies> + <!-- + AWS SDK v1 for the built-in S3 credential adapters (spark-3.x source set). Use the s3 + module, not just core: it supplies com.amazonaws.auth.* for the adapter AND the + com.amazonaws.services.s3.model.* classes that hadoop-aws 3.3.4's S3AUtils API references + (the incremental compiler extracts that API). commons-logging is excluded because + jcl-over-slf4j already provides those classes (duplicate-class enforcer). + --> + <dependency> Review Comment: The spark-3.x adapter tests fail on both 3.4 and 3.5 with this dependency set. Six of the eleven error out with `NoClassDefFoundError: com/amazonaws/services/dynamodbv2/model/AmazonDynamoDBException` from the first `S3AUtils.propagateBucketOptions` call. hadoop-aws 3.3.4's `S3AUtils.translateException` still has the S3Guard branch for `AmazonDynamoDBException`, so `S3AUtils` can't link without the DynamoDB module. On main the 3.x test classpath gets it from `spark-hadoop-cloud` -> `hadoop-aws` -> `aws-java-sdk-bundle`. The new direct `hadoop-aws` dependency wins Maven's mediation, and because it excludes the bundle, the bundle drops out entirely. That also means S3A can't initialize in the existing MinIO suites on 3.x. Production is fine, since hadoop-aws 3.3.4 normally arrives with the bundle. Adding `com.amazonaws:aws-java-sdk-dynamodb:${aws-java-sdk-v1.version}` at `provided` scope next to this one, with the same `commons-logging` exclusion, made them all pass for me on 3.5. Could you add it to both 3.x profiles? These are also the bodies #6022 was reported against, so it would be good to see the MinIO suite run once on 3.4 or 3.5. ########## docs/source/user-guide/latest/s3-credential-providers.md: ########## @@ -36,6 +36,39 @@ You probably do, if any of these are true: - You have a custom Iceberg `client.factory` that injects a configured S3 client. - Spark queries against your S3 paths work, but the same queries with Comet enabled fail with 403. +## Built-in adapters + +If a native Parquet scan fails with `Unsupported credential provider: <class>` (for example `com.amazonaws.auth.DefaultAWSCredentialsProviderChain`), the class you named in `fs.s3a.aws.credentials.provider` is one that plain Spark/Hadoop accepts but Comet's native reader does not reimplement. Comet ships two built-in `CometS3CredentialProvider` adapters that fix this with a one-line config change; you leave your existing `fs.s3a.aws.credentials.provider` untouched. + +These adapters cover the Parquet native scan path only. Enabling one is opt-in: naming it is what activates it, and Comet's existing native provider handling is unchanged for everyone else. Review Comment: "Comet's existing native provider handling is unchanged for everyone else" isn't quite true for existing bridges. `forward_catalog_properties` runs for any provider class named on the Parquet path, not only these two. A vendor provider that got an empty map from `initialize()` in 1.0, as the 1.0 javadoc promised, now gets the whole `fs.s3a.*` subset including the static keys. It also gets one instance per distinct `fs.s3a.*` config rather than one per bucket. I'm fine with that as an additive change. But a vendor that logs the map, or treats an empty one as the Parquet path, needs to be able to find out. Could this sentence say so? And could the vendor contract paragraph further down, which describes what the Iceberg path's `catalogProperties` carries, say what the Parquet path carries now? ########## docs/source/user-guide/latest/s3-credential-providers.md: ########## @@ -36,6 +36,39 @@ You probably do, if any of these are true: - You have a custom Iceberg `client.factory` that injects a configured S3 client. - Spark queries against your S3 paths work, but the same queries with Comet enabled fail with 403. +## Built-in adapters + +If a native Parquet scan fails with `Unsupported credential provider: <class>` (for example `com.amazonaws.auth.DefaultAWSCredentialsProviderChain`), the class you named in `fs.s3a.aws.credentials.provider` is one that plain Spark/Hadoop accepts but Comet's native reader does not reimplement. Comet ships two built-in `CometS3CredentialProvider` adapters that fix this with a one-line config change; you leave your existing `fs.s3a.aws.credentials.provider` untouched. + +These adapters cover the Parquet native scan path only. Enabling one is opt-in: naming it is what activates it, and Comet's existing native provider handling is unchanged for everyone else. + +### `HadoopS3ACredentialProviderAdapter` (recommended) + +Delegates to Hadoop S3A's own provider construction, so it accepts everything the `fs.s3a.aws.credentials.provider` chain accepts (the default chain, web-identity, assumed-role, custom signers, per-bucket config). This is the general answer for the failure above. + +``` +spark.hadoop.fs.s3a.comet.credential.provider.class=org.apache.comet.cloud.s3.HadoopS3ACredentialProviderAdapter +# leave your existing config as-is, for example: +spark.hadoop.fs.s3a.aws.credentials.provider=com.amazonaws.auth.DefaultAWSCredentialsProviderChain +``` + +It needs no extra config: it reads the standard `fs.s3a.aws.credentials.provider` (and the per-bucket `fs.s3a.bucket.<bucket>.aws.credentials.provider`) itself, and Comet forwards the full `fs.s3a.*` config to it, so a provider chain (including static keys and assumed-role) resolves the same way it would under Spark. + +### `AwsSdkCredentialProviderAdapter` + +Wraps a single raw AWS SDK credential-provider class that is not registered through S3A. Name the delegate in a separate key: + +``` +spark.hadoop.fs.s3a.comet.credential.provider.class=org.apache.comet.cloud.s3.AwsSdkCredentialProviderAdapter +spark.hadoop.fs.s3a.comet.credential.adapter.class=<FQCN of your credential provider> +# per-bucket variant: +spark.hadoop.fs.s3a.bucket.<bucket>.comet.credential.adapter.class=<FQCN> +``` + +### Which one, and which Spark version + +Use `HadoopS3ACredentialProviderAdapter` unless you have a plain SDK provider not wired through S3A. Both class names are the same on every Comet build; each build automatically uses the AWS SDK its Hadoop line ships (v1 on the Spark 3.4/3.5 builds, v2 on 4.0+), so you configure one name and get the right implementation. Review Comment: The adapters link against hadoop-aws and the AWS SDK through the class loader that loaded Comet, so those jars have to be visible from there. With the installation guide's layout (Comet on `spark.executor.extraClassPath`) and hadoop-aws supplied through `--packages`, they aren't. I split a test classpath that way with two loaders, and `ensureInitialized` fails at planning with `NoClassDefFoundError: software/amazon/awssdk/auth/credentials/AwsCredentialsProvider`. That happens before the adapter's `LinkageError` handler can run, so the error gives no hint. Could this section say that hadoop-aws and the matching SDK need to be on the same classpath as Comet, for example `extraClassPath` or `$SPARK_HOME/jars`? ########## docs/source/about/versioning_policy.md: ########## @@ -212,6 +212,15 @@ The SPI consists of: - `CometS3Credentials`, the value a provider returns. - `CometS3CredentialContext` and `CometS3AccessMode`, describing the request being served. +Comet also ships two built-in implementations of the SPI as public API, so their class names are a Review Comment: These two are named in config rather than compiled against, which is what the "Class Names Referenced From Configuration" table above is for. That table pins the name and says the internal structure carries no guarantee. Under the SPI heading they pick up "both source and binary compatibility matter here" instead, which puts their whole public surface under the 1.x contract. Given how much their internals have moved in this review alone, could they go in the table instead? ########## docs/source/user-guide/latest/s3-credential-providers.md: ########## @@ -36,6 +36,39 @@ You probably do, if any of these are true: - You have a custom Iceberg `client.factory` that injects a configured S3 client. - Spark queries against your S3 paths work, but the same queries with Comet enabled fail with 403. +## Built-in adapters + +If a native Parquet scan fails with `Unsupported credential provider: <class>` (for example `com.amazonaws.auth.DefaultAWSCredentialsProviderChain`), the class you named in `fs.s3a.aws.credentials.provider` is one that plain Spark/Hadoop accepts but Comet's native reader does not reimplement. Comet ships two built-in `CometS3CredentialProvider` adapters that fix this with a one-line config change; you leave your existing `fs.s3a.aws.credentials.provider` untouched. + +These adapters cover the Parquet native scan path only. Enabling one is opt-in: naming it is what activates it, and Comet's existing native provider handling is unchanged for everyone else. + +### `HadoopS3ACredentialProviderAdapter` (recommended) + +Delegates to Hadoop S3A's own provider construction, so it accepts everything the `fs.s3a.aws.credentials.provider` chain accepts (the default chain, web-identity, assumed-role, custom signers, per-bucket config). This is the general answer for the failure above. Review Comment: Custom signers should come out of this list. They aren't part of the provider chain. The S3A client reads `fs.s3a.custom.signers` when it signs a request, and the native reader always signs SigV4 itself with whatever the chain returns. So the adapter never sees the signer. Someone who switches to it because of this sentence reads with the chain's credentials, not the identity their signer would have used. The "Do I need this?" section at the top still says a custom signer means you need a vendor bridge, which is right. Could the paragraph below also say that `fs.s3a.delegation.token.binding` is refused? That's the one configuration the adapter now rejects outright, and someone on a DT cluster should find that here rather than in an executor log. ########## spark/src/main/java/org/apache/comet/cloud/s3/AdapterSupport.java: ########## @@ -0,0 +1,208 @@ +/* + * 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.comet.cloud.s3; + +import java.lang.reflect.Constructor; +import java.lang.reflect.Method; +import java.lang.reflect.Modifier; +import java.net.URI; +import java.util.Map; + +import scala.Tuple2; + +import org.apache.hadoop.conf.Configuration; +import org.apache.spark.SparkConf; +import org.apache.spark.SparkEnv; + +/** Config and reflection helpers shared by the built-in S3 credential provider adapters. */ +final class AdapterSupport { + + private AdapterSupport() {} + + /** + * Rebuilds a Hadoop {@link Configuration} for the adapter to hand to the S3A provider factory. + * Seeds from the executor's own Spark-derived Hadoop conf (so keys the provider reads that Comet + * does not forward -- e.g. {@code hadoop.security.credential.provider.path} set via {@code + * spark.hadoop.*}, which is not an {@code fs.s3a.*} key -- are present), then overlays the + * forwarded {@code fs.s3a.*} map on top. Off the executor (e.g. in unit tests) there is no {@code + * SparkEnv}, so it falls back to a bare Configuration that still loads {@code core-site.xml}. + */ + static Configuration toConfiguration(Map<String, String> props) { + Configuration conf = new Configuration(); + SparkEnv env = SparkEnv.get(); + if (env != null) { + SparkConf sparkConf = env.conf(); + // spark.hadoop.<k>=<v> maps to Hadoop conf key <k>, matching SparkHadoopUtil. + for (Tuple2<String, String> kv : sparkConf.getAllWithPrefix("spark.hadoop.")) { + conf.set(kv._1(), kv._2()); + } + } + for (Map.Entry<String, String> entry : props.entrySet()) { + if (entry.getValue() != null) { + conf.set(entry.getKey(), entry.getValue()); + } + } + return conf; + } + + /** + * Per-bucket then global lookup, mirroring Comet's native {@code fs.s3a} resolution: {@code + * fs.s3a.bucket.<bucket>.<property>} wins over {@code fs.s3a.<property>}. Returns null if neither + * is set (after trimming). + */ + static String lookup(Map<String, String> props, String bucket, String property) { + String perBucket = props.get("fs.s3a.bucket." + bucket + "." + property); + if (perBucket != null && !perBucket.trim().isEmpty()) { + return perBucket.trim(); + } + String global = props.get("fs.s3a." + property); + if (global != null && !global.trim().isEmpty()) { + return global.trim(); + } + return null; + } + + /** Returns the public static no-arg method {@code name} on {@code clazz}, or null if absent. */ + private static Method staticMethod(Class<?> clazz, String name) { + try { + Method m = clazz.getMethod(name); + return Modifier.isStatic(m.getModifiers()) ? m : null; + } catch (NoSuchMethodException e) { + return null; + } + } + + /** + * Instantiates a credential-provider delegate, trying the same ordered conventions for both the + * v1 and v2 adapters so their {@code @Public} contract is identical: the Hadoop-style {@code + * (URI, Configuration)} and {@code (Configuration)} constructors first (matching {@code + * S3AUtils.getInstanceFromReflection}), then the SDK static factories {@code create()} / {@code + * builder().build()} / {@code getInstance()}, then a public no-arg constructor. + * + * <p>Factory return types must be assignable to {@code targetType} (as Hadoop's {@code + * getFactoryMethod} requires), so an unrelated {@code static String create()} is skipped rather + * than invoked and failing later with a {@code ClassCastException}. Returns an untyped instance; + * the caller casts to its SDK provider interface. + */ + static Object instantiateDelegate( + Class<?> targetType, Class<?> clazz, URI uri, Configuration conf) throws Exception { + Constructor<?> uriConf = constructor(clazz, URI.class, Configuration.class); + if (uriConf != null) { + return uriConf.newInstance(uri, conf); + } + Constructor<?> confOnly = constructor(clazz, Configuration.class); + if (confOnly != null) { + return confOnly.newInstance(conf); + } + Method create = factoryMethod(clazz, "create", targetType); + if (create != null) { + return create.invoke(null); + } + Method builder = staticMethod(clazz, "builder"); + if (builder != null) { + // build()'s declared return type may be erased to Object (a public Builder that inherits + // build() from a generic SdkBuilder<B, T> without redeclaring it), so check the built + // instance's runtime type rather than the declared return type. Fall through if the builder + // does not yield the target type. + Object built = tryBuild(builder); + if (targetType.isInstance(built)) { + return built; + } + } + Method getInstance = factoryMethod(clazz, "getInstance", targetType); + if (getInstance != null) { + return getInstance.invoke(null); + } + return clazz.getDeclaredConstructor().newInstance(); + } + + /** + * Invokes {@code builder().build()} and returns the built object, or null if there is no public + * no-arg {@code build()} or the builder yields null. Resolves {@code build()} off {@code + * builder()}'s declared (public) return type, not the runtime object's class, which may be a + * non-public implementation. + */ + private static Object tryBuild(Method builder) throws Exception { + Object b = builder.invoke(null); + if (b == null) { + return null; + } + Method build; + try { + build = builder.getReturnType().getMethod("build"); + } catch (NoSuchMethodException e) { + return null; + } + return build.invoke(b); + } + + private static Constructor<?> constructor(Class<?> clazz, Class<?>... params) { + try { + return clazz.getConstructor(params); + } catch (NoSuchMethodException e) { + return null; + } + } + + /** A public static no-arg factory whose return type is assignable to {@code targetType}. */ + private static Method factoryMethod(Class<?> clazz, String name, Class<?> targetType) { + Method m = staticMethod(clazz, name); + return (m != null && targetType.isAssignableFrom(m.getReturnType())) ? m : null; + } + + /** + * Replicates the step {@code S3AFileSystem.initialize} performs before building the provider + * list: promote the S3A credential-store path ({@code fs.s3a.security.credential.provider.path}) + * into Hadoop's generic {@code hadoop.security.credential.provider.path}, so a provider that + * looks up a secret through Hadoop's credential-provider API can see the configured store. The + * factory methods the adapters call do not do this on their own. The S3A path takes precedence + * over any generic path already set. Call after {@code propagateBucketOptions} so per-bucket + * store paths are already promoted to the base key. + */ + static void patchSecurityCredentialProviders(Configuration conf) { + String s3aPath = conf.getTrimmed("fs.s3a.security.credential.provider.path"); + if (s3aPath == null || s3aPath.isEmpty()) { + return; + } + String generic = conf.getTrimmed("hadoop.security.credential.provider.path"); + String merged = (generic == null || generic.isEmpty()) ? s3aPath : s3aPath + "," + generic; + conf.set("hadoop.security.credential.provider.path", merged); + } + + /** + * Fails if S3A delegation tokens are configured. {@code S3AFileSystem.initialize} switches to the + * delegation-token provider and bypasses the configured credential-provider chain; the Hadoop + * adapter always builds the chain, so on a DT cluster it would resolve a different identity than + * Spark. Rather than silently do that, refuse. Call after {@code propagateBucketOptions} so a + * per-bucket binding is already promoted to the base key. + */ + static void checkNoDelegationTokenBinding(Configuration conf) { Review Comment: The guard works. I checked it with both `fs.s3a.delegation.token.binding` and the per-bucket spelling. But nothing tests it, so deleting the call from either `buildDelegate` leaves every test green. Could both `HadoopS3ACredentialProviderAdapterTest` classes get a case that sets the per-bucket spelling and asserts the failure? The per-bucket form also pins that the check runs after `propagateBucketOptions`. -- 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]
