HDFS-17210. Optimize AvailableSpaceBlockPlacementPolicy. (#6113). Contributed by GuoPhilipse.

Reviewed-by:  He Xiaoqiao <hexiaoqiao@apache.org>
Signed-off-by: Shuyan Zhang <zhangshuyan@apache.org>

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java

@@ -1241,6 +1241,12 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
   public static final int
       DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_TOLERANCE_DEFAULT =
       5;
+  public static final String
+      DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_TOLERANCE_LIMIT_KEY =
+      "dfs.namenode.available-space-block-placement-policy.balanced-space-tolerance-limit";
+  public static final int
+      DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_TOLERANCE_LIMIT_DEFAULT =
+      100;
   public static final String
       DFS_NAMENODE_AVAILABLE_SPACE_RACK_FAULT_TOLERANT_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_PREFERENCE_FRACTION_KEY =
       "dfs.namenode.available-space-rack-fault-tolerant-block-placement-policy"

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/AvailableSpaceBlockPlacementPolicy.java

@@ -22,6 +22,8 @@
 import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_PREFERENCE_FRACTION_DEFAULT;
 import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_TOLERANCE_DEFAULT;
 import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_TOLERANCE_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_TOLERANCE_LIMIT_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_TOLERANCE_LIMIT_DEFAULT;
 
 import java.util.Collection;
 import java.util.EnumMap;
@@ -50,7 +52,10 @@ public class AvailableSpaceBlockPlacementPolicy extends
   private int balancedPreference =
       (int) (100 * DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_PREFERENCE_FRACTION_DEFAULT);
   private int balancedSpaceTolerance =
-          DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_TOLERANCE_DEFAULT;
+      DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_TOLERANCE_DEFAULT;
+
+  private int balancedSpaceToleranceLimit =
+      DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_TOLERANCE_LIMIT_DEFAULT;
   private boolean optimizeLocal;
 
   @Override
@@ -59,8 +64,8 @@ public void initialize(Configuration conf, FSClusterStats stats,
     super.initialize(conf, stats, clusterMap, host2datanodeMap);
     float balancedPreferencePercent =
         conf.getFloat(
-          DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_PREFERENCE_FRACTION_KEY,
+        DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_PREFERENCE_FRACTION_KEY,
-          DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_PREFERENCE_FRACTION_DEFAULT);
+        DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_PREFERENCE_FRACTION_DEFAULT);
 
     LOG.info("Available space block placement policy initialized: "
         + DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_PREFERENCE_FRACTION_KEY
@@ -71,6 +76,11 @@ public void initialize(Configuration conf, FSClusterStats stats,
         DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_TOLERANCE_KEY,
         DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_TOLERANCE_DEFAULT);
 
+    balancedSpaceToleranceLimit =
+      conf.getInt(
+      DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_TOLERANCE_LIMIT_KEY,
+      DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_TOLERANCE_LIMIT_DEFAULT);
+
     optimizeLocal = conf.getBoolean(
         DFSConfigKeys.DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCE_LOCAL_NODE_KEY,
         DFSConfigKeys.DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCE_LOCAL_NODE_DEFAULT);
@@ -87,6 +97,17 @@ public void initialize(Configuration conf, FSClusterStats stats,
           + " receive  more block allocations.");
     }
 
+    if (balancedSpaceToleranceLimit > 100 || balancedSpaceToleranceLimit < 0) {
+      LOG.warn("The value of "
+          + DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_TOLERANCE_LIMIT_KEY
+          + " is invalid, Current value is " + balancedSpaceToleranceLimit + ", Default value "
+          + DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_TOLERANCE_LIMIT_DEFAULT
+          + " will be used instead.");
+
+      balancedSpaceToleranceLimit =
+          DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_TOLERANCE_LIMIT_DEFAULT;
+    }
+
     if (balancedSpaceTolerance > 20 || balancedSpaceTolerance < 0) {
       LOG.warn("The value of "
           + DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_TOLERANCE_KEY
@@ -201,8 +222,12 @@ private DatanodeDescriptor select(DatanodeDescriptor a, DatanodeDescriptor b,
    */
   protected int compareDataNode(final DatanodeDescriptor a,
       final DatanodeDescriptor b, boolean isBalanceLocal) {
+
+    boolean toleranceLimit = Math.max(a.getDfsUsedPercent(), b.getDfsUsedPercent())
+        < balancedSpaceToleranceLimit;
     if (a.equals(b)
-        || Math.abs(a.getDfsUsedPercent() - b.getDfsUsedPercent()) < balancedSpaceTolerance || ((
+        || (toleranceLimit && Math.abs(a.getDfsUsedPercent() - b.getDfsUsedPercent())
+            < balancedSpaceTolerance) || ((
         isBalanceLocal && a.getDfsUsedPercent() < 50))) {
       return 0;
     }

hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml

@@ -5173,6 +5173,17 @@
     </description>
   </property>
 
+  <property>
+    <name>dfs.namenode.available-space-block-placement-policy.balanced-space-tolerance-limit</name>
+    <value>100</value>
+    <description>
+      Only used when the dfs.block.replicator.classname is set to
+      org.apache.hadoop.hdfs.server.blockmanagement.AvailableSpaceBlockPlacementPolicy.
+      Special value between 0 and 100, inclusive. if the value is set beyond the scope,
+      this value will be set as 100 by default.
+    </description>
+  </property>
+
 <property>
   <name>
     dfs.namenode.available-space-block-placement-policy.balance-local-node

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestAvailableSpaceBlockPlacementPolicy.java

@@ -58,9 +58,12 @@ public class TestAvailableSpaceBlockPlacementPolicy {
   @BeforeClass
   public static void setupCluster() throws Exception {
     conf = new HdfsConfiguration();
-    conf.setFloat(
+    conf.setFloat(DFSConfigKeys.
-      DFSConfigKeys.DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_PREFERENCE_FRACTION_KEY,
+        DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_PREFERENCE_FRACTION_KEY,
-      0.6f);
+            0.6f);
+    conf.setInt(DFSConfigKeys.
+        DFS_NAMENODE_AVAILABLE_SPACE_BLOCK_PLACEMENT_POLICY_BALANCED_SPACE_TOLERANCE_LIMIT_KEY,
+            93);
     String[] racks = new String[numRacks];
     for (int i = 0; i < numRacks; i++) {
       racks[i] = "/rack" + i;
@@ -219,6 +222,78 @@ public void testChooseSimilarDataNode() {
             tolerateDataNodes[0], false) == 1);
   }
 
+
+  @Test
+  public void testCompareDataNode() {
+    DatanodeDescriptor[] tolerateDataNodes;
+    DatanodeStorageInfo[] tolerateStorages;
+    int capacity  = 5;
+    Collection<Node> allTolerateNodes = new ArrayList<>(capacity);
+    String[] ownerRackOfTolerateNodes = new String[capacity];
+    for (int i = 0; i < capacity; i++) {
+      ownerRackOfTolerateNodes[i] = "rack"+i;
+    }
+    tolerateStorages = DFSTestUtil.createDatanodeStorageInfos(ownerRackOfTolerateNodes);
+    tolerateDataNodes = DFSTestUtil.toDatanodeDescriptor(tolerateStorages);
+
+    Collections.addAll(allTolerateNodes, tolerateDataNodes);
+    final BlockManager bm = namenode.getNamesystem().getBlockManager();
+    AvailableSpaceBlockPlacementPolicy toleratePlacementPolicy =
+        (AvailableSpaceBlockPlacementPolicy)bm.getBlockPlacementPolicy();
+
+    //96.6%
+    updateHeartbeatWithUsage(tolerateDataNodes[0],
+        30 * HdfsServerConstants.MIN_BLOCKS_FOR_WRITE * blockSize,
+        29 * HdfsServerConstants.MIN_BLOCKS_FOR_WRITE * blockSize,
+        HdfsServerConstants.MIN_BLOCKS_FOR_WRITE
+        * blockSize, 0L, 0L, 0L, 0, 0);
+
+    //93.3%
+    updateHeartbeatWithUsage(tolerateDataNodes[1],
+        30 * HdfsServerConstants.MIN_BLOCKS_FOR_WRITE * blockSize,
+        28 * HdfsServerConstants.MIN_BLOCKS_FOR_WRITE * blockSize,
+        HdfsServerConstants.MIN_BLOCKS_FOR_WRITE
+        * blockSize, 0L, 0L, 0L, 0, 0);
+
+    //90.0%
+    updateHeartbeatWithUsage(tolerateDataNodes[2],
+        30 * HdfsServerConstants.MIN_BLOCKS_FOR_WRITE * blockSize,
+        27 * HdfsServerConstants.MIN_BLOCKS_FOR_WRITE * blockSize,
+        HdfsServerConstants.MIN_BLOCKS_FOR_WRITE
+        * blockSize, 0L, 0L, 0L, 0, 0);
+
+    //86.6%
+    updateHeartbeatWithUsage(tolerateDataNodes[3],
+        30 * HdfsServerConstants.MIN_BLOCKS_FOR_WRITE * blockSize,
+        26 * HdfsServerConstants.MIN_BLOCKS_FOR_WRITE * blockSize,
+        HdfsServerConstants.MIN_BLOCKS_FOR_WRITE
+        * blockSize, 0L, 0L, 0L, 0, 0);
+
+    //83.3%
+    updateHeartbeatWithUsage(tolerateDataNodes[4],
+        30 * HdfsServerConstants.MIN_BLOCKS_FOR_WRITE * blockSize,
+        25 * HdfsServerConstants.MIN_BLOCKS_FOR_WRITE * blockSize,
+        HdfsServerConstants.MIN_BLOCKS_FOR_WRITE
+        * blockSize, 0L, 0L, 0L, 0, 0);
+
+    assertTrue(toleratePlacementPolicy.compareDataNode(tolerateDataNodes[0],
+        tolerateDataNodes[1], false) == 1);
+    assertTrue(toleratePlacementPolicy.compareDataNode(tolerateDataNodes[1],
+        tolerateDataNodes[0], false) == -1);
+    assertTrue(toleratePlacementPolicy.compareDataNode(tolerateDataNodes[1],
+        tolerateDataNodes[2], false) == 1);
+    assertTrue(toleratePlacementPolicy.compareDataNode(tolerateDataNodes[2],
+        tolerateDataNodes[1], false) == -1);
+    assertTrue(toleratePlacementPolicy.compareDataNode(tolerateDataNodes[2],
+        tolerateDataNodes[3], false) == 0);
+    assertTrue(toleratePlacementPolicy.compareDataNode(tolerateDataNodes[3],
+        tolerateDataNodes[2], false) == 0);
+    assertTrue(toleratePlacementPolicy.compareDataNode(tolerateDataNodes[2],
+        tolerateDataNodes[4], false) == 1);
+    assertTrue(toleratePlacementPolicy.compareDataNode(tolerateDataNodes[4],
+        tolerateDataNodes[2], false) == -1);
+  }
+
   @AfterClass
   public static void teardownCluster() {
     if (namenode != null) {