From 067fbaa5b5b56dd68700f805f192457f0a0d6120 Mon Sep 17 00:00:00 2001
From: tedyu <yuzhihong@gmail.com>
Date: Sun, 12 Aug 2018 09:29:37 -0700
Subject: [PATCH] HBASE-21038 SAXParseException when
 hbase.spark.use.hbasecontext=false (Ajith S)

---
 .../hadoop/hbase/spark/DefaultSource.scala    |  4 ++--
 .../hbase/spark/DefaultSourceSuite.scala      | 23 +++++++++++++++++++
 2 files changed, 25 insertions(+), 2 deletions(-)

diff --git a/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/DefaultSource.scala b/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/DefaultSource.scala
index 08df63546cf..9e139540e5b 100644
--- a/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/DefaultSource.scala
+++ b/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/DefaultSource.scala
@@ -105,7 +105,7 @@ case class HBaseRelation (
 
   val catalog = HBaseTableCatalog(parameters)
   def tableName = catalog.name
-  val configResources = parameters.getOrElse(HBaseSparkConf.HBASE_CONFIG_LOCATION, "")
+  val configResources = parameters.get(HBaseSparkConf.HBASE_CONFIG_LOCATION)
   val useHBaseContext =  parameters.get(HBaseSparkConf.USE_HBASECONTEXT).map(_.toBoolean).getOrElse(HBaseSparkConf.DEFAULT_USE_HBASECONTEXT)
   val usePushDownColumnFilter = parameters.get(HBaseSparkConf.PUSHDOWN_COLUMN_FILTER)
     .map(_.toBoolean).getOrElse(HBaseSparkConf.DEFAULT_PUSHDOWN_COLUMN_FILTER)
@@ -132,7 +132,7 @@ case class HBaseRelation (
     LatestHBaseContextCache.latest
   } else {
     val config = HBaseConfiguration.create()
-    configResources.split(",").foreach( r => config.addResource(r))
+    configResources.map(resource => resource.split(",").foreach(r => config.addResource(r)))
     new HBaseContext(sqlContext.sparkContext, config)
   }
 
diff --git a/hbase-spark/src/test/scala/org/apache/hadoop/hbase/spark/DefaultSourceSuite.scala b/hbase-spark/src/test/scala/org/apache/hadoop/hbase/spark/DefaultSourceSuite.scala
index 6298ed06d9f..89b68f4cde2 100644
--- a/hbase-spark/src/test/scala/org/apache/hadoop/hbase/spark/DefaultSourceSuite.scala
+++ b/hbase-spark/src/test/scala/org/apache/hadoop/hbase/spark/DefaultSourceSuite.scala
@@ -28,6 +28,7 @@ import org.apache.spark.sql.functions._
 import org.apache.spark.sql.{DataFrame, SQLContext}
 import org.apache.spark.{SparkConf, SparkContext}
 import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach, FunSuite}
+import org.xml.sax.SAXParseException
 
 case class HBaseRecord(
   col0: String,
@@ -1038,4 +1039,26 @@ BeforeAndAfterEach with BeforeAndAfterAll with Logging {
     s.show()
     assert(s.count() == 7)
   }
+
+  test("test create HBaseRelation with new context throws SAXParseException") {
+    val catalog = s"""{
+                     |"table":{"namespace":"default", "name":"t1NotThere"},
+                     |"rowkey":"key",
+                     |"columns":{
+                     |"KEY_FIELD":{"cf":"rowkey", "col":"key", "type":"string"},
+                     |"A_FIELD":{"cf":"c", "col":"a", "type":"string"},
+                     |"B_FIELD":{"cf":"c", "col":"c", "type":"string"}
+                     |}
+                     |}""".stripMargin
+    try {
+      HBaseRelation(Map(HBaseTableCatalog.tableCatalog -> catalog,
+        HBaseSparkConf.USE_HBASECONTEXT -> "false"), None)(sqlContext)
+    } catch {
+        case e: Throwable => if(e.getCause.isInstanceOf[SAXParseException]) {
+          fail("SAXParseException due to configuration loading empty resource")
+        } else {
+          println("Failed due to some other exception, ignore " + e.getMessage)
+        }
+    }
+  }
 }