More changes based on PR comments.

Author: tdas

streaming/src/main/scala/org/apache/spark/streaming/rdd/WriteAheadLogBackedBlockRDD.scala

@@ -103,6 +103,11 @@ class WriteAheadLogBackedBlockRDD[T: ClassTag](
         var dataRead: ByteBuffer = null
         var writeAheadLog: WriteAheadLog = null
         try {
+          // The WriteAheadLogUtils.createLog*** method needs a directory to create a
+          // WriteAheadLog object as the default FileBasedWriteAheadLog needs a directory for
+          // writing log data. However, the directory is not needed if data needs to be read, hence
+          // a dummy path is provided to satisfy the method parameter requirements.
+          // FileBasedWriteAheadLog will not create any file or directory at that path. 
           val dummyDirectory = FileUtils.getTempDirectoryPath()
           writeAheadLog = WriteAheadLogUtils.createLogForReceiver(
             SparkEnv.get.conf, dummyDirectory, hadoopConf)
@@ -114,6 +119,7 @@ class WriteAheadLogBackedBlockRDD[T: ClassTag](
         } finally {
           if (writeAheadLog != null) {
             writeAheadLog.close()
+            writeAheadLog = null
           }
         }
         if (dataRead == null) {

streaming/src/main/scala/org/apache/spark/streaming/util/FileBasedWriteAheadLogRandomReader.scala

@@ -24,7 +24,7 @@ import org.apache.hadoop.conf.Configuration
 /**
  * A random access reader for reading write ahead log files written using
  * [[org.apache.spark.streaming.util.FileBasedWriteAheadLogWriter]]. Given the file segment info,
- * this reads the record (bytebuffer) from the log file.
+ * this reads the record (ByteBuffer) from the log file.
  */
 private[streaming] class FileBasedWriteAheadLogRandomReader(path: String, conf: Configuration)
   extends Closeable {

streaming/src/test/scala/org/apache/spark/streaming/util/WriteAheadLogSuite.scala

@@ -288,6 +288,19 @@ class WriteAheadLogSuite extends FunSuite with BeforeAndAfter {
     val readData = readDataUsingWriteAheadLog(testDir)
     assert(readData === dataToWrite2)
   }
+
+  test("FileBasedWriteAheadLog - do not create directories or files unless write") {
+    val nonexistentTempPath = File.createTempFile("test", "")
+    nonexistentTempPath.delete()
+    assert(!nonexistentTempPath.exists())
+
+    val writtenSegment = writeDataManually(generateRandomData(), testFile)
+    val wal = new FileBasedWriteAheadLog(
+      new SparkConf(), tempDir.getAbsolutePath, new Configuration(), 1, 1)
+    assert(!nonexistentTempPath.exists(), "Directory created just by creating log object")
+    wal.read(writtenSegment.head)
+    assert(!nonexistentTempPath.exists(), "Directory created just by attempting to read segment")
+  }
 }
 
 object WriteAheadLogSuite {