set fs.hdfs.impl.disable.cache true fix #80

src/main/scala/org/apache/spark/KyuubiSparkUtil.scala

@@ -69,6 +69,8 @@ object KyuubiSparkUtil extends Logging {
   val USE_DB: Regex = """use:([^=]+)""".r
   val QUEUE = SPARK_PREFIX + YARN_PREFIX + "queue"
   val DEPRECATED_QUEUE = "mapred.job.queue.name"
+  val HDFS_CLIENT_CACHE = SPARK_HADOOP_PREFIX + "fs.hdfs.impl.disable.cache"
+  val HDFS_CLIENT_CACHE_DEFAULT = "true"
 
   // Runtime Spark Version
   val SPARK_VERSION = org.apache.spark.SPARK_VERSION

src/main/scala/yaooqinn/kyuubi/server/KyuubiServer.scala

@@ -125,6 +125,11 @@ object KyuubiServer extends Logging {
 
     conf.setIfMissing(
       KyuubiSparkUtil.SPARK_LOCAL_DIR, conf.get(KyuubiConf.BACKEND_SESSION_LOCAL_DIR.key))
+
+    if (UserGroupInformation.isSecurityEnabled) {
+      conf.setIfMissing(KyuubiSparkUtil.HDFS_CLIENT_CACHE,
+        KyuubiSparkUtil.HDFS_CLIENT_CACHE_DEFAULT)
+    }
   }
 
   private[kyuubi] def validate(): Unit = {

src/test/scala/org/apache/spark/KyuubiSparkUtilSuite.scala

@@ -191,4 +191,9 @@ class KyuubiSparkUtilSuite extends SparkFunSuite with Logging {
     KyuubiSparkUtil.addShutdownHook(f)
     assert(y === 0)
   }
+
+  test("testHDFS_CLIENT_CACHE") {
+    assert(KyuubiSparkUtil.HDFS_CLIENT_CACHE === "spark.hadoop.fs.hdfs.impl.disable.cache")
+    assert(KyuubiSparkUtil.HDFS_CLIENT_CACHE_DEFAULT.toBoolean)
+  }
 }