[CARMEL-6030][FollowUp] Fix OutOfMemoryError-original fix may cause task thread stuck (#986)

Author: allenma

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileScanRDD.scala

@@ -86,6 +86,7 @@ class FileScanRDD(
   protected val fileScanMaxThreadNum = sparkSession.sessionState.conf.fileScanMaxThreadNum
   protected val maxInConsumingObjNum = sparkSession.sessionState.conf.parquetColumnBatchNum
   @volatile protected var usingMultiThread = false
+  @transient protected var fileScanExecutor: Option[ThreadPoolExecutor] = None
 
   protected val indexEnabled = sparkSession.sessionState.conf.indexEnabled
 
@@ -373,20 +374,20 @@ class FileScanRDD(
     if (fileNum <= maxThreadNum) {
       workersNum = fileNum
     }
-
-    private val executor = getExecutor
+    if (fileScanExecutor == null || fileScanExecutor.isEmpty) {
+      fileScanExecutor = Some(createExecutor)
+    }
+    private val executor = fileScanExecutor.get
     private val notCompleteFiles = new AtomicInteger(fileNum)
 
     @volatile private var scanException : Exception = null
     import scala.collection.JavaConverters._
     private val workingInputStatsMap = new ConcurrentHashMap[String, (Long, Long)].asScala
     private val completedInputStats = (new LongAccumulator, new LongAccumulator)
 
-    private def getExecutor: ThreadPoolExecutor = {
-      context.getOrCreateObjFromEnv("fileScanExecutor",
-        {ThreadUtils.newDaemonFixedThreadPool(workersNum,
-          s"FileScanForPartition ${context.partitionId()}")})
-        .asInstanceOf[ThreadPoolExecutor]
+    private def createExecutor: ThreadPoolExecutor = {
+      ThreadUtils.newDaemonFixedThreadPool(workersNum,
+          s"FileScanForPartition ${context.partitionId()}")
     }
 
     // start to submit scan task