Throw HyperspaceException with path for JSON parsing error (#472)

src/main/scala/com/microsoft/hyperspace/index/IndexLogManager.scala

@@ -24,6 +24,7 @@ import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.{FileStatus, FileSystem, FileUtil, Path}
 import org.apache.spark.internal.Logging
 
+import com.microsoft.hyperspace.HyperspaceException
 import com.microsoft.hyperspace.actions.Constants
 import com.microsoft.hyperspace.util.{FileUtils, JsonUtils}
 
@@ -71,8 +72,16 @@ class IndexLogManagerImpl(indexPath: Path, hadoopConfiguration: Configuration =
     if (!fs.exists(path)) {
       return None
     }
+
     val contents = FileUtils.readContents(fs, path)
-    Some(LogEntry.fromJson(contents))
+
+    try {
+      Some(LogEntry.fromJson(contents))
+    } catch {
+      case e: Exception =>
+        throw HyperspaceException(
+          s"Cannot parse JSON in ${path}: ${e.getMessage}")
+    }
   }
 
   override def getLog(id: Int): Option[LogEntry] = {

src/test/scala/com/microsoft/hyperspace/index/IndexLogManagerImplTest.scala

@@ -22,6 +22,7 @@ import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.Path
 import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
 
+import com.microsoft.hyperspace.HyperspaceException
 import com.microsoft.hyperspace.TestUtils
 import com.microsoft.hyperspace.index.IndexConstants.HYPERSPACE_LOG
 import com.microsoft.hyperspace.util.{FileUtils, JsonUtils}
@@ -104,6 +105,23 @@ class IndexLogManagerImplTest extends HyperspaceSuite {
     assert(actual.equals(expected))
   }
 
+  test("testGetLog fails with Exception if json is not in proper form") {
+    val path = new Path(testRoot, "testPath")
+
+    // find position to insert \0
+    val jsonContent = JsonUtils.toJson(sampleIndexLogEntry)
+    val sourceIndex = jsonContent.indexOf("\"source\"")
+    val damagedJsonContent = jsonContent.substring(0, sourceIndex + 8) + "\0" + jsonContent
+      .substring(sourceIndex + 8);
+
+    FileUtils.createFile(
+      path.getFileSystem(new Configuration),
+      new Path(path, s"$HYPERSPACE_LOG/0"),
+      damagedJsonContent)
+
+    assertThrows[HyperspaceException](new IndexLogManagerImpl(path).getLog(0).get)
+  }
+
   test("testGetLog for path") {}
 
   test("testWriteNextLog") {}
@@ -223,8 +241,7 @@ class IndexLogManagerImplTest extends HyperspaceSuite {
     val path = new Path(testRoot, UUID.randomUUID().toString)
     val fs = path.getFileSystem(new Configuration)
     FileUtils.createFile(fs, new Path(path, s"$HYPERSPACE_LOG/0"), "Invalid Log Entry")
-    assertThrows[com.fasterxml.jackson.core.JsonParseException](
+    assertThrows[HyperspaceException](new IndexLogManagerImpl(path).createLatestStableLog(0))
-      new IndexLogManagerImpl(path).createLatestStableLog(0))
   }
 
   // TODO: Test the case where the id does not exist.