HDFS-16638. Add isDebugEnabled check for debug blockLogs in BlockManager. (#4480). Contributed by dzcxzl.

Signed-off-by: Ayush Saxena <ayushsaxena@apache.org>

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

@@ -1893,8 +1893,10 @@ private void markBlockAsCorrupt(BlockToMarkCorrupt b,
       DatanodeStorageInfo storageInfo,
       DatanodeDescriptor node) throws IOException {
     if (b.getStored().isDeleted()) {
+      if(blockLog.isDebugEnabled()) {
         blockLog.debug("BLOCK markBlockAsCorrupt: {} cannot be marked as" +
             " corrupt as it does not belong to any file", b);
+      }
       addToInvalidates(b.getCorrupted(), node);
       return;
     }
@@ -1975,10 +1977,12 @@ private boolean invalidateBlock(BlockToMarkCorrupt b, DatanodeInfo dn,
 
     // Check how many copies we have of the block
     if (nr.replicasOnStaleNodes() > 0 && !deleteCorruptReplicaImmediately) {
+      if(blockLog.isDebugEnabled()) {
         blockLog.debug("BLOCK* invalidateBlocks: postponing " +
             "invalidation of {} on {} because {} replica(s) are located on " +
             "nodes with potentially out-of-date block reports", b, dn,
             nr.replicasOnStaleNodes());
+      }
       postponeBlock(b.getCorrupted());
       return false;
     } else {
@@ -2220,8 +2224,10 @@ BlockReconstructionWork scheduleReconstruction(BlockInfo block,
     int pendingNum = pendingReconstruction.getNumReplicas(block);
     if (hasEnoughEffectiveReplicas(block, numReplicas, pendingNum)) {
       neededReconstruction.remove(block, priority);
+      if(blockLog.isDebugEnabled()) {
         blockLog.debug("BLOCK* Removing {} from neededReconstruction as" +
             " it has enough replicas", block);
+      }
       NameNode.getNameNodeMetrics().incNumTimesReReplicationNotScheduled();
       return null;
     }
@@ -2317,8 +2323,10 @@ boolean validateReconstructionWork(BlockReconstructionWork rw) {
     if (hasEnoughEffectiveReplicas(block, numReplicas, pendingNum)) {
       neededReconstruction.remove(block, priority);
       rw.resetTargets();
+      if(blockLog.isDebugEnabled()) {
         blockLog.debug("BLOCK* Removing {} from neededReconstruction as" +
             " it has enough replicas", block);
+      }
       return false;
     }
 
@@ -2349,8 +2357,10 @@ boolean validateReconstructionWork(BlockReconstructionWork rw) {
     // The reason we use 'pending' is so we can retry
     // reconstructions that fail after an appropriate amount of time.
     pendingReconstruction.increment(block, targets);
+    if(blockLog.isDebugEnabled()) {
       blockLog.debug("BLOCK* block {} is moved from neededReconstruction to "
           + "pendingReconstruction", block);
+    }
 
     int numEffectiveReplicas = numReplicas.liveReplicas() + pendingNum;
     // remove from neededReconstruction
@@ -2896,11 +2906,13 @@ public boolean processReport(final DatanodeID nodeID,
 
     if(blockLog.isDebugEnabled()) {
       for (Block b : invalidatedBlocks) {
+        if(blockLog.isDebugEnabled()) {
           blockLog.debug("BLOCK* processReport 0x{} with lease ID 0x{}: {} on node {} size {} " +
                           "does not belong to any file.", strBlockReportId, fullBrLeaseId, b,
                            node, b.getNumBytes());
         }
       }
+    }
 
     // Log the block report processing stats from Namenode perspective
     final NameNodeMetrics metrics = NameNode.getNameNodeMetrics();
@@ -3056,9 +3068,11 @@ public void markBlockReplicasAsCorrupt(Block oldBlock,
         }
       }
       if (isCorrupt) {
+        if(blockLog.isDebugEnabled()) {
           blockLog.debug("BLOCK* markBlockReplicasAsCorrupt: mark block replica" +
               " {} on {} as corrupt because the dn is not in the new committed " +
               "storage list.", b, storage.getDatanodeDescriptor());
+        }
         markBlockAsCorrupt(b, storage, storage.getDatanodeDescriptor());
       }
     }
@@ -3575,8 +3589,10 @@ private Block addStoredBlock(final BlockInfo block,
     }
     if (storedBlock == null || storedBlock.isDeleted()) {
       // If this block does not belong to anyfile, then we are done.
+      if(blockLog.isDebugEnabled()) {
         blockLog.debug("BLOCK* addStoredBlock: {} on {} size {} but it does not" +
             " belong to any file", block, node, block.getNumBytes());
+      }
 
       // we could add this block to invalidate set of this datanode.
       // it will happen in next block report otherwise.
@@ -3605,10 +3621,12 @@ private Block addStoredBlock(final BlockInfo block,
       corruptReplicas.removeFromCorruptReplicasMap(block, node,
           Reason.GENSTAMP_MISMATCH);
       curReplicaDelta = 0;
+      if(blockLog.isDebugEnabled()) {
         blockLog.debug("BLOCK* addStoredBlock: Redundant addStoredBlock request"
                 + " received for {} on node {} size {}", storedBlock, node,
             storedBlock.getNumBytes());
       }
+    }
 
     // Now check for completion of blocks and safe block count
     NumberReplicas num = countNodes(storedBlock);
@@ -3708,8 +3726,10 @@ private void invalidateCorruptReplicas(BlockInfo blk, Block reported,
           removedFromBlocksMap = false;
         }
       } catch (IOException e) {
+        if(blockLog.isDebugEnabled()) {
           blockLog.debug("invalidateCorruptReplicas error in deleting bad block"
               + " {} on {}", blk, node, e);
+        }
         removedFromBlocksMap = false;
       }
     }
@@ -4158,9 +4178,11 @@ private void processChosenExcessRedundancy(
     //
     final Block blockToInvalidate = getBlockOnStorage(storedBlock, chosen);
     addToInvalidates(blockToInvalidate, chosen.getDatanodeDescriptor());
+    if(blockLog.isDebugEnabled()) {
       blockLog.debug("BLOCK* chooseExcessRedundancies: "
           + "({}, {}) is added to invalidated blocks set", chosen, storedBlock);
     }
+  }
 
   private void removeStoredBlock(DatanodeStorageInfo storageInfo, Block block,
       DatanodeDescriptor node) {
@@ -4220,8 +4242,7 @@ private void removeStaleReplicas(List<ReplicaUnderConstruction> staleReplicas,
     for (ReplicaUnderConstruction r : staleReplicas) {
       removeStoredBlock(block,
           r.getExpectedStorageLocation().getDatanodeDescriptor());
-      NameNode.blockStateChangeLog
-          .debug("BLOCK* Removing stale replica {}" + " of {}", r,
+      blockLog.debug("BLOCK* Removing stale replica {} of {}", r,
               Block.toString(r));
     }
   }
@@ -4350,8 +4371,10 @@ private boolean processAndHandleReportedBlock(
           maxNumBlocksToLog, numBlocksLogged);
     }
     for (Block b : toInvalidate) {
+      if(blockLog.isDebugEnabled()) {
         blockLog.debug("BLOCK* addBlock: block {} on node {} size {} does not " +
             "belong to any file", b, node, b.getNumBytes());
+      }
       addToInvalidates(b, node);
     }
     for (BlockToMarkCorrupt b : toCorrupt) {
@@ -4432,10 +4455,12 @@ private void processIncrementalBlockReport(final DatanodeDescriptor node,
       blockLog.debug("BLOCK* block {}: {} is received from {}",
           rdbi.getStatus(), rdbi.getBlock(), node);
     }
+    if(blockLog.isDebugEnabled()) {
       blockLog.debug("*BLOCK* NameNode.processIncrementalBlockReport: from "
               + "{} receiving: {}, received: {}, deleted: {}", node, receiving,
           received, deleted);
     }
+  }
 
   /**
    * Return the number of nodes hosting a given block, grouped