HDFS-17218. Modify patch based on comments

Author: haiyang1987

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

@@ -30,6 +30,7 @@
 import java.util.BitSet;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -50,6 +51,7 @@
 import java.util.concurrent.ConcurrentLinkedQueue;
 
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
 import javax.management.ObjectName;
 
 import org.apache.hadoop.HadoopIllegalArgumentException;
@@ -3098,15 +3100,18 @@ void processTimedOutExcessBlocks() {
     long now = Time.monotonicNow();
     int processed = 0;
     try {
-      Iterator<Map.Entry<String, LightWeightHashSet<ExcessBlockInfo>>> iter =
+      Iterator<Map.Entry<String, LightWeightHashSet<Block>>> iter =
           excessRedundancyMap.getExcessRedundancyMap().entrySet().iterator();
       while (iter.hasNext() && processed < excessRedundancyTimeoutCheckLimit) {
-        Map.Entry<String, LightWeightHashSet<ExcessBlockInfo>> entry = iter.next();
+        Map.Entry<String, LightWeightHashSet<Block>> entry = iter.next();
         String datanodeUuid = entry.getKey();
-        LightWeightHashSet<ExcessBlockInfo> blocks = entry.getValue();
-        List<ExcessBlockInfo> sortedBlocks = new ArrayList<>(blocks);
+        LightWeightHashSet<Block> blocks = entry.getValue();
         // Sort blocks by timestamp in descending order.
-        Collections.sort(sortedBlocks);
+        List<ExcessBlockInfo> sortedBlocks = blocks.stream()
+            .filter(block -> block instanceof ExcessBlockInfo)
+            .map(block -> (ExcessBlockInfo) block)
+            .sorted(Comparator.comparingLong(ExcessBlockInfo::getTimeStamp))
+            .collect(Collectors.toList());
 
         for (ExcessBlockInfo excessBlockInfo : sortedBlocks) {
           if (processed >= excessRedundancyTimeoutCheckLimit) {
@@ -3132,11 +3137,11 @@ void processTimedOutExcessBlocks() {
             DatanodeDescriptor datanodeDescriptor = datanodeStorageInfo.getDatanodeDescriptor();
             if (datanodeDescriptor.getDatanodeUuid().equals(datanodeUuid) &&
                 datanodeStorageInfo.getState().equals(State.NORMAL)) {
-              final Block block = getBlockOnStorage(blockInfo, datanodeStorageInfo);
-              if (!containsInvalidateBlock(datanodeDescriptor, block)) {
-                addToInvalidates(block, datanodeDescriptor);
+              final Block b = getBlockOnStorage(blockInfo, datanodeStorageInfo);
+              if (!containsInvalidateBlock(datanodeDescriptor, b)) {
+                addToInvalidates(b, datanodeDescriptor);
                 LOG.debug("Excess block timeout ({}, {}) is added to invalidated.",
-                    block, datanodeDescriptor);
+                    b, datanodeDescriptor);
               }
               excessBlockInfo.setTimeStamp();
               break;

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

@@ -21,6 +21,7 @@
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicLong;
 
+import org.apache.hadoop.hdfs.protocol.Block;
 import org.apache.hadoop.hdfs.server.namenode.NameNode;
 import org.apache.hadoop.hdfs.util.LightWeightHashSet;
 import org.slf4j.Logger;
@@ -37,7 +38,7 @@
 class ExcessRedundancyMap {
   public static final Logger blockLog = NameNode.blockStateChangeLog;
 
-  private final Map<String, LightWeightHashSet<ExcessBlockInfo>> map = new HashMap<>();
+  private final Map<String, LightWeightHashSet<Block>> map = new HashMap<>();
   private final AtomicLong size = new AtomicLong(0L);
 
   /**
@@ -52,7 +53,7 @@ long size() {
    */
   @VisibleForTesting
   synchronized int getSize4Testing(String dnUuid) {
-    final LightWeightHashSet<ExcessBlockInfo> set = map.get(dnUuid);
+    final LightWeightHashSet<Block> set = map.get(dnUuid);
     return set == null? 0: set.size();
   }
 
@@ -66,7 +67,7 @@ synchronized void clear() {
    *         datanode and the given block?
    */
   synchronized boolean contains(DatanodeDescriptor dn, BlockInfo blk) {
-    final LightWeightHashSet<ExcessBlockInfo> set = map.get(dn.getDatanodeUuid());
+    final LightWeightHashSet<Block> set = map.get(dn.getDatanodeUuid());
     return set != null && set.contains(blk);
   }
 
@@ -77,7 +78,7 @@ synchronized boolean contains(DatanodeDescriptor dn, BlockInfo blk) {
    * @return true if the block is added.
    */
   synchronized boolean add(DatanodeDescriptor dn, BlockInfo blk) {
-    LightWeightHashSet<ExcessBlockInfo> set = map.get(dn.getDatanodeUuid());
+    LightWeightHashSet<Block> set = map.get(dn.getDatanodeUuid());
     if (set == null) {
       set = new LightWeightHashSet<>();
       map.put(dn.getDatanodeUuid(), set);
@@ -97,11 +98,10 @@ synchronized boolean add(DatanodeDescriptor dn, BlockInfo blk) {
    * @return true if the block is removed.
    */
   synchronized boolean remove(DatanodeDescriptor dn, BlockInfo blk) {
-    final LightWeightHashSet<ExcessBlockInfo> set = map.get(dn.getDatanodeUuid());
+    final LightWeightHashSet<Block> set = map.get(dn.getDatanodeUuid());
     if (set == null) {
       return false;
     }
-
     final boolean removed = set.remove(blk);
     if (removed) {
       size.decrementAndGet();
@@ -114,19 +114,20 @@ synchronized boolean remove(DatanodeDescriptor dn, BlockInfo blk) {
     return removed;
   }
 
-  synchronized Map<String, LightWeightHashSet<ExcessBlockInfo>> getExcessRedundancyMap() {
+  synchronized Map<String, LightWeightHashSet<Block>> getExcessRedundancyMap() {
     return map;
   }
 
   /**
    * An object that contains information about a block that is being excess redundancy.
    * It records the timestamp when added excess redundancy map of this block.
    */
-  static class ExcessBlockInfo implements Comparable<ExcessBlockInfo> {
+  static class ExcessBlockInfo extends Block {
     private long timeStamp;
-    private BlockInfo blockInfo;
+    private final BlockInfo blockInfo;
 
     ExcessBlockInfo(BlockInfo blockInfo) {
+      super(blockInfo.getBlockId(), blockInfo.getNumBytes(), blockInfo.getGenerationStamp());
       this.timeStamp = monotonicNow();
       this.blockInfo = blockInfo;
     }
@@ -145,29 +146,12 @@ void setTimeStamp() {
 
     @Override
     public int hashCode() {
-      return blockInfo.hashCode();
+      return super.hashCode();
     }
 
     @Override
     public boolean equals(Object obj) {
-      if (this == obj) {
-        return true;
-      }
-
-      if (obj instanceof ExcessBlockInfo) {
-        ExcessBlockInfo other = (ExcessBlockInfo) obj;
-        return this.blockInfo.equals(other.blockInfo);
-      }
-
-      if (obj instanceof BlockInfo) {
-       return this.blockInfo.equals(obj);
-      }
-      return false;
-    }
-
-    @Override
-    public int compareTo(ExcessBlockInfo o) {
-      return Long.compare(this.timeStamp, o.timeStamp);
+      return super.equals(obj);
     }
   }
 }