Merge -r 772875:772876 from trunk onto 0.20 branch. Fixes HADOOP-5349.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/branches/branch-0.20@772877 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -53,6 +53,10 @@ Release 0.20.1 - Unreleased
     kills itself if it ever discovers that the port to which jetty is actually
     bound is invalid (-1). (ddas)
 
+    HADOOP-5349. Fixes a problem in LocalDirAllocator to check for the return
+    path value that is returned for the case where the file we want to write
+    is of an unknown size. (Vinod Kumar Vavilapalli via ddas)
+
 Release 0.20.0 - 2009-04-15
 
   INCOMPATIBLE CHANGES

src/core/org/apache/hadoop/fs/LocalDirAllocator.java

@@ -306,16 +306,24 @@ public class LocalDirAllocator {
           availableOnDisk[i] = dirDF[i].getAvailable();
           totalAvailable += availableOnDisk[i];
         }
-            // "roll the ball" -- pick a directory
+
+        // Keep rolling the wheel till we get a valid path
         Random r = new java.util.Random();
-        long randomPosition = Math.abs(r.nextLong()) % totalAvailable;
-        int dir=0;
-        while(randomPosition > availableOnDisk[dir]) {
-          randomPosition -= availableOnDisk[dir];
-          dir++;
+        while (numDirsSearched < numDirs && returnPath == null) {
+          long randomPosition = Math.abs(r.nextLong()) % totalAvailable;
+          int dir = 0;
+          while (randomPosition > availableOnDisk[dir]) {
+            randomPosition -= availableOnDisk[dir];
+            dir++;
+          }
+          dirNumLastAccessed = dir;
+          returnPath = createPath(pathStr);
+          if (returnPath == null) {
+            totalAvailable -= availableOnDisk[dir];
+            availableOnDisk[dir] = 0; // skip this disk
+            numDirsSearched++;
+          }
         }
-        dirNumLastAccessed = dir;
-        returnPath = createPath(pathStr);
       } else {
         while (numDirsSearched < numDirs && returnPath == null) {
           long capacity = dirDF[dirNumLastAccessed].getAvailable();