fix: Validating object store configs should not throw exception (#2308)

Author: andygrove

spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala

@@ -434,20 +434,18 @@ object CometScanRule extends Logging {
     configValidityMap.get(cacheKey) match {
       case Some(Some(reason)) =>
         fallbackReasons += reason
-        throw new CometNativeException(reason)
       case Some(None) =>
       // previously validated
       case _ =>
         try {
           val objectStoreOptions = JavaConverters.mapAsJavaMap(objectStoreConfigMap)
           Native.validateObjectStoreConfig(filePath, objectStoreOptions)
         } catch {
-          case e: Exception =>
+          case e: CometNativeException =>
             val reason = "Object store config not supported by " +
               s"$SCAN_NATIVE_ICEBERG_COMPAT: ${e.getMessage}"
             fallbackReasons += reason
             configValidityMap.put(cacheKey, Some(reason))
-            throw e
         }
     }
 

spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala

@@ -22,15 +22,13 @@ package org.apache.comet.objectstore
 import java.net.URI
 
 import scala.collection.mutable
-import scala.util.Try
 
 import org.scalatest.BeforeAndAfterEach
 import org.scalatest.funsuite.AnyFunSuite
 import org.scalatest.matchers.should.Matchers
 
 import org.apache.hadoop.conf.Configuration
 
-import org.apache.comet.CometNativeException
 import org.apache.comet.rules.CometScanRule
 
 class NativeConfigSuite extends AnyFunSuite with Matchers with BeforeAndAfterEach {
@@ -98,23 +96,20 @@ class NativeConfigSuite extends AnyFunSuite with Matchers with BeforeAndAfterEac
   test("validate object store config - invalid provider") {
     val hadoopConf = new Configuration()
     hadoopConf.set("fs.s3a.aws.credentials.provider", "invalid")
-    val e = intercept[CometNativeException] {
-      validate(hadoopConf)
-    }
-    assert(e.getMessage.contains("Unsupported credential provider: invalid"))
+    val fallbackReasons = validate(hadoopConf)
+    val expectedError = "Unsupported credential provider: invalid"
+    assert(fallbackReasons.exists(_.contains(expectedError)))
   }
 
   test("validate object store config - mixed anonymous providers") {
     val hadoopConf = new Configuration()
     val provider1 = "com.amazonaws.auth.AnonymousAWSCredentials"
     val provider2 = "software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider"
     hadoopConf.set("fs.s3a.aws.credentials.provider", Seq(provider1, provider2).mkString(","))
-    val e = intercept[CometNativeException] {
-      validate(hadoopConf)
-    }
+    val fallbackReasons = validate(hadoopConf)
     val expectedError =
       "Anonymous credential provider cannot be mixed with other credential providers"
-    assert(e.getMessage.contains(expectedError))
+    assert(fallbackReasons.exists(_.contains(expectedError)))
   }
 
   test("validity cache") {
@@ -125,19 +120,20 @@ class NativeConfigSuite extends AnyFunSuite with Matchers with BeforeAndAfterEac
 
     assert(CometScanRule.configValidityMap.isEmpty)
     for (_ <- 0 until 5) {
-      assert(Try(validate(hadoopConf)).isFailure)
+      assert(validate(hadoopConf).nonEmpty)
       assert(CometScanRule.configValidityMap.size == 1)
     }
 
     // set the same providers but in a different order
     hadoopConf.set("fs.s3a.aws.credentials.provider", Seq(provider2, provider1).mkString(","))
-    assert(Try(validate(hadoopConf)).isFailure)
+    assert(validate(hadoopConf).nonEmpty)
     assert(CometScanRule.configValidityMap.size == 2)
   }
 
-  private def validate(hadoopConf: Configuration): Unit = {
+  private def validate(hadoopConf: Configuration): Set[String] = {
     val path = "s3a://path/to/file.parquet"
     val fallbackReasons = mutable.ListBuffer[String]()
     CometScanRule.validateObjectStoreConfig(path, hadoopConf, fallbackReasons)
+    fallbackReasons.toSet
   }
 }