HDFS-15525. Make trash root inside each snapshottable directory for WebHDFS

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/web/resources/NamenodeWebHdfsMethods.java

@@ -1345,19 +1345,55 @@ protected Response get(
     }
   }
 
+  /**
+   * Get the snapshot root of a given file or directory if it exists.
+   * e.g. if /snapdir1 is a snapshottable directory and path given is
+   * /snapdir1/path/to/file, this method would return /snapdir1
+   * @param pathStr String of path to a file or a directory.
+   * @return Not null if found in a snapshot root directory.
+   * @throws IOException
+   */
+  String getSnapshotRoot(String pathStr) throws IOException {
+    SnapshottableDirectoryStatus[] dirStatusList =
+        getRpcClientProtocol().getSnapshottableDirListing();
+    if (dirStatusList == null) {
+      return null;
+    }
+    for (SnapshottableDirectoryStatus dirStatus : dirStatusList) {
+      String currDir = dirStatus.getFullPath().toString();
+      if (pathStr.startsWith(currDir)) {
+        return currDir;
+      }
+    }
+    return null;
+  }
+
   private String getTrashRoot(Configuration conf, String fullPath)
       throws IOException {
-    UserGroupInformation ugi= UserGroupInformation.getCurrentUser();
+    UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
     String parentSrc = getParent(fullPath);
+    String ssTrashRoot = "";
+    boolean isSnapshotTrashRootEnabled = getRpcClientProtocol()
+        .getServerDefaults().getSnapshotTrashRootEnabled();
+    if (isSnapshotTrashRootEnabled) {
+      String ssRoot = getSnapshotRoot(fullPath);
+      if (ssRoot != null) {
+        ssTrashRoot = DFSUtilClient.getSnapshotTrashRoot(ssRoot, ugi);
+      }
+    }
     EncryptionZone ez = getRpcClientProtocol().getEZForPath(
         parentSrc != null ? parentSrc : fullPath);
-    String trashRoot;
+    String ezTrashRoot = "";
     if (ez != null) {
-      trashRoot = DFSUtilClient.getEZTrashRoot(ez, ugi);
+      ezTrashRoot = DFSUtilClient.getEZTrashRoot(ez, ugi);
+    }
+    // Choose the longest path
+    if (ssTrashRoot.isEmpty() && ezTrashRoot.isEmpty()) {
+      return DFSUtilClient.getTrashRoot(conf, ugi);
     } else {
-      trashRoot = DFSUtilClient.getTrashRoot(conf, ugi);
+      return ssTrashRoot.length() > ezTrashRoot.length() ?
+          ssTrashRoot : ezTrashRoot;
     }
-    return trashRoot;
   }
 
   /**

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

@@ -1569,6 +1569,34 @@ public void testGetTrashRoot() throws Exception {
     assertEquals(expectedPath.toUri().getPath(), trashPath.toUri().getPath());
   }
 
+  @Test
+  public void testGetSnapshotTrashRoot() throws Exception {
+    final Configuration conf = WebHdfsTestUtil.createConf();
+    conf.setBoolean("dfs.namenode.snapshot.trashroot.enabled", true);
+    final String currentUser =
+        UserGroupInformation.getCurrentUser().getShortUserName();
+    cluster = new MiniDFSCluster.Builder(conf).numDataNodes(0).build();
+    final WebHdfsFileSystem webFS = WebHdfsTestUtil.getWebHdfsFileSystem(conf,
+        WebHdfsConstants.WEBHDFS_SCHEME);
+    Path ssDir1 = new Path("/ssDir1");
+    assertTrue(webFS.mkdirs(ssDir1));
+
+    Path trashPath = webFS.getTrashRoot(ssDir1);
+    Path expectedPath = new Path(FileSystem.USER_HOME_PREFIX,
+        new Path(currentUser, FileSystem.TRASH_PREFIX));
+    assertEquals(expectedPath.toUri().getPath(), trashPath.toUri().getPath());
+    // Enable snapshot
+    webFS.allowSnapshot(ssDir1);
+    Path trashPathAfter = webFS.getTrashRoot(ssDir1);
+    Path expectedPathAfter = new Path(ssDir1,
+        new Path(FileSystem.TRASH_PREFIX, currentUser));
+    assertEquals(expectedPathAfter.toUri().getPath(),
+        trashPathAfter.toUri().getPath());
+    // Cleanup
+    webFS.disallowSnapshot(ssDir1);
+    webFS.delete(ssDir1, true);
+  }
+
   @Test
   public void testGetEZTrashRoot() throws Exception {
     final Configuration conf = WebHdfsTestUtil.createConf();