Revert "HDFS-12914. Block report leases cause missing blocks until next report. Contributed by Santosh Marella, He Xiaoqiao."

This reverts commit ae4143a529.

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

@@ -2572,21 +2572,6 @@ public class BlockManager implements BlockStatsMXBean {
     }
   }
 
-  /**
-   * Check block report lease.
-   * @return true if lease exist and not expire
-   */
-  public boolean checkBlockReportLease(BlockReportContext context,
-      final DatanodeID nodeID) throws UnregisteredNodeException {
-    if (context == null) {
-      return true;
-    }
-    DatanodeDescriptor node = datanodeManager.getDatanode(nodeID);
-    final long startTime = Time.monotonicNow();
-    return blockReportLeaseManager.checkLease(node, startTime,
-        context.getLeaseId());
-  }
-
   /**
    * The given storage is reporting all its blocks.
    * Update the (storage{@literal -->}block list) and
@@ -2634,6 +2619,12 @@ public class BlockManager implements BlockStatsMXBean {
         blockReportLeaseManager.removeLease(node);
         return !node.hasStaleStorages();
       }
+      if (context != null) {
+        if (!blockReportLeaseManager.checkLease(node, startTime,
+              context.getLeaseId())) {
+          return false;
+        }
+      }
 
       if (storageInfo.getBlockReportCount() == 0) {
         // The first block report can be processed a lot more efficiently than

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

@@ -45,6 +45,7 @@ import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.Callable;
 
 import com.google.common.collect.Lists;
 
@@ -174,7 +175,6 @@ import org.apache.hadoop.hdfs.server.protocol.NamenodeProtocols;
 import org.apache.hadoop.hdfs.server.protocol.NamenodeRegistration;
 import org.apache.hadoop.hdfs.server.protocol.NamespaceInfo;
 import org.apache.hadoop.hdfs.server.protocol.NodeRegistration;
-import org.apache.hadoop.hdfs.server.protocol.RegisterCommand;
 import org.apache.hadoop.hdfs.server.protocol.RemoteEditLogManifest;
 import org.apache.hadoop.hdfs.server.protocol.SlowDiskReports;
 import org.apache.hadoop.hdfs.server.protocol.SlowPeerReports;
@@ -1591,8 +1591,6 @@ public class NameNodeRpcServer implements NamenodeProtocols {
     }
     final BlockManager bm = namesystem.getBlockManager(); 
     boolean noStaleStorages = false;
-    try {
-      if (bm.checkBlockReportLease(context, nodeReg)) {
     for (int r = 0; r < reports.length; r++) {
       final BlockListAsLongs blocks = reports[r].getBlocks();
       //
@@ -1601,15 +1599,13 @@ public class NameNodeRpcServer implements NamenodeProtocols {
       // call of this loop is the final updated value for noStaleStorage.
       //
       final int index = r;
-          noStaleStorages = bm.runBlockOp(() ->
+      noStaleStorages = bm.runBlockOp(new Callable<Boolean>() {
-            bm.processReport(nodeReg, reports[index].getStorage(),
+        @Override
-                blocks, context));
+        public Boolean call() throws IOException {
+          return bm.processReport(nodeReg, reports[index].getStorage(),
+              blocks, context);
         }
-      }
+      });
-    } catch (UnregisteredNodeException une) {
-      LOG.debug("Datanode {} is attempting to report but not register yet.",
-          nodeReg);
-      return RegisterCommand.REGISTER;
     }
     bm.removeBRLeaseIfNeeded(nodeReg, context);