HDDS-14177. OFS#getTrashRoot should use the internal FS username instead of current UGI (#9502)

Author: ivandika3

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/OFSPath.java

@@ -365,13 +365,25 @@ public static String getTempMountBucketNameOfCurrentUser()
    * @return trash root for the given path.
    */
   public Path getTrashRoot() {
+    return getTrashRoot(null);
+  }
+
+  /**
+   * Return trash root for the given path and username.
+   * The username can be specified to use the proxy user instead of {@link UserGroupInformation#getCurrentUser()}.
+   * @param username the username used to get the trash root path. If it is not specified,
+   *                 will fall back to {@link UserGroupInformation#getCurrentUser()}.
+   * @return trash root for the given path and username.
+   */
+  public Path getTrashRoot(String username) {
     if (!this.isKey()) {
       throw new RuntimeException("Recursive rm of volume or bucket with trash" +
           " enabled is not permitted. Consider using the -skipTrash option.");
     }
     try {
-      final String username =
-              UserGroupInformation.getCurrentUser().getShortUserName();
+      if (StringUtils.isEmpty(username)) {
+        username = UserGroupInformation.getCurrentUser().getShortUserName();
+      }
       final Path pathRoot = new Path(
           OZONE_OFS_URI_SCHEME, authority, OZONE_URI_DELIMITER);
       final Path pathToVolume = new Path(pathRoot, volumeName);

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/AbstractRootedOzoneFileSystemTest.java

@@ -1719,6 +1719,14 @@ private void checkFileStatusOwner(int expectedSize, String expectedOwner,
     res.forEach(e -> assertEquals(expectedOwner, e.getOwner()));
   }
 
+  @Test
+  void testGetTrashRoot() {
+    String testKeyName = "keyToBeDeleted";
+    Path keyPath1 = new Path(bucketPath, testKeyName);
+    assertEquals(new Path(rootPath + volumeName + "/" + bucketName + "/" +
+        TRASH_PREFIX + "/" +  USER1 + "/"), userOfs.getTrashRoot(keyPath1));
+  }
+
   /**
    * Test getTrashRoots() in OFS. Different from the existing test for o3fs.
    */

hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicRootedOzoneFileSystem.java

@@ -1018,7 +1018,7 @@ public String getUsername() {
   public Path getTrashRoot(Path path) {
     OFSPath ofsPath = new OFSPath(path,
         ozoneConfiguration);
-    return this.makeQualified(ofsPath.getTrashRoot());
+    return this.makeQualified(ofsPath.getTrashRoot(getUsername()));
   }
 
   /**