Merge branch 'apache:trunk' into YARN-7707-V2

Author: slfan1989

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/impl/prefetch/SingleFilePerBlockCache.java

@@ -271,12 +271,12 @@ public void get(int blockNumber, ByteBuffer buffer) throws IOException {
   protected int readFile(Path path, ByteBuffer buffer) throws IOException {
     int numBytesRead = 0;
     int numBytes;
-    FileChannel channel = FileChannel.open(path, StandardOpenOption.READ);
-    while ((numBytes = channel.read(buffer)) > 0) {
-      numBytesRead += numBytes;
+    try (FileChannel channel = FileChannel.open(path, StandardOpenOption.READ)) {
+      while ((numBytes = channel.read(buffer)) > 0) {
+        numBytesRead += numBytes;
+      }
+      buffer.limit(buffer.position());
     }
-    buffer.limit(buffer.position());
-    channel.close();
     return numBytesRead;
   }
 
@@ -460,11 +460,11 @@ private void deleteBlockFileAndEvictCache(Entry elementToPurge) {
 
   protected void writeFile(Path path, ByteBuffer buffer) throws IOException {
     buffer.rewind();
-    WritableByteChannel writeChannel = Files.newByteChannel(path, CREATE_OPTIONS);
-    while (buffer.hasRemaining()) {
-      writeChannel.write(buffer);
+    try (WritableByteChannel writeChannel = Files.newByteChannel(path, CREATE_OPTIONS)) {
+      while (buffer.hasRemaining()) {
+        writeChannel.write(buffer);
+      }
     }
-    writeChannel.close();
   }
 
   /**

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/store/LogExactlyOnce.java

@@ -49,4 +49,15 @@ public void error(String format, Object...args) {
       log.error(format, args);
     }
   }
+
+  /**
+   * Log at DEBUG if nothing has been logged yet.
+   * @param format format string
+   * @param args arguments
+   */
+  public void debug(String format, Object...args) {
+    if (!logged.getAndSet(true)) {
+      log.debug(format, args);
+    }
+  }
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSUtil.java

@@ -139,13 +139,20 @@ public static SecureRandom getSecureRandom() {
   public static class ServiceComparator implements Comparator<DatanodeInfo> {
     @Override
     public int compare(DatanodeInfo a, DatanodeInfo b) {
-      // Decommissioned nodes will still be moved to the end of the list
+      // Decommissioned nodes will be moved to the end of the list.
       if (a.isDecommissioned()) {
         return b.isDecommissioned() ? 0 : 1;
       } else if (b.isDecommissioned()) {
         return -1;
       }
 
+      // Decommissioning nodes will be placed before decommissioned nodes.
+      if (a.isDecommissionInProgress()) {
+        return b.isDecommissionInProgress() ? 0 : 1;
+      } else if (b.isDecommissionInProgress()) {
+        return -1;
+      }
+
       // ENTERING_MAINTENANCE nodes should be after live nodes.
       if (a.isEnteringMaintenance()) {
         return b.isEnteringMaintenance() ? 0 : 1;
@@ -159,9 +166,9 @@ public int compare(DatanodeInfo a, DatanodeInfo b) {
 
   /**
    * Comparator for sorting DataNodeInfo[] based on
-   * slow, stale, entering_maintenance and decommissioned states.
+   * slow, stale, entering_maintenance, decommissioning and decommissioned states.
    * Order: live {@literal ->} slow {@literal ->} stale {@literal ->}
-   * entering_maintenance {@literal ->} decommissioned
+   * entering_maintenance {@literal ->} decommissioning {@literal ->} decommissioned
    */
   @InterfaceAudience.Private 
   public static class StaleAndSlowComparator extends ServiceComparator {

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

@@ -147,6 +147,14 @@ public byte[] getBlockIndices() {
     return indices;
   }
 
+  public byte[] getBlockIndicesForSpecifiedStorages(List<Integer> storageIdx) {
+    byte[] indices = new byte[storageIdx.size()];
+    for (int i = 0; i < indices.length; i++) {
+      indices[i] = BlockIdManager.getBlockIndex(replicas[storageIdx.get(i)]);
+    }
+    return indices;
+  }
+
   public int getNumExpectedLocations() {
     return replicas.length;
   }

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

@@ -511,7 +511,9 @@ public DatanodeStatistics getDatanodeStatistics() {
   }
 
   private boolean isInactive(DatanodeInfo datanode) {
-    return datanode.isDecommissioned() || datanode.isEnteringMaintenance() ||
+    return datanode.isDecommissioned() ||
+        datanode.isDecommissionInProgress() ||
+        datanode.isEnteringMaintenance() ||
         (avoidStaleDataNodesForRead && datanode.isStale(staleInterval));
   }
 
@@ -540,7 +542,7 @@ public int getMaxSlowpeerCollectNodes() {
   /**
    * Sort the non-striped located blocks by the distance to the target host.
    *
-   * For striped blocks, it will only move decommissioned/stale/slow
+   * For striped blocks, it will only move decommissioned/decommissioning/stale/slow
    * nodes to the bottom. For example, assume we have storage list:
    * d0, d1, d2, d3, d4, d5, d6, d7, d8, d9
    * mapping to block indices:
@@ -570,7 +572,7 @@ public void sortLocatedBlocks(final String targetHost,
   }
 
   /**
-   * Move decommissioned/entering_maintenance/stale/slow
+   * Move decommissioned/decommissioning/entering_maintenance/stale/slow
    * datanodes to the bottom. After sorting it will
    * update block indices and block tokens respectively.
    *
@@ -588,7 +590,8 @@ private void sortLocatedStripedBlock(final LocatedBlock lb,
       locToIndex.put(di[i], lsb.getBlockIndices()[i]);
       locToToken.put(di[i], lsb.getBlockTokens()[i]);
     }
-    // Move decommissioned/stale datanodes to the bottom
+    // Arrange the order of datanodes as follows:
+    // live(in-service) -> stale -> entering_maintenance -> decommissioning -> decommissioned
     Arrays.sort(di, comparator);
 
     // must update cache since we modified locations array
@@ -602,7 +605,7 @@ private void sortLocatedStripedBlock(final LocatedBlock lb,
   }
 
   /**
-   * Move decommissioned/entering_maintenance/stale/slow
+   * Move decommissioned/decommissioning/entering_maintenance/stale/slow
    * datanodes to the bottom. Also, sort nodes by network
    * distance.
    *
@@ -634,8 +637,8 @@ private void sortLocatedBlock(final LocatedBlock lb, String targetHost,
     }
 
     DatanodeInfoWithStorage[] di = lb.getLocations();
-    // Move decommissioned/entering_maintenance/stale/slow
-    // datanodes to the bottom
+    // Arrange the order of datanodes as follows:
+    // live(in-service) -> stale -> entering_maintenance -> decommissioning -> decommissioned
     Arrays.sort(di, comparator);
 
     // Sort nodes by network distance only for located blocks
@@ -1717,9 +1720,11 @@ private BlockRecoveryCommand getBlockRecoveryCommand(String blockPoolId,
       // Skip stale nodes during recovery
       final List<DatanodeStorageInfo> recoveryLocations =
           new ArrayList<>(storages.length);
-      for (DatanodeStorageInfo storage : storages) {
-        if (!storage.getDatanodeDescriptor().isStale(staleInterval)) {
-          recoveryLocations.add(storage);
+      final List<Integer> storageIdx = new ArrayList<>(storages.length);
+      for (int i = 0; i < storages.length; ++i) {
+        if (!storages[i].getDatanodeDescriptor().isStale(staleInterval)) {
+          recoveryLocations.add(storages[i]);
+          storageIdx.add(i);
         }
       }
       // If we are performing a truncate recovery than set recovery fields
@@ -1752,7 +1757,8 @@ private BlockRecoveryCommand getBlockRecoveryCommand(String blockPoolId,
         rBlock = new RecoveringBlock(primaryBlock, recoveryInfos,
             uc.getBlockRecoveryId());
         if (b.isStriped()) {
-          rBlock = new RecoveringStripedBlock(rBlock, uc.getBlockIndices(),
+          rBlock = new RecoveringStripedBlock(rBlock,
+              uc.getBlockIndicesForSpecifiedStorages(storageIdx),
               ((BlockInfoStriped) b).getErasureCodingPolicy());
         }
       }

hadoop-hdfs-project/hadoop-hdfs/src/site/markdown/HDFSCommands.md

@@ -452,6 +452,7 @@ Usage:
 
       hdfs dfsrouteradmin
           [-add <source> <nameservice1, nameservice2, ...> <destination> [-readonly] [-faulttolerant] [-order HASH|LOCAL|RANDOM|HASH_ALL] -owner <owner> -group <group> -mode <mode>]
+          [-addAll <source1> <nameservice1,nameservice2,...> <destination1> [-readonly] [-faulttolerant][-order HASH|LOCAL|RANDOM|HASH_ALL|SPACE] -owner <owner1> -group <group1> -mode <mode1> , <source2> <nameservice1,nameservice2,...> <destination2> [-readonly] [-faulttolerant] [-order HASH|LOCAL|RANDOM|HASH_ALL|SPACE] -owner <owner2> -group <group2> -mode <mode2> , ...]
           [-update <source> [<nameservice1, nameservice2, ...> <destination>] [-readonly true|false] [-faulttolerant true|false] [-order HASH|LOCAL|RANDOM|HASH_ALL] -owner <owner> -group <group> -mode <mode>]
           [-rm <source>]
           [-ls [-d] <path>]
@@ -460,6 +461,7 @@ Usage:
           [-setStorageTypeQuota <path> -storageType <storage type> <quota in bytes or quota size string>]
           [-clrQuota <path>]
           [-clrStorageTypeQuota <path>]
+          [-dumpState]
           [-safemode enter | leave | get]
           [-nameservice disable | enable <nameservice>]
           [-getDisabledNameservices]
@@ -517,7 +519,7 @@ Runs the diskbalancer CLI. See [HDFS Diskbalancer](./HDFSDiskbalancer.html) for
 Usage:
 
        hdfs ec [generic options]
-         [-setPolicy -policy <policyName> -path <path>]
+         [-setPolicy -policy <policyName> -path <path> [-replicate]]
          [-getPolicy -path <path>]
          [-unsetPolicy -path <path>]
          [-listPolicies]
@@ -718,11 +720,13 @@ Recover the lease on the specified path. The path must reside on an HDFS file sy
 
 ### `verifyEC`
 
-Usage: `hdfs debug verifyEC -file <file>`
+Usage: `hdfs debug verifyEC -file <file> [-blockId <blk_Id>] [-skipFailureBlocks]`
 
 | COMMAND\_OPTION | Description |
 |:---- |:---- |
 | [`-file` *EC-file*] | HDFS EC file to be verified. |
+| [`-blockId` *blk_Id*] | Specify the blk_Id to verify a block group of the file. |
+| [`-skipFailureBlocks`] | Specify will skip any block group failures during verify and continues verify all block groups of the file, the default is not to skip failure blocks . |
 
 Verify the correctness of erasure coding on an erasure coded file.
 

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestLeaseRecoveryStriped.java

@@ -30,6 +30,10 @@
 import org.apache.hadoop.hdfs.server.datanode.BlockRecoveryWorker;
 import org.apache.hadoop.hdfs.server.datanode.DataNode;
 import org.apache.hadoop.hdfs.util.StripedBlockUtil;
+import org.apache.hadoop.hdfs.protocol.LocatedBlock;
+import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeDescriptor;
+import org.apache.hadoop.hdfs.server.datanode.DataNodeTestUtils;
+import org.apache.hadoop.hdfs.server.datanode.fsdataset.impl.TestInterDatanodeProtocol;
 import org.apache.hadoop.io.IOUtils;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.test.GenericTestUtils;
@@ -188,6 +192,62 @@ public void testLeaseRecovery() throws Exception {
     }
   }
 
+  /**
+   * Test lease recovery for EC policy when one internal block located on
+   * stale datanode.
+   */
+  @Test
+  public void testLeaseRecoveryWithStaleDataNode() {
+    LOG.info("blockLengthsSuite: " +
+        Arrays.toString(blockLengthsSuite));
+    long staleInterval = conf.getLong(
+        DFSConfigKeys.DFS_NAMENODE_STALE_DATANODE_INTERVAL_KEY,
+        DFSConfigKeys.DFS_NAMENODE_STALE_DATANODE_INTERVAL_DEFAULT);
+
+    for (int i = 0; i < blockLengthsSuite.length; i++) {
+      BlockLengths blockLengths = blockLengthsSuite[i];
+      try {
+        writePartialBlocks(blockLengths.getBlockLengths());
+
+        // Get block info for the last block and mark corresponding datanode
+        // as stale.
+        LocatedBlock locatedblock =
+            TestInterDatanodeProtocol.getLastLocatedBlock(
+                dfs.dfs.getNamenode(), p.toString());
+        DatanodeInfo firstDataNode = locatedblock.getLocations()[0];
+        DatanodeDescriptor dnDes = cluster.getNameNode().getNamesystem()
+            .getBlockManager().getDatanodeManager()
+            .getDatanode(firstDataNode);
+        DataNodeTestUtils.setHeartbeatsDisabledForTests(
+            cluster.getDataNode(dnDes.getIpcPort()), true);
+        DFSTestUtil.resetLastUpdatesWithOffset(dnDes, -(staleInterval + 1));
+
+        long[] longArray = new long[blockLengths.getBlockLengths().length - 1];
+        for (int j = 0; j < longArray.length; ++j) {
+          longArray[j] = blockLengths.getBlockLengths()[j + 1];
+        }
+        int safeLength = (int) StripedBlockUtil.getSafeLength(ecPolicy,
+            longArray);
+        int checkDataLength = Math.min(testFileLength, safeLength);
+        recoverLease();
+        List<Long> oldGS = new ArrayList<>();
+        oldGS.add(1001L);
+        StripedFileTestUtil.checkData(dfs, p, checkDataLength,
+            new ArrayList<>(), oldGS, blockGroupSize);
+
+        DataNodeTestUtils.setHeartbeatsDisabledForTests(
+            cluster.getDataNode(dnDes.getIpcPort()), false);
+        DFSTestUtil.resetLastUpdatesWithOffset(dnDes, 0);
+
+      } catch (Throwable e) {
+        String msg = "failed testCase at i=" + i + ", blockLengths="
+            + blockLengths + "\n"
+            + StringUtils.stringifyException(e);
+        Assert.fail(msg);
+      }
+    }
+  }
+
   @Test
   public void testSafeLength() {
     checkSafeLength(0, 0); // Length of: 0