[SC-5631] Fix scala 2.10 build and use thread-pool util

## What changes were proposed in this pull request?

Fix scala 2.10 build and use thread-pool util

## How was this patch tested?

Built manually.

Author: Eric Liang <ekl@databricks.com>

Closes apache#176 from ericl/fix-build-2.

Author: ericl

sql/core/src/main/scala/com/databricks/sql/transaction/DatabricksAtomicCommitProtocol.scala

@@ -23,6 +23,7 @@ import org.apache.spark.SparkEnv
 import org.apache.spark.internal.Logging
 import org.apache.spark.internal.io.FileCommitProtocol
 import org.apache.spark.sql.SparkSession
+import org.apache.spark.util.ThreadUtils
 
 /**
  * File commit protocol optimized for cloud storage. Files are written directly to their final
@@ -159,24 +160,9 @@ object DatabricksAtomicCommitProtocol extends Logging {
   private val sparkSession = SparkSession.builder.getOrCreate()
 
   import scala.collection.parallel.ThreadPoolTaskSupport
-  import java.util.concurrent.{LinkedBlockingQueue, ThreadFactory, ThreadPoolExecutor, TimeUnit}
-
-  private lazy val tasksupport = new ThreadPoolTaskSupport({
-    val pool = new ThreadPoolExecutor(
-      100,
-      100,
-      100L,
-      TimeUnit.SECONDS,
-      new LinkedBlockingQueue[Runnable])
-    pool.setThreadFactory(new ThreadFactory {
-      override def newThread(task: Runnable): Thread = {
-        val thread = new Thread(task, "DatabricksAtomicCommitProtocolWorker")
-        thread.setDaemon(true)
-        thread
-      }
-    })
-    pool
-  })
+
+  private lazy val tasksupport = new ThreadPoolTaskSupport(
+    ThreadUtils.newDaemonCachedThreadPool("db-atomic-commit-worker", 100))
 
   /**
    * Traverses the given directories and cleans up uncommitted or garbage files and markers. A

sql/core/src/main/scala/com/databricks/sql/transaction/DatabricksAtomicReadProtocol.scala

@@ -8,7 +8,7 @@
 
 package org.apache.spark.sql.transaction
 
-import java.io._
+import java.io.{File, InputStream, InputStreamReader, IOException, OutputStream}
 import java.nio.charset.StandardCharsets
 
 import scala.collection.mutable

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

@@ -402,7 +402,8 @@ object PartitioningAwareFileIndex extends Logging {
       }
 
       val allLeafStatuses = {
-        val (dirs, topLevelFiles) = statuses.partition(_.isDirectory)
+        val (dirs: Array[FileStatus], topLevelFiles: Array[FileStatus]) =
+          statuses.partition(_.isDirectory)
         val nestedFiles: Seq[FileStatus] = sessionOpt match {
           case Some(session) =>
             bulkListLeafFiles(dirs.map(_.getPath), hadoopConf, filter, session).flatMap(_._2)

sql/core/src/test/scala/com/databricks/sql/acl/InMemoryAclClient.scala

@@ -30,7 +30,7 @@ class InMemoryAclClient(var underlyingPrincipal: Principal) extends AclClient {
       principal: Option[Principal] = None,
       securable: Securable): Seq[Permission] = {
     allPermissions.filter { perm =>
-      (principal.isEmpty || principal.contains(perm.principal)) && (securable == perm.securable)
+      (principal.isEmpty || principal == Some(perm.principal)) && (securable == perm.securable)
     }.toSeq
   }