fix(job-server): Fix for issue spark-jobserver#786 - Concurrent access to the jars w… (spark-jobserver#872)

* fix(job-server): Fix for issue spark-jobserver#786 - Concurrent access to the jars when cashing on the file system resulting in ClassNotFoundError

* Modify/extend the FileCacher.cacheBinary(...) method to store the JAR/binary data into a temporary file first, and only after this completes, try to rename it to the expected full original name (with a new timestamp and so on). If the rename succeeds - everything is fine. If the rename fails (ex. there is already a file with the same target full filename), then it means that the other process was faster than us, and we're OK with this (but we have to delete our temporary file, its not needed anymore).

* fix(job-server): Fix for issue spark-jobserver#786 - Concurrent access to the jars when cashing on the file system resulting in ClassNotFoundError

* Modify/extend the FileCacher.cacheBinary(...) method to store the JAR/binary data into a temporary file first, and only after this completes, try to rename it to the expected full original name (with a new timestamp and so on). If the rename succeeds - everything is fine. If the rename fails (ex. there is already a file with the same target full filename), then it means that the other process was faster than us, and we're OK with this (but we have to delete our temporary file, its not needed anymore).

* fix(job-server): Fix for issue spark-jobserver#786 - Concurrent access to the jars when cashing on the file system resulting in ClassNotFoundError

* Modify/extend the FileCacher.cacheBinary(...) method to store the JAR/binary data into a temporary file first, and only after this completes, try to rename it to the expected full original name (with a new timestamp and so on). If the rename succeeds - everything is fine. If the rename fails (ex. there is already a file with the same target full filename), then it means that the other process was faster than us, and we're OK with this (but we have to delete our temporary file, its not needed anymore).

Author: Meehau

job-server/src/main/scala/spark/jobserver/io/FileCacher.scala

@@ -29,19 +29,34 @@ trait FileCacher {
 
   // Cache the jar file into local file system.
   protected def cacheBinary(appName: String,
-                          binaryType: BinaryType,
-                          uploadTime: DateTime,
-                          binBytes: Array[Byte]) {
-    val outFile =
-      new File(rootDir, createBinaryName(appName, binaryType, uploadTime))
-    val bos = new BufferedOutputStream(new FileOutputStream(outFile))
+                            binaryType: BinaryType,
+                            uploadTime: DateTime,
+                            binBytes: Array[Byte]) {
+    val targetFullBinaryName = createBinaryName(appName, binaryType, uploadTime)
+    val tempSuffix = ".tmp"
+    val tempOutFile = File.createTempFile(targetFullBinaryName + "-", tempSuffix, new File(rootDir))
+    val tempOutFileName = tempOutFile.getName
+    val bos = new BufferedOutputStream(new FileOutputStream(tempOutFile))
+
     try {
-      logger.debug("Writing {} bytes to file {}", binBytes.length, outFile.getPath)
+      logger.debug("Writing {} bytes to a temporary file {}", binBytes.length, tempOutFile.getPath)
       bos.write(binBytes)
       bos.flush()
     } finally {
       bos.close()
     }
+
+    logger.debug("Renaming the temporary file {} to the target full binary name {}",
+      tempOutFileName, targetFullBinaryName: Any)
+
+    val tempFile = new File(rootDir, tempOutFileName)
+    if( ! tempFile.renameTo(new File(rootDir, targetFullBinaryName))) {
+      logger.debug("Renaming the temporary file {} failed, another process has probably already updated " +
+        "the target file - deleting the redundant temp file", tempOutFileName)
+      if( ! tempFile.delete()) {
+        logger.warn("Could not delete the temporary file {}", tempOutFileName)
+      }
+    }
   }
 
   protected def cleanCacheBinaries(appName: String): Unit = {