volume-balancer-next-subdir.patch

diff --git a/src/main/java/org/apache/hadoop/hdfs/server/datanode/VolumeBalancer.java b/src/main/java/org/apache/hadoop/hdfs/server/datanode/VolumeBalancer.java
index 59d32ae..7937fbd 100644
--- a/src/main/java/org/apache/hadoop/hdfs/server/datanode/VolumeBalancer.java
+++ b/src/main/java/org/apache/hadoop/hdfs/server/datanode/VolumeBalancer.java
@@ -316,10 +316,7 @@ public class VolumeBalancer {
             /*
              * Generate the final name of the destination
              */
-            final File[] existingSubDirs = findSubdirs(leastUsedBlockSubdir);
-
-            final File finalLeastUsedBlockSubdir = new File(leastUsedBlockSubdir, DataStorage.BLOCK_SUBDIR_PREFIX
-                + existingSubDirs.length);
+            final File finalLeastUsedBlockSubdir = new File(leastUsedBlockSubdir, nextSubdir(leastUsedBlockSubdir, DataStorage.BLOCK_SUBDIR_PREFIX));
 
             /*
              * Schedule the two subdir for a move.
@@ -348,6 +345,31 @@ public class VolumeBalancer {
 
     }
 
+    private static String nextSubdir(File dir, final String subdirPrefix){
+      File[] existing = dir.listFiles(new FileFilter() {
+        @Override
+        public boolean accept(File pathname) {
+          return pathname.getName().startsWith(subdirPrefix);
+        }
+      });
+      if (existing == null || existing.length == 0)
+        return subdirPrefix + "0";
+
+      //  listFiles doesn't guarantee ordering
+      int lastIndex = -1;
+      for (int i = 0; i < existing.length; i++){
+        String name = existing[i].getName();
+        try{
+          int index = Integer.parseInt(name.substring(subdirPrefix.length()));
+          if (lastIndex < index)
+            lastIndex = index;
+        }catch(NumberFormatException e){
+          // ignore
+        }
+      }
+      return subdirPrefix + (lastIndex +1);
+    }
+
     private static File[] findSubdirs(File parent) {
         return parent.listFiles(new FileFilter() {
             @Override