HDFS-17358. EC: infinite lease recovery caused by the length of RWR e…

…quals to zero.

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

@@ -403,7 +403,7 @@ protected void recover() throws IOException {
 
           if (info != null &&
               info.getGenerationStamp() >= block.getGenerationStamp() &&
-              info.getNumBytes() > 0) {
+              info.getNumBytes() >= 0) {
             final BlockRecord existing = syncBlocks.get(blockId);
             if (existing == null ||
                 info.getNumBytes() > existing.rInfo.getNumBytes()) {
@@ -474,6 +474,13 @@ protected void recover() throws IOException {
       ExtendedBlock newBlock = new ExtendedBlock(bpid, block.getBlockId(),
           safeLength, recoveryId);
       DatanodeProtocolClientSideTranslatorPB nn = getActiveNamenodeForBP(bpid);
+      if (newBlock.getNumBytes() == 0) {
+        nn.commitBlockSynchronization(block, newBlock.getGenerationStamp(),
+            newBlock.getNumBytes(), true, true, newLocs, newStorages);
+        LOG.info("After block recovery, the length of new block is 0. " +
+            "Will remove this block: {} from file.", newBlock);
+        return;
+      }
       nn.commitBlockSynchronization(block, newBlock.getGenerationStamp(),
           newBlock.getNumBytes(), true, false, newLocs, newStorages);
     }