diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileSystem.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileSystem.java
index 176362c54f7..2b15295b7eb 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileSystem.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileSystem.java
@@ -746,6 +746,7 @@ protected FileSystem() {
    *
    */
   protected void checkPath(Path path) {
+    Preconditions.checkArgument(path != null, "null path");
     URI uri = path.toUri();
     String thatScheme = uri.getScheme();
     if (thatScheme == null)                // fs is relative
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/Constants.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/Constants.java
index 7f9277cdd3d..0c0e8a34174 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/Constants.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/Constants.java
@@ -75,4 +75,10 @@ public interface Constants {
   String CONFIG_VIEWFS_ENABLE_INNER_CACHE = "fs.viewfs.enable.inner.cache";
 
   boolean CONFIG_VIEWFS_ENABLE_INNER_CACHE_DEFAULT = true;
+
+  /**
+   * Enable ViewFileSystem to return a trashRoot which is local to mount point.
+   */
+  String CONFIG_VIEWFS_MOUNT_POINT_LOCAL_TRASH = "fs.viewfs.mount.point.local.trash";
+  boolean CONFIG_VIEWFS_MOUNT_POINT_LOCAL_TRASH_DEFAULT = false;
 }
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ViewFileSystem.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ViewFileSystem.java
index c832caea973..a1b9c9e857d 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ViewFileSystem.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ViewFileSystem.java
@@ -20,6 +20,8 @@
 import static org.apache.hadoop.fs.viewfs.Constants.PERMISSION_555;
 import static org.apache.hadoop.fs.viewfs.Constants.CONFIG_VIEWFS_ENABLE_INNER_CACHE;
 import static org.apache.hadoop.fs.viewfs.Constants.CONFIG_VIEWFS_ENABLE_INNER_CACHE_DEFAULT;
+import static org.apache.hadoop.fs.viewfs.Constants.CONFIG_VIEWFS_MOUNT_POINT_LOCAL_TRASH;
+import static org.apache.hadoop.fs.viewfs.Constants.CONFIG_VIEWFS_MOUNT_POINT_LOCAL_TRASH_DEFAULT;
 
 import com.google.common.base.Function;
 import java.io.FileNotFoundException;
@@ -27,7 +29,9 @@
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.security.PrivilegedExceptionAction;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.EnumSet;
 import java.util.HashMap;
@@ -925,8 +929,130 @@ public void deleteSnapshot(Path path, String snapshotName)
     res.targetFileSystem.deleteSnapshot(res.remainingPath, snapshotName);
   }
 
-  /*
-   * An instance of this class represents an internal dir of the viewFs 
+  /**
+   * Get the trash root directory for current user when the path
+   * specified is deleted.
+   *
+   * If CONFIG_VIEWFS_MOUNT_POINT_LOCAL_TRASH is not set, return
+   * the default trash root from targetFS.
+   *
+   * When CONFIG_VIEWFS_MOUNT_POINT_LOCAL_TRASH is set to true,
+   * 1) If path p is in fallback FS or from the same mount point as the default
+   *    trash root for targetFS, return the default trash root for targetFS.
+   * 2) else, return a trash root in the mounted targetFS
+   *    (/mntpoint/.Trash/{user})
+   *
+   * @param path the trash root of the path to be determined.
+   * @return the trash root path.
+   */
+  @Override
+  public Path getTrashRoot(Path path) {
+    boolean useMountPointLocalTrash =
+        config.getBoolean(CONFIG_VIEWFS_MOUNT_POINT_LOCAL_TRASH,
+            CONFIG_VIEWFS_MOUNT_POINT_LOCAL_TRASH_DEFAULT);
+
+    try {
+      InodeTree.ResolveResult<FileSystem> res =
+          fsState.resolve(getUriPath(path), true);
+
+      Path trashRoot = res.targetFileSystem.getTrashRoot(res.remainingPath);
+      if (!useMountPointLocalTrash) {
+        return trashRoot;
+      } else {
+        // Path p is either in a mount point or in the fallback FS
+
+        if (ROOT_PATH.equals(new Path(res.resolvedPath))
+            || trashRoot.toUri().getPath().startsWith(res.resolvedPath)) {
+          // Path p is in the fallback FS or targetFileSystem.trashRoot is in
+          // the same mount point as Path p
+          return trashRoot;
+        } else {
+          // targetFileSystem.trashRoot is in a different mount point from
+          // Path p. Return the trash root for the mount point.
+          Path mountPointRoot =
+              res.targetFileSystem.getFileStatus(new Path("/")).getPath();
+          return new Path(mountPointRoot,
+              TRASH_PREFIX + "/" + ugi.getShortUserName());
+        }
+      }
+    } catch (IOException | IllegalArgumentException e) {
+      throw new NotInMountpointException(path, "getTrashRoot");
+    }
+  }
+
+  /**
+   * Get all the trash roots for current user or all users.
+   *
+   * @param allUsers return trash roots for all users if true.
+   * @return all Trash root directories.
+   */
+  @Override
+  public Collection<FileStatus> getTrashRoots(boolean allUsers) {
+    List<FileStatus> trashRoots = new ArrayList<>();
+    for (FileSystem fs : getChildFileSystems()) {
+      trashRoots.addAll(fs.getTrashRoots(allUsers));
+    }
+
+    // Add trash dirs for each mount point
+    boolean useMountPointLocalTrash =
+        config.getBoolean(CONFIG_VIEWFS_MOUNT_POINT_LOCAL_TRASH,
+            CONFIG_VIEWFS_MOUNT_POINT_LOCAL_TRASH_DEFAULT);
+    if (useMountPointLocalTrash) {
+
+      Set<Path> currentTrashPaths = new HashSet<>();
+      for (FileStatus file : trashRoots) {
+        currentTrashPaths.add(file.getPath());
+      }
+
+      MountPoint[] mountPoints = getMountPoints();
+      try {
+        for (int i = 0; i < mountPoints.length; i++) {
+          Path trashRoot = makeQualified(
+              new Path(mountPoints[i].getSrc() + "/" + TRASH_PREFIX));
+
+          // Continue if trashRoot does not exist for this filesystem
+          if (!exists(trashRoot)) {
+            continue;
+          }
+
+          InodeTree.ResolveResult<FileSystem> res =
+              fsState.resolve(getUriPath(trashRoot), true);
+
+          if (!allUsers) {
+            Path userTrash =
+                new Path("/" + TRASH_PREFIX + "/" + ugi.getShortUserName());
+            try {
+              FileStatus file = res.targetFileSystem.getFileStatus(userTrash);
+              if (!currentTrashPaths.contains(file.getPath())) {
+                trashRoots.add(file);
+                currentTrashPaths.add(file.getPath());
+              }
+            } catch (FileNotFoundException ignored) {
+            }
+          } else {
+            FileStatus[] targetFsTrashRoots =
+                res.targetFileSystem.listStatus(new Path("/" + TRASH_PREFIX));
+            for (FileStatus file : targetFsTrashRoots) {
+              // skip if we already include it in currentTrashPaths
+              if (currentTrashPaths.contains(file.getPath())) {
+                continue;
+              }
+
+              trashRoots.add(file);
+              currentTrashPaths.add(file.getPath());
+            }
+          }
+        }
+      } catch (IOException e) {
+        LOG.warn("Exception in get all trash roots", e);
+      }
+    }
+
+    return trashRoots;
+  }
+
+  /**
+   * An instance of this class represents an internal dir of the viewFs
    * that is internal dir of the mount table.
    * It is a read only mount tables and create, mkdir or delete operations
    * are not allowed.
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/viewfs/ViewFileSystemBaseTest.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/viewfs/ViewFileSystemBaseTest.java
index 1f01ea349b4..6b398bc626f 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/viewfs/ViewFileSystemBaseTest.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/viewfs/ViewFileSystemBaseTest.java
@@ -30,6 +30,7 @@
 import org.apache.hadoop.fs.BlockLocation;
 import org.apache.hadoop.fs.LocalFileSystem;
 import org.apache.hadoop.fs.TestFileUtil;
+import org.apache.hadoop.fs.Trash;
 import org.apache.hadoop.fs.contract.ContractTestUtils;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FileSystemTestHelper;
@@ -37,6 +38,8 @@
 import org.apache.hadoop.fs.permission.AclEntry;
 import static org.apache.hadoop.fs.viewfs.Constants.PERMISSION_555;
 import static org.apache.hadoop.fs.viewfs.Constants.CONFIG_VIEWFS_ENABLE_INNER_CACHE;
+import static org.apache.hadoop.fs.viewfs.Constants.CONFIG_VIEWFS_MOUNT_POINT_LOCAL_TRASH;
+import static org.apache.hadoop.fs.FileSystem.TRASH_PREFIX;
 import static org.apache.hadoop.test.GenericTestUtils.*;
 import static org.junit.Assert.*;
 
@@ -951,6 +954,223 @@ public void testInternalDeleteSnapshot() throws IOException {
     fsView.deleteSnapshot(new Path("/internalDir"), "snap1");
   }
 
+  @Test
+  public void testTrashRoot() throws IOException {
+
+    Path mountDataRootPath = new Path("/data");
+    Path fsTargetFilePath = new Path("debug.log");
+    Path mountDataFilePath = new Path(mountDataRootPath, fsTargetFilePath);
+    Path mountDataNonExistingFilePath = new Path(mountDataRootPath, "no.log");
+    fileSystemTestHelper.createFile(fsTarget, fsTargetFilePath);
+
+    // Get Trash roots for paths via ViewFileSystem handle
+    Path mountDataRootTrashPath = fsView.getTrashRoot(mountDataRootPath);
+    Path mountDataFileTrashPath = fsView.getTrashRoot(mountDataFilePath);
+
+    // Get Trash roots for the same set of paths via the mounted filesystem
+    Path fsTargetRootTrashRoot = fsTarget.getTrashRoot(mountDataRootPath);
+    Path fsTargetFileTrashPath = fsTarget.getTrashRoot(mountDataFilePath);
+
+    // Verify if Trash roots from ViewFileSystem matches that of the ones
+    // from the target mounted FileSystem.
+    assertEquals(mountDataRootTrashPath.toUri().getPath(),
+        fsTargetRootTrashRoot.toUri().getPath());
+    assertEquals(mountDataFileTrashPath.toUri().getPath(),
+        fsTargetFileTrashPath.toUri().getPath());
+    assertEquals(mountDataRootTrashPath.toUri().getPath(),
+        mountDataFileTrashPath.toUri().getPath());
+
+
+    // Verify trash root for an non-existing file but on a valid mountpoint.
+    Path trashRoot = fsView.getTrashRoot(mountDataNonExistingFilePath);
+    assertEquals(mountDataRootTrashPath.toUri().getPath(),
+        trashRoot.toUri().getPath());
+
+    // Verify trash root for invalid mounts.
+    Path invalidMountRootPath = new Path("/invalid_mount");
+    Path invalidMountFilePath = new Path(invalidMountRootPath, "debug.log");
+    try {
+      fsView.getTrashRoot(invalidMountRootPath);
+      fail("ViewFileSystem getTashRoot should fail for non-mountpoint paths.");
+    } catch (NotInMountpointException e) {
+      //expected exception
+    }
+    try {
+      fsView.getTrashRoot(invalidMountFilePath);
+      fail("ViewFileSystem getTashRoot should fail for non-mountpoint paths.");
+    } catch (NotInMountpointException e) {
+      //expected exception
+    }
+    try {
+      fsView.getTrashRoot(null);
+      fail("ViewFileSystem getTashRoot should fail for empty paths.");
+    } catch (NotInMountpointException e) {
+      //expected exception
+    }
+
+    // Move the file to trash
+    FileStatus fileStatus = fsTarget.getFileStatus(fsTargetFilePath);
+    Configuration newConf = new Configuration(conf);
+    newConf.setLong("fs.trash.interval", 1000);
+    Trash lTrash = new Trash(fsTarget, newConf);
+    boolean trashed = lTrash.moveToTrash(fsTargetFilePath);
+    Assert.assertTrue("File " + fileStatus + " move to " +
+        "trash failed.", trashed);
+
+    // Verify ViewFileSystem trash roots shows the ones from
+    // target mounted FileSystem.
+    Assert.assertTrue("", fsView.getTrashRoots(true).size() > 0);
+  }
+
+  /**
+   * Test the localized trash root for getTrashRoot.
+   */
+  @Test
+  public void testTrashRootLocalizedTrash() throws IOException {
+    UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
+    Configuration conf2 = new Configuration(conf);
+    conf2.setBoolean(CONFIG_VIEWFS_MOUNT_POINT_LOCAL_TRASH, true);
+    FileSystem fsView2 = FileSystem.get(FsConstants.VIEWFS_URI, conf2);
+
+    // Case 1: path p not in the default FS.
+    // Return a trash root within the mount point.
+    Path dataTestPath = new Path("/data/dir/file");
+    Path dataTrashRoot = new Path(targetTestRoot,
+        "data/" + TRASH_PREFIX + "/" + ugi.getShortUserName());
+    Assert.assertEquals(dataTrashRoot, fsView2.getTrashRoot(dataTestPath));
+
+    // Case 2: turn off the CONFIG_VIEWFS_MOUNT_POINT_LOCAL_TRASH flag.
+    // Return a trash root in user home dir.
+    Path nonExistentPath = new Path("/nonExistentDir/nonExistentFile");
+    Path userTrashRoot = new Path(fsTarget.getHomeDirectory(), TRASH_PREFIX);
+    conf2.setBoolean(CONFIG_VIEWFS_MOUNT_POINT_LOCAL_TRASH, false);
+    fsView2 = FileSystem.get(FsConstants.VIEWFS_URI, conf2);
+    Assert.assertEquals(userTrashRoot, fsView2.getTrashRoot(dataTestPath));
+
+    // Case 3: viewFS without fallback. Expect exception for a nonExistent path
+    conf2 = new Configuration(conf);
+    fsView2 = FileSystem.get(FsConstants.VIEWFS_URI, conf2);
+    try {
+      fsView2.getTrashRoot(nonExistentPath);
+    } catch (NotInMountpointException ignored) {
+    }
+
+    // Case 4: path p is in the same mount point as targetFS.getTrashRoot().
+    // Return targetFS.getTrashRoot()
+    // Use a new Configuration object, so that we can start with an empty
+    // mount table. This would avoid a conflict between the /user link in
+    // setupMountPoints() and homeDir we will need to setup for this test.
+    // default homeDir for hdfs is /user/.
+    Configuration conf3 = ViewFileSystemTestSetup.createConfig();
+    Path homeDir = fsTarget.getHomeDirectory();
+    String homeParentDir = homeDir.getParent().toUri().getPath();
+    conf3.setBoolean(CONFIG_VIEWFS_MOUNT_POINT_LOCAL_TRASH, true);
+    ConfigUtil.addLink(conf3, homeParentDir,
+        new Path(targetTestRoot, homeParentDir).toUri());
+    Path homeTestPath = new Path(homeDir.toUri().getPath(), "testuser/file");
+    FileSystem fsView3 = FileSystem.get(FsConstants.VIEWFS_URI, conf3);
+    Assert.assertEquals(userTrashRoot, fsView3.getTrashRoot(homeTestPath));
+  }
+
+  /**
+   * A mocked FileSystem which returns a deep trash dir.
+   */
+  static class MockTrashRootFS extends MockFileSystem {
+    public static final Path TRASH =
+        new Path("/mnt/very/deep/deep/trash/dir/.Trash");
+
+    @Override
+    public Path getTrashRoot(Path path) {
+      return TRASH;
+    }
+  }
+
+  /**
+   * Test a trash root that is inside a mount point for getTrashRoot.
+   */
+  @Test
+  public void testTrashRootDeepTrashDir() throws IOException {
+
+    Configuration conf2 = ViewFileSystemTestSetup.createConfig();
+    conf2.setBoolean(CONFIG_VIEWFS_MOUNT_POINT_LOCAL_TRASH, true);
+    conf2.setClass("fs.mocktrashfs.impl", MockTrashRootFS.class,
+        FileSystem.class);
+    ConfigUtil.addLink(conf2, "/mnt", URI.create("mocktrashfs://mnt/path"));
+    Path testPath = new Path(MockTrashRootFS.TRASH, "projs/proj");
+    FileSystem fsView2 = FileSystem.get(FsConstants.VIEWFS_URI, conf2);
+    Assert.assertEquals(MockTrashRootFS.TRASH, fsView2.getTrashRoot(testPath));
+  }
+
+  /**
+   * Test localized trash roots in getTrashRoots() for all users.
+   */
+  @Test
+  public void testTrashRootsAllUsers() throws IOException {
+    Configuration conf2 = new Configuration(conf);
+    conf2.setBoolean(CONFIG_VIEWFS_MOUNT_POINT_LOCAL_TRASH, true);
+    FileSystem fsView2 = FileSystem.get(FsConstants.VIEWFS_URI, conf2);
+
+    // Case 1: verify correct trash roots from fsView and fsView2
+    int beforeTrashRootNum = fsView.getTrashRoots(true).size();
+    int beforeTrashRootNum2 = fsView2.getTrashRoots(true).size();
+    Assert.assertEquals(beforeTrashRootNum, beforeTrashRootNum2);
+
+    fsView.mkdirs(new Path("/data/" + TRASH_PREFIX + "/user1"));
+    fsView.mkdirs(new Path("/data/" + TRASH_PREFIX + "/user2"));
+    fsView.mkdirs(new Path("/user/" + TRASH_PREFIX + "/user3"));
+    fsView.mkdirs(new Path("/user/" + TRASH_PREFIX + "/user4"));
+    fsView.mkdirs(new Path("/user2/" + TRASH_PREFIX + "/user5"));
+    int afterTrashRootsNum = fsView.getTrashRoots(true).size();
+    int afterTrashRootsNum2 = fsView2.getTrashRoots(true).size();
+    Assert.assertEquals(beforeTrashRootNum, afterTrashRootsNum);
+    Assert.assertEquals(beforeTrashRootNum2 + 5, afterTrashRootsNum2);
+
+    // Case 2: per-user mount point
+    fsTarget.mkdirs(new Path(targetTestRoot, "Users/userA/.Trash/userA"));
+    Configuration conf3 = new Configuration(conf2);
+    ConfigUtil.addLink(conf3, "/Users/userA",
+        new Path(targetTestRoot, "Users/userA").toUri());
+    FileSystem fsView3 = FileSystem.get(FsConstants.VIEWFS_URI, conf3);
+    int trashRootsNum3 = fsView3.getTrashRoots(true).size();
+    Assert.assertEquals(afterTrashRootsNum2 + 1, trashRootsNum3);
+
+    // Case 3: single /Users mount point for all users
+    fsTarget.mkdirs(new Path(targetTestRoot, "Users/.Trash/user1"));
+    fsTarget.mkdirs(new Path(targetTestRoot, "Users/.Trash/user2"));
+    Configuration conf4 = new Configuration(conf2);
+    ConfigUtil.addLink(conf4, "/Users",
+        new Path(targetTestRoot, "Users").toUri());
+    FileSystem fsView4 = FileSystem.get(FsConstants.VIEWFS_URI, conf4);
+    int trashRootsNum4 = fsView4.getTrashRoots(true).size();
+    Assert.assertEquals(afterTrashRootsNum2 + 2, trashRootsNum4);
+  }
+
+  /**
+   * Test localized trash roots in getTrashRoots() for current user.
+   */
+  @Test
+  public void testTrashRootsCurrentUser() throws IOException {
+    String currentUser =
+        UserGroupInformation.getCurrentUser().getShortUserName();
+    Configuration conf2 = new Configuration(conf);
+    conf2.setBoolean(CONFIG_VIEWFS_MOUNT_POINT_LOCAL_TRASH, true);
+    FileSystem fsView2 = FileSystem.get(FsConstants.VIEWFS_URI, conf2);
+
+    int beforeTrashRootNum = fsView.getTrashRoots(false).size();
+    int beforeTrashRootNum2 = fsView2.getTrashRoots(false).size();
+    Assert.assertEquals(beforeTrashRootNum, beforeTrashRootNum2);
+
+    fsView.mkdirs(new Path("/data/" + TRASH_PREFIX + "/" + currentUser));
+    fsView.mkdirs(new Path("/data/" + TRASH_PREFIX + "/user2"));
+    fsView.mkdirs(new Path("/user/" + TRASH_PREFIX + "/" + currentUser));
+    fsView.mkdirs(new Path("/user/" + TRASH_PREFIX + "/user4"));
+    fsView.mkdirs(new Path("/user2/" + TRASH_PREFIX + "/user5"));
+    int afterTrashRootsNum = fsView.getTrashRoots(false).size();
+    int afterTrashRootsNum2 = fsView2.getTrashRoots(false).size();
+    Assert.assertEquals(beforeTrashRootNum, afterTrashRootsNum);
+    Assert.assertEquals(beforeTrashRootNum2 + 2, afterTrashRootsNum2);
+  }
+
   @Test
   public void testCheckOwnerWithFileStatus()
       throws IOException, InterruptedException {