Addressed comments

Author: tdas

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/CheckpointFileManager.scala

@@ -237,7 +237,12 @@ class FileSystemBasedCheckpointFileManager(path: Path, hadoopConf: Configuration
   }
 
   override def exists(path: Path): Boolean = {
-    fs.exists(path)
+    try
+      return fs.getFileStatus(path) != null
+    catch {
+      case e: FileNotFoundException =>
+        return false
+    }
   }
 
   override def renameTempFile(srcPath: Path, dstPath: Path, overwriteIfPossible: Boolean): Unit = {
@@ -247,17 +252,14 @@ class FileSystemBasedCheckpointFileManager(path: Path, hadoopConf: Configuration
     }
 
     if (!fs.rename(srcPath, dstPath)) {
-      // If overwriteIfPossible = false, then we want to find out why the rename failed and
-      // try to throw the right error.
+      // FileSystem.rename() returning false is very ambiguous as it can be for many reasons.
+      // This tries to make a best effort attempt to return the most appropriate exception.
       if (fs.exists(dstPath)) {
-        // Some implementations of FileSystem may only return false instead of throwing
-        // FileAlreadyExistsException. In that case, explicitly throw the error the error
-        // if overwriteIfPossible = false. Note that this is definitely not atomic.
-        // This is only a best-effort attempt to identify the situation when rename returned
-        // false.
         if (!overwriteIfPossible) {
-          throw new FileAlreadyExistsException(s"$dstPath already exists")
+          throw new FileAlreadyExistsException(s"Failed to rename as $dstPath already exists")
         }
+      } else if (!fs.exists(srcPath)) {
+        throw new FileNotFoundException(s"Failed to rename as $srcPath was not found")
       } else {
         val msg = s"Failed to rename temp file $srcPath to $dstPath as rename returned false"
         logWarning(msg)

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLog.scala

@@ -137,22 +137,6 @@ class HDFSMetadataLog[T <: AnyRef : ClassTag](sparkSession: SparkSession, path:
     }
   }
 
-  /**
-   * @return the deserialized metadata in a batch file, or None if file not exist.
-   * @throws IllegalArgumentException when path does not point to a batch file.
-   */
-  def get(batchFile: Path): Option[T] = {
-    if (fileManager.exists(batchFile)) {
-      if (isBatchFile(batchFile)) {
-        get(pathToBatchId(batchFile))
-      } else {
-        throw new IllegalArgumentException(s"File ${batchFile} is not a batch file!")
-      }
-    } else {
-      None
-    }
-  }
-
   override def get(batchId: Long): Option[T] = {
     val batchMetadataFile = batchIdToPath(batchId)
     if (fileManager.exists(batchMetadataFile)) {

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/CheckpointFileManagerSuite.scala

@@ -34,7 +34,7 @@ abstract class CheckpointFileManagerTests extends SparkFunSuite {
 
   def createManager(path: Path): CheckpointFileManager
 
-  test("mkdirs, list, createAtomic, open, delete") {
+  test("mkdirs, list, createAtomic, open, delete, exists") {
     withTempPath { p =>
       val basePath = new Path(p.getAbsolutePath)
       val fm = createManager(basePath)