diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/StorageType.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/StorageType.java
index 2ecd206a04080..78a05fa41b766 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/StorageType.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/StorageType.java
@@ -34,13 +34,15 @@
 @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) {
-    this.isTransient = isTransient;
+  StorageType(boolean isTransient, boolean isRAM) {
+    this.isTransient = isTransienit;
+    this.isRAM = isRAM;
   }
 
   public boolean isTransient() {
     return isTransient;
   }
 
+  public boolean isRAM() {
+    return isRAM;
+  }
+
   public boolean supportTypeQuota() {
     return !isTransient;
   }
@@ -93,4 +100,4 @@ private static List<StorageType> getNonTransientTypes() {
     }
     return nonTransientTypes;
   }
-}
\ No newline at end of file
+}
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsConstants.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsConstants.java
index a025b9bad2e69..e3452f651b4c2 100755
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsConstants.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsConstants.java
@@ -37,6 +37,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;
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelperClient.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelperClient.java
index 9fc302464271d..d31c263aa3d45 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelperClient.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelperClient.java
@@ -472,6 +472,8 @@ public static StorageTypeProto convertStorageType(StorageType type) {
       return StorageTypeProto.RAM_DISK;
     case PROVIDED:
       return StorageTypeProto.PROVIDED;
+    case NVDIMM:
+      return StorageTypeProto.NVDIMM;
     default:
       throw new IllegalStateException(
           "BUG: StorageType not found, type=" + type);
@@ -490,6 +492,8 @@ public static StorageType convertStorageType(StorageTypeProto type) {
       return StorageType.RAM_DISK;
     case PROVIDED:
       return StorageType.PROVIDED;
+    case NVDIMM:
+      return StorageType.NVDIMM;
     default:
       throw new IllegalStateException(
           "BUG: StorageTypeProto not found, type=" + type);
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/proto/hdfs.proto b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/proto/hdfs.proto
index 82fe329c9ce5e..1523c3906e3b4 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/proto/hdfs.proto
+++ b/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;
 }
 
 /**
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockStoragePolicySuite.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockStoragePolicySuite.java
index d62cb727c6c8d..afb95daacbd41 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockStoragePolicySuite.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockStoragePolicySuite.java
@@ -63,6 +63,12 @@ public static BlockStoragePolicySuite createDefaultSuite(
         new StorageType[]{StorageType.DISK},
         new StorageType[]{StorageType.DISK},
         true);    // Cannot be changed on regular files, but inherited.
+    final byte allNVDIMMId = HdfsConstants.ALLNVDIMM_STORAGE_POLICY_ID;
+    policies[allNVDIMMId] = new BlockStoragePolicy(allNVDIMMId,
+        HdfsConstants.ALLNVDIMM_STORAGE_POLICY_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(),
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/FsVolumeSpi.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/FsVolumeSpi.java
index be978d75e9a34..195343cc35cf2 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/FsVolumeSpi.java
+++ b/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 NOT 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.
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java
index a083012a2cf7a..3b784f130535d 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java
@@ -2286,9 +2286,9 @@ private void cacheBlock(String bpid, long blockId) {
               ": volume was not an instance of FsVolumeImpl.");
           return;
         }
-        if (volume.isTransientStorage()) {
+        if (volume.isRAMStorage()) {
           LOG.warn("Caching not supported on block with id " + blockId +
-              " since the volume is backed by RAM.");
+              " since the volume is backed by RAM_DISK or NVDIMM.");
           return;
         }
         success = true;
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeImpl.java
index 5843c7d6696b7..9fd0d1179cbd3 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeImpl.java
+++ b/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 boolean isTransientStorage() {
     return storageType.isTransient();
   }
 
+  @Override
+  public boolean isRAMStorage() {
+    return storageType.isRAM();
+  }
+
   @VisibleForTesting
   public File getFinalizedDir(String bpid) throws IOException {
     return getBlockPoolSlice(bpid).getFinalizedDir();
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/DFSAdmin.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/DFSAdmin.java
index ec5fa0afddc60..f087d404a3472 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/DFSAdmin.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/DFSAdmin.java
@@ -239,7 +239,8 @@ private static class ClearSpaceQuotaCommand extends DFSAdminCommand {
         "\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;
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/SimulatedFSDataset.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/SimulatedFSDataset.java
index 8069b64246bc3..2da78c802fce1 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/SimulatedFSDataset.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/SimulatedFSDataset.java
@@ -606,6 +606,11 @@ public boolean isTransientStorage() {
       return false;
     }
 
+    @Override
+    public boolean isRAMStorage() {
+      return false;
+    }
+
     @Override
     public void reserveSpaceForReplica(long bytesToReserve) {
     }
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDirectoryScanner.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDirectoryScanner.java
index 25470018a9628..d5e3931dd513c 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDirectoryScanner.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDirectoryScanner.java
@@ -942,6 +942,11 @@ public boolean isTransientStorage() {
       return false;
     }
 
+    @Override
+    public boolean isRAMStorage() {
+      return false;
+    }
+
     @Override
     public void reserveSpaceForReplica(long bytesToReserve) {
     }
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/extdataset/ExternalVolumeImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/extdataset/ExternalVolumeImpl.java
index 9d62499eb5a3e..6c8e828f3689e 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/extdataset/ExternalVolumeImpl.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/extdataset/ExternalVolumeImpl.java
@@ -67,6 +67,11 @@ public boolean isTransientStorage() {
     return false;
   }
 
+  @Override
+  public boolean isRAMStorage() {
+    return false;
+  }
+
   @Override
   public void reserveSpaceForReplica(long bytesToReserve) {
   }