HDFS-13162. Create Replica Trash directory on DN startup. Contributed by Bharat Viswanadham.

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

@@ -1079,6 +1079,11 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
   public static final boolean DFS_REJECT_UNRESOLVED_DN_TOPOLOGY_MAPPING_DEFAULT =
       false;
 
+  // Datanode Replica Trash
+  public static final String DFS_DATANODE_ENABLE_REPLICA_TRASH_KEY =
+      "dfs.datanode.enable.replica.trash";
+  public static final boolean DFS_DATANODE_ENABLE_REPLICA_TRASH_DEFAULT = false;
+
   // Slow io warning log threshold settings for dfsclient and datanode.
   public static final String DFS_DATANODE_SLOW_IO_WARNING_THRESHOLD_KEY =
     "dfs.datanode.slow.io.warning.threshold.ms";

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

@@ -81,6 +81,7 @@ public class DataStorage extends Storage {
   public final static String STORAGE_DIR_FINALIZED = "finalized";
   public final static String STORAGE_DIR_LAZY_PERSIST = "lazypersist";
   public final static String STORAGE_DIR_TMP = "tmp";
+  public final static String STORAGE_DIR_REPLICA_TRASH = "replica-trash";
 
   /**
    * Set of bpids for which 'trash' is currently enabled.
@@ -333,7 +334,7 @@ public class DataStorage extends Storage {
     VolumeBuilder builder =
         new VolumeBuilder(this, sd);
     for (NamespaceInfo nsInfo : nsInfos) {
-      location.makeBlockPoolDir(nsInfo.getBlockPoolID(), null);
+      location.makeBlockPoolDir(nsInfo.getBlockPoolID(), datanode.getConf());
 
       final BlockPoolSliceStorage bpStorage = getBlockPoolSliceStorage(nsInfo);
       final List<StorageDirectory> dirs = bpStorage.loadBpStorageDirectories(
@@ -448,7 +449,7 @@ public class DataStorage extends Storage {
     final List<StorageDirectory> success = Lists.newArrayList();
     final List<UpgradeTask> tasks = Lists.newArrayList();
     for (StorageLocation dataDir : dataDirs) {
-      dataDir.makeBlockPoolDir(bpid, null);
+      dataDir.makeBlockPoolDir(bpid, datanode.getConf());
       try {
         final List<Callable<StorageDirectory>> callables = Lists.newArrayList();
         final List<StorageDirectory> dirs = bpStorage.recoverTransitionRead(

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

@@ -198,6 +198,7 @@ public class StorageLocation
 
   /**
    * Create physical directory for block pools on the data node.
+   * If replica trash is enabled, create replica-trash folder.
    *
    * @param blockPoolID
    *          the block pool id
@@ -223,8 +224,19 @@ public class StorageLocation
         DFSConfigKeys.DFS_DATANODE_DATA_DIR_PERMISSION_KEY,
         DFSConfigKeys.DFS_DATANODE_DATA_DIR_PERMISSION_DEFAULT));
     File data = new File(getBpURI(blockPoolID, Storage.STORAGE_DIR_CURRENT));
+
+    boolean replicaTrashEnable = conf.getBoolean(
+        DFSConfigKeys.DFS_DATANODE_ENABLE_REPLICA_TRASH_KEY,
+        DFSConfigKeys.DFS_DATANODE_ENABLE_REPLICA_TRASH_DEFAULT);
+
     try {
       DiskChecker.checkDir(localFS, new Path(data.toURI()), permission);
+      if (replicaTrashEnable) {
+        File replicaTrash = new File(data,
+            DataStorage.STORAGE_DIR_REPLICA_TRASH);
+        DiskChecker.checkDir(localFS,
+            new Path(replicaTrash.toURI()), permission);
+      }
     } catch (IOException e) {
       DataStorage.LOG.warn("Invalid directory in: " + data.getCanonicalPath() +
           ": " + e.getMessage());

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

@@ -3799,6 +3799,16 @@
   </description>
 </property>
 
+<property>
+  <name>dfs.datanode.enable.replica.trash</name>
+  <value>false</value>
+  <description>
+    If true, datanode replica trash feature is enabled. This will move the
+    blocks invalidated to replica-trash folder under block pool directory,
+    which will help in the recovery of accidentally deleted data.
+  </description>
+</property>
+
 <property>
   <name>dfs.datanode.balance.max.concurrent.moves</name>
   <value>50</value>

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

@@ -20,6 +20,7 @@ package org.apache.hadoop.hdfs.server.datanode;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileUtil;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.hdfs.HdfsConfiguration;
 import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.StartupOption;
 import org.apache.hadoop.hdfs.server.common.Storage;
@@ -40,6 +41,7 @@ import java.util.List;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
+import static org.junit.Assert.assertFalse;
 
 public class TestDataStorage {
   private final static String DEFAULT_BPID = "bp-0";
@@ -51,12 +53,13 @@ public class TestDataStorage {
   private final static StartupOption START_OPT = StartupOption.REGULAR;
 
   private DataNode mockDN = Mockito.mock(DataNode.class);
+  private Configuration conf;
   private NamespaceInfo nsInfo;
   private DataStorage storage;
 
   @Before
   public void setUp() throws IOException {
-    Configuration conf = new HdfsConfiguration();
+    conf = new HdfsConfiguration();
     storage = new DataStorage();
     nsInfo = new NamespaceInfo(0, CLUSTER_ID, DEFAULT_BPID, CTIME,
         BUILD_VERSION, SOFTWARE_VERSION);
@@ -130,6 +133,56 @@ public class TestDataStorage {
     assertTrue(bpSd.getVersionFile().isFile());
   }
 
+
+  @Test
+  public void testReplicaTrashDirectory() throws IOException {
+    final int numLocations = 3;
+    final int numNamespace = 3;
+    List<StorageLocation> locations = createStorageLocations(numLocations);
+
+    // Add volumes for multiple namespaces.
+    List<NamespaceInfo> namespaceInfos = createNamespaceInfos(numNamespace);
+
+    conf.setBoolean(DFSConfigKeys.DFS_DATANODE_ENABLE_REPLICA_TRASH_KEY,
+        true);
+    mockDN.setConf(conf);
+
+    for (NamespaceInfo ni : namespaceInfos) {
+      storage.addStorageLocations(mockDN, ni, locations, START_OPT);
+      for (StorageLocation sl : locations) {
+        File currentDir = new File(new File(sl.getUri()),
+            Storage.STORAGE_DIR_CURRENT);
+        File bpDir = new File(currentDir, ni.getBlockPoolID());
+        File replicaTrashDir = new File(bpDir, DataStorage
+            .STORAGE_DIR_REPLICA_TRASH);
+        assertTrue(replicaTrashDir.isDirectory());
+      }
+    }
+  }
+
+  @Test
+  public void testNoReplicaTrashDirectory() throws IOException {
+    final int numLocations = 3;
+    final int numNamespace = 3;
+    List<StorageLocation> locations = createStorageLocations(numLocations);
+
+    // Add volumes for multiple namespaces.
+    List<NamespaceInfo> namespaceInfos = createNamespaceInfos(numNamespace);
+
+    for (NamespaceInfo ni : namespaceInfos) {
+      storage.addStorageLocations(mockDN, ni, locations, START_OPT);
+      for (StorageLocation sl : locations) {
+        File currentDir = new File(new File(sl.getUri()),
+            Storage.STORAGE_DIR_CURRENT);
+        File bpDir = new File(currentDir, ni.getBlockPoolID());
+        File replicaTrashDir = new File(bpDir, DataStorage
+            .STORAGE_DIR_REPLICA_TRASH);
+        assertFalse(replicaTrashDir.exists());
+      }
+    }
+  }
+
+
   @Test
   public void testAddStorageDirectories() throws IOException,
       URISyntaxException {