Add failing regression test for SPARK-8062

Author: JoshRosen

core/src/test/scala/org/apache/spark/metrics/InputOutputMetricsSuite.scala

@@ -106,4 +106,28 @@ class InputOutputMetricsSuite extends FunSuite with SharedSparkContext with Shou
       }
     }
   }
+
+  test("exceptions while getting IO thread statistics should not fail tasks / jobs (SPARK-8062)") {
+    // For some reason, the following code needs to be called in order for this regression test to
+    // fail and reproduce the bug. The fact that this is necessary suggests that there may be other
+    // bugs in our InputOutputMetrics code; SPARK-8086 tracks progress towards investigating this
+    // issue, since fixing it is out of scope for SPARK-8062.
+    val fs = FileSystem.getLocal(new Configuration())
+    val outPath = new Path(fs.getWorkingDirectory, "outdir")
+    SparkHadoopUtil.get.getFSBytesWrittenOnThreadCallback(outPath, fs.getConf)
+
+    // Intentionally call this method with a null scheme, which will store an entry for a FileSystem
+    // with a null scheme into Hadoop's global `FileSystem.statisticsTable`.
+    FileSystem.getStatistics(null, classOf[FileSystem])
+
+    // Prior to fixing SPARK-8062, this would fail with a NullPointerException in
+    // SparkHadoopUtil.getFileSystemThreadStatistics
+    val rdd = sc.parallelize(Array("a", "b", "c", "d"), 2)
+    try {
+      rdd.saveAsTextFile(outPath.toString)
+      sc.textFile(outPath.toString).count()
+    } finally {
+      fs.delete(outPath, true)
+    }
+  }
 }