HDFS-17641. Add badly distributed blocks metric (#7123)

hadoop-common-project/hadoop-common/src/site/markdown/Metrics.md

@@ -304,6 +304,7 @@ Each metrics record contains tags such as HAState and Hostname as additional inf
 | `StaleDataNodes` | Current number of DataNodes marked stale due to delayed heartbeat |
 | `NumStaleStorages` | Number of storages marked as content stale (after NameNode restart/failover before first block report is received) |
 | `MissingReplOneBlocks` | Current number of missing blocks with replication factor 1 |
+| `BadlyDistributedBlocks` | Current number of blocks that are badly distributed across racks. |
 | `HighestPriorityLowRedundancyReplicatedBlocks` | Current number of non-corrupt, low redundancy replicated blocks with the highest risk of loss (have 0 or 1 replica). Will be recovered with the highest priority. |
 | `HighestPriorityLowRedundancyECBlocks` | Current number of non-corrupt, low redundancy EC blocks with the highest risk of loss. Will be recovered with the highest priority. |
 | `NumFilesUnderConstruction` | Current number of files under construction |

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/ECBlockGroupStats.java

@@ -38,24 +38,28 @@ public final class ECBlockGroupStats {
   private final long missingBlockGroups;
   private final long bytesInFutureBlockGroups;
   private final long pendingDeletionBlocks;
+  private final long badlyDistributedBlocks;
   private final Long highestPriorityLowRedundancyBlocks;
 
   public ECBlockGroupStats(long lowRedundancyBlockGroups,
       long corruptBlockGroups, long missingBlockGroups,
-      long bytesInFutureBlockGroups, long pendingDeletionBlocks) {
+      long bytesInFutureBlockGroups, long pendingDeletionBlocks,
+      long badlyDistributedBlocks) {
     this(lowRedundancyBlockGroups, corruptBlockGroups, missingBlockGroups,
-        bytesInFutureBlockGroups, pendingDeletionBlocks, null);
+        bytesInFutureBlockGroups, pendingDeletionBlocks,
+        badlyDistributedBlocks, null);
   }
 
   public ECBlockGroupStats(long lowRedundancyBlockGroups,
       long corruptBlockGroups, long missingBlockGroups,
       long bytesInFutureBlockGroups, long pendingDeletionBlocks,
-      Long highestPriorityLowRedundancyBlocks) {
+      long badlyDistributedBlocks, Long highestPriorityLowRedundancyBlocks) {
     this.lowRedundancyBlockGroups = lowRedundancyBlockGroups;
     this.corruptBlockGroups = corruptBlockGroups;
     this.missingBlockGroups = missingBlockGroups;
     this.bytesInFutureBlockGroups = bytesInFutureBlockGroups;
     this.pendingDeletionBlocks = pendingDeletionBlocks;
+    this.badlyDistributedBlocks = badlyDistributedBlocks;
     this.highestPriorityLowRedundancyBlocks
         = highestPriorityLowRedundancyBlocks;
   }
@@ -80,6 +84,10 @@ public final class ECBlockGroupStats {
     return pendingDeletionBlocks;
   }
 
+  public long getBadlyDistributedBlocks() {
+    return badlyDistributedBlocks;
+  }
+
   public boolean hasHighestPriorityLowRedundancyBlocks() {
     return getHighestPriorityLowRedundancyBlocks() != null;
   }
@@ -99,7 +107,8 @@ public final class ECBlockGroupStats {
         .append(", BytesInFutureBlockGroups=").append(
             getBytesInFutureBlockGroups())
         .append(", PendingDeletionBlocks=").append(
-            getPendingDeletionBlocks());
+            getPendingDeletionBlocks())
+        .append(" , BadlyDistributedBlocks=").append(getBadlyDistributedBlocks());
     if (hasHighestPriorityLowRedundancyBlocks()) {
       statsBuilder.append(", HighestPriorityLowRedundancyBlocks=")
           .append(getHighestPriorityLowRedundancyBlocks());
@@ -116,6 +125,7 @@ public final class ECBlockGroupStats {
         .append(missingBlockGroups)
         .append(bytesInFutureBlockGroups)
         .append(pendingDeletionBlocks)
+        .append(badlyDistributedBlocks)
         .append(highestPriorityLowRedundancyBlocks)
         .toHashCode();
   }
@@ -135,6 +145,7 @@ public final class ECBlockGroupStats {
         .append(missingBlockGroups, other.missingBlockGroups)
         .append(bytesInFutureBlockGroups, other.bytesInFutureBlockGroups)
         .append(pendingDeletionBlocks, other.pendingDeletionBlocks)
+        .append(badlyDistributedBlocks, other.badlyDistributedBlocks)
         .append(highestPriorityLowRedundancyBlocks,
             other.highestPriorityLowRedundancyBlocks)
         .isEquals();
@@ -151,6 +162,7 @@ public final class ECBlockGroupStats {
     long missingBlockGroups = 0;
     long bytesInFutureBlockGroups = 0;
     long pendingDeletionBlocks = 0;
+    long badlyDistributedBlocks = 0;
     long highestPriorityLowRedundancyBlocks = 0;
     boolean hasHighestPriorityLowRedundancyBlocks = false;
 
@@ -160,6 +172,7 @@ public final class ECBlockGroupStats {
       missingBlockGroups += stat.getMissingBlockGroups();
       bytesInFutureBlockGroups += stat.getBytesInFutureBlockGroups();
       pendingDeletionBlocks += stat.getPendingDeletionBlocks();
+      badlyDistributedBlocks += stat.getBadlyDistributedBlocks();
       if (stat.hasHighestPriorityLowRedundancyBlocks()) {
         hasHighestPriorityLowRedundancyBlocks = true;
         highestPriorityLowRedundancyBlocks +=
@@ -169,9 +182,10 @@ public final class ECBlockGroupStats {
     if (hasHighestPriorityLowRedundancyBlocks) {
       return new ECBlockGroupStats(lowRedundancyBlockGroups, corruptBlockGroups,
           missingBlockGroups, bytesInFutureBlockGroups, pendingDeletionBlocks,
-          highestPriorityLowRedundancyBlocks);
+          badlyDistributedBlocks, highestPriorityLowRedundancyBlocks);
     }
     return new ECBlockGroupStats(lowRedundancyBlockGroups, corruptBlockGroups,
-        missingBlockGroups, bytesInFutureBlockGroups, pendingDeletionBlocks);
+        missingBlockGroups, bytesInFutureBlockGroups, pendingDeletionBlocks,
+        badlyDistributedBlocks);
   }
 }

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/ReplicatedBlockStats.java

@@ -37,27 +37,30 @@ public final class ReplicatedBlockStats {
   private final long missingReplicationOneBlocks;
   private final long bytesInFutureBlocks;
   private final long pendingDeletionBlocks;
+  private final long badlyDistributedBlocks;
   private final Long highestPriorityLowRedundancyBlocks;
 
   public ReplicatedBlockStats(long lowRedundancyBlocks,
       long corruptBlocks, long missingBlocks,
       long missingReplicationOneBlocks, long bytesInFutureBlocks,
-      long pendingDeletionBlocks) {
+      long pendingDeletionBlocks, long badlyDistributedBlocks) {
     this(lowRedundancyBlocks, corruptBlocks, missingBlocks,
         missingReplicationOneBlocks, bytesInFutureBlocks, pendingDeletionBlocks,
-        null);
+        badlyDistributedBlocks, null);
   }
 
   public ReplicatedBlockStats(long lowRedundancyBlocks,
       long corruptBlocks, long missingBlocks,
       long missingReplicationOneBlocks, long bytesInFutureBlocks,
-      long pendingDeletionBlocks, Long highestPriorityLowRedundancyBlocks) {
+      long pendingDeletionBlocks, long badlyDistributedBlocks,
+      Long highestPriorityLowRedundancyBlocks) {
     this.lowRedundancyBlocks = lowRedundancyBlocks;
     this.corruptBlocks = corruptBlocks;
     this.missingBlocks = missingBlocks;
     this.missingReplicationOneBlocks = missingReplicationOneBlocks;
     this.bytesInFutureBlocks = bytesInFutureBlocks;
     this.pendingDeletionBlocks = pendingDeletionBlocks;
+    this.badlyDistributedBlocks = badlyDistributedBlocks;
     this.highestPriorityLowRedundancyBlocks
         = highestPriorityLowRedundancyBlocks;
   }
@@ -86,6 +89,10 @@ public final class ReplicatedBlockStats {
     return pendingDeletionBlocks;
   }
 
+  public long getBadlyDistributedBlocks() {
+    return badlyDistributedBlocks;
+  }
+
   public boolean hasHighestPriorityLowRedundancyBlocks() {
     return getHighestPriorityLowRedundancyBlocks() != null;
   }
@@ -94,6 +101,7 @@ public final class ReplicatedBlockStats {
     return highestPriorityLowRedundancyBlocks;
   }
 
+
   @Override
   public String toString() {
     StringBuilder statsBuilder = new StringBuilder();
@@ -105,7 +113,8 @@ public final class ReplicatedBlockStats {
             getMissingReplicationOneBlocks())
         .append(", BytesInFutureBlocks=").append(getBytesInFutureBlocks())
         .append(", PendingDeletionBlocks=").append(
-            getPendingDeletionBlocks());
+            getPendingDeletionBlocks())
+        .append(" , badlyDistributedBlocks=").append(getBadlyDistributedBlocks());
     if (hasHighestPriorityLowRedundancyBlocks()) {
         statsBuilder.append(", HighestPriorityLowRedundancyBlocks=").append(
             getHighestPriorityLowRedundancyBlocks());
@@ -127,6 +136,7 @@ public final class ReplicatedBlockStats {
     long missingReplicationOneBlocks = 0;
     long bytesInFutureBlocks = 0;
     long pendingDeletionBlocks = 0;
+    long badlyDistributedBlocks = 0;
     long highestPriorityLowRedundancyBlocks = 0;
     boolean hasHighestPriorityLowRedundancyBlocks = false;
 
@@ -138,6 +148,7 @@ public final class ReplicatedBlockStats {
       missingReplicationOneBlocks += stat.getMissingReplicationOneBlocks();
       bytesInFutureBlocks += stat.getBytesInFutureBlocks();
       pendingDeletionBlocks += stat.getPendingDeletionBlocks();
+      badlyDistributedBlocks += stat.getBadlyDistributedBlocks();
       if (stat.hasHighestPriorityLowRedundancyBlocks()) {
         hasHighestPriorityLowRedundancyBlocks = true;
         highestPriorityLowRedundancyBlocks +=
@@ -147,10 +158,10 @@ public final class ReplicatedBlockStats {
     if (hasHighestPriorityLowRedundancyBlocks) {
       return new ReplicatedBlockStats(lowRedundancyBlocks, corruptBlocks,
           missingBlocks, missingReplicationOneBlocks, bytesInFutureBlocks,
-          pendingDeletionBlocks, highestPriorityLowRedundancyBlocks);
+          pendingDeletionBlocks, badlyDistributedBlocks, highestPriorityLowRedundancyBlocks);
     }
     return new ReplicatedBlockStats(lowRedundancyBlocks, corruptBlocks,
         missingBlocks, missingReplicationOneBlocks, bytesInFutureBlocks,
-        pendingDeletionBlocks);
+        pendingDeletionBlocks, badlyDistributedBlocks);
   }
 }

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelperClient.java

@@ -2037,13 +2037,13 @@ public class PBHelperClient {
       return new ReplicatedBlockStats(res.getLowRedundancy(),
           res.getCorruptBlocks(), res.getMissingBlocks(),
           res.getMissingReplOneBlocks(), res.getBlocksInFuture(),
-          res.getPendingDeletionBlocks(),
+          res.getPendingDeletionBlocks(), res.getBadlyDistributedBlocks(),
           res.getHighestPrioLowRedundancyBlocks());
     }
     return new ReplicatedBlockStats(res.getLowRedundancy(),
         res.getCorruptBlocks(), res.getMissingBlocks(),
         res.getMissingReplOneBlocks(), res.getBlocksInFuture(),
-        res.getPendingDeletionBlocks());
+        res.getBadlyDistributedBlocks(), res.getPendingDeletionBlocks());
   }
 
   public static ECBlockGroupStats convert(
@@ -2052,11 +2052,12 @@ public class PBHelperClient {
       return new ECBlockGroupStats(res.getLowRedundancy(),
           res.getCorruptBlocks(), res.getMissingBlocks(),
           res.getBlocksInFuture(), res.getPendingDeletionBlocks(),
-          res.getHighestPrioLowRedundancyBlocks());
+          res.getBadlyDistributedBlocks(), res.getHighestPrioLowRedundancyBlocks());
     }
     return new ECBlockGroupStats(res.getLowRedundancy(),
         res.getCorruptBlocks(), res.getMissingBlocks(),
-        res.getBlocksInFuture(), res.getPendingDeletionBlocks());
+        res.getBlocksInFuture(), res.getPendingDeletionBlocks(),
+        res.getBadlyDistributedBlocks());
   }
 
   public static DatanodeReportTypeProto convert(DatanodeReportType t) {
@@ -2525,6 +2526,8 @@ public class PBHelperClient {
         replicatedBlockStats.getBytesInFutureBlocks());
     result.setPendingDeletionBlocks(
         replicatedBlockStats.getPendingDeletionBlocks());
+    result.setBadlyDistributedBlocks(
+        replicatedBlockStats.getBadlyDistributedBlocks());
     if (replicatedBlockStats.hasHighestPriorityLowRedundancyBlocks()) {
       result.setHighestPrioLowRedundancyBlocks(
           replicatedBlockStats.getHighestPriorityLowRedundancyBlocks());
@@ -2544,6 +2547,8 @@ public class PBHelperClient {
         ecBlockGroupStats.getBytesInFutureBlockGroups());
     result.setPendingDeletionBlocks(
         ecBlockGroupStats.getPendingDeletionBlocks());
+    result.setBadlyDistributedBlocks(
+        ecBlockGroupStats.getBadlyDistributedBlocks());
     if (ecBlockGroupStats.hasHighestPriorityLowRedundancyBlocks()) {
       result.setHighestPrioLowRedundancyBlocks(
           ecBlockGroupStats.getHighestPriorityLowRedundancyBlocks());

hadoop-hdfs-project/hadoop-hdfs-client/src/main/proto/ClientNamenodeProtocol.proto

@@ -372,6 +372,7 @@ message GetFsReplicatedBlockStatsResponseProto {
   required uint64 blocks_in_future = 5;
   required uint64 pending_deletion_blocks = 6;
   optional uint64 highest_prio_low_redundancy_blocks = 7;
+  required uint64 badly_distributed_blocks = 8;
 
 }
 
@@ -385,6 +386,7 @@ message GetFsECBlockGroupStatsResponseProto {
   required uint64 blocks_in_future = 4;
   required uint64 pending_deletion_blocks = 5;
   optional uint64 highest_prio_low_redundancy_blocks = 6;
+  required uint64 badly_distributed_blocks = 7;
 }
 
 enum DatanodeReportTypeProto {  // type of the datanode report

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/metrics/FederationMBean.java

@@ -326,6 +326,13 @@ public interface FederationMBean {
    */
   long getNumberOfMissingBlocksWithReplicationFactorOne();
 
+  /**
+   * Gets the total number of badly distributed blocks.
+   *
+   * @return the total number of badly distrubted blocks.
+   */
+  long getNumberOfBadlyDistributedBlocks();
+
   /**
    * Gets the total number of replicated low redundancy blocks on the cluster
    * with the highest risk of loss.

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/metrics/NamenodeBeanMetrics.java

@@ -384,6 +384,16 @@ public class NamenodeBeanMetrics
     return 0;
   }
 
+  @Override
+  public long getNumberOfBadlyDistributedBlocks() {
+    try {
+      return getRBFMetrics().getNumberOfBadlyDistributedBlocks();
+    } catch (IOException e) {
+      LOG.debug("Failed to get number of badly distributed blocks", e);
+    }
+    return 0;
+  }
+
   @Override
   public long getHighestPriorityLowRedundancyReplicatedBlocks() {
     try {

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/metrics/RBFMetrics.java

@@ -756,6 +756,12 @@ public class RBFMetrics implements RouterMBean, FederationMBean {
         MembershipStats::getHighestPriorityLowRedundancyReplicatedBlocks);
   }
 
+  @Override
+  public long getNumberOfBadlyDistributedBlocks() {
+    return getNameserviceAggregatedLong(
+        MembershipStats::getNumberOfBadlyDistributedBlocks);
+  }
+
   @Override
   public long getHighestPriorityLowRedundancyECBlocks() {
     return getNameserviceAggregatedLong(

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/MembershipNamenodeResolver.java

@@ -363,6 +363,8 @@ public class MembershipNamenodeResolver
           report.getScheduledReplicationBlocks());
       stats.setNumberOfMissingBlocksWithReplicationFactorOne(
           report.getNumberOfMissingBlocksWithReplicationFactorOne());
+      stats.setNumberOfBadlyDistributedBlocks(
+          report.getNumberOfBadlyDistributedBlocks());
       stats.setHighestPriorityLowRedundancyReplicatedBlocks(
           report.getHighestPriorityLowRedundancyReplicatedBlocks());
       stats.setHighestPriorityLowRedundancyECBlocks(

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/NamenodeStatusReport.java

@@ -73,6 +73,7 @@ public class NamenodeStatusReport {
   private int corruptFilesCount = -1;
   private long scheduledReplicationBlocks = -1;
   private long numberOfMissingBlocksWithReplicationFactorOne = -1;
+  private long numberOfBadlyDistributedBlocks = -1;
   private long highestPriorityLowRedundancyReplicatedBlocks = -1;
   private long highestPriorityLowRedundancyECBlocks = -1;
   private int pendingSPSPaths = -1;
@@ -394,6 +395,7 @@ public class NamenodeStatusReport {
    * @param numCorruptFiles number of corrupt files.
    * @param numOfMissingBlocksWithReplicationFactorOne number of missing
    * blocks with rep one.
+   * @param numOfBadlyDistributedBlocks number of badly distributed blocks
    * @param highestPriorityLowRedundancyRepBlocks number of high priority low
    * redundancy rep blocks.
    * @param highPriorityLowRedundancyECBlocks number of high priority low
@@ -401,11 +403,14 @@ public class NamenodeStatusReport {
    */
   public void setNamenodeInfo(int numCorruptFiles,
       long numOfMissingBlocksWithReplicationFactorOne,
+      long numOfBadlyDistributedBlocks,
       long highestPriorityLowRedundancyRepBlocks,
       long highPriorityLowRedundancyECBlocks) {
     this.corruptFilesCount = numCorruptFiles;
     this.numberOfMissingBlocksWithReplicationFactorOne =
         numOfMissingBlocksWithReplicationFactorOne;
+    this.numberOfBadlyDistributedBlocks =
+        numOfBadlyDistributedBlocks;
     this.highestPriorityLowRedundancyReplicatedBlocks =
         highestPriorityLowRedundancyRepBlocks;
     this.highestPriorityLowRedundancyECBlocks =
@@ -441,6 +446,16 @@ public class NamenodeStatusReport {
     return this.numberOfMissingBlocksWithReplicationFactorOne;
   }
 
+  /**
+   * Gets the total number of badly distributed blocks.
+   *
+   * @return the total number of badly distrubted blocks.
+   */
+  public long getNumberOfBadlyDistributedBlocks() {
+    return this.numberOfBadlyDistributedBlocks;
+  }
+
+
   /**
    * Gets the total number of replicated low redundancy blocks on the cluster
    * with the highest risk of loss.

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/NamenodeHeartbeatService.java

@@ -543,7 +543,8 @@ public class NamenodeHeartbeatService extends PeriodicService {
                 .optLong("NumberOfMissingBlocksWithReplicationFactorOne"),
             jsonObject
                 .optLong("HighestPriorityLowRedundancyReplicatedBlocks"),
-            jsonObject.optLong("HighestPriorityLowRedundancyECBlocks"));
+            jsonObject.optLong("HighestPriorityLowRedundancyECBlocks"),
+            jsonObject.optLong("BadlyDistributedBlocks"));
       }
     }
   }
@@ -608,4 +609,4 @@ public class NamenodeHeartbeatService extends PeriodicService {
     }
     super.serviceStop();
   }
-}
+}

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/records/MembershipStats.java

@@ -122,6 +122,11 @@ public abstract class MembershipStats extends BaseRecord {
 
   public abstract long getNumberOfMissingBlocksWithReplicationFactorOne();
 
+  public abstract void setNumberOfBadlyDistributedBlocks(
+      long blocks);
+
+  public abstract long getNumberOfBadlyDistributedBlocks();
+
   public abstract void setHighestPriorityLowRedundancyReplicatedBlocks(
       long blocks);
 
@@ -171,4 +176,4 @@ public abstract class MembershipStats extends BaseRecord {
     // We don't store this record directly
     return 0;
   }
-}
+}

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/records/impl/pb/MembershipStatsPBImpl.java

@@ -274,6 +274,18 @@ public class MembershipStatsPBImpl extends MembershipStats
         .getNumberOfMissingBlocksWithReplicationFactorOne();
   }
 
+  @Override
+  public void setNumberOfBadlyDistributedBlocks(long blocks) {
+    this.translator.getBuilder()
+        .setBadlyDistributedBlocks(blocks);
+  }
+
+  @Override
+  public long getNumberOfBadlyDistributedBlocks() {
+    return this.translator.getProtoOrBuilder()
+        .getBadlyDistributedBlocks();
+  }
+
   @Override
   public void setHighestPriorityLowRedundancyReplicatedBlocks(long blocks) {
     this.translator.getBuilder()
@@ -307,4 +319,4 @@ public class MembershipStatsPBImpl extends MembershipStats
   public int getPendingSPSPaths() {
     return this.translator.getProtoOrBuilder().getPendingSPSPaths();
   }
-}
+}

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/proto/FederationProtocol.proto

@@ -55,6 +55,7 @@ message NamenodeMembershipStatsRecordProto {
   optional uint64 highestPriorityLowRedundancyReplicatedBlocks = 32;
   optional uint64 HighestPriorityLowRedundancyECBlocks = 33;
   optional uint32 pendingSPSPaths = 34;
+  optional uint64 badlyDistributedBlocks = 35;
 }
 
 message NamenodeMembershipRecordProto {
@@ -321,4 +322,4 @@ message GetDisabledNameservicesRequestProto {
 
 message GetDisabledNameservicesResponseProto {
   repeated string nameServiceIds = 1;
-}
+}

hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/metrics/TestRBFMetrics.java

@@ -295,6 +295,7 @@ public class TestRBFMetrics extends TestMetricsBase {
     int numCorruptsFilesCount = 0;
     long scheduledReplicationBlocks = 0;
     long numberOfMissingBlocksWithReplicationFactorOne = 0;
+    long numberOfBadlyDistributedBlocks = 0;
     long highestPriorityLowRedundancyReplicatedBlocks = 0;
     long highestPriorityLowRedundancyECBlocks = 0;
     long numFiles = 0;
@@ -315,6 +316,7 @@ public class TestRBFMetrics extends TestMetricsBase {
       scheduledReplicationBlocks += stats.getScheduledReplicationBlocks();
       numberOfMissingBlocksWithReplicationFactorOne +=
           stats.getNumberOfMissingBlocksWithReplicationFactorOne();
+      numberOfBadlyDistributedBlocks += stats.getNumberOfBadlyDistributedBlocks();
       highestPriorityLowRedundancyReplicatedBlocks +=
           stats.getHighestPriorityLowRedundancyReplicatedBlocks();
       highestPriorityLowRedundancyECBlocks +=
@@ -342,6 +344,8 @@ public class TestRBFMetrics extends TestMetricsBase {
         bean.getScheduledReplicationBlocks());
     assertEquals(numberOfMissingBlocksWithReplicationFactorOne,
         bean.getNumberOfMissingBlocksWithReplicationFactorOne());
+    assertEquals(numberOfBadlyDistributedBlocks,
+        bean.getNumberOfBadlyDistributedBlocks());
     assertEquals(highestPriorityLowRedundancyReplicatedBlocks,
         bean.getHighestPriorityLowRedundancyReplicatedBlocks());
     assertEquals(highestPriorityLowRedundancyECBlocks,

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

@@ -267,6 +267,11 @@ public class BlockManager implements BlockStatsMXBean {
     return neededReconstruction.getCorruptReplicationOneBlocks();
   }
 
+  /** Used by metrics. */
+  public long getBadlyDistributedBlocks() {
+    return neededReconstruction.getBadlyDistributedBlocks();
+  }
+
   /** Used by metrics. */
   public long getPendingDeletionReplicatedBlocks() {
     return invalidateBlocks.getBlocks();
@@ -777,7 +782,7 @@ public class BlockManager implements BlockStatsMXBean {
     return storagePolicySuite;
   }
 
-  /** get the BlockTokenSecretManager */
+  /** @return get the BlockTokenSecretManager */
   @VisibleForTesting
   public BlockTokenSecretManager getBlockTokenSecretManager() {
     return blockTokenSecretManager;
@@ -5186,6 +5191,11 @@ public class BlockManager implements BlockStatsMXBean {
     return this.neededReconstruction.getCorruptReplicationOneBlockSize();
   }
 
+  public long getBadlyDistributedBlocksCount() {
+    // not locking
+    return this.neededReconstruction.getBadlyDistributedBlocks();
+  }
+
   public long getHighestPriorityReplicatedBlockCount(){
     return this.neededReconstruction.getHighestPriorityReplicatedBlockCount();
   }

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

@@ -93,6 +93,7 @@ class LowRedundancyBlocks implements Iterable<BlockInfo> {
   private final LongAdder corruptReplicationOneBlocks = new LongAdder();
   private final LongAdder lowRedundancyECBlockGroups = new LongAdder();
   private final LongAdder corruptECBlockGroups = new LongAdder();
+  private final LongAdder badlyDistributedBlocks = new LongAdder();
   private final LongAdder highestPriorityLowRedundancyReplicatedBlocks
       = new LongAdder();
   private final LongAdder highestPriorityLowRedundancyECBlocks
@@ -169,6 +170,11 @@ class LowRedundancyBlocks implements Iterable<BlockInfo> {
     return corruptReplicationOneBlocks.longValue();
   }
 
+  /** Return badly distributed block count. */
+  long getBadlyDistributedBlocks() {
+    return badlyDistributedBlocks.longValue();
+  }
+
   /** Return the number of under replicated blocks
    * with the highest priority to recover */
   long getHighestPriorityReplicatedBlockCount() {
@@ -322,6 +328,9 @@ class LowRedundancyBlocks implements Iterable<BlockInfo> {
       if (priLevel == QUEUE_HIGHEST_PRIORITY) {
         highestPriorityLowRedundancyECBlocks.increment();
       }
+      if (priLevel == QUEUE_REPLICAS_BADLY_DISTRIBUTED) {
+        badlyDistributedBlocks.increment();
+      }
     } else {
       lowRedundancyBlocks.increment();
       if (priLevel == QUEUE_WITH_CORRUPT_BLOCKS) {
@@ -333,6 +342,9 @@ class LowRedundancyBlocks implements Iterable<BlockInfo> {
       if (priLevel == QUEUE_HIGHEST_PRIORITY) {
         highestPriorityLowRedundancyReplicatedBlocks.increment();
       }
+      if (priLevel == QUEUE_REPLICAS_BADLY_DISTRIBUTED) {
+        badlyDistributedBlocks.increment();
+      }
     }
   }
 
@@ -409,6 +421,9 @@ class LowRedundancyBlocks implements Iterable<BlockInfo> {
       if (priLevel == QUEUE_HIGHEST_PRIORITY) {
         highestPriorityLowRedundancyECBlocks.decrement();
       }
+      if (priLevel == QUEUE_REPLICAS_BADLY_DISTRIBUTED) {
+        badlyDistributedBlocks.decrement();
+      }
     } else {
       lowRedundancyBlocks.decrement();
       if (priLevel == QUEUE_WITH_CORRUPT_BLOCKS) {
@@ -423,6 +438,9 @@ class LowRedundancyBlocks implements Iterable<BlockInfo> {
       if (priLevel == QUEUE_HIGHEST_PRIORITY) {
         highestPriorityLowRedundancyReplicatedBlocks.decrement();
       }
+      if (priLevel == QUEUE_REPLICAS_BADLY_DISTRIBUTED) {
+        badlyDistributedBlocks.decrement();
+      }
     }
   }
 
@@ -590,4 +608,4 @@ class LowRedundancyBlocks implements Iterable<BlockInfo> {
       }
     };
   }
-}
+}

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java

@@ -4738,6 +4738,12 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
     // not locking
     return blockManager.getMissingReplOneBlocksCount();
   }
+
+  @Metric({"BadlyDistBlocks", "Number of Badly Distributed Blocks"})
+  public long getBadlyDistributedBlocksCount() {
+    // not locking
+    return blockManager.getBadlyDistributedBlocksCount();
+  }
   
   @Metric(value = {"ExpiredHeartbeats", "Number of expired heartbeats"},
       type = Metric.Type.COUNTER)
@@ -4816,7 +4822,8 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
         getCorruptReplicatedBlocks(), getMissingReplicatedBlocks(),
         getMissingReplicationOneBlocks(), getBytesInFutureReplicatedBlocks(),
         getPendingDeletionReplicatedBlocks(),
-        getHighestPriorityLowRedundancyReplicatedBlocks());
+        getHighestPriorityLowRedundancyReplicatedBlocks(),
+        getBadlyDistributedBlocks());
   }
 
   /**
@@ -4829,7 +4836,7 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
     return new ECBlockGroupStats(getLowRedundancyECBlockGroups(),
         getCorruptECBlockGroups(), getMissingECBlockGroups(),
         getBytesInFutureECBlockGroups(), getPendingDeletionECBlocks(),
-        getHighestPriorityLowRedundancyECBlocks());
+        getBadlyDistributedBlocks(), getHighestPriorityLowRedundancyECBlocks());
   }
 
   @Override // FSNamesystemMBean
@@ -5495,6 +5502,12 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
     return blockManager.getMissingReplicationOneBlocks();
   }
 
+  @Override // ReplicatedBlocksMBean
+  @Metric({"BadlyDistributedBlocks", "Number of badly distributed blocks"})
+  public long getBadlyDistributedBlocks() {
+    return blockManager.getBadlyDistributedBlocks();
+  }
+
   @Override // ReplicatedBlocksMBean
   @Metric({"HighestPriorityLowRedundancyReplicatedBlocks", "Number of " +
       "replicated blocks which have the highest risk of loss."})
@@ -6652,6 +6665,11 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
     return getMissingReplOneBlocksCount();
   }
 
+  @Override // NameNodeMXBean
+  public long getNumberOfBadlyDistributedBlocks() {
+    return getBadlyDistributedBlocks();
+  }
+
   @Override // NameNodeMXBean
   public int getThreads() {
     return ManagementFactory.getThreadMXBean().getThreadCount();

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeMXBean.java

@@ -171,6 +171,14 @@ public interface NameNodeMXBean {
    */
   long getNumberOfMissingBlocksWithReplicationFactorOne();
 
+
+  /**
+   * Gets the total number of badly distributed blocks.
+   *
+   * @return the total number of badly distrubted blocks.
+   */
+  long getNumberOfBadlyDistributedBlocks();
+
   /**
    * Gets the total number of replicated low redundancy blocks on the cluster
    * with the highest risk of loss.

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/metrics/ReplicatedBlocksMBean.java

@@ -51,6 +51,11 @@ public interface ReplicatedBlocksMBean {
    */
   long getMissingReplicationOneBlocks();
 
+  /**
+    * Return count of badly distributed blocks
+   */
+  long getBadlyDistributedBlocks();
+
   /**
    * Return total bytes of future blocks.
    */

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/DFSAdmin.java

@@ -123,7 +123,7 @@ public class DFSAdmin extends FsShell {
   private static final Logger LOG = LoggerFactory.getLogger(DFSAdmin.class);
 
   /**
-   * An abstract class for the execution of a file system command
+   * An abstract class for the execution of a file system command.
    */
   abstract private static class DFSAdminCommand extends Command {
     protected DistributedFileSystem dfs;
@@ -554,6 +554,8 @@ public class DFSAdmin extends FsShell {
         replicatedBlockStats.getMissingReplicaBlocks());
     System.out.println("\tMissing blocks (with replication factor 1): " +
         replicatedBlockStats.getMissingReplicationOneBlocks());
+    System.out.println("\tBadly Distributed Blocks: " +
+        replicatedBlockStats.getBadlyDistributedBlocks());
     if (replicatedBlockStats.hasHighestPriorityLowRedundancyBlocks()) {
       System.out.println("\tLow redundancy blocks with highest priority " +
           "to recover: " +
@@ -571,6 +573,8 @@ public class DFSAdmin extends FsShell {
         ecBlockGroupStats.getCorruptBlockGroups());
     System.out.println("\tMissing block groups: " +
         ecBlockGroupStats.getMissingBlockGroups());
+    System.out.println("\tBadly Distributed Blocks: " +
+        ecBlockGroupStats.getBadlyDistributedBlocks());
     if (ecBlockGroupStats.hasHighestPriorityLowRedundancyBlocks()) {
       System.out.println("\tLow redundancy blocks with highest priority " +
           "to recover: " +

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

@@ -75,7 +75,7 @@ public class TestLowRedundancyBlockQueues {
       int lowRedundancyReplicaCount, int corruptReplicaCount,
       int corruptReplicationOneCount, int lowRedundancyStripedCount,
       int corruptStripedCount, int highestPriorityReplicatedBlockCount,
-      int highestPriorityECBlockCount) {
+      int highestPriorityECBlockCount, int badlyDistributedBlockCount) {
     assertEquals("Low redundancy replica count incorrect!",
         lowRedundancyReplicaCount, queues.getLowRedundancyBlocks());
     assertEquals("Corrupt replica count incorrect!",
@@ -93,6 +93,8 @@ public class TestLowRedundancyBlockQueues {
     assertEquals("LowRedundancyBlocks queue size incorrect!",
         (lowRedundancyReplicaCount + corruptReplicaCount +
         lowRedundancyStripedCount + corruptStripedCount), queues.size());
+    assertEquals("Badly Distributed Blocks queue size incorrect!",
+        badlyDistributedBlockCount, queues.getBadlyDistributedBlocks());
     assertEquals("Highest priority replicated low redundancy " +
             "blocks count is incorrect!",
         highestPriorityReplicatedBlockCount,
@@ -177,50 +179,58 @@ public class TestLowRedundancyBlockQueues {
     BlockInfo block_very_low_redundancy = genBlockInfo(3);
     BlockInfo block_corrupt = genBlockInfo(4);
     BlockInfo block_corrupt_repl_one = genBlockInfo(5);
+    BlockInfo blockBadlyDistributed = genBlockInfo(6);
 
     // Add a block with a single entry
     assertAdded(queues, block1, 1, 0, 3);
     assertInLevel(queues, block1, LowRedundancyBlocks.QUEUE_HIGHEST_PRIORITY);
-    verifyBlockStats(queues, 1, 0, 0, 0, 0, 1, 0);
+    verifyBlockStats(queues, 1, 0, 0, 0, 0, 1, 0, 0);
 
     // Repeated additions fail
     assertFalse(queues.add(block1, 1, 0, 0, 3));
-    verifyBlockStats(queues, 1, 0, 0, 0, 0, 1, 0);
+    verifyBlockStats(queues, 1, 0, 0, 0, 0, 1, 0, 0);
 
     // Add a second block with two replicas
     assertAdded(queues, block2, 2, 0, 3);
     assertInLevel(queues, block2, LowRedundancyBlocks.QUEUE_LOW_REDUNDANCY);
-    verifyBlockStats(queues, 2, 0, 0, 0, 0, 1, 0);
+    verifyBlockStats(queues, 2, 0, 0, 0, 0, 1, 0, 0);
 
     // Now try to add a block that is corrupt
     assertAdded(queues, block_corrupt, 0, 0, 3);
     assertInLevel(queues, block_corrupt,
                   LowRedundancyBlocks.QUEUE_WITH_CORRUPT_BLOCKS);
-    verifyBlockStats(queues, 2, 1, 0, 0, 0, 1, 0);
+    verifyBlockStats(queues, 2, 1, 0, 0, 0, 1, 0, 0);
 
     // Insert a very insufficiently redundancy block
     assertAdded(queues, block_very_low_redundancy, 4, 0, 25);
     assertInLevel(queues, block_very_low_redundancy,
                   LowRedundancyBlocks.QUEUE_VERY_LOW_REDUNDANCY);
-    verifyBlockStats(queues, 3, 1, 0, 0, 0, 1, 0);
+    verifyBlockStats(queues, 3, 1, 0, 0, 0, 1, 0, 0);
 
     // Insert a corrupt block with replication factor 1
     assertAdded(queues, block_corrupt_repl_one, 0, 0, 1);
-    verifyBlockStats(queues, 3, 2, 1, 0, 0, 1, 0);
+    verifyBlockStats(queues, 3, 2, 1, 0, 0, 1, 0, 0);
 
     // Bump up the expected count for corrupt replica one block from 1 to 3
     queues.update(block_corrupt_repl_one, 0, 0, 0, 3, 0, 2);
-    verifyBlockStats(queues, 3, 2, 0, 0, 0, 1, 0);
+    verifyBlockStats(queues, 3, 2, 0, 0, 0, 1, 0, 0);
 
     // Reduce the expected replicas to 1
     queues.update(block_corrupt, 0, 0, 0, 1, 0, -2);
-    verifyBlockStats(queues, 3, 2, 1, 0, 0, 1, 0);
+    verifyBlockStats(queues, 3, 2, 1, 0, 0, 1, 0, 0);
     queues.update(block_very_low_redundancy, 0, 0, 0, 1, -4, -24);
-    verifyBlockStats(queues, 2, 3, 2, 0, 0, 1, 0);
+    verifyBlockStats(queues, 2, 3, 2, 0, 0, 1, 0, 0);
 
     // Reduce the expected replicas to 1 for block1
     queues.update(block1, 1, 0, 0, 1, 0, 0);
-    verifyBlockStats(queues, 2, 3, 2, 0, 0, 0, 0);
+    // expect 1 badly distributed block
+    verifyBlockStats(queues, 2, 3, 2, 0, 0, 0, 0, 1);
+
+    // insert a block with too many replicas to make badly distributed
+    assertAdded(queues, blockBadlyDistributed, 2, 0, 1);
+    assertInLevel(queues, blockBadlyDistributed,
+        LowRedundancyBlocks.QUEUE_REPLICAS_BADLY_DISTRIBUTED);
+    verifyBlockStats(queues, 3, 3, 2, 0, 0, 0, 0, 2);
   }
 
   @Test
@@ -230,12 +240,12 @@ public class TestLowRedundancyBlockQueues {
     assertAdded(queues, corruptBlock, 0, 0, 3);
     assertInLevel(queues, corruptBlock,
         LowRedundancyBlocks.QUEUE_WITH_CORRUPT_BLOCKS);
-    verifyBlockStats(queues, 0, 1, 0, 0, 0, 0, 0);
+    verifyBlockStats(queues, 0, 1, 0, 0, 0, 0, 0, 0);
 
     // Remove with wrong priority
     queues.remove(corruptBlock, LowRedundancyBlocks.QUEUE_LOW_REDUNDANCY);
     // Verify the number of corrupt block is decremented
-    verifyBlockStats(queues, 0, 0, 0, 0, 0, 0, 0);
+    verifyBlockStats(queues, 0, 0, 0, 0, 0, 0, 0, 0);
   }
 
   @Test
@@ -271,17 +281,17 @@ public class TestLowRedundancyBlockQueues {
         assertInLevel(queues, block,
             LowRedundancyBlocks.QUEUE_LOW_REDUNDANCY);
       }
-      verifyBlockStats(queues, 0, 0, 0, numUR, 0, 0, 1);
+      verifyBlockStats(queues, 0, 0, 0, numUR, 0, 0, 1, 0);
     }
 
     // add a corrupted block
     BlockInfo block_corrupt = genStripedBlockInfo(-10, numBytes);
     assertEquals(numCorrupt, queues.getCorruptBlockSize());
-    verifyBlockStats(queues, 0, 0, 0, numUR, numCorrupt, 0, 1);
+    verifyBlockStats(queues, 0, 0, 0, numUR, numCorrupt, 0, 1, 0);
 
     assertAdded(queues, block_corrupt, dataBlkNum - 1, 0, groupSize);
     numCorrupt++;
-    verifyBlockStats(queues, 0, 0, 0, numUR, numCorrupt, 0, 1);
+    verifyBlockStats(queues, 0, 0, 0, numUR, numCorrupt, 0, 1, 0);
 
     assertInLevel(queues, block_corrupt,
         LowRedundancyBlocks.QUEUE_WITH_CORRUPT_BLOCKS);

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/metrics/TestNameNodeMetrics.java

@@ -460,6 +460,9 @@ public class TestNameNodeMetrics {
     assertEquals("Missing blocks with replication factor one not matching!",
         namesystem.getMissingReplOneBlocksCount(),
         namesystem.getMissingReplicationOneBlocks());
+    assertEquals("Blocks with badly distributed are not matching!",
+        namesystem.getBadlyDistributedBlocksCount(),
+        namesystem.getBadlyDistributedBlocks());
     assertEquals("Bytes in future blocks metrics not matching!",
         namesystem.getBytesInFuture(),
         namesystem.getBytesInFutureReplicatedBlocks() +
@@ -510,6 +513,7 @@ public class TestNameNodeMetrics {
     assertGauge("LowRedundancyReplicatedBlocks", 1L, rb);
     assertGauge("CorruptReplicatedBlocks", 1L, rb);
     assertGauge("HighestPriorityLowRedundancyReplicatedBlocks", 1L, rb);
+    assertGauge("BadlyDistributedBlocks", 0L, rb);
     // Verify striped blocks metrics
     assertGauge("LowRedundancyECBlockGroups", 0L, rb);
     assertGauge("CorruptECBlockGroups", 0L, rb);
@@ -537,6 +541,7 @@ public class TestNameNodeMetrics {
     assertGauge("LowRedundancyReplicatedBlocks", 0L, rb);
     assertGauge("CorruptReplicatedBlocks", 0L, rb);
     assertGauge("HighestPriorityLowRedundancyReplicatedBlocks", 0L, rb);
+    assertGauge("BadlyDistributedBlocks", 0L, rb);
     // Verify striped blocks metrics
     assertGauge("LowRedundancyECBlockGroups", 0L, rb);
     assertGauge("CorruptECBlockGroups", 0L, rb);
@@ -602,6 +607,7 @@ public class TestNameNodeMetrics {
     assertGauge("LowRedundancyReplicatedBlocks", 0L, rb);
     assertGauge("CorruptReplicatedBlocks", 0L, rb);
     assertGauge("HighestPriorityLowRedundancyReplicatedBlocks", 0L, rb);
+    assertGauge("BadlyDistributedBlocks", 0L, rb);
     // Verify striped block groups metrics
     assertGauge("LowRedundancyECBlockGroups", 1L, rb);
     assertGauge("CorruptECBlockGroups", 1L, rb);
@@ -695,6 +701,7 @@ public class TestNameNodeMetrics {
     assertGauge("MissingReplOneBlocks", 1L, rb);
     assertGauge("HighestPriorityLowRedundancyReplicatedBlocks", 0L, rb);
     assertGauge("HighestPriorityLowRedundancyECBlocks", 0L, rb);
+    assertGauge("BadlyDistributedBlocks", 0L, rb);
     fs.delete(file, true);
     waitForDnMetricValue(NS_METRICS, "UnderReplicatedBlocks", 0L);
   }