Review fix.

Author: gaborgsomogyi

core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManager.scala

@@ -249,7 +249,7 @@ private[spark] class HadoopDelegationTokenManager(
   private def loadProviders(): Map[String, HadoopDelegationTokenProvider] = {
     val loader = ServiceLoader.load(classOf[HadoopDelegationTokenProvider],
       Utils.getContextOrSparkClassLoader)
-    var providers = mutable.ArrayBuffer[HadoopDelegationTokenProvider]()
+    val providers = mutable.ArrayBuffer[HadoopDelegationTokenProvider]()
 
     val iterator = loader.iterator
     while (iterator.hasNext) {

core/src/test/resources/META-INF/services/org.apache.spark.deploy.security.HadoopDelegationTokenProvider

@@ -0,0 +1 @@
+org.apache.spark.deploy.security.ExceptionThrowingDelegationTokenProvider

core/src/test/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManagerSuite.scala

@@ -18,16 +18,36 @@
 package org.apache.spark.deploy.security
 
 import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.FileSystem
+import org.apache.hadoop.security.Credentials
 
 import org.apache.spark.{SparkConf, SparkFunSuite}
 
+private class ExceptionThrowingDelegationTokenProvider extends HadoopDelegationTokenProvider {
+  throw new IllegalArgumentException
+
+  override def serviceName: String = "throw"
+
+  override def delegationTokensRequired(
+    sparkConf: SparkConf,
+    hadoopConf: Configuration): Boolean = throw new IllegalArgumentException
+
+  override def obtainDelegationTokens(
+    hadoopConf: Configuration,
+    sparkConf: SparkConf,
+    fileSystems: Set[FileSystem],
+    creds: Credentials): Option[Long] = throw new IllegalArgumentException
+}
+
 class HadoopDelegationTokenManagerSuite extends SparkFunSuite {
   private val hadoopConf = new Configuration()
 
   test("default configuration") {
     val manager = new HadoopDelegationTokenManager(new SparkConf(false), hadoopConf, null)
     assert(manager.isProviderLoaded("hadoopfs"))
     assert(manager.isProviderLoaded("hbase"))
+    // This checks that providers are loaded independently and they have no effect on each other
+    assert(!manager.isProviderLoaded("throw"))
   }
 
   test("using deprecated configurations") {

sql/hive/src/test/scala/org/apache/spark/sql/hive/security/HiveHadoopDelegationTokenManagerSuite.scala

@@ -91,6 +91,8 @@ private object NoHiveTest {
   def runTest(): Unit = {
     try {
       val manager = new HadoopDelegationTokenManager(new SparkConf(), new Configuration(), null)
+      assert(manager.isProviderLoaded("hadoopfs"))
+      assert(manager.isProviderLoaded("hbase"))
       require(!manager.isProviderLoaded("hive"))
     } catch {
       case e: Throwable =>