HDFS-10309 Balancer doesn't honor dfs.blocksize value defined with suffix k(kilo), m(mega), g(giga). Contributed by Amit Anand

Author: Tsz-Wo Nicholas Sze

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Balancer.java

@@ -230,6 +230,15 @@ static long getLong(Configuration conf, String key, long defaultValue) {
     return v;
   }
 
+  static long getLongBytes(Configuration conf, String key, long defaultValue) {
+    final long v = conf.getLongBytes(key, defaultValue);
+    LOG.info(key + " = " + v + " (default=" + defaultValue + ")");
+    if (v <= 0) {
+      throw new HadoopIllegalArgumentException(key + " = " + v  + " <= " + 0);
+    }
+    return v;
+  }
+
   static int getInt(Configuration conf, String key, int defaultValue) {
     final int v = conf.getInt(key, defaultValue);
     LOG.info(key + " = " + v + " (default=" + defaultValue + ")");
@@ -261,10 +270,10 @@ static int getInt(Configuration conf, String key, int defaultValue) {
         DFSConfigKeys.DFS_DATANODE_BALANCE_MAX_NUM_CONCURRENT_MOVES_KEY,
         DFSConfigKeys.DFS_DATANODE_BALANCE_MAX_NUM_CONCURRENT_MOVES_DEFAULT);
 
-    final long getBlocksSize = getLong(conf,
+    final long getBlocksSize = getLongBytes(conf,
         DFSConfigKeys.DFS_BALANCER_GETBLOCKS_SIZE_KEY,
         DFSConfigKeys.DFS_BALANCER_GETBLOCKS_SIZE_DEFAULT);
-    final long getBlocksMinBlockSize = getLong(conf,
+    final long getBlocksMinBlockSize = getLongBytes(conf,
         DFSConfigKeys.DFS_BALANCER_GETBLOCKS_MIN_BLOCK_SIZE_KEY,
         DFSConfigKeys.DFS_BALANCER_GETBLOCKS_MIN_BLOCK_SIZE_DEFAULT);
 
@@ -279,10 +288,10 @@ static int getInt(Configuration conf, String key, int defaultValue) {
     this.sourceNodes = p.getSourceNodes();
     this.runDuringUpgrade = p.getRunDuringUpgrade();
 
-    this.maxSizeToMove = getLong(conf,
+    this.maxSizeToMove = getLongBytes(conf,
         DFSConfigKeys.DFS_BALANCER_MAX_SIZE_TO_MOVE_KEY,
         DFSConfigKeys.DFS_BALANCER_MAX_SIZE_TO_MOVE_DEFAULT);
-    this.defaultBlockSize = getLong(conf,
+    this.defaultBlockSize = getLongBytes(conf,
         DFSConfigKeys.DFS_BLOCK_SIZE_KEY,
         DFSConfigKeys.DFS_BLOCK_SIZE_DEFAULT);
   }