change config name and simplify the code

Author: tomscut

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

@@ -698,9 +698,9 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
       "dfs.datanode.max.disks.to.report";
   public static final int DFS_DATANODE_MAX_DISKS_TO_REPORT_DEFAULT =
       5;
-  public static final String DFS_DATANODE_MAX_SLOWDISKS_TO_BE_EXCLUDED_KEY =
-      "dfs.datanode.max.slowdisks.to.be.excluded";
-  public static final int DFS_DATANODE_MAX_SLOWDISKS_TO_BE_EXCLUDED_DEFAULT =
+  public static final String DFS_DATANODE_MAX_SLOWDISKS_TO_EXCLUDE_KEY =
+      "dfs.datanode.max.slowdisks.to.exclude";
+  public static final int DFS_DATANODE_MAX_SLOWDISKS_TO_EXCLUDE_DEFAULT =
       0;
   public static final String  DFS_DATANODE_HOST_NAME_KEY =
       HdfsClientConfigKeys.DeprecatedKeys.DFS_DATANODE_HOST_NAME_KEY;

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeList.java

@@ -107,9 +107,9 @@ private FsVolumeReference chooseVolume(List<FsVolumeImpl> list,
 
     // Exclude slow disks when choosing volume.
     if (diskMetrics != null) {
-      List<String> slowDisksToBeExcluded = diskMetrics.getSlowDisksToBeExcluded();
+      List<String> slowDisksToExclude = diskMetrics.getSlowDisksToExclude();
       list = list.stream()
-          .filter(volume -> !slowDisksToBeExcluded.contains(volume.getBaseURI().getPath()))
+          .filter(volume -> !slowDisksToExclude.contains(volume.getBaseURI().getPath()))
           .collect(Collectors.toList());
     }
 

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/metrics/DataNodeDiskMetrics.java

@@ -36,6 +36,7 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
@@ -77,11 +78,11 @@ public class DataNodeDiskMetrics {
   /**
    * The number of slow disks that needs to be excluded.
    */
-  private int maxSlowDisksToBeExcluded;
+  private int maxSlowDisksToExclude;
   /**
    * List of slow disks that need to be excluded.
    */
-  private List<String> slowDisksToBeExcluded = new ArrayList<>();
+  private List<String> slowDisksToExclude = new ArrayList<>();
 
   public DataNodeDiskMetrics(DataNode dn, long diskOutlierDetectionIntervalMs,
       Configuration conf) {
@@ -93,9 +94,9 @@ public DataNodeDiskMetrics(DataNode dn, long diskOutlierDetectionIntervalMs,
     lowThresholdMs =
         conf.getLong(DFSConfigKeys.DFS_DATANODE_SLOWDISK_LOW_THRESHOLD_MS_KEY,
             DFSConfigKeys.DFS_DATANODE_SLOWDISK_LOW_THRESHOLD_MS_DEFAULT);
-    maxSlowDisksToBeExcluded =
-        conf.getInt(DFSConfigKeys.DFS_DATANODE_MAX_SLOWDISKS_TO_BE_EXCLUDED_KEY,
-            DFSConfigKeys.DFS_DATANODE_MAX_SLOWDISKS_TO_BE_EXCLUDED_DEFAULT);
+    maxSlowDisksToExclude =
+        conf.getInt(DFSConfigKeys.DFS_DATANODE_MAX_SLOWDISKS_TO_EXCLUDE_KEY,
+            DFSConfigKeys.DFS_DATANODE_MAX_SLOWDISKS_TO_EXCLUDE_DEFAULT);
     slowDiskDetector =
         new OutlierDetector(minOutlierDetectionDisks, lowThresholdMs);
     shouldRun = true;
@@ -144,14 +145,19 @@ public void run() {
             detectAndUpdateDiskOutliers(metadataOpStats, readIoStats,
                 writeIoStats);
 
-            // Sort the slow disks by latency.
-            if (maxSlowDisksToBeExcluded > 0) {
+            // Sort the slow disks by latency and .
+            if (maxSlowDisksToExclude > 0) {
               ArrayList<DiskLatency> diskLatencies = new ArrayList<>();
               for (Map.Entry<String, Map<DiskOp, Double>> diskStats :
                   diskOutliersStats.entrySet()) {
                 diskLatencies.add(new DiskLatency(diskStats.getKey(), diskStats.getValue()));
               }
-              sortSlowDisks(diskLatencies);
+
+              Collections.sort(diskLatencies, (o1, o2)
+                  -> Double.compare(o2.getMaxLatency(), o1.getMaxLatency()));
+
+              slowDisksToExclude = diskLatencies.stream().limit(maxSlowDisksToExclude)
+                  .map(DiskLatency::getSlowDisk).collect(Collectors.toList());
             }
           }
 
@@ -197,29 +203,6 @@ private void detectAndUpdateDiskOutliers(Map<String, Double> metadataOpStats,
     }
   }
 
-  private void sortSlowDisks(ArrayList<DiskLatency> diskLatencies) {
-    if (diskOutliersStats.isEmpty()) {
-      return;
-    }
-
-    final PriorityQueue<DiskLatency> topNReports = new PriorityQueue<>(
-        diskLatencies.size(),
-        (o1, o2) -> Doubles.compare(
-            o1.getMaxLatency(), o2.getMaxLatency()));
-
-    for (DiskLatency diskLatency : diskLatencies) {
-      if (topNReports.size() < maxSlowDisksToBeExcluded) {
-        topNReports.add(diskLatency);
-      } else if (topNReports.peek().getMaxLatency() <
-          diskLatency.getMaxLatency()) {
-        topNReports.poll();
-        topNReports.add(diskLatency);
-      }
-    }
-    slowDisksToBeExcluded =
-        topNReports.stream().map(DiskLatency::getSlowDisk).collect(Collectors.toList());
-  }
-
   /**
    * This structure is a wrapper over disk latencies.
    */
@@ -285,7 +268,7 @@ public void addSlowDiskForTesting(String slowDiskPath,
     }
   }
 
-  public List<String> getSlowDisksToBeExcluded() {
-    return slowDisksToBeExcluded;
+  public List<String> getSlowDisksToExclude() {
+    return slowDisksToExclude;
   }
 }

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

@@ -2484,7 +2484,7 @@
 </property>
 
 <property>
-  <name>dfs.datanode.max.slowdisks.to.be.excluded</name>
+  <name>dfs.datanode.max.slowdisks.to.exclude</name>
   <value>0</value>
   <description>
     The number of slow disks that needs to be excluded. By default, this parameter is set to 0,

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/TestFsVolumeList.java

@@ -80,7 +80,7 @@ public class TestFsVolumeList {
   private String baseDir;
   private BlockScanner blockScanner;
   private final static int NUM_DATANODES = 3;
-  private final static int STORAGES_PER_DATANODE = 5;
+  private final static int STORAGES_PER_DATANODE = 3;
   private final static int DEFAULT_BLOCK_SIZE = 102400;
   private final static int BUFFER_LENGTH = 1024;
 
@@ -640,7 +640,7 @@ public void testExcludeSlowDiskWhenChoosingVolume() throws Exception {
     // Enable datanode disk metrics collector.
     conf.setInt(DFSConfigKeys.DFS_DATANODE_FILEIO_PROFILING_SAMPLING_PERCENTAGE_KEY, 30);
     // Enable excluding slow disks when choosing volume.
-    conf.setInt(DFSConfigKeys.DFS_DATANODE_MAX_SLOWDISKS_TO_BE_EXCLUDED_KEY, 1);
+    conf.setInt(DFSConfigKeys.DFS_DATANODE_MAX_SLOWDISKS_TO_EXCLUDE_KEY, 1);
     // Ensure that each volume capacity is larger than the DEFAULT_BLOCK_SIZE.
     long capacity = 10 * DEFAULT_BLOCK_SIZE;
     long[][] capacities = new long[NUM_DATANODES][STORAGES_PER_DATANODE];
@@ -704,9 +704,9 @@ public void testExcludeSlowDiskWhenChoosingVolume() throws Exception {
     // Wait until the data on the slow disk is collected successfully.
     GenericTestUtils.waitFor(new Supplier<Boolean>() {
       @Override public Boolean get() {
-        return dn0.getDiskMetrics().getSlowDisksToBeExcluded().size() == 1 &&
-            dn1.getDiskMetrics().getSlowDisksToBeExcluded().size() == 1 &&
-            dn2.getDiskMetrics().getSlowDisksToBeExcluded().size() == 1;
+        return dn0.getDiskMetrics().getSlowDisksToExclude().size() == 1 &&
+            dn1.getDiskMetrics().getSlowDisksToExclude().size() == 1 &&
+            dn2.getDiskMetrics().getSlowDisksToExclude().size() == 1;
       }
     }, 1000, 5000);