Replace Class.forName with ShimLoader.loadClass

api_validation/src/main/scala/com/nvidia/spark/rapids/api/ApiValidation.scala

@@ -101,7 +101,7 @@ object ApiValidation extends Logging {
         }
 
         // TODO: Add error handling if Type is not present
-        val gpuTypes = classToTypeTag(Class.forName(gpu))
+        val gpuTypes = classToTypeTag(ShimLoader.loadClass(gpu))
 
         val sparkToGpuExecMap = Map(
           "org.apache.spark.sql.catalyst.expressions.Expression" ->

shims/spark311cdh/src/main/scala/com/nvidia/spark/rapids/shims/spark311cdh/Spark311CDHShims.scala

@@ -74,7 +74,7 @@ class Spark311CDHShims extends SparkBaseShims {
   override def getGpuColumnarToRowTransition(plan: SparkPlan,
      exportColumnRdd: Boolean): GpuColumnarToRowExecParent = {
     val serName = plan.conf.getConf(StaticSQLConf.SPARK_CACHE_SERIALIZER)
-    val serClass = Class.forName(serName)
+    val serClass = ShimLoader.loadClass(serName)
     if (serClass == classOf[ParquetCachedBatchSerializer]) {
       GpuColumnarToRowTransitionExec(plan)
     } else {

shims/spark320/src/main/scala/com/nvidia/spark/rapids/shims/spark320/Spark320Shims.scala

@@ -778,7 +778,7 @@ class Spark320Shims extends Spark32XShims {
   override def getGpuColumnarToRowTransition(plan: SparkPlan,
       exportColumnRdd: Boolean): GpuColumnarToRowExecParent = {
     val serName = plan.conf.getConf(StaticSQLConf.SPARK_CACHE_SERIALIZER)
-    val serClass = Class.forName(serName)
+    val serClass = ShimLoader.loadClass(serName)
     if (serClass == classOf[ParquetCachedBatchSerializer]) {
       GpuColumnarToRowTransitionExec(plan)
     } else {

sql-plugin/src/main/311db/scala/com/nvidia/spark/rapids/shims/v2/SparkBaseShims.scala

@@ -663,7 +663,7 @@ abstract class SparkBaseShims extends Spark30XShims {
   override def getGpuColumnarToRowTransition(plan: SparkPlan,
      exportColumnRdd: Boolean): GpuColumnarToRowExecParent = {
     val serName = plan.conf.getConf(StaticSQLConf.SPARK_CACHE_SERIALIZER)
-    val serClass = Class.forName(serName)
+    val serClass = ShimLoader.loadClass(serName)
     if (serClass == classOf[com.nvidia.spark.ParquetCachedBatchSerializer]) {
       GpuColumnarToRowTransitionExec(plan)
     } else {

sql-plugin/src/main/311until320-nondb/scala/com/nvidia/spark/rapids/shims/v2/SparkBaseShims.scala

@@ -636,7 +636,7 @@ abstract class SparkBaseShims extends Spark31XShims {
   override def getGpuColumnarToRowTransition(plan: SparkPlan,
      exportColumnRdd: Boolean): GpuColumnarToRowExecParent = {
     val serName = plan.conf.getConf(StaticSQLConf.SPARK_CACHE_SERIALIZER)
-    val serClass = Class.forName(serName)
+    val serClass = ShimLoader.loadClass(serName)
     if (serClass == classOf[com.nvidia.spark.ParquetCachedBatchSerializer]) {
       GpuColumnarToRowTransitionExec(plan)
     } else {

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

@@ -3688,7 +3688,7 @@ case class GpuOverrides() extends Rule[SparkPlan] with Logging {
         // is impacted by forcing operators onto CPU due to other rules that we have
         wrap.runAfterTagRules()
         val optimizer = try {
-          Class.forName(conf.optimizerClassName).newInstance().asInstanceOf[Optimizer]
+          ShimLoader.newInstanceOf[Optimizer](conf.optimizerClassName)
         } catch {
           case e: Exception =>
             throw new RuntimeException(s"Failed to create optimizer ${conf.optimizerClassName}", e)

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

@@ -40,7 +40,7 @@ object HostColumnarToGpu extends Logging {
 
   // use reflection to get access to a private field in a class
   private def getClassFieldAccessible(className: String, fieldName: String) = {
-    val classObj = Class.forName(className)
+    val classObj = ShimLoader.loadClass(className)
     val fields = classObj.getDeclaredFields.toList
     val field = fields.filter( x => {
       x.getName.contains(fieldName)

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

@@ -42,7 +42,7 @@ object PlanUtils {
     val execNameWithoutPackage = getBaseNameFromClass(planClass.getName)
     execNameWithoutPackage == fallbackCpuClass ||
       plan.getClass.getName == fallbackCpuClass ||
-      Try(java.lang.Class.forName(fallbackCpuClass))
+      Try(ShimLoader.loadClass(fallbackCpuClass))
         .map(_.isAssignableFrom(planClass))
         .getOrElse(false)
   }

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

@@ -21,7 +21,7 @@ import java.util.concurrent.ConcurrentLinkedQueue
 import java.util.concurrent.atomic.AtomicInteger
 
 import ai.rapids.cudf.{MemoryBuffer, NvtxColor, NvtxRange}
-import com.nvidia.spark.rapids.RapidsConf
+import com.nvidia.spark.rapids.{RapidsConf, ShimLoader}
 
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.rapids.storage.RapidsStorageUtils
@@ -560,7 +560,7 @@ object RapidsShuffleTransport extends Logging {
   def makeTransport(shuffleServerId: BlockManagerId,
                     rapidsConf: RapidsConf): RapidsShuffleTransport = {
     val transportClass = try {
-      Class.forName(rapidsConf.shuffleTransportClassName)
+      ShimLoader.loadClass(rapidsConf.shuffleTransportClassName)
     } catch {
       case classNotFoundException: ClassNotFoundException =>
         logError(s"Unable to find RapidsShuffleTransport class " +

sql-plugin/src/main/scala/org/apache/spark/sql/hive/rapids/GpuHiveOverrides.scala

@@ -17,19 +17,17 @@
 package org.apache.spark.sql.hive.rapids
 
 import com.nvidia.spark.RapidsUDF
-import com.nvidia.spark.rapids.{ExprChecks, ExprMeta, ExprRule, GpuExpression, GpuOverrides, RepeatingParamCheck, TypeSig}
+import com.nvidia.spark.rapids.{ExprChecks, ExprMeta, ExprRule, GpuExpression, GpuOverrides, RepeatingParamCheck, ShimLoader, TypeSig}
 import com.nvidia.spark.rapids.GpuUserDefinedFunction.udfTypeSig
 
 import org.apache.spark.sql.catalyst.expressions.Expression
 import org.apache.spark.sql.hive.{HiveGenericUDF, HiveSimpleUDF}
 
 object GpuHiveOverrides {
   def isSparkHiveAvailable: Boolean = {
-    // Using the same approach as SparkSession.hiveClassesArePresent
-    val loader = Thread.currentThread().getContextClassLoader
     try {
-      Class.forName("org.apache.spark.sql.hive.HiveSessionStateBuilder", true, loader)
-      Class.forName("org.apache.hadoop.hive.conf.HiveConf", true, loader)
+      ShimLoader.loadClass("org.apache.spark.sql.hive.HiveSessionStateBuilder")
+      ShimLoader.loadClass("org.apache.hadoop.hive.conf.HiveConf")
       true
     } catch {
       case _: ClassNotFoundException | _: NoClassDefFoundError => false

udf-compiler/src/main/scala/com/nvidia/spark/udf/LambdaReflection.scala

@@ -16,11 +16,11 @@
 
 package com.nvidia.spark.udf
 
-import Repr.UnknownCapturedArg
 import java.lang.invoke.SerializedLambda
+
+import com.nvidia.spark.rapids.ShimLoader
 import javassist.{ClassClassPath, ClassPool, CtBehavior, CtClass, CtField, CtMethod}
-import javassist.bytecode.{AccessFlag, CodeIterator, ConstPool,
-                           Descriptor, MethodParametersAttribute}
+import javassist.bytecode.{CodeIterator, ConstPool, Descriptor}
 
 import org.apache.spark.SparkException
 import org.apache.spark.sql.catalyst.expressions.Expression
@@ -152,9 +152,7 @@ object LambdaReflection {
   }
 
   def getClass(name: String): Class[_] = {
-    // scalastyle:off classforname
-    Class.forName(name, true, Thread.currentThread().getContextClassLoader)
-    // scalastyle:on classforname
+    ShimLoader.loadClass(name)
   }
 
   def parseTypeSig(sig: String): Option[DataType] = {