HDFS-15025. Applying NVDIMM storage media to HDFS (#2189)

* HDFS-15025. Applying NVDIMM storage media to HDFS
Co-authored-by: YaYun-Wang <yywangyayun@163.com>
Co-authored-by: YaYun-Wang <34060507+YaYun-Wang@users.noreply.github.com>

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

@@ -34,13 +34,15 @@ import org.apache.hadoop.util.StringUtils;
 @InterfaceStability.Unstable
 public enum StorageType {
   // sorted by the speed of the storage types, from fast to slow
-  RAM_DISK(true),
-  SSD(false),
-  DISK(false),
-  ARCHIVE(false),
-  PROVIDED(false);
+  RAM_DISK(true, true),
+  NVDIMM(false, true),
+  SSD(false, false),
+  DISK(false, false),
+  ARCHIVE(false, false),
+  PROVIDED(false, false);
 
   private final boolean isTransient;
+  private final boolean isRAM;
 
   public static final StorageType DEFAULT = DISK;
 
@@ -48,14 +50,19 @@ public enum StorageType {
 
   private static final StorageType[] VALUES = values();
 
-  StorageType(boolean isTransient) {
+  StorageType(boolean isTransient, boolean isRAM) {
     this.isTransient = isTransient;
+    this.isRAM = isRAM;
   }
 
   public boolean isTransient() {
     return isTransient;
   }
 
+  public boolean isRAM() {
+    return isRAM;
+  }
+
   public boolean supportTypeQuota() {
     return !isTransient;
   }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/shell/Count.java

@@ -89,7 +89,7 @@ public class Count extends FsCommand {
           "Otherwise, it displays the quota and usage for all the storage \n" +
           "types that support quota. The list of possible storage " +
           "types(case insensitive):\n" +
-          "ram_disk, ssd, disk and archive.\n" +
+          "ram_disk, ssd, disk, archive and nvdimm.\n" +
           "It can also pass the value '', 'all' or 'ALL' to specify all " +
           "the storage types.\n" +
           "The -" + OPTION_QUOTA_AND_USAGE + " option shows the quota and \n" +

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/shell/TestCount.java

@@ -283,6 +283,7 @@ public class TestCount {
     count.processOptions(options);
     String withStorageTypeHeader =
         // <----13---> <-------17------> <----13-----> <------17------->
+        " NVDIMM_QUOTA  REM_NVDIMM_QUOTA " +
         "    SSD_QUOTA     REM_SSD_QUOTA    DISK_QUOTA    REM_DISK_QUOTA " +
         // <----13---> <-------17------>
         "ARCHIVE_QUOTA REM_ARCHIVE_QUOTA PROVIDED_QUOTA REM_PROVIDED_QUOTA " +
@@ -337,6 +338,7 @@ public class TestCount {
     count.processOptions(options);
     String withStorageTypeHeader =
         // <----13---> <-------17------>
+        " NVDIMM_QUOTA  REM_NVDIMM_QUOTA " +
         "    SSD_QUOTA     REM_SSD_QUOTA " +
         "   DISK_QUOTA    REM_DISK_QUOTA " +
         "ARCHIVE_QUOTA REM_ARCHIVE_QUOTA " +
@@ -495,7 +497,7 @@ public class TestCount {
         + "Otherwise, it displays the quota and usage for all the storage \n"
         + "types that support quota. The list of possible storage "
         + "types(case insensitive):\n"
-        + "ram_disk, ssd, disk and archive.\n"
+        + "ram_disk, ssd, disk, archive and nvdimm.\n"
         + "It can also pass the value '', 'all' or 'ALL' to specify all the "
         + "storage types.\n"
         + "The -u option shows the quota and \n"

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsConstants.java

@@ -38,6 +38,8 @@ public final class HdfsConstants {
 
   public static final byte MEMORY_STORAGE_POLICY_ID = 15;
   public static final String MEMORY_STORAGE_POLICY_NAME = "LAZY_PERSIST";
+  public static final byte ALLNVDIMM_STORAGE_POLICY_ID = 14;
+  public static final String ALLNVDIMM_STORAGE_POLICY_NAME = "ALL_NVDIMM";
   public static final byte ALLSSD_STORAGE_POLICY_ID = 12;
   public static final String ALLSSD_STORAGE_POLICY_NAME = "ALL_SSD";
   public static final byte ONESSD_STORAGE_POLICY_ID = 10;
@@ -65,6 +67,7 @@ public final class HdfsConstants {
     HOT(HOT_STORAGE_POLICY_ID),
     ONE_SSD(ONESSD_STORAGE_POLICY_ID),
     ALL_SSD(ALLSSD_STORAGE_POLICY_ID),
+    ALL_NVDIMM(ALLNVDIMM_STORAGE_POLICY_ID),
     LAZY_PERSIST(MEMORY_STORAGE_POLICY_ID);
 
     private byte value;
@@ -86,6 +89,8 @@ public final class HdfsConstants {
         return ONE_SSD;
       case 12:
         return ALL_SSD;
+      case 14:
+        return ALL_NVDIMM;
       case 15:
         return LAZY_PERSIST;
       default:

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelperClient.java

@@ -475,6 +475,8 @@ public class PBHelperClient {
       return StorageTypeProto.RAM_DISK;
     case PROVIDED:
       return StorageTypeProto.PROVIDED;
+    case NVDIMM:
+      return StorageTypeProto.NVDIMM;
     default:
       throw new IllegalStateException(
           "BUG: StorageType not found, type=" + type);
@@ -493,6 +495,8 @@ public class PBHelperClient {
       return StorageType.RAM_DISK;
     case PROVIDED:
       return StorageType.PROVIDED;
+    case NVDIMM:
+      return StorageType.NVDIMM;
     default:
       throw new IllegalStateException(
           "BUG: StorageTypeProto not found, type=" + type);

hadoop-hdfs-project/hadoop-hdfs-client/src/main/proto/hdfs.proto

@@ -221,6 +221,7 @@ enum StorageTypeProto {
   ARCHIVE = 3;
   RAM_DISK = 4;
   PROVIDED = 5;
+  NVDIMM = 6;
 }
 
 /**

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

@@ -63,6 +63,12 @@ public class BlockStoragePolicySuite {
         new StorageType[]{StorageType.DISK},
         new StorageType[]{StorageType.DISK},
         true);    // Cannot be changed on regular files, but inherited.
+    final byte allnvdimmId = HdfsConstants.StoragePolicy.ALL_NVDIMM.value();
+    policies[allnvdimmId] = new BlockStoragePolicy(allnvdimmId,
+        HdfsConstants.StoragePolicy.ALL_NVDIMM.name(),
+        new StorageType[]{StorageType.NVDIMM},
+        new StorageType[]{StorageType.DISK},
+        new StorageType[]{StorageType.DISK});
     final byte allssdId = HdfsConstants.StoragePolicy.ALL_SSD.value();
     policies[allssdId] = new BlockStoragePolicy(allssdId,
         HdfsConstants.StoragePolicy.ALL_SSD.name(),

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

@@ -77,6 +77,9 @@ public interface FsVolumeSpi
   /** Returns true if the volume is NOT backed by persistent storage. */
   boolean isTransientStorage();
 
+  /** Returns true if the volume is backed by RAM storage. */
+  boolean isRAMStorage();
+
   /**
    * Reserve disk space for a block (RBW or Re-replicating)
    * so a writer does not run out of space before the block is full.

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java

@@ -2299,9 +2299,9 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {
               ": volume was not an instance of FsVolumeImpl.");
           return;
         }
-        if (volume.isTransientStorage()) {
-          LOG.warn("Caching not supported on block with id " + blockId +
-              " since the volume is backed by RAM.");
+        if (volume.isRAMStorage()) {
+          LOG.warn("Caching not supported on block with id {} since the " +
+              "volume is backed by {} which is RAM.", blockId, volume);
           return;
         }
         success = true;

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeImpl.java

@@ -193,7 +193,7 @@ public class FsVolumeImpl implements FsVolumeSpi {
   }
 
   protected ThreadPoolExecutor initializeCacheExecutor(File parent) {
-    if (storageType.isTransient()) {
+    if (storageType.isRAM()) {
       return null;
     }
     if (dataset.datanode == null) {
@@ -533,6 +533,11 @@ public class FsVolumeImpl implements FsVolumeSpi {
     return storageType.isTransient();
   }
 
+  @Override
+  public boolean isRAMStorage() {
+    return storageType.isRAM();
+  }
+
   @VisibleForTesting
   public File getFinalizedDir(String bpid) throws IOException {
     return getBlockPoolSlice(bpid).getFinalizedDir();

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/DFSAdmin.java

@@ -239,7 +239,8 @@ public class DFSAdmin extends FsShell {
         "\t\t- DISK\n" +
         "\t\t- SSD\n" +
         "\t\t- ARCHIVE\n" +
-        "\t\t- PROVIDED";
+        "\t\t- PROVIDED\n" +
+        "\t\t- NVDIMM";
 
 
     private StorageType type;
@@ -303,7 +304,8 @@ public class DFSAdmin extends FsShell {
         "\t\t- DISK\n" +
         "\t\t- SSD\n" +
         "\t\t- ARCHIVE\n" +
-        "\t\t- PROVIDED";
+        "\t\t- PROVIDED\n" +
+        "\t\t- NVDIMM";
 
     private long quota; // the quota to be set
     private StorageType type;

hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml

@@ -377,7 +377,7 @@
   <value>0</value>
   <description>Reserved space in bytes per volume. Always leave this much space free for non dfs use.
       Specific storage type based reservation is also supported. The property can be followed with
-      corresponding storage types ([ssd]/[disk]/[archive]/[ram_disk]) for cluster with heterogeneous storage.
+      corresponding storage types ([ssd]/[disk]/[archive]/[ram_disk]/[nvdimm]) for cluster with heterogeneous storage.
       For example, reserved space for RAM_DISK storage can be configured using property
       'dfs.datanode.du.reserved.ram_disk'. If specific storage type reservation is not configured
       then dfs.datanode.du.reserved will be used. Support multiple size unit suffix(case insensitive),
@@ -395,7 +395,7 @@
     when this takes effect. The actual number of bytes reserved will be calculated by using the
     total capacity of the data directory in question. Specific storage type based reservation
     is also supported. The property can be followed with corresponding storage types
-    ([ssd]/[disk]/[archive]/[ram_disk]) for cluster with heterogeneous storage.
+    ([ssd]/[disk]/[archive]/[ram_disk]/[nvdimm]) for cluster with heterogeneous storage.
     For example, reserved percentage space for RAM_DISK storage can be configured using property
     'dfs.datanode.du.reserved.pct.ram_disk'. If specific storage type reservation is not configured
     then dfs.datanode.du.reserved.pct will be used.
@@ -604,7 +604,7 @@
   should store its blocks.  If this is a comma-delimited
   list of directories, then data will be stored in all named
   directories, typically on different devices. The directories should be tagged
-  with corresponding storage types ([SSD]/[DISK]/[ARCHIVE]/[RAM_DISK]) for HDFS
+  with corresponding storage types ([SSD]/[DISK]/[ARCHIVE]/[RAM_DISK]/[NVDIMM]) for HDFS
   storage policies. The default storage type will be DISK if the directory does
   not have a storage type tagged explicitly. Directories that do not exist will
   be created if local filesystem permission allows.

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

@@ -27,7 +27,7 @@ The frameworks provided by Heterogeneous Storage and Archival Storage generalize
 Storage Types and Storage Policies
 ----------------------------------
 
-### Storage Types: ARCHIVE, DISK, SSD and RAM\_DISK
+### Storage Types: ARCHIVE, DISK, SSD, RAM\_DISK and NVDIMM
 
 The first phase of [Heterogeneous Storage (HDFS-2832)](https://issues.apache.org/jira/browse/HDFS-2832) changed datanode storage model from a single storage, which may correspond to multiple physical storage medias, to a collection of storages with each storage corresponding to a physical storage media. It also added the notion of storage types, DISK and SSD, where DISK is the default storage type.
 
@@ -35,7 +35,9 @@ A new storage type *ARCHIVE*, which has high storage density (petabyte of storag
 
 Another new storage type *RAM\_DISK* is added for supporting writing single replica files in memory.
 
-### Storage Policies: Hot, Warm, Cold, All\_SSD, One\_SSD, Lazy\_Persist and Provided
+From Hadoop 3.4, a new storage type *NVDIMM* is added for supporting writing replica files in non-volatile memory that has the capability to hold saved data even if the power is turned off.
+
+### Storage Policies: Hot, Warm, Cold, All\_SSD, One\_SSD, Lazy\_Persist, Provided and All\_NVDIMM
 
 A new concept of storage policies is introduced in order to allow files to be stored in different storage types according to the storage policy.
 
@@ -48,6 +50,7 @@ We have the following storage policies:
 * **One\_SSD** - for storing one of the replicas in SSD. The remaining replicas are stored in DISK.
 * **Lazy\_Persist** - for writing blocks with single replica in memory. The replica is first written in RAM\_DISK and then it is lazily persisted in DISK.
 * **Provided** - for storing data outside HDFS. See also [HDFS Provided Storage](./HdfsProvidedStorage.html).
+* **All\_NVDIMM** - for storing all replicas in NVDIMM.
 
 More formally, a storage policy consists of the following fields:
 
@@ -64,6 +67,7 @@ The following is a typical storage policy table.
 | **Policy** **ID** | **Policy** **Name** | **Block Placement** **(n  replicas)** | **Fallback storages** **for creation** | **Fallback storages** **for replication** |
 |:---- |:---- |:---- |:---- |:---- |
 | 15 | Lazy\_Persist | RAM\_DISK: 1, DISK: *n*-1 | DISK | DISK |
+| 14 | All\_NVDIMM | NVDIMM: *n* | DISK | DISK |
 | 12 | All\_SSD | SSD: *n* | DISK | DISK |
 | 10 | One\_SSD | SSD: 1, DISK: *n*-1 | SSD, DISK | SSD, DISK |
 | 7 | Hot (default) | DISK: *n* | \<none\> | ARCHIVE |
@@ -73,7 +77,7 @@ The following is a typical storage policy table.
 
 Note 1: The Lazy\_Persist policy is useful only for single replica blocks. For blocks with more than one replicas, all the replicas will be written to DISK since writing only one of the replicas to RAM\_DISK does not improve the overall performance.
 
-Note 2: For the erasure coded files with striping layout, the suitable storage policies are All\_SSD, Hot, Cold. So, if user sets the policy for striped EC files other than the mentioned policies, it will not follow that policy while creating or moving block.
+Note 2: For the erasure coded files with striping layout, the suitable storage policies are All\_SSD, Hot, Cold and All\_NVDIMM. So, if user sets the policy for striped EC files other than the mentioned policies, it will not follow that policy while creating or moving block.
 
 ### Storage Policy Resolution
 
@@ -88,13 +92,14 @@ The effective storage policy can be retrieved by the "[`storagepolicies -getStor
 ### Configuration
 
 * **dfs.storage.policy.enabled** - for enabling/disabling the storage policy feature. The default value is `true`.
-* **dfs.storage.default.policy** - Set the default storage policy with the policy name. The default value is `HOT`.  All possible policies are defined in enum StoragePolicy, including `LAZY_PERSIST` `ALL_SSD` `ONE_SSD` `HOT` `WARM` `COLD` and `PROVIDED`.
+* **dfs.storage.default.policy** - Set the default storage policy with the policy name. The default value is `HOT`.  All possible policies are defined in enum StoragePolicy, including `LAZY_PERSIST` `ALL_SSD` `ONE_SSD` `HOT` `WARM` `COLD` `PROVIDED` and `ALL_NVDIMM`.
 * **dfs.datanode.data.dir** - on each data node, the comma-separated storage locations should be tagged with their storage types. This allows storage policies to place the blocks on different storage types according to policy. For example:
 
     1.  A datanode storage location /grid/dn/disk0 on DISK should be configured with `[DISK]file:///grid/dn/disk0`
     2.  A datanode storage location /grid/dn/ssd0 on SSD can should configured with `[SSD]file:///grid/dn/ssd0`
     3.  A datanode storage location /grid/dn/archive0 on ARCHIVE should be configured with `[ARCHIVE]file:///grid/dn/archive0`
     4.  A datanode storage location /grid/dn/ram0 on RAM_DISK should be configured with `[RAM_DISK]file:///grid/dn/ram0`
+    5.  A datanode storage location /grid/dn/nvdimm0 on NVDIMM should be configured with `[NVDIMM]file:///grid/dn/nvdimm0`
 
     The default storage type of a datanode storage location will be DISK if it does not have a storage type tagged explicitly.
 

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

@@ -39,7 +39,7 @@ Quotas are persistent with the fsimage. When starting, if the fsimage is immedia
 Storage Type Quotas
 ------------------
 
-The storage type quota is a hard limit on the usage of specific storage type (SSD, DISK, ARCHIVE) by files in the tree rooted at the directory. It works similar to storage space quota in many aspects but offers fine-grain control over the cluster storage space usage. To set storage type quota on a directory, storage policies must be configured on the directory in order to allow files to be stored in different storage types according to the storage policy. See the [HDFS Storage Policy Documentation](./ArchivalStorage.html) for more information.
+The storage type quota is a hard limit on the usage of specific storage type (SSD, DISK, ARCHIVE, NVDIMM) by files in the tree rooted at the directory. It works similar to storage space quota in many aspects but offers fine-grain control over the cluster storage space usage. To set storage type quota on a directory, storage policies must be configured on the directory in order to allow files to be stored in different storage types according to the storage policy. See the [HDFS Storage Policy Documentation](./ArchivalStorage.html) for more information.
 
 The storage type quota can be combined with the space quotas and name quotas to efficiently manage the cluster storage usage. For example,
 
@@ -96,7 +96,7 @@ Quotas are managed by a set of commands available only to the administrator.
     integer, the directory does not exist or it is a file, or the
     directory would immediately exceed the new quota. The storage type
     specific quota is set when -storageType option is specified. Available
-    storageTypes are DISK,SSD,ARCHIVE,PROVIDED.
+    storageTypes are DISK,SSD,ARCHIVE,PROVIDED,NVDIMM.
 
 *   `hdfs dfsadmin -clrSpaceQuota -storageType <storagetype> <directory>...<directory>`
 
@@ -104,7 +104,7 @@ Quotas are managed by a set of commands available only to the administrator.
     for each directory, with faults reported if the directory does not exist or
     it is a file. It is not a fault if the directory has no storage type quota on
     for storage type specified. The storage type specific quota is cleared when -storageType
-    option is specified. Available storageTypes are DISK,SSD,ARCHIVE,PROVIDED.
+    option is specified. Available storageTypes are DISK,SSD,ARCHIVE,PROVIDED,NVDIMM.
 
 Reporting Command
 -----------------

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

@@ -1163,6 +1163,14 @@ Storage Policy Operations
                        "replicationFallbacks": ["DISK"],
                        "storageTypes": ["SSD"]
                    },
+                   {
+                       "copyOnCreateFile": false,
+                       "creationFallbacks": ["DISK"],
+                       "id": 14,
+                       "name": "ALL_NVDIMM",
+                       "replicationFallbacks": ["DISK"],
+                       "storageTypes": ["NVDIMM"]
+                   },
                    {
                        "copyOnCreateFile": true,
                        "creationFallbacks": ["DISK"],

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

@@ -70,10 +70,11 @@ public class TestBlockStoragePolicy {
   static final EnumSet<StorageType> archive = EnumSet.of(StorageType.ARCHIVE);
   static final EnumSet<StorageType> disk = EnumSet.of(StorageType.DISK);
   static final EnumSet<StorageType> ssd = EnumSet.of(StorageType.SSD);
+  static final EnumSet<StorageType> nvdimm = EnumSet.of(StorageType.NVDIMM);
   static final EnumSet<StorageType> disk_archive = EnumSet.of(StorageType.DISK,
       StorageType.ARCHIVE);
   static final EnumSet<StorageType> all = EnumSet.of(StorageType.SSD,
-      StorageType.DISK, StorageType.ARCHIVE);
+      StorageType.DISK, StorageType.ARCHIVE, StorageType.NVDIMM);
 
   static final long FILE_LEN = 1024;
   static final short REPLICATION = 3;
@@ -85,6 +86,7 @@ public class TestBlockStoragePolicy {
   static final byte ALLSSD  = HdfsConstants.ALLSSD_STORAGE_POLICY_ID;
   static final byte LAZY_PERSIST  = HdfsConstants.MEMORY_STORAGE_POLICY_ID;
   static final byte PROVIDED  = HdfsConstants.PROVIDED_STORAGE_POLICY_ID;
+  static final byte ALLNVDIMM = HdfsConstants.ALLNVDIMM_STORAGE_POLICY_ID;
 
   @Test (timeout=300000)
   public void testConfigKeyEnabled() throws IOException {
@@ -149,6 +151,11 @@ public class TestBlockStoragePolicy {
             + ", storageTypes=[PROVIDED, DISK], "
             + "creationFallbacks=[PROVIDED, DISK], "
             + "replicationFallbacks=[PROVIDED, DISK]}");
+    expectedPolicyStrings.put(ALLNVDIMM,
+        "BlockStoragePolicy{ALL_NVDIMM:" + ALLNVDIMM
+            + ", storageTypes=[NVDIMM], "
+            + "creationFallbacks=[DISK], "
+            + "replicationFallbacks=[DISK]}");
 
     for(byte i = 1; i < 16; i++) {
       final BlockStoragePolicy policy = POLICY_SUITE.getPolicy(i); 
@@ -159,79 +166,86 @@ public class TestBlockStoragePolicy {
     }
     Assert.assertEquals(POLICY_SUITE.getPolicy(HOT), POLICY_SUITE.getDefaultPolicy());
     
-    { // check Cold policy
-      final BlockStoragePolicy cold = POLICY_SUITE.getPolicy(COLD);
-      for(short replication = 1; replication < 6; replication++) {
-        final List<StorageType> computed = cold.chooseStorageTypes(replication);
-        assertStorageType(computed, replication, StorageType.ARCHIVE);
-      }
-      assertCreationFallback(cold, null, null, null, null, null);
-      assertReplicationFallback(cold, null, null, null, null);
+    // check Cold policy
+    final BlockStoragePolicy cold = POLICY_SUITE.getPolicy(COLD);
+    for(short replication = 1; replication < 6; replication++) {
+      final List<StorageType> computed = cold.chooseStorageTypes(replication);
+      assertStorageType(computed, replication, StorageType.ARCHIVE);
     }
+    assertCreationFallback(cold, null, null, null, null, null, null);
+    assertReplicationFallback(cold, null, null, null, null, null);
     
-    { // check Warm policy
-      final BlockStoragePolicy warm = POLICY_SUITE.getPolicy(WARM);
-      for(short replication = 1; replication < 6; replication++) {
-        final List<StorageType> computed = warm.chooseStorageTypes(replication);
-        assertStorageType(computed, replication, StorageType.DISK, StorageType.ARCHIVE);
-      }
-      assertCreationFallback(warm, StorageType.DISK, StorageType.DISK,
-          StorageType.ARCHIVE, StorageType.DISK, null);
-      assertReplicationFallback(warm, StorageType.DISK, StorageType.DISK,
-          StorageType.ARCHIVE, StorageType.DISK);
-    }
-
-    { // check Hot policy
-      final BlockStoragePolicy hot = POLICY_SUITE.getPolicy(HOT);
-      for(short replication = 1; replication < 6; replication++) {
-        final List<StorageType> computed = hot.chooseStorageTypes(replication);
-        assertStorageType(computed, replication, StorageType.DISK);
-      }
-      assertCreationFallback(hot, null, null, null, null, null);
-      assertReplicationFallback(hot, StorageType.ARCHIVE, null,
-          StorageType.ARCHIVE, StorageType.ARCHIVE);
+    // check Warm policy
+    final BlockStoragePolicy warm = POLICY_SUITE.getPolicy(WARM);
+    for(short replication = 1; replication < 6; replication++) {
+      final List<StorageType> computed = warm.chooseStorageTypes(replication);
+      assertStorageType(computed, replication, StorageType.DISK,
+        StorageType.ARCHIVE);
     }
+    assertCreationFallback(warm, StorageType.DISK, StorageType.DISK,
+      StorageType.ARCHIVE, StorageType.DISK, null, StorageType.DISK);
+    assertReplicationFallback(warm, StorageType.DISK, StorageType.DISK,
+      StorageType.ARCHIVE, StorageType.DISK, StorageType.DISK);
 
-    { // check ONE_SSD policy
-      final BlockStoragePolicy onessd = POLICY_SUITE.getPolicy(ONESSD);
-      for (short replication = 1; replication < 6; replication++) {
-        final List<StorageType> computed = onessd
-            .chooseStorageTypes(replication);
-        assertStorageType(computed, replication, StorageType.SSD,
-            StorageType.DISK);
-      }
-      assertCreationFallback(onessd, StorageType.SSD, StorageType.SSD,
-          StorageType.SSD, StorageType.DISK, StorageType.SSD);
-      assertReplicationFallback(onessd, StorageType.SSD, StorageType.SSD,
-          StorageType.SSD, StorageType.DISK);
-    }
-
-    { // check ALL_SSD policy
-      final BlockStoragePolicy allssd = POLICY_SUITE.getPolicy(ALLSSD);
-      for (short replication = 1; replication < 6; replication++) {
-        final List<StorageType> computed = allssd
-            .chooseStorageTypes(replication);
-        assertStorageType(computed, replication, StorageType.SSD);
-      }
-      assertCreationFallback(allssd, StorageType.DISK, StorageType.DISK, null,
-          StorageType.DISK, null);
-      assertReplicationFallback(allssd, StorageType.DISK, StorageType.DISK,
-          null, StorageType.DISK);
-    }
-
-    { // check LAZY_PERSIST policy
-      final BlockStoragePolicy lazyPersist = POLICY_SUITE
-          .getPolicy(LAZY_PERSIST);
-      for (short replication = 1; replication < 6; replication++) {
-        final List<StorageType> computed = lazyPersist
-            .chooseStorageTypes(replication);
-        assertStorageType(computed, replication, StorageType.DISK);
-      }
-      assertCreationFallback(lazyPersist, StorageType.DISK, StorageType.DISK,
-          null, StorageType.DISK, null);
-      assertReplicationFallback(lazyPersist, StorageType.DISK,
-          StorageType.DISK, null, StorageType.DISK);
-    }
+    // check Hot policy
+    final BlockStoragePolicy hot = POLICY_SUITE.getPolicy(HOT);
+    for(short replication = 1; replication < 6; replication++) {
+      final List<StorageType> computed = hot.chooseStorageTypes(replication);
+      assertStorageType(computed, replication, StorageType.DISK);
+    }
+    assertCreationFallback(hot, null, null, null, null, null, null);
+    assertReplicationFallback(hot, StorageType.ARCHIVE, null,
+      StorageType.ARCHIVE, StorageType.ARCHIVE, StorageType.ARCHIVE);
+
+    // check ONE_SSD policy
+    final BlockStoragePolicy onessd = POLICY_SUITE.getPolicy(ONESSD);
+    for (short replication = 1; replication < 6; replication++) {
+      final List<StorageType> computed = onessd
+        .chooseStorageTypes(replication);
+      assertStorageType(computed, replication, StorageType.SSD,
+        StorageType.DISK);
+    }
+    assertCreationFallback(onessd, StorageType.SSD, StorageType.SSD,
+      StorageType.SSD, StorageType.DISK, StorageType.SSD, StorageType.SSD);
+    assertReplicationFallback(onessd, StorageType.SSD, StorageType.SSD,
+      StorageType.SSD, StorageType.DISK, StorageType.SSD);
+
+    // check ALL_SSD policy
+    final BlockStoragePolicy allssd = POLICY_SUITE.getPolicy(ALLSSD);
+    for (short replication = 1; replication < 6; replication++) {
+      final List<StorageType> computed = allssd
+        .chooseStorageTypes(replication);
+      assertStorageType(computed, replication, StorageType.SSD);
+    }
+    assertCreationFallback(allssd, StorageType.DISK, StorageType.DISK, null,
+      StorageType.DISK, null, StorageType.DISK);
+    assertReplicationFallback(allssd, StorageType.DISK, StorageType.DISK,
+      null, StorageType.DISK, StorageType.DISK);
+
+    // check LAZY_PERSIST policy
+    final BlockStoragePolicy lazyPersist = POLICY_SUITE
+      .getPolicy(LAZY_PERSIST);
+    for (short replication = 1; replication < 6; replication++) {
+      final List<StorageType> computed = lazyPersist
+        .chooseStorageTypes(replication);
+      assertStorageType(computed, replication, StorageType.DISK);
+    }
+    assertCreationFallback(lazyPersist, StorageType.DISK, StorageType.DISK,
+      null, StorageType.DISK, null, StorageType.DISK);
+    assertReplicationFallback(lazyPersist, StorageType.DISK,
+      StorageType.DISK, null, StorageType.DISK, StorageType.DISK);
+
+    // check ALL_NVDIMM policy
+    final BlockStoragePolicy allnvdimm = POLICY_SUITE.getPolicy(ALLNVDIMM);
+    for (short replication = 1; replication < 6; replication++) {
+      final List<StorageType> computed = allnvdimm
+        .chooseStorageTypes(replication);
+      assertStorageType(computed, replication, StorageType.NVDIMM);
+    }
+    assertCreationFallback(allnvdimm, StorageType.DISK, StorageType.DISK,
+      null, StorageType.DISK, null, StorageType.DISK);
+    assertReplicationFallback(allnvdimm, StorageType.DISK,
+      StorageType.DISK, null, StorageType.DISK, StorageType.DISK);
   }
 
   static StorageType[] newStorageTypes(int nDisk, int nArchive) {
@@ -258,11 +272,12 @@ public class TestBlockStoragePolicy {
   static void assertCreationFallback(BlockStoragePolicy policy,
       StorageType noneExpected, StorageType archiveExpected,
       StorageType diskExpected, StorageType ssdExpected,
-      StorageType disk_archiveExpected) {
+      StorageType disk_archiveExpected, StorageType nvdimmExpected) {
     Assert.assertEquals(noneExpected, policy.getCreationFallback(none));
     Assert.assertEquals(archiveExpected, policy.getCreationFallback(archive));
     Assert.assertEquals(diskExpected, policy.getCreationFallback(disk));
     Assert.assertEquals(ssdExpected, policy.getCreationFallback(ssd));
+    Assert.assertEquals(nvdimmExpected, policy.getCreationFallback(nvdimm));
     Assert.assertEquals(disk_archiveExpected,
         policy.getCreationFallback(disk_archive));
     Assert.assertEquals(null, policy.getCreationFallback(all));
@@ -270,12 +285,14 @@ public class TestBlockStoragePolicy {
 
   static void assertReplicationFallback(BlockStoragePolicy policy,
       StorageType noneExpected, StorageType archiveExpected,
-      StorageType diskExpected, StorageType ssdExpected) {
+      StorageType diskExpected, StorageType ssdExpected,
+      StorageType nvdimmExpected) {
     Assert.assertEquals(noneExpected, policy.getReplicationFallback(none));
     Assert
         .assertEquals(archiveExpected, policy.getReplicationFallback(archive));
     Assert.assertEquals(diskExpected, policy.getReplicationFallback(disk));
     Assert.assertEquals(ssdExpected, policy.getReplicationFallback(ssd));
+    Assert.assertEquals(nvdimmExpected, policy.getReplicationFallback(nvdimm));
     Assert.assertEquals(null, policy.getReplicationFallback(all));
   }
 
@@ -377,6 +394,7 @@ public class TestBlockStoragePolicy {
     final BlockStoragePolicy hot = POLICY_SUITE.getPolicy(HOT);
     final BlockStoragePolicy warm = POLICY_SUITE.getPolicy(WARM);
     final BlockStoragePolicy cold = POLICY_SUITE.getPolicy(COLD);
+    final BlockStoragePolicy allnvdimm = POLICY_SUITE.getPolicy(ALLNVDIMM);
 
     final short replication = 3;
     {
@@ -387,6 +405,8 @@ public class TestBlockStoragePolicy {
           StorageType.DISK, StorageType.ARCHIVE, StorageType.ARCHIVE);
       method.checkChooseStorageTypes(cold, replication, chosen,
           StorageType.ARCHIVE, StorageType.ARCHIVE, StorageType.ARCHIVE);
+      method.checkChooseStorageTypes(allnvdimm, replication, chosen,
+          StorageType.NVDIMM, StorageType.NVDIMM, StorageType.NVDIMM);
     }
 
     {
@@ -397,6 +417,8 @@ public class TestBlockStoragePolicy {
           StorageType.ARCHIVE, StorageType.ARCHIVE);
       method.checkChooseStorageTypes(cold, replication, chosen,
           StorageType.ARCHIVE, StorageType.ARCHIVE, StorageType.ARCHIVE);
+      method.checkChooseStorageTypes(allnvdimm, replication, chosen,
+          StorageType.NVDIMM, StorageType.NVDIMM, StorageType.NVDIMM);
     }
 
     {
@@ -407,6 +429,8 @@ public class TestBlockStoragePolicy {
           StorageType.DISK, StorageType.ARCHIVE);
       method.checkChooseStorageTypes(cold, replication, chosen,
           StorageType.ARCHIVE, StorageType.ARCHIVE);
+      method.checkChooseStorageTypes(allnvdimm, replication, chosen,
+          StorageType.NVDIMM, StorageType.NVDIMM, StorageType.NVDIMM);
     }
 
     {
@@ -418,6 +442,8 @@ public class TestBlockStoragePolicy {
           StorageType.ARCHIVE, StorageType.ARCHIVE);
       method.checkChooseStorageTypes(cold, replication, chosen,
           StorageType.ARCHIVE, StorageType.ARCHIVE, StorageType.ARCHIVE);
+      method.checkChooseStorageTypes(allnvdimm, replication, chosen,
+          StorageType.NVDIMM, StorageType.NVDIMM, StorageType.NVDIMM);
     }
 
     {
@@ -429,6 +455,8 @@ public class TestBlockStoragePolicy {
           StorageType.ARCHIVE);
       method.checkChooseStorageTypes(cold, replication, chosen,
           StorageType.ARCHIVE, StorageType.ARCHIVE);
+      method.checkChooseStorageTypes(allnvdimm, replication, chosen,
+          StorageType.NVDIMM, StorageType.NVDIMM, StorageType.NVDIMM);
     }
 
     {
@@ -440,6 +468,8 @@ public class TestBlockStoragePolicy {
           StorageType.DISK);
       method.checkChooseStorageTypes(cold, replication, chosen,
           StorageType.ARCHIVE);
+      method.checkChooseStorageTypes(allnvdimm, replication, chosen,
+          StorageType.NVDIMM, StorageType.NVDIMM, StorageType.NVDIMM);
     }
 
     {
@@ -450,6 +480,8 @@ public class TestBlockStoragePolicy {
           StorageType.ARCHIVE, StorageType.ARCHIVE);
       method.checkChooseStorageTypes(cold, replication, chosen,
           StorageType.ARCHIVE, StorageType.ARCHIVE, StorageType.ARCHIVE);
+      method.checkChooseStorageTypes(allnvdimm, replication, chosen,
+          StorageType.NVDIMM, StorageType.NVDIMM, StorageType.NVDIMM);
     }
 
     {
@@ -461,6 +493,8 @@ public class TestBlockStoragePolicy {
           StorageType.ARCHIVE);
       method.checkChooseStorageTypes(cold, replication, chosen,
           StorageType.ARCHIVE, StorageType.ARCHIVE);
+      method.checkChooseStorageTypes(allnvdimm, replication, chosen,
+          StorageType.NVDIMM, StorageType.NVDIMM, StorageType.NVDIMM);
     }
 
     {
@@ -471,6 +505,8 @@ public class TestBlockStoragePolicy {
       method.checkChooseStorageTypes(warm, replication, chosen);
       method.checkChooseStorageTypes(cold, replication, chosen,
           StorageType.ARCHIVE);
+      method.checkChooseStorageTypes(allnvdimm, replication, chosen,
+          StorageType.NVDIMM, StorageType.NVDIMM, StorageType.NVDIMM);
     }
 
     {
@@ -481,6 +517,8 @@ public class TestBlockStoragePolicy {
       method.checkChooseStorageTypes(warm, replication, chosen,
           StorageType.DISK);
       method.checkChooseStorageTypes(cold, replication, chosen);
+      method.checkChooseStorageTypes(allnvdimm, replication, chosen,
+          StorageType.NVDIMM, StorageType.NVDIMM, StorageType.NVDIMM);
     }
   }
 
@@ -1387,10 +1425,12 @@ public class TestBlockStoragePolicy {
     map.put(StorageType.DISK, 1);
     map.put(StorageType.SSD, 1);
     map.put(StorageType.RAM_DISK, 1);
+    map.put(StorageType.NVDIMM, 1);
 
     {
       final Iterator<StorageType> i = map.keySet().iterator();
       Assert.assertEquals(StorageType.RAM_DISK, i.next());
+      Assert.assertEquals(StorageType.NVDIMM, i.next());
       Assert.assertEquals(StorageType.SSD, i.next());
       Assert.assertEquals(StorageType.DISK, i.next());
       Assert.assertEquals(StorageType.ARCHIVE, i.next());
@@ -1400,6 +1440,7 @@ public class TestBlockStoragePolicy {
       final Iterator<Map.Entry<StorageType, Integer>> i
           = map.entrySet().iterator();
       Assert.assertEquals(StorageType.RAM_DISK, i.next().getKey());
+      Assert.assertEquals(StorageType.NVDIMM, i.next().getKey());
       Assert.assertEquals(StorageType.SSD, i.next().getKey());
       Assert.assertEquals(StorageType.DISK, i.next().getKey());
       Assert.assertEquals(StorageType.ARCHIVE, i.next().getKey());
@@ -1475,6 +1516,15 @@ public class TestBlockStoragePolicy {
         new StorageType[]{StorageType.DISK},
         false);
 
+    testStorageTypeCheckAccessResult(
+        new StorageType[]{StorageType.DISK, StorageType.NVDIMM},
+        new StorageType[]{StorageType.NVDIMM}, true);
+
+    testStorageTypeCheckAccessResult(
+        new StorageType[]{StorageType.RAM_DISK, StorageType.NVDIMM,
+            StorageType.ARCHIVE},
+        new StorageType[]{StorageType.SSD}, false);
+
   }
 
   private void testStorageTypeCheckAccessResult(StorageType[] requested,

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/net/TestDFSNetworkTopology.java

@@ -64,37 +64,43 @@ public class TestDFSNetworkTopology {
     final String[] racks = {
         "/l1/d1/r1", "/l1/d1/r1", "/l1/d1/r2", "/l1/d1/r2", "/l1/d1/r2",
 
-        "/l1/d2/r3", "/l1/d2/r3", "/l1/d2/r3",
+        "/l1/d2/r3", "/l1/d2/r3", "/l1/d2/r3", "/l1/d2/r3",
 
         "/l2/d3/r1", "/l2/d3/r2", "/l2/d3/r3", "/l2/d3/r4", "/l2/d3/r5",
 
         "/l2/d4/r1", "/l2/d4/r1", "/l2/d4/r1", "/l2/d4/r1", "/l2/d4/r1",
-        "/l2/d4/r1", "/l2/d4/r1"};
+        "/l2/d4/r1", "/l2/d4/r1", "/l2/d4/r2",
+
+        "/l3/d5/r1", "/l3/d5/r1", "/l3/d5/r2"};
     final String[] hosts = {
         "host1", "host2", "host3", "host4", "host5",
-        "host6", "host7", "host8",
-        "host9", "host10", "host11", "host12", "host13",
-        "host14", "host15", "host16", "host17", "host18", "host19", "host20"};
+        "host6", "host7", "host8", "host9",
+        "host10", "host11", "host12", "host13", "host14",
+        "host15", "host16", "host17", "host18", "host19", "host20",
+        "host21", "host22", "host23", "host24", "host25"};
     final StorageType[] types = {
         StorageType.ARCHIVE, StorageType.DISK, StorageType.ARCHIVE,
         StorageType.DISK, StorageType.DISK,
 
         StorageType.DISK, StorageType.RAM_DISK, StorageType.SSD,
+        StorageType.NVDIMM,
 
         StorageType.DISK, StorageType.RAM_DISK, StorageType.DISK,
         StorageType.ARCHIVE, StorageType.ARCHIVE,
 
         StorageType.DISK, StorageType.DISK, StorageType.RAM_DISK,
         StorageType.RAM_DISK, StorageType.ARCHIVE, StorageType.ARCHIVE,
-        StorageType.SSD};
+        StorageType.SSD, StorageType.NVDIMM,
+
+        StorageType.NVDIMM, StorageType.ARCHIVE, StorageType.DISK};
     final DatanodeStorageInfo[] storages =
-        DFSTestUtil.createDatanodeStorageInfos(20, racks, hosts, types);
+        DFSTestUtil.createDatanodeStorageInfos(25, racks, hosts, types);
     dataNodes = DFSTestUtil.toDatanodeDescriptor(storages);
     for (int i = 0; i < dataNodes.length; i++) {
       CLUSTER.add(dataNodes[i]);
     }
-    dataNodes[9].setDecommissioned();
     dataNodes[10].setDecommissioned();
+    dataNodes[11].setDecommissioned();
   }
 
   /**
@@ -120,10 +126,11 @@ public class TestDFSNetworkTopology {
     HashMap<String, EnumMap<StorageType, Integer>> d2info =
         d2.getChildrenStorageInfo();
     assertEquals(1, d2info.keySet().size());
-    assertTrue(d2info.get("r3").size() == 3);
+    assertEquals(4, d2info.get("r3").size());
     assertEquals(1, (int)d2info.get("r3").get(StorageType.DISK));
     assertEquals(1, (int)d2info.get("r3").get(StorageType.RAM_DISK));
     assertEquals(1, (int)d2info.get("r3").get(StorageType.SSD));
+    assertEquals(1, (int)d2info.get("r3").get(StorageType.NVDIMM));
 
     DFSTopologyNodeImpl d3 =
         (DFSTopologyNodeImpl) CLUSTER.getNode("/l2/d3");
@@ -140,11 +147,23 @@ public class TestDFSNetworkTopology {
         (DFSTopologyNodeImpl) CLUSTER.getNode("/l2/d4");
     HashMap<String, EnumMap<StorageType, Integer>> d4info =
         d4.getChildrenStorageInfo();
-    assertEquals(1, d4info.keySet().size());
+    assertEquals(2, d4info.keySet().size());
     assertEquals(2, (int)d4info.get("r1").get(StorageType.DISK));
     assertEquals(2, (int)d4info.get("r1").get(StorageType.RAM_DISK));
     assertEquals(2, (int)d4info.get("r1").get(StorageType.ARCHIVE));
     assertEquals(1, (int)d4info.get("r1").get(StorageType.SSD));
+    assertEquals(1, (int)d4info.get("r2").get(StorageType.NVDIMM));
+
+    DFSTopologyNodeImpl d5 =
+        (DFSTopologyNodeImpl) CLUSTER.getNode("/l3/d5");
+    System.out.println(d5);
+    HashMap<String, EnumMap<StorageType, Integer>> d5info =
+        d5.getChildrenStorageInfo();
+    assertEquals(2, d5info.keySet().size());
+    assertEquals(1, (int)d5info.get("r1").get(StorageType.ARCHIVE));
+    assertEquals(1, (int)d5info.get("r1").get(StorageType.NVDIMM));
+    assertEquals(1, (int)d5info.get("r2").get(StorageType.DISK));
+
 
     DFSTopologyNodeImpl l1 =
         (DFSTopologyNodeImpl) CLUSTER.getNode("/l1");
@@ -152,12 +171,13 @@ public class TestDFSNetworkTopology {
         l1.getChildrenStorageInfo();
     assertEquals(2, l1info.keySet().size());
     assertTrue(l1info.get("d1").size() == 2
-        && l1info.get("d2").size() == 3);
+        && l1info.get("d2").size() == 4);
     assertEquals(2, (int)l1info.get("d1").get(StorageType.ARCHIVE));
     assertEquals(3, (int)l1info.get("d1").get(StorageType.DISK));
     assertEquals(1, (int)l1info.get("d2").get(StorageType.DISK));
     assertEquals(1, (int)l1info.get("d2").get(StorageType.RAM_DISK));
     assertEquals(1, (int)l1info.get("d2").get(StorageType.SSD));
+    assertEquals(1, (int)l1info.get("d2").get(StorageType.NVDIMM));
 
     // checking level = 1 nodes
     DFSTopologyNodeImpl l2 =
@@ -165,7 +185,7 @@ public class TestDFSNetworkTopology {
     HashMap<String, EnumMap<StorageType, Integer>> l2info =
         l2.getChildrenStorageInfo();
     assertTrue(l2info.get("d3").size() == 3
-        && l2info.get("d4").size() == 4);
+        && l2info.get("d4").size() == 5);
     assertEquals(2, l2info.keySet().size());
     assertEquals(2, (int)l2info.get("d3").get(StorageType.DISK));
     assertEquals(2, (int)l2info.get("d3").get(StorageType.ARCHIVE));
@@ -174,6 +194,17 @@ public class TestDFSNetworkTopology {
     assertEquals(2, (int)l2info.get("d4").get(StorageType.ARCHIVE));
     assertEquals(2, (int)l2info.get("d4").get(StorageType.RAM_DISK));
     assertEquals(1, (int)l2info.get("d4").get(StorageType.SSD));
+    assertEquals(1, (int)l2info.get("d4").get(StorageType.NVDIMM));
+
+    DFSTopologyNodeImpl l3 =
+        (DFSTopologyNodeImpl) CLUSTER.getNode("/l3");
+    HashMap<String, EnumMap<StorageType, Integer>> l3info =
+        l3.getChildrenStorageInfo();
+    assertTrue(l3info.get("d5").size() == 3);
+    assertEquals(1, l3info.keySet().size());
+    assertEquals(1, (int)l3info.get("d5").get(StorageType.NVDIMM));
+    assertEquals(1, (int)l3info.get("d5").get(StorageType.ARCHIVE));
+    assertEquals(1, (int)l3info.get("d5").get(StorageType.DISK));
   }
 
   /**
@@ -182,15 +213,16 @@ public class TestDFSNetworkTopology {
    */
   @Test
   public void testAddAndRemoveTopology() throws Exception {
-    String[] newRack = {"/l1/d1/r1", "/l1/d1/r3", "/l1/d3/r3", "/l1/d3/r3"};
-    String[] newHost = {"nhost1", "nhost2", "nhost3", "nhost4"};
+    String[] newRack = {"/l1/d1/r1", "/l1/d1/r3", "/l1/d3/r3", "/l1/d3/r3",
+        "/l1/d3/r4"};
+    String[] newHost = {"nhost1", "nhost2", "nhost3", "nhost4", "nhost5"};
     String[] newips = {"30.30.30.30", "31.31.31.31", "32.32.32.32",
-        "33.33.33.33"};
+        "33.33.33.33", "34.34.34.34"};
     StorageType[] newTypes = {StorageType.DISK, StorageType.SSD,
-        StorageType.SSD, StorageType.SSD};
-    DatanodeDescriptor[] newDD = new DatanodeDescriptor[4];
+        StorageType.SSD, StorageType.SSD, StorageType.NVDIMM};
+    DatanodeDescriptor[] newDD = new DatanodeDescriptor[5];
 
-    for (int i = 0; i<4; i++) {
+    for (int i = 0; i < 5; i++) {
       DatanodeStorageInfo dsi = DFSTestUtil.createDatanodeStorageInfo(
           "s" + newHost[i], newips[i], newRack[i], newHost[i],
           newTypes[i], null);
@@ -215,9 +247,10 @@ public class TestDFSNetworkTopology {
         (DFSTopologyNodeImpl) CLUSTER.getNode("/l1/d3");
     HashMap<String, EnumMap<StorageType, Integer>> d3info =
         d3.getChildrenStorageInfo();
-    assertEquals(1, d3info.keySet().size());
-    assertTrue(d3info.get("r3").size() == 1);
+    assertEquals(2, d3info.keySet().size());
+    assertTrue(d3info.get("r3").size() == 1 && d3info.get("r4").size() == 1);
     assertEquals(2, (int)d3info.get("r3").get(StorageType.SSD));
+    assertEquals(1, (int)d3info.get("r4").get(StorageType.NVDIMM));
 
     DFSTopologyNodeImpl l1 =
         (DFSTopologyNodeImpl) CLUSTER.getNode("/l1");
@@ -225,7 +258,7 @@ public class TestDFSNetworkTopology {
         l1.getChildrenStorageInfo();
     assertEquals(3, l1info.keySet().size());
     assertTrue(l1info.get("d1").size() == 3 &&
-        l1info.get("d2").size() == 3 && l1info.get("d3").size() == 1);
+        l1info.get("d2").size() == 4 && l1info.get("d3").size() == 2);
     assertEquals(4, (int)l1info.get("d1").get(StorageType.DISK));
     assertEquals(2, (int)l1info.get("d1").get(StorageType.ARCHIVE));
     assertEquals(1, (int)l1info.get("d1").get(StorageType.SSD));
@@ -233,8 +266,9 @@ public class TestDFSNetworkTopology {
     assertEquals(1, (int)l1info.get("d2").get(StorageType.RAM_DISK));
     assertEquals(1, (int)l1info.get("d2").get(StorageType.DISK));
     assertEquals(2, (int)l1info.get("d3").get(StorageType.SSD));
+    assertEquals(1, (int)l1info.get("d3").get(StorageType.NVDIMM));
 
-    for (int i = 0; i<4; i++) {
+    for (int i = 0; i < 5; i++) {
       CLUSTER.remove(newDD[i]);
     }
 
@@ -257,12 +291,13 @@ public class TestDFSNetworkTopology {
         nl1.getChildrenStorageInfo();
     assertEquals(2, nl1info.keySet().size());
     assertTrue(l1info.get("d1").size() == 2
-        && l1info.get("d2").size() == 3);
+        && l1info.get("d2").size() == 4);
     assertEquals(2, (int)nl1info.get("d1").get(StorageType.ARCHIVE));
     assertEquals(3, (int)nl1info.get("d1").get(StorageType.DISK));
     assertEquals(1, (int)l1info.get("d2").get(StorageType.DISK));
     assertEquals(1, (int)l1info.get("d2").get(StorageType.RAM_DISK));
     assertEquals(1, (int)l1info.get("d2").get(StorageType.SSD));
+    assertEquals(1, (int)l1info.get("d2").get(StorageType.NVDIMM));
 
     assertNull(CLUSTER.getNode("/l1/d3"));
   }
@@ -278,6 +313,7 @@ public class TestDFSNetworkTopology {
     Set<String> archiveUnderL1 = Sets.newHashSet("host1", "host3");
     Set<String> ramdiskUnderL1 = Sets.newHashSet("host7");
     Set<String> ssdUnderL1 = Sets.newHashSet("host8");
+    Set<String> nvdimmUnderL1 = Sets.newHashSet("host9");
     for (int i = 0; i < 10; i++) {
       n = CLUSTER.chooseRandomWithStorageType("/l1", null, null,
           StorageType.DISK);
@@ -319,35 +355,35 @@ public class TestDFSNetworkTopology {
     // exclude the host on r4 (since there is only one host, no randomness here)
     excluded.add(n);
 
-    for (int i = 0; i<10; i++) {
+    for (int i = 0; i < 10; i++) {
       n = CLUSTER.chooseRandomWithStorageType("/l2/d3", null, null,
           StorageType.ARCHIVE);
       assertTrue(n instanceof DatanodeDescriptor);
       dd = (DatanodeDescriptor) n;
-      assertTrue(dd.getHostName().equals("host12") ||
-          dd.getHostName().equals("host13"));
+      assertTrue(dd.getHostName().equals("host13") ||
+          dd.getHostName().equals("host14"));
     }
 
     // test exclude nodes
-    for (int i = 0; i<10; i++) {
+    for (int i = 0; i < 10; i++) {
       n = CLUSTER.chooseRandomWithStorageType("/l2/d3", null, excluded,
           StorageType.ARCHIVE);
       assertTrue(n instanceof DatanodeDescriptor);
       dd = (DatanodeDescriptor) n;
-      assertTrue(dd.getHostName().equals("host13"));
+      assertTrue(dd.getHostName().equals("host14"));
     }
 
     // test exclude scope
-    for (int i = 0; i<10; i++) {
+    for (int i = 0; i < 10; i++) {
       n = CLUSTER.chooseRandomWithStorageType("/l2/d3", "/l2/d3/r4", null,
           StorageType.ARCHIVE);
       assertTrue(n instanceof DatanodeDescriptor);
       dd = (DatanodeDescriptor) n;
-      assertTrue(dd.getHostName().equals("host13"));
+      assertTrue(dd.getHostName().equals("host14"));
     }
 
     // test exclude scope + excluded node with expected null return node
-    for (int i = 0; i<10; i++) {
+    for (int i = 0; i < 10; i++) {
       n = CLUSTER.chooseRandomWithStorageType("/l2/d3", "/l2/d3/r5", excluded,
           StorageType.ARCHIVE);
       assertNull(n);
@@ -361,7 +397,7 @@ public class TestDFSNetworkTopology {
     // exclude the host on r4 (since there is only one host, no randomness here)
     excluded.add(n);
     Set<String> expectedSet = Sets.newHashSet("host4", "host5");
-    for (int i = 0; i<10; i++) {
+    for (int i = 0; i < 10; i++) {
       // under l1, there are four hosts with DISK:
       // /l1/d1/r1/host2, /l1/d1/r2/host4, /l1/d1/r2/host5 and /l1/d2/r3/host6
       // host6 is excludedNode, host2 is under excluded range scope /l1/d1/r1
@@ -403,21 +439,21 @@ public class TestDFSNetworkTopology {
     excluded.add(n);
 
     // search with given scope being desired scope
-    for (int i = 0; i<10; i++) {
+    for (int i = 0; i < 10; i++) {
       n = CLUSTER.chooseRandomWithStorageType(
           "/l2/d3", null, StorageType.ARCHIVE);
       assertTrue(n instanceof DatanodeDescriptor);
       dd = (DatanodeDescriptor) n;
-      assertTrue(dd.getHostName().equals("host12") ||
-          dd.getHostName().equals("host13"));
+      assertTrue(dd.getHostName().equals("host13") ||
+          dd.getHostName().equals("host14"));
     }
 
-    for (int i = 0; i<10; i++) {
+    for (int i = 0; i < 10; i++) {
       n = CLUSTER.chooseRandomWithStorageType(
           "/l2/d3", excluded, StorageType.ARCHIVE);
       assertTrue(n instanceof DatanodeDescriptor);
       dd = (DatanodeDescriptor) n;
-      assertTrue(dd.getHostName().equals("host13"));
+      assertTrue(dd.getHostName().equals("host14"));
     }
 
     // search with given scope being exclude scope
@@ -425,13 +461,13 @@ public class TestDFSNetworkTopology {
     // a total of 4 ramdisk nodes:
     // /l1/d2/r3/host7, /l2/d3/r2/host10, /l2/d4/r1/host7 and /l2/d4/r1/host10
     // so if we exclude /l2/d4/r1, if should be always either host7 or host10
-    for (int i = 0; i<10; i++) {
+    for (int i = 0; i < 10; i++) {
       n = CLUSTER.chooseRandomWithStorageType(
           "~/l2/d4", null, StorageType.RAM_DISK);
       assertTrue(n instanceof DatanodeDescriptor);
       dd = (DatanodeDescriptor) n;
       assertTrue(dd.getHostName().equals("host7") ||
-          dd.getHostName().equals("host10"));
+          dd.getHostName().equals("host11"));
     }
 
     // similar to above, except that we also exclude host10 here. so it should
@@ -440,7 +476,7 @@ public class TestDFSNetworkTopology {
         StorageType.RAM_DISK);
     // add host10 to exclude
     excluded.add(n);
-    for (int i = 0; i<10; i++) {
+    for (int i = 0; i < 10; i++) {
       n = CLUSTER.chooseRandomWithStorageType(
           "~/l2/d4", excluded, StorageType.RAM_DISK);
       assertTrue(n instanceof DatanodeDescriptor);
@@ -537,21 +573,21 @@ public class TestDFSNetworkTopology {
     excluded.add(n);
 
     // search with given scope being desired scope
-    for (int i = 0; i<10; i++) {
+    for (int i = 0; i < 10; i++) {
       n = CLUSTER.chooseRandomWithStorageTypeTwoTrial(
           "/l2/d3", null, StorageType.ARCHIVE);
       assertTrue(n instanceof DatanodeDescriptor);
       dd = (DatanodeDescriptor) n;
-      assertTrue(dd.getHostName().equals("host12") ||
-          dd.getHostName().equals("host13"));
+      assertTrue(dd.getHostName().equals("host13") ||
+          dd.getHostName().equals("host14"));
     }
 
-    for (int i = 0; i<10; i++) {
+    for (int i = 0; i < 10; i++) {
       n = CLUSTER.chooseRandomWithStorageTypeTwoTrial(
           "/l2/d3", excluded, StorageType.ARCHIVE);
       assertTrue(n instanceof DatanodeDescriptor);
       dd = (DatanodeDescriptor) n;
-      assertTrue(dd.getHostName().equals("host13"));
+      assertTrue(dd.getHostName().equals("host14"));
     }
 
     // search with given scope being exclude scope
@@ -559,13 +595,13 @@ public class TestDFSNetworkTopology {
     // a total of 4 ramdisk nodes:
     // /l1/d2/r3/host7, /l2/d3/r2/host10, /l2/d4/r1/host7 and /l2/d4/r1/host10
     // so if we exclude /l2/d4/r1, if should be always either host7 or host10
-    for (int i = 0; i<10; i++) {
+    for (int i = 0; i < 10; i++) {
       n = CLUSTER.chooseRandomWithStorageTypeTwoTrial(
           "~/l2/d4", null, StorageType.RAM_DISK);
       assertTrue(n instanceof DatanodeDescriptor);
       dd = (DatanodeDescriptor) n;
       assertTrue(dd.getHostName().equals("host7") ||
-          dd.getHostName().equals("host10"));
+          dd.getHostName().equals("host11"));
     }
 
     // similar to above, except that we also exclude host10 here. so it should
@@ -574,7 +610,7 @@ public class TestDFSNetworkTopology {
         StorageType.RAM_DISK);
     // add host10 to exclude
     excluded.add(n);
-    for (int i = 0; i<10; i++) {
+    for (int i = 0; i < 10; i++) {
       n = CLUSTER.chooseRandomWithStorageTypeTwoTrial(
           "~/l2/d4", excluded, StorageType.RAM_DISK);
       assertTrue(n instanceof DatanodeDescriptor);

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/protocolPB/TestPBHelper.java

@@ -529,14 +529,18 @@ public class TestPBHelper {
             AdminStates.NORMAL),
         DFSTestUtil.getLocalDatanodeInfo("127.0.0.1", "h4",
             AdminStates.NORMAL),
+        DFSTestUtil.getLocalDatanodeInfo("127.0.0.1", "h5",
+            AdminStates.NORMAL),
     };
-    String[] storageIDs = {"s1", "s2", "s3", "s4"};
+    String[] storageIDs = {"s1", "s2", "s3", "s4", "s5"};
     StorageType[] media = {
         StorageType.DISK,
         StorageType.SSD,
         StorageType.DISK,
-        StorageType.RAM_DISK
+        StorageType.RAM_DISK,
+        StorageType.NVDIMM,
     };
+
     LocatedBlock lb = new LocatedBlock(
         new ExtendedBlock("bp12", 12345, 10, 53),
         dnInfos, storageIDs, media, 5, false, new DatanodeInfo[]{});

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/security/token/block/TestBlockToken.java

@@ -792,7 +792,7 @@ public class TestBlockToken {
         EnumSet.allOf(BlockTokenIdentifier.AccessMode.class);
     StorageType[] storageTypes =
         new StorageType[]{StorageType.RAM_DISK, StorageType.SSD,
-            StorageType.DISK, StorageType.ARCHIVE};
+            StorageType.DISK, StorageType.ARCHIVE, StorageType.NVDIMM};
     BlockTokenIdentifier ident = new BlockTokenIdentifier("user", "bpool",
         123, accessModes, storageTypes, new String[] {"fake-storage-id"},
         useProto);

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestBlockStatsMXBean.java

@@ -63,7 +63,7 @@ public class TestBlockStatsMXBean {
     conf.setTimeDuration(DFSConfigKeys.DFS_DATANODE_DISK_CHECK_MIN_GAP_KEY,
         0, TimeUnit.MILLISECONDS);
     cluster = null;
-    StorageType[][] types = new StorageType[6][];
+    StorageType[][] types = new StorageType[7][];
     for (int i=0; i<3; i++) {
       types[i] = new StorageType[] {StorageType.RAM_DISK, StorageType.DISK};
     }
@@ -72,8 +72,9 @@ public class TestBlockStatsMXBean {
     }
     types[5] = new StorageType[] {StorageType.RAM_DISK, StorageType.ARCHIVE,
         StorageType.ARCHIVE};
+    types[6] = new StorageType[]{StorageType.RAM_DISK, StorageType.NVDIMM};
 
-    cluster = new MiniDFSCluster.Builder(conf).numDataNodes(6).
+    cluster = new MiniDFSCluster.Builder(conf).numDataNodes(7).
         storageTypes(types).storagesPerDatanode(3).build();
     cluster.waitActive();
   }
@@ -93,16 +94,20 @@ public class TestBlockStatsMXBean {
     assertTrue(storageTypeStatsMap.containsKey(StorageType.RAM_DISK));
     assertTrue(storageTypeStatsMap.containsKey(StorageType.DISK));
     assertTrue(storageTypeStatsMap.containsKey(StorageType.ARCHIVE));
+    assertTrue(storageTypeStatsMap.containsKey(StorageType.NVDIMM));
 
     StorageTypeStats storageTypeStats =
         storageTypeStatsMap.get(StorageType.RAM_DISK);
-    assertEquals(6, storageTypeStats.getNodesInService());
+    assertEquals(7, storageTypeStats.getNodesInService());
 
     storageTypeStats = storageTypeStatsMap.get(StorageType.DISK);
     assertEquals(3, storageTypeStats.getNodesInService());
 
     storageTypeStats = storageTypeStatsMap.get(StorageType.ARCHIVE);
     assertEquals(3, storageTypeStats.getNodesInService());
+
+    storageTypeStats = storageTypeStatsMap.get(StorageType.NVDIMM);
+    assertEquals(1, storageTypeStats.getNodesInService());
   }
 
   protected static String readOutput(URL url) throws IOException {
@@ -136,7 +141,7 @@ public class TestBlockStatsMXBean {
     Object[] storageTypeStatsList =
         (Object[])blockStats.get("StorageTypeStats");
     assertNotNull(storageTypeStatsList);
-    assertEquals (3, storageTypeStatsList.length);
+    assertEquals(4, storageTypeStatsList.length);
 
     Set<String> typesPresent = new HashSet<> ();
     for (Object obj : storageTypeStatsList) {
@@ -144,12 +149,18 @@ public class TestBlockStatsMXBean {
       String storageType = (String)entry.get("key");
       Map<String,Object> storageTypeStats = (Map<String,Object>)entry.get("value");
       typesPresent.add(storageType);
-      if (storageType.equals("ARCHIVE") || storageType.equals("DISK") ) {
-        assertEquals(3l, storageTypeStats.get("nodesInService"));
-      } else if (storageType.equals("RAM_DISK")) {
-        assertEquals(6l, storageTypeStats.get("nodesInService"));
-      }
-      else {
+      switch (storageType) {
+      case "ARCHIVE":
+      case "DISK":
+        assertEquals(3L, storageTypeStats.get("nodesInService"));
+        break;
+      case "RAM_DISK":
+        assertEquals(7L, storageTypeStats.get("nodesInService"));
+        break;
+      case "NVDIMM":
+        assertEquals(1L, storageTypeStats.get("nodesInService"));
+        break;
+      default:
         fail();
       }
     }
@@ -157,6 +168,7 @@ public class TestBlockStatsMXBean {
     assertTrue(typesPresent.contains("ARCHIVE"));
     assertTrue(typesPresent.contains("DISK"));
     assertTrue(typesPresent.contains("RAM_DISK"));
+    assertTrue(typesPresent.contains("NVDIMM"));
   }
 
   @Test
@@ -172,19 +184,24 @@ public class TestBlockStatsMXBean {
 
     StorageTypeStats storageTypeStats = storageTypeStatsMap
         .get(StorageType.RAM_DISK);
-    assertEquals(6, storageTypeStats.getNodesInService());
+    assertEquals(7, storageTypeStats.getNodesInService());
 
     storageTypeStats = storageTypeStatsMap.get(StorageType.DISK);
     assertEquals(3, storageTypeStats.getNodesInService());
 
     storageTypeStats = storageTypeStatsMap.get(StorageType.ARCHIVE);
     assertEquals(3, storageTypeStats.getNodesInService());
+
+    storageTypeStats = storageTypeStatsMap.get(StorageType.NVDIMM);
+    assertEquals(1, storageTypeStats.getNodesInService());
     File dn1ArcVol1 = cluster.getInstanceStorageDir(0, 1);
     File dn2ArcVol1 = cluster.getInstanceStorageDir(1, 1);
     File dn3ArcVol1 = cluster.getInstanceStorageDir(2, 1);
+    File dn4ArcVol1 = cluster.getInstanceStorageDir(3, 1);
     DataNodeTestUtils.injectDataDirFailure(dn1ArcVol1);
     DataNodeTestUtils.injectDataDirFailure(dn2ArcVol1);
     DataNodeTestUtils.injectDataDirFailure(dn3ArcVol1);
+    DataNodeTestUtils.injectDataDirFailure(dn4ArcVol1);
     try {
       DFSTestUtil.createFile(cluster.getFileSystem(), new Path(
           "/blockStatsFile2"), 1024, (short) 1, 0L);
@@ -202,7 +219,8 @@ public class TestBlockStatsMXBean {
     DataNodeTestUtils.restoreDataDirFromFailure(dn1ArcVol1);
     DataNodeTestUtils.restoreDataDirFromFailure(dn2ArcVol1);
     DataNodeTestUtils.restoreDataDirFromFailure(dn3ArcVol1);
-    for (int i = 0; i < 3; i++) {
+    DataNodeTestUtils.restoreDataDirFromFailure(dn4ArcVol1);
+    for (int i = 0; i < 4; i++) {
       cluster.restartDataNode(0, true);
     }
     // wait for heartbeat
@@ -210,12 +228,15 @@ public class TestBlockStatsMXBean {
     storageTypeStatsMap = cluster.getNamesystem().getBlockManager()
         .getStorageTypeStats();
     storageTypeStats = storageTypeStatsMap.get(StorageType.RAM_DISK);
-    assertEquals(6, storageTypeStats.getNodesInService());
+    assertEquals(7, storageTypeStats.getNodesInService());
 
     storageTypeStats = storageTypeStatsMap.get(StorageType.DISK);
     assertEquals(3, storageTypeStats.getNodesInService());
 
     storageTypeStats = storageTypeStatsMap.get(StorageType.ARCHIVE);
     assertEquals(3, storageTypeStats.getNodesInService());
+
+    storageTypeStats = storageTypeStatsMap.get(StorageType.NVDIMM);
+    assertEquals(1, storageTypeStats.getNodesInService());
   }
 }

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestDatanodeManager.java

@@ -375,15 +375,16 @@ public class TestDatanodeManager {
     }
     DatanodeManager dm = mockDatanodeManager(fsn, conf);
 
-    int totalDNs = 5 + providedStorages;
+    int totalDNs = 6 + providedStorages;
 
-    // register 5 datanodes, each with different storage ID and type
+    // register 6 datanodes, each with different storage ID and type
     DatanodeInfo[] locs = new DatanodeInfo[totalDNs];
     String[] storageIDs = new String[totalDNs];
     List<StorageType> storageTypesList = new ArrayList<>(
         Arrays.asList(StorageType.ARCHIVE,
             StorageType.DEFAULT,
             StorageType.DISK,
+            StorageType.NVDIMM,
             StorageType.RAM_DISK,
             StorageType.SSD));
 
@@ -420,7 +421,7 @@ public class TestDatanodeManager {
     List<LocatedBlock> blocks = new ArrayList<>();
     blocks.add(block);
 
-    final String targetIp = locs[4].getIpAddr();
+    final String targetIp = locs[5].getIpAddr();
 
     // sort block locations
     dm.sortLocatedBlocks(targetIp, blocks);
@@ -511,7 +512,7 @@ public class TestDatanodeManager {
     assertEquals(DatanodeInfo.AdminStates.DECOMMISSIONED,
         sortedLocs[sortedLocs.length - 2].getAdminState());
 
-    // test client not in cluster but same rack with locs[4]
+    // test client not in cluster but same rack with locs[5]
     final String targetIpNotInCluster = locs[4].getIpAddr() + "-client";
     dm.sortLocatedBlocks(targetIpNotInCluster, blocks);
     DatanodeInfo[] sortedLocs2 = block.getLocations();

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/SimulatedFSDataset.java

@@ -606,6 +606,11 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> {
       return false;
     }
 
+    @Override
+    public boolean isRAMStorage() {
+      return false;
+    }
+
     @Override
     public void reserveSpaceForReplica(long bytesToReserve) {
     }

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataDirs.java

@@ -43,14 +43,15 @@ public class TestDataDirs {
 
     File dir5 = new File("/dir5");
     File dir6 = new File("/dir6");
+    File dir7 = new File("/dir7");
     // Verify that a valid string is correctly parsed, and that storage
     // type is not case-sensitive and we are able to handle white-space between
     // storage type and URI.
     String locations1 = "[disk]/dir0,[DISK]/dir1,[sSd]/dir2,[disK]/dir3," +
-            "[ram_disk]/dir4,[disk]/dir5, [disk] /dir6, [disk] ";
+            "[ram_disk]/dir4,[disk]/dir5, [disk] /dir6, [disk] , [nvdimm]/dir7";
     conf.set(DFS_DATANODE_DATA_DIR_KEY, locations1);
     locations = DataNode.getStorageLocations(conf);
-    assertThat(locations.size(), is(8));
+    assertThat(locations.size(), is(9));
     assertThat(locations.get(0).getStorageType(), is(StorageType.DISK));
     assertThat(locations.get(0).getUri(), is(dir0.toURI()));
     assertThat(locations.get(1).getStorageType(), is(StorageType.DISK));
@@ -70,6 +71,9 @@ public class TestDataDirs {
     // test set to make sure that we don't fail if we get URIs like that.
     assertThat(locations.get(7).getStorageType(), is(StorageType.DISK));
 
+    assertThat(locations.get(8).getStorageType(), is(StorageType.NVDIMM));
+    assertThat(locations.get(8).getUri(), is(dir7.toURI()));
+
     // Verify that an unrecognized storage type result in an exception.
     String locations2 = "[BadMediaType]/dir0,[ssd]/dir1,[disk]/dir2";
     conf.set(DFS_DATANODE_DATA_DIR_KEY, locations2);

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDirectoryScanner.java

@@ -942,6 +942,11 @@ public class TestDirectoryScanner {
       return false;
     }
 
+    @Override
+    public boolean isRAMStorage() {
+      return false;
+    }
+
     @Override
     public void reserveSpaceForReplica(long bytesToReserve) {
     }

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/extdataset/ExternalVolumeImpl.java

@@ -67,6 +67,11 @@ public class ExternalVolumeImpl implements FsVolumeSpi {
     return false;
   }
 
+  @Override
+  public boolean isRAMStorage() {
+    return false;
+  }
+
   @Override
   public void reserveSpaceForReplica(long bytesToReserve) {
   }

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/TestFsVolumeList.java

@@ -170,6 +170,9 @@ public class TestFsVolumeList {
     conf.setLong(
         DFSConfigKeys.DFS_DATANODE_DU_RESERVED_KEY + "."
             + StringUtils.toLowerCase(StorageType.SSD.toString()), 2L);
+    conf.setLong(
+        DFSConfigKeys.DFS_DATANODE_DU_RESERVED_KEY + "."
+            + StringUtils.toLowerCase(StorageType.NVDIMM.toString()), 3L);
     FsVolumeImpl volume1 = new FsVolumeImplBuilder().setDataset(dataset)
         .setStorageDirectory(
             new StorageDirectory(
@@ -202,6 +205,14 @@ public class TestFsVolumeList {
         .setConf(conf)
         .build();
     assertEquals("", 100L, volume4.getReserved());
+    FsVolumeImpl volume5 = new FsVolumeImplBuilder().setDataset(dataset)
+        .setStorageDirectory(
+            new StorageDirectory(
+                StorageLocation.parse("[NVDIMM]"+volDir.getPath())))
+        .setStorageID("storage-id")
+        .setConf(conf)
+        .build();
+    assertEquals(3L, volume5.getReserved());
   }
 
   @Test
@@ -286,6 +297,9 @@ public class TestFsVolumeList {
     conf.setLong(
         DFS_DATANODE_DU_RESERVED_PERCENTAGE_KEY + "."
             + StringUtils.toLowerCase(StorageType.SSD.toString()), 50);
+    conf.setLong(
+        DFS_DATANODE_DU_RESERVED_PERCENTAGE_KEY + "."
+            + StringUtils.toLowerCase(StorageType.NVDIMM.toString()), 20);
     FsVolumeImpl volume1 = new FsVolumeImplBuilder()
         .setConf(conf)
         .setDataset(dataset)
@@ -329,6 +343,18 @@ public class TestFsVolumeList {
         .setUsage(usage)
         .build();
     assertEquals(600, volume4.getReserved());
+    FsVolumeImpl volume5 = new FsVolumeImplBuilder()
+        .setConf(conf)
+        .setDataset(dataset)
+        .setStorageID("storage-id")
+        .setStorageDirectory(
+            new StorageDirectory(StorageLocation.parse(
+                "[NVDIMM]" + volDir.getPath())))
+        .setUsage(usage)
+        .build();
+    assertEquals(800, volume5.getReserved());
+    assertEquals(3200, volume5.getCapacity());
+    assertEquals(200, volume5.getAvailable());
   }
 
   @Test(timeout = 60000)

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/TestReservedSpaceCalculator.java

@@ -61,6 +61,7 @@ public class TestReservedSpaceCalculator {
     checkReserved(StorageType.DISK, 10000, 900);
     checkReserved(StorageType.SSD, 10000, 900);
     checkReserved(StorageType.ARCHIVE, 10000, 900);
+    checkReserved(StorageType.NVDIMM, 10000, 900);
   }
 
   @Test
@@ -76,6 +77,10 @@ public class TestReservedSpaceCalculator {
     // Test SSD
     conf.setLong(DFS_DATANODE_DU_RESERVED_KEY + ".ssd", 750);
     checkReserved(StorageType.SSD, 1550, 750);
+
+    // Test NVDIMM
+    conf.setLong(DFS_DATANODE_DU_RESERVED_KEY + ".nvdimm", 300);
+    checkReserved(StorageType.NVDIMM, 1000, 300);
   }
 
   @Test
@@ -89,11 +94,13 @@ public class TestReservedSpaceCalculator {
     checkReserved(StorageType.DISK, 10000, 1000);
     checkReserved(StorageType.SSD, 10000, 1000);
     checkReserved(StorageType.ARCHIVE, 10000, 1000);
+    checkReserved(StorageType.NVDIMM, 10000, 1000);
 
     conf.setLong(DFS_DATANODE_DU_RESERVED_PERCENTAGE_KEY, 50);
     checkReserved(StorageType.DISK, 4000, 2000);
     checkReserved(StorageType.SSD, 4000, 2000);
     checkReserved(StorageType.ARCHIVE, 4000, 2000);
+    checkReserved(StorageType.NVDIMM, 4000, 2000);
   }
 
   @Test
@@ -109,6 +116,10 @@ public class TestReservedSpaceCalculator {
     // Test SSD
     conf.setLong(DFS_DATANODE_DU_RESERVED_PERCENTAGE_KEY + ".ssd", 50);
     checkReserved(StorageType.SSD, 8001, 4000);
+
+    // Test NVDIMM
+    conf.setLong(DFS_DATANODE_DU_RESERVED_PERCENTAGE_KEY + ".nvdimm", 30);
+    checkReserved(StorageType.NVDIMM, 1000, 300);
   }
 
   @Test
@@ -129,6 +140,12 @@ public class TestReservedSpaceCalculator {
     conf.setLong(DFS_DATANODE_DU_RESERVED_KEY + ".archive", 1300);
     conf.setLong(DFS_DATANODE_DU_RESERVED_PERCENTAGE_KEY + ".archive", 50);
     checkReserved(StorageType.ARCHIVE, 6200, 3100);
+
+    // Test NVDIMM + taking reserved space based on the percentage,
+    // as that gives more reserved space
+    conf.setLong(DFS_DATANODE_DU_RESERVED_KEY + ".nvdimm", 500);
+    conf.setLong(DFS_DATANODE_DU_RESERVED_PERCENTAGE_KEY + ".nvdimm", 20);
+    checkReserved(StorageType.NVDIMM, 3000, 600);
   }
 
   @Test

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestNamenodeStorageDirectives.java

@@ -201,7 +201,8 @@ public class TestNamenodeStorageDirectives {
             {StorageType.SSD, StorageType.DISK}},
         "ONE_SSD",
         new StorageType[]{StorageType.SSD, StorageType.DISK},
-        new StorageType[]{StorageType.RAM_DISK, StorageType.ARCHIVE});
+        new StorageType[]{StorageType.RAM_DISK, StorageType.ARCHIVE,
+            StorageType.NVDIMM});
     // only on SSD.
     testStorageTypes(new StorageType[][]{
             {StorageType.SSD, StorageType.DISK},
@@ -209,7 +210,7 @@ public class TestNamenodeStorageDirectives {
         "ALL_SSD",
         new StorageType[]{StorageType.SSD},
         new StorageType[]{StorageType.RAM_DISK, StorageType.DISK,
-            StorageType.ARCHIVE});
+            StorageType.ARCHIVE, StorageType.NVDIMM});
     // only on SSD.
     testStorageTypes(new StorageType[][]{
             {StorageType.SSD, StorageType.DISK, StorageType.DISK},
@@ -218,7 +219,7 @@ public class TestNamenodeStorageDirectives {
         "ALL_SSD",
         new StorageType[]{StorageType.SSD},
         new StorageType[]{StorageType.RAM_DISK, StorageType.DISK,
-            StorageType.ARCHIVE});
+            StorageType.ARCHIVE, StorageType.NVDIMM});
 
     // DISK and not anything else.
     testStorageTypes(new StorageType[][] {
@@ -228,7 +229,7 @@ public class TestNamenodeStorageDirectives {
         "HOT",
         new StorageType[]{StorageType.DISK},
         new StorageType[] {StorageType.RAM_DISK, StorageType.SSD,
-            StorageType.ARCHIVE});
+            StorageType.ARCHIVE, StorageType.NVDIMM});
 
     testStorageTypes(new StorageType[][] {
             {StorageType.RAM_DISK, StorageType.SSD},
@@ -237,7 +238,8 @@ public class TestNamenodeStorageDirectives {
             {StorageType.ARCHIVE, StorageType.ARCHIVE}},
         "WARM",
         new StorageType[]{StorageType.DISK, StorageType.ARCHIVE},
-        new StorageType[]{StorageType.RAM_DISK, StorageType.SSD});
+        new StorageType[]{StorageType.RAM_DISK, StorageType.SSD,
+            StorageType.NVDIMM});
 
     testStorageTypes(new StorageType[][] {
             {StorageType.RAM_DISK, StorageType.SSD},
@@ -247,7 +249,7 @@ public class TestNamenodeStorageDirectives {
         "COLD",
         new StorageType[]{StorageType.ARCHIVE},
         new StorageType[]{StorageType.RAM_DISK, StorageType.SSD,
-            StorageType.DISK});
+            StorageType.DISK, StorageType.NVDIMM});
 
     // We wait for Lasy Persist to write to disk.
     testStorageTypes(new StorageType[][] {
@@ -257,7 +259,15 @@ public class TestNamenodeStorageDirectives {
         "LAZY_PERSIST",
         new StorageType[]{StorageType.DISK},
         new StorageType[]{StorageType.RAM_DISK, StorageType.SSD,
-            StorageType.ARCHIVE});
+            StorageType.ARCHIVE, StorageType.NVDIMM});
+
+    testStorageTypes(new StorageType[][] {
+            {StorageType.NVDIMM, StorageType.DISK, StorageType.SSD},
+            {StorageType.NVDIMM, StorageType.DISK, StorageType.SSD}},
+            "ALL_NVDIMM",
+            new StorageType[]{StorageType.NVDIMM},
+            new StorageType[]{StorageType.RAM_DISK, StorageType.SSD,
+                StorageType.DISK, StorageType.ARCHIVE});
   }
 
   /**

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/sps/TestExternalStoragePolicySatisfier.java

@@ -446,6 +446,33 @@ public class TestExternalStoragePolicySatisfier {
         dfs);
   }
 
+  @Test(timeout = 300000)
+  public void testWhenStoragePolicySetToALLNVDIMM()
+          throws Exception {
+    try {
+      createCluster();
+      // Change policy to ALL_NVDIMM
+      dfs.setStoragePolicy(new Path(FILE), "ALL_NVDIMM");
+
+      StorageType[][] newtypes =
+          new StorageType[][]{{StorageType.NVDIMM, StorageType.DISK},
+              {StorageType.NVDIMM, StorageType.DISK},
+              {StorageType.NVDIMM, StorageType.DISK}};
+
+      startAdditionalDNs(config, 3, NUM_OF_DATANODES, newtypes,
+              STORAGES_PER_DATANODE, CAPACITY, hdfsCluster);
+
+      dfs.satisfyStoragePolicy(new Path(FILE));
+      hdfsCluster.triggerHeartbeats();
+      // Wait till StorgePolicySatisfier Identified that block
+      // to move to MVDIMM areas
+      DFSTestUtil.waitExpectedStorageType(FILE, StorageType.NVDIMM, 3,
+          30000, dfs);
+    } finally {
+      shutdownCluster();
+    }
+  }
+
   @Test(timeout = 300000)
   public void testWhenStoragePolicySetToALLSSD()
       throws Exception {

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/web/TestWebHDFS.java

@@ -1084,10 +1084,12 @@ public class TestWebHDFS {
     webHdfs.setQuotaByStorageType(path, StorageType.DISK, spaceQuota);
     webHdfs.setQuotaByStorageType(path, StorageType.ARCHIVE, spaceQuota);
     webHdfs.setQuotaByStorageType(path, StorageType.SSD, spaceQuota);
+    webHdfs.setQuotaByStorageType(path, StorageType.NVDIMM, spaceQuota);
     quotaUsage = dfs.getQuotaUsage(path);
     assertEquals(spaceQuota, quotaUsage.getTypeQuota(StorageType.DISK));
     assertEquals(spaceQuota, quotaUsage.getTypeQuota(StorageType.ARCHIVE));
     assertEquals(spaceQuota, quotaUsage.getTypeQuota(StorageType.SSD));
+    assertEquals(spaceQuota, quotaUsage.getTypeQuota(StorageType.NVDIMM));
 
     // Test invalid parameters
 
@@ -1103,6 +1105,8 @@ public class TestWebHDFS {
         () -> webHdfs.setQuotaByStorageType(path, StorageType.SSD, -100));
     LambdaTestUtils.intercept(IllegalArgumentException.class,
         () -> webHdfs.setQuotaByStorageType(path, StorageType.RAM_DISK, 100));
+    LambdaTestUtils.intercept(IllegalArgumentException.class,
+        () -> webHdfs.setQuotaByStorageType(path, StorageType.NVDIMM, -100));
   }