Fix Mocked FS parse non-default container root path issue

Also introduced Windows file status class with fixing reference Class
RawLocalFileSystem method loadPermissionInfo() issue.

Signed-off-by: Wei Zhang <[email protected]>

Author: wezhang

Utils/hdinsight-node-common/Test/java/com/microsoft/azure/hdinsight/spark/mock/SparkLocalRunnerITScenario.kt

@@ -23,6 +23,7 @@
 package com.microsoft.azure.hdinsight.spark.mock
 
 import com.microsoft.azure.hdinsight.spark.common.SparkLocalJvmProcess
+import com.microsoft.azure.hdinsight.spark.mock.jobapp.CatCmd
 import cucumber.api.java.en.And
 import cucumber.api.java.en.Given
 import cucumber.api.java.en.Then
@@ -46,14 +47,10 @@ class SparkLocalRunnerITScenario {
         sparkLocalJob = jvmProcess.createProcess("", SparkLocalRunner::class.java, args)
     }
 
-    private fun runToGetStdoutLines(): List<String> {
-        assertThat(sparkLocalJob)
-                .describedAs("Run Spark Job locally firstly")
-                .isNotNull()
+    private fun runToGetStdoutLines(job: ProcessBuilder): List<String> {
+        job.redirectOutput(ProcessBuilder.Redirect.PIPE)
 
-        sparkLocalJob!!.redirectOutput(ProcessBuilder.Redirect.PIPE)
-
-        val process = sparkLocalJob!!.start()
+        val process = job.start()
 
         assertThat(process.waitFor())
                 .describedAs("Spark job exist with error.")
@@ -66,15 +63,15 @@ class SparkLocalRunnerITScenario {
 
     @Then("^locally run stand output should be")
     fun checkLocallyRunStdout(expectOutputs: List<String>) {
-        val outputLines = runToGetStdoutLines()
+        val outputLines = runToGetStdoutLines(sparkLocalJob!!)
 
         assertThat(outputLines).containsAll(expectOutputs)
         assertThat(outputLines).hasSameSizeAs(expectOutputs)
     }
 
     @Then("^locally run stand output table should be")
     fun checkLocallyRunStdoutTable(expectOutputs: DataTable) {
-        val outputTable = runToGetStdoutLines()
+        val outputTable = runToGetStdoutLines(sparkLocalJob!!)
                 .filter { !it.matches("""^\+[-+]*\+$""".toRegex()) }
                 .map {
                     it.split("|")
@@ -86,4 +83,20 @@ class SparkLocalRunnerITScenario {
 
         assertThat(outputTable).isEqualTo(expectOutputs.raw())
     }
+
+    @Then("^locally cat '(.*)' should be")
+    fun checkLocallyRunFile(fileUri: String, expectOutputs: List<String>) {
+        runToGetStdoutLines(sparkLocalJob!!)
+
+        // Start another job to read the file
+
+        val args = arrayOf("--master local[1]", CatCmd::class.java.canonicalName, fileUri)
+
+        val catJob = jvmProcess.createProcess("", SparkLocalRunner::class.java, args)
+
+        val outputLines = runToGetStdoutLines(catJob!!)
+
+        assertThat(outputLines).containsAll(expectOutputs)
+        assertThat(outputLines).hasSameSizeAs(expectOutputs)
+    }
 }

Utils/hdinsight-node-common/Test/resources/com/microsoft/azure/hdinsight/spark/mock/MockRawLocalFileSystemScenario.feature

@@ -18,3 +18,7 @@ Feature: Mock File System Unit Test
   Scenario: WASB container blob path
     Given set mocked file system local working directory to '/data/__default__/user/current'
     Then convert mocked file system path 'wasb://account@blob1/abc/def' to File should be '/data/account@blob1/abc/def'
+
+  Scenario: WASB container blob root path
+    Given set mocked file system local working directory to '/data/__default__/user/current'
+    Then convert mocked file system path 'wasb://account@blob1/' to File should be '/data/account@blob1/'

Utils/hdinsight-node-common/Test/resources/com/microsoft/azure/hdinsight/spark/mock/SparkLocalRunnerScenario.feature

@@ -38,6 +38,20 @@ Feature: Spark Local Runner Integration Test
       | local,2  |
       | world,1  |
 
+  Scenario: Mocked blob container file write
+    Given locally run job 'com.microsoft.azure.hdinsight.spark.mock.jobapp.WordCountTest' with args
+      | /word_count_input.txt | wasb://container1/wc_result |
+    Then locally cat 'wasb://container1/wc_result' should be
+      | (a,1)      |
+      | (mocked,1) |
+      | (fs,1)     |
+      | (with,1)   |
+      | (Spark,1)  |
+      | (run,1)    |
+      | (Hello,1)  |
+      | (local,2)  |
+      | (world,1)  |
+
   Scenario: SparkSQL test
     Given locally run job 'com.microsoft.azure.hdinsight.spark.mock.jobapp.SparkSQLTest' with args
       | wasb:///people.json | people | SELECT * FROM people |

Utils/hdinsight-node-common/Test/scala/com/microsoft/azure/hdinsight/spark/mock/jobapp/CatCmd.scala

@@ -0,0 +1,40 @@
+/*
+ * Copyright (c) Microsoft Corporation
+ * <p/>
+ * All rights reserved.
+ * <p/>
+ * MIT License
+ * <p/>
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
+ * documentation files (the "Software"), to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and
+ * to permit persons to whom the Software is furnished to do so, subject to the following conditions:
+ * <p/>
+ * The above copyright notice and this permission notice shall be included in all copies or substantial portions of
+ * the Software.
+ * <p/>
+ * THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO
+ * THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
+ * TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package com.microsoft.azure.hdinsight.spark.mock.jobapp
+
+import org.apache.spark.{SparkConf, SparkContext}
+
+object CatCmd {
+  def main(args: Array[String]) {
+    val inputFile = args(0)
+
+    val conf = new SparkConf().setAppName("DFS cat command")
+    // Create a Scala Spark Context.
+    val sc = new SparkContext(conf)
+    // Load our input data.
+    val input = sc.textFile(inputFile)
+    // output line by line
+    input
+      .foreach(l => println(l))
+  }
+}

Utils/hdinsight-node-common/Test/scala/com/microsoft/azure/hdinsight/spark/mock/jobapp/WordCountTest.scala

@@ -23,7 +23,6 @@
 package com.microsoft.azure.hdinsight.spark.mock.jobapp
 
 import org.apache.spark._
-import org.apache.spark.SparkContext._
 
 object WordCountTest {
   def main(args: Array[String]) {
@@ -41,7 +40,12 @@ object WordCountTest {
 
     outputFileOption match {
       // Save the word count back out to a text file, causing evaluation.
-      case Some(outputFile) => counts.saveAsTextFile(outputFile)
+      case Some(outputFile) => {
+        val dfs = org.apache.hadoop.fs.FileSystem.get(new java.net.URI(outputFile), sc.hadoopConfiguration)
+        dfs.delete(new org.apache.hadoop.fs.Path(outputFile), true)
+
+        counts.saveAsTextFile(outputFile)
+      }
       // just output to stdout
       case None =>
         counts.collect().foreach { case (word, count) => println(s"$word,$count") }