HADOOP-18444 Add Support for localized trash for ViewFileSystem in Trash.moveToAppropriateTrash (#4869)

* HADOOP-18444 Add Support for localized trash for ViewFileSystem in Trash.moveToAppropriateTrash

Signed-off-by: Xing Lin <xinglin@linkedin.com>

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

@@ -23,8 +23,10 @@ import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.conf.Configured;
+import org.apache.hadoop.fs.viewfs.ViewFileSystem;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import static org.apache.hadoop.fs.viewfs.Constants.*;
 
 /** 
  * Provides a trash facility which supports pluggable Trash policies. 
@@ -94,6 +96,27 @@ public class Trash extends Configured {
       LOG.warn("Failed to get server trash configuration", e);
       throw new IOException("Failed to get server trash configuration", e);
     }
+
+    /*
+     * In HADOOP-18144, we changed getTrashRoot() in ViewFileSystem to return a
+     * viewFS path, instead of a targetFS path. moveToTrash works for
+     * ViewFileSystem now. ViewFileSystem will do path resolution internally by
+     * itself.
+     *
+     * When localized trash flag is enabled:
+     *    1). if fs is a ViewFileSystem, we can initialize Trash() with a
+     *        ViewFileSystem object;
+     *    2). When fs is not a ViewFileSystem, the only place we would need to
+     *        resolve a path is for symbolic links. However, symlink is not
+     *        enabled in Hadoop due to the complexity to support it
+     *        (HADOOP-10019).
+     */
+    if (conf.getBoolean(CONFIG_VIEWFS_TRASH_FORCE_INSIDE_MOUNT_POINT,
+        CONFIG_VIEWFS_TRASH_FORCE_INSIDE_MOUNT_POINT_DEFAULT)) {
+      Trash trash = new Trash(fs, conf);
+      return trash.moveToTrash(p);
+    }
+
     Trash trash = new Trash(fullyResolvedFs, conf);
     return trash.moveToTrash(fullyResolvedPath);
   }

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/viewfs/TestViewFsTrash.java

@@ -17,15 +17,24 @@
  */
 package org.apache.hadoop.fs.viewfs;
 
+import java.io.DataOutputStream;
+import java.io.IOException;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FileSystemTestHelper;
 import org.apache.hadoop.fs.FsConstants;
+import org.apache.hadoop.fs.LocalFileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.TestTrash;
+import org.apache.hadoop.fs.Trash;
+import org.apache.hadoop.fs.TrashPolicyDefault;
+import org.apache.hadoop.fs.contract.ContractTestUtils;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
+import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.*;
+import static org.apache.hadoop.fs.viewfs.Constants.*;
+import static org.junit.Assert.*;
 
 public class TestViewFsTrash {
   FileSystem fsTarget;  // the target file system - the mount will point here
@@ -65,5 +74,37 @@ public class TestViewFsTrash {
     TestTrash.trashShell(conf, fileSystemTestHelper.getTestRootPath(fsView),
         fsView, new Path(fileSystemTestHelper.getTestRootPath(fsView), ".Trash/Current"));
   }
-  
+
+  @Test
+  public void testLocalizedTrashInMoveToAppropriateTrash() throws IOException {
+    Configuration conf2 = new Configuration(conf);
+    Path testFile = new Path("/data/testfile.txt");
+
+    // Enable moveToTrash and add a mount point for /data
+    conf2.setLong(FS_TRASH_INTERVAL_KEY, 1);
+    ConfigUtil.addLink(conf2, "/data", new Path(fileSystemTestHelper.getAbsoluteTestRootPath(fsTarget), "data").toUri());
+
+    // Default case. file should be moved to fsTarget.getTrashRoot()/resolvedPath
+    conf2.setBoolean(CONFIG_VIEWFS_TRASH_FORCE_INSIDE_MOUNT_POINT, false);
+    try (FileSystem fsView2 = FileSystem.get(conf2)) {
+      FileSystemTestHelper.createFile(fsView2, testFile);
+      Path resolvedFile = fsView2.resolvePath(testFile);
+
+      Trash.moveToAppropriateTrash(fsView2, testFile, conf2);
+      Trash trash = new Trash(fsTarget, conf2);
+      Path movedPath = Path.mergePaths(trash.getCurrentTrashDir(testFile), resolvedFile);
+      ContractTestUtils.assertPathExists(fsTarget, "File not in trash", movedPath);
+    }
+
+    // Turn on localized trash. File should be moved to viewfs:/data/.Trash/{user}/Current.
+    conf2.setBoolean(CONFIG_VIEWFS_TRASH_FORCE_INSIDE_MOUNT_POINT, true);
+    try (FileSystem fsView2 = FileSystem.get(conf2)) {
+      FileSystemTestHelper.createFile(fsView2, testFile);
+
+      Trash.moveToAppropriateTrash(fsView2, testFile, conf2);
+      Trash trash = new Trash(fsView2, conf2);
+      Path movedPath = Path.mergePaths(trash.getCurrentTrashDir(testFile), testFile);
+      ContractTestUtils.assertPathExists(fsView2, "File not in localized trash", movedPath);
+    }
+  }
 }