Detect multiple jars on the classpath when init plugin [databricks]

sql-plugin/src/main/scala/com/nvidia/spark/rapids/Plugin.scala

@@ -111,6 +111,31 @@ object RapidsPluginUtils extends Logging {
     }
   }
 
+  private def detectMultipleJar(propName: String, jarName: String, 
+      complainFun: (Boolean, String) => Unit): Unit = {
+    val classloader = ShimLoader.getShimClassLoader()
+    val rapidsJarURLs = classloader.getResources(propName).asScala.toSet
+    lazy val rapidsJars = rapidsJarURLs.map(_.toString.split("!").head).mkString(",")
+    lazy val rapidsJarsVers = rapidsJarURLs.map { 
+      url => scala.io.Source.fromInputStream(url.openStream()).mkString("") 
+    }.mkString(",")
+    complainFun(rapidsJarURLs.size <= 1,
+        s"Multiple $jarName jars found in the classpath: $rapidsJars, please make sure there " +
+        s"is only one $jarName jar in the classpath. If it is impossible to fix the classpath " + 
+        s"you can suppress the error by setting ${RapidsConf.ALLOW_MULTIPLE_JARS.key} to true. " + 
+        s"Version info: \n$rapidsJarsVers")
+  }
+
+  def detectMultipleJars(conf: RapidsConf): Unit = {
+    val complainFun: (Boolean, String) => Unit = conf.allowMultipleJars match {
+      case true => (request: Boolean, msg: String) => if (!request) logWarning(msg)
+      case false => (request: Boolean, msg: String) => require(request, msg)
+    }
+    detectMultipleJar(PLUGIN_PROPS_FILENAME, "rapids4spark", complainFun)
+    detectMultipleJar(JNI_PROPS_FILENAME, "spark-rapids-jni", complainFun)
+    detectMultipleJar(CUDF_PROPS_FILENAME, "cudf", complainFun)
+  }
+
   // This assumes Apache Spark logic, if CSPs are setting defaults differently, we may need
   // to handle.
   def estimateCoresOnExec(conf: SparkConf): Int = {
@@ -309,6 +334,7 @@ class RapidsDriverPlugin extends DriverPlugin with Logging {
     val sparkConf = pluginContext.conf
     RapidsPluginUtils.fixupConfigsOnDriver(sparkConf)
     val conf = new RapidsConf(sparkConf)
+    RapidsPluginUtils.detectMultipleJars(conf)
     RapidsPluginUtils.logPluginMode(conf)
     GpuCoreDumpHandler.driverInit(sc, conf)
 
@@ -352,6 +378,7 @@ class RapidsExecutorPlugin extends ExecutorPlugin with Logging {
       pluginContext: PluginContext,
       extraConf: java.util.Map[String, String]): Unit = {
     try {
+
       if (Cuda.getComputeCapabilityMajor < 6) {
         throw new RuntimeException(s"GPU compute capability ${Cuda.getComputeCapabilityMajor}" +
           " is unsupported, requires 6.0+")
@@ -363,6 +390,9 @@ class RapidsExecutorPlugin extends ExecutorPlugin with Logging {
       val numCores = RapidsPluginUtils.estimateCoresOnExec(sparkConf)
       val conf = new RapidsConf(extraConf.asScala.toMap)
 
+      // Fail if there are multiple plugin jars in the classpath.
+      RapidsPluginUtils.detectMultipleJars(conf)
+
       // Compare if the cudf version mentioned in the classpath is equal to the version which
       // plugin expects. If there is a version mismatch, throw error. This check can be disabled
       // by setting this config spark.rapids.cudfVersionOverride=true

sql-plugin/src/main/scala/com/nvidia/spark/rapids/RapidsConf.scala

@@ -1839,6 +1839,14 @@ object RapidsConf {
     .booleanConf
     .createWithDefault(false)
 
+  val ALLOW_MULTIPLE_JARS = conf("spark.rapids.sql.allowMultipleJars")
+    .internal()
+    .startupOnly()
+    .doc("Allow multiple rapids-4-spark, spark-rapids-jni, and cudf jars on the classpath. " +
+      "Spark will take the first one it finds, so the version may not be expected. ")
+    .booleanConf
+    .createWithDefault(false)
+
   val ALLOW_DISABLE_ENTIRE_PLAN = conf("spark.rapids.allowDisableEntirePlan")
     .internal()
     .doc("The plugin has the ability to detect possibe incompatibility with some specific " +
@@ -2634,6 +2642,8 @@ class RapidsConf(conf: Map[String, String]) extends Logging {
 
   lazy val cudfVersionOverride: Boolean = get(CUDF_VERSION_OVERRIDE)
 
+  lazy val allowMultipleJars: Boolean = get(ALLOW_MULTIPLE_JARS)
+
   lazy val allowDisableEntirePlan: Boolean = get(ALLOW_DISABLE_ENTIRE_PLAN)
 
   lazy val useArrowCopyOptimization: Boolean = get(USE_ARROW_OPT)