Merge -r 605867:605868 from trunk to branch-0.15 to fix HADOOP-2437.

git-svn-id: https://svn.apache.org/repos/asf/lucene/hadoop/branches/branch-0.15@605870 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -35,6 +35,10 @@ Branch 0.15 (unreleased)
     free-space on any one of the available disks. (Amareshwari Sri Ramadasu
     via acmurthy)
 
+    HADOOP-2437.  Fix the LocalDirAllocator to choose the seed for the
+    round-robin disk selections randomly. This helps in spreading data across
+    multiple partitions much better. (acmurhty)
+
   IMPROVEMENTS
 
     HADOOP-2160.  Remove project-level, non-user documentation from

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

@@ -177,12 +177,22 @@ public class LocalDirAllocator {
     return context.ifExists(pathStr, conf);
   }
 
+  /**
+   * Get the current directory index for the given configuration item.
+   * @return the current directory index for the given configuration item.
+   */
+  int getCurrentDirectoryIndex() {
+    AllocatorPerContext context = obtainContext(contextCfgItemName);
+    return context.getCurrentDirectoryIndex();
+  }
+  
   private static class AllocatorPerContext {
 
     private final Log LOG =
       LogFactory.getLog("org.apache.hadoop.fs.AllocatorPerContext");
 
     private int dirNumLastAccessed;
+    private Random dirIndexRandomizer = new Random();
     private FileSystem localFS;
     private DF[] dirDF;
     private String contextCfgItemName;
@@ -227,8 +237,10 @@ public class LocalDirAllocator {
         }
         localDirs = dirs.toArray(new String[dirs.size()]);
         dirDF = dfList.toArray(new DF[dirs.size()]);
-        dirNumLastAccessed = 0;
         savedLocalDirs = newLocalDirs;
+        
+        // randomize the first disk picked in the round-robin selection 
+        dirNumLastAccessed = dirIndexRandomizer.nextInt(dirs.size());
       }
     }
 
@@ -246,6 +258,14 @@ public class LocalDirAllocator {
       }
     }
 
+    /**
+     * Get the current directory index.
+     * @return the current directory index.
+     */
+    int getCurrentDirectoryIndex() {
+      return dirNumLastAccessed;
+    }
+    
     /** Get a path from the local FS. This method should be used if the size of 
      *  the file is not known apriori. We go round-robin over the set of disks
      *  (via the configured dirs) and return the first complete path where

src/test/org/apache/hadoop/fs/TestLocalDirAllocator.java

@@ -68,8 +68,8 @@ public class TestLocalDirAllocator extends TestCase {
   
   private void validateTempDirCreation(int i) throws IOException {
     File result = createTempFile();
-    assertTrue(result.getPath().startsWith(
-        new File(BUFFER_DIR[i], FILENAME).getPath()));
+    assertTrue("Checking for " + BUFFER_DIR[i] + " in " + result + " - FAILED!", 
+        result.getPath().startsWith(new File(BUFFER_DIR[i], FILENAME).getPath()));
   }
   
   private File createTempFile() throws IOException {
@@ -119,10 +119,16 @@ public class TestLocalDirAllocator extends TestCase {
     if (isWindows) return;
     try {
       conf.set(CONTEXT, BUFFER_DIR[2]+","+BUFFER_DIR[3]);
-      validateTempDirCreation(2);
-      validateTempDirCreation(3);
-      validateTempDirCreation(2);
-      validateTempDirCreation(3);
+
+      // create the first file, and then figure the round-robin sequence
+      createTempFile();
+      int firstDirIdx = (dirAllocator.getCurrentDirectoryIndex() == 0) ? 2 : 3;
+      int secondDirIdx = (firstDirIdx == 2) ? 3 : 2;
+      
+      // check if tmp dirs are allocated in a round-robin manner
+      validateTempDirCreation(firstDirIdx);
+      validateTempDirCreation(secondDirIdx);
+      validateTempDirCreation(firstDirIdx);
     } finally {
       rmBufferDirs();
     }
@@ -139,8 +145,11 @@ public class TestLocalDirAllocator extends TestCase {
       assertTrue(localFs.mkdirs(BUFFER_PATH[3]));
       assertTrue(localFs.mkdirs(BUFFER_PATH[4]));
       
-      validateTempDirCreation(3);
-      validateTempDirCreation(4);
+      // create the first file, and then figure the round-robin sequence
+      createTempFile();
+
+      int nextDirIdx = (dirAllocator.getCurrentDirectoryIndex() == 0) ? 3 : 4;
+      validateTempDirCreation(nextDirIdx);
 
       // change buffer directory 2 to be read only
       new File(BUFFER_DIR[4]).setReadOnly();