HADOOP-671. Fix file cache to check for pre-existence before creating symlinks. Contributed by Mahadev.

git-svn-id: https://svn.apache.org/repos/asf/lucene/hadoop/trunk@470226 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -137,6 +137,9 @@ Trunk (unreleased changes)
 38. HADOOP-647.  Permit map outputs to use a different compression
     type than the job output.  (omalley via cutting)
 
+39. HADOOP-671.  File cache should not create symlinks when they
+    already exist.  (Mahadev Konar via cutting)
+
 
 Release 0.7.2 - 2006-10-18
 

src/java/org/apache/hadoop/filecache/DistributedCache.java

@@ -164,19 +164,23 @@ public class DistributedCache {
     boolean doSymlink = getSymlink(conf);
     FileSystem dfs = getFileSystem(cache, conf);
     b = ifExistsAndFresh(cacheStatus, cache, dfs, md5, conf);
+    String link = currentWorkDir.toString() + Path.SEPARATOR + cache.getFragment();
+    File flink = new File(link);
     if (b) {
       if (isArchive) {
-        if (doSymlink)
-        FileUtil.symLink(cacheStatus.localLoadPath.toString(), 
-            currentWorkDir.toString() + Path.SEPARATOR + cache.getFragment());
-        
+        if (doSymlink){
+          if (!flink.exists())
+            FileUtil.symLink(cacheStatus.localLoadPath.toString(), 
+                link);
+        }
         return cacheStatus.localLoadPath;
       }
       else {
-        if (doSymlink)
-          FileUtil.symLink(cacheFilePath(cacheStatus.localLoadPath).toString(), 
-              currentWorkDir.toString() + Path.SEPARATOR + cache.getFragment());
-       
+        if (doSymlink){
+          if (!flink.exists())
+            FileUtil.symLink(cacheFilePath(cacheStatus.localLoadPath).toString(), 
+              link);
+        }
         return cacheFilePath(cacheStatus.localLoadPath);
       }
     } else {
@@ -219,16 +223,21 @@ public class DistributedCache {
       cacheStatus.currentStatus = true;
       cacheStatus.md5 = checkSum;
     }
+    
     if (isArchive){
-      if (doSymlink)
-        FileUtil.symLink(cacheStatus.localLoadPath.toString(), 
-            currentWorkDir.toString() + Path.SEPARATOR + cache.getFragment());
+      if (doSymlink){
+        if (!flink.exists())
+          FileUtil.symLink(cacheStatus.localLoadPath.toString(), 
+            link);
+      }
       return cacheStatus.localLoadPath;
     }
     else {
-      if (doSymlink)
-        FileUtil.symLink(cacheFilePath(cacheStatus.localLoadPath).toString(), 
-            currentWorkDir.toString() + Path.SEPARATOR + cache.getFragment());
+      if (doSymlink){
+        if (!flink.exists())
+          FileUtil.symLink(cacheFilePath(cacheStatus.localLoadPath).toString(), 
+            link);
+      }
       return cacheFilePath(cacheStatus.localLoadPath);
     }
   }