Use url.hashCode + timestamp as cachedFileName

Author: li-zhihui

core/src/main/scala/org/apache/spark/SparkContext.scala

@@ -809,7 +809,7 @@ class SparkContext(config: SparkConf) extends Logging {
     addedFiles(key) = timestamp
 
     // Fetch the file locally in case a job is executed using DAGScheduler.runLocally().
-    Utils.fetchCachedFile(path, new File(SparkFiles.getRootDirectory()), conf, env.securityManager,
+    Utils.fetchFile(path, new File(SparkFiles.getRootDirectory()), conf, env.securityManager,
       hadoopConfiguration, timestamp, useCache = false)
 
     logInfo("Added file " + path + " at " + key + " with timestamp " + addedFiles(key))

core/src/main/scala/org/apache/spark/executor/Executor.scala

@@ -322,13 +322,13 @@ private[spark] class Executor(
       // Fetch missing dependencies
       for ((name, timestamp) <- newFiles if currentFiles.getOrElse(name, -1L) < timestamp) {
         logInfo("Fetching " + name + " with timestamp " + timestamp)
-        Utils.fetchCachedFile(name, new File(SparkFiles.getRootDirectory), conf,
+        Utils.fetchFile(name, new File(SparkFiles.getRootDirectory), conf,
           env.securityManager, hadoopConf, timestamp, true)
         currentFiles(name) = timestamp
       }
       for ((name, timestamp) <- newJars if currentJars.getOrElse(name, -1L) < timestamp) {
         logInfo("Fetching " + name + " with timestamp " + timestamp)
-        Utils.fetchCachedFile(name, new File(SparkFiles.getRootDirectory), conf,
+        Utils.fetchFile(name, new File(SparkFiles.getRootDirectory), conf,
           env.securityManager, hadoopConf, timestamp, true)
         currentJars(name) = timestamp
         // Add it to our class loader

core/src/main/scala/org/apache/spark/util/Utils.scala

@@ -322,7 +322,7 @@ private[spark] object Utils extends Logging {
    * Throws SparkException if the target file already exists and has different contents than
    * the requested file.
    */
-  def fetchCachedFile(
+  def fetchFile(
       url: String,
       targetDir: File,
       conf: SparkConf,
@@ -333,27 +333,27 @@ private[spark] object Utils extends Logging {
     val fileName = url.split("/").last
     val targetFile = new File(targetDir, fileName)
     if (useCache) {
-      val cachedFileName = fileName + timestamp
-      val lockFileName = fileName + timestamp + "_lock"
+      val cachedFileName = url.hashCode + timestamp + "_cach"
+      val lockFileName = url.hashCode + timestamp + "_lock"
       val localDir = new File(getLocalDir(conf))
       val lockFile = new File(localDir, lockFileName)
       val raf = new RandomAccessFile(lockFile, "rw")
       // Only one executor entry.
       // The FileLock is only used to control synchronization for executors download file,
       // it's always safe regardless of lock type(mandatory or advisory).
       val lock = raf.getChannel().lock()
-      val cachedFile = new File(SparkFiles.getRootDirectory + "../", cachedFileName)
+      val cachedFile = new File(localDir, cachedFileName)
       try {
         if (!cachedFile.exists()) {
-          fetchFile(url, localDir, conf, securityMgr, hadoopConf)
+          doFetchFile(url, localDir, conf, securityMgr, hadoopConf)
           Files.move(new File(localDir, fileName), cachedFile)
         }
       } finally {
         lock.release()
       }
       Files.copy(cachedFile, targetFile)
     } else {
-      fetchFile(url, targetDir, conf, securityMgr, hadoopConf)
+      doFetchFile(url, targetDir, conf, securityMgr, hadoopConf)
     }
 
     // Decompress the file if it's a .tar or .tar.gz
@@ -375,8 +375,12 @@ private[spark] object Utils extends Logging {
    * Throws SparkException if the target file already exists and has different contents than
    * the requested file.
    */
-  private def fetchFile(url: String, targetDir: File, conf: SparkConf, securityMgr: SecurityManager,
-    hadoopConf: Configuration) {
+  private def doFetchFile(
+      url: String,
+      targetDir: File,
+      conf: SparkConf,
+      securityMgr: SecurityManager,
+      hadoopConf: Configuration) {
     val filename = url.split("/").last
     val tempDir = getLocalDir(conf)
     val tempFile =  File.createTempFile("fetchFileTemp", null, new File(tempDir))