HADOOP-19217. Introduce getTrashPolicy to FileSystem API

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

@@ -3429,6 +3429,29 @@ public Collection<? extends BlockStoragePolicySpi> getAllStoragePolicies()
         + " doesn't support getAllStoragePolicies");
   }
 
+  /**
+   * Get the trash policy implementation used by this FileSystem. This trash policy
+   * is used by classes of {@link Trash} to implement the trash behavior.
+   * <p>
+   * FileSystem implementation can consider overriding this method to handle
+   * situation where a single FileSystem client shares a configuration, but
+   * each FileSystem scheme requires a distinct TrashPolicy implementation.
+   *
+   * @param conf configuration which can be used to choose the TrashPolicy
+   *             implementation.
+   * @return TrashPolicy implementation by this filesystem.
+   *         The default implementation returns the configured TrashPolicy
+   *         based on the value of the configuration parameter fs.trash.classname
+   *         of the passed configuration.
+   */
+  @InterfaceAudience.Public
+  @InterfaceStability.Unstable
+  public TrashPolicy getTrashPolicy(Configuration conf) {
+    Class<? extends TrashPolicy> trashClass = conf.getClass(
+        "fs.trash.classname", TrashPolicyDefault.class, TrashPolicy.class);
+    return ReflectionUtils.newInstance(trashClass, conf);
+  }
+
   /**
    * Get the root directory of Trash for current user when the path specified
    * is deleted.

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

@@ -686,6 +686,11 @@ public Collection<? extends BlockStoragePolicySpi> getAllStoragePolicies()
     return fs.getAllStoragePolicies();
   }
 
+  @Override
+  public TrashPolicy getTrashPolicy(Configuration conf) {
+    return fs.getTrashPolicy(conf);
+  }
+
   @Override
   public Path getTrashRoot(Path path) {
     return fs.getTrashRoot(path);

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

@@ -126,6 +126,16 @@ public Path getCurrentTrashDir(Path path) throws IOException {
    */
   public abstract Runnable getEmptier() throws IOException;
 
+  /**
+   * Return the deletion interval associated with the trash policy. Deletion interval
+   * can be used as trash emptier {@link #getEmptier()} check interval, as well as used
+   * as the trash expiry configuration used in the checkpoint {@link #deleteCheckpoint()}
+   * @return deletion interval.
+   */
+  public final long getDeletionInterval() {
+    return deletionInterval;
+  }
+
   /**
    * Get an instance of the configured TrashPolicy based on the value
    * of the configuration parameter fs.trash.classname.
@@ -146,18 +156,19 @@ public static TrashPolicy getInstance(Configuration conf, FileSystem fs, Path ho
   }
 
   /**
-   * Get an instance of the configured TrashPolicy based on the value
-   * of the configuration parameter fs.trash.classname.
+   * Get an instance of the TrashPolicy associated with the FileSystem implementation of
+   * {@link FileSystem#getTrashPolicy(Configuration)}. The configuration passed might be used
+   * by the FileSystem implementation to pick the {@link TrashPolicy} implementation. The default
+   * {@link FileSystem#getTrashPolicy(Configuration)} checks fs.trash.classname to pick the
+   * {@link TrashPolicy} implementation.
    *
    * @param conf the configuration to be used
    * @param fs the file system to be used
    * @return an instance of TrashPolicy
    */
   public static TrashPolicy getInstance(Configuration conf, FileSystem fs) {
-    Class<? extends TrashPolicy> trashClass = conf.getClass(
-        "fs.trash.classname", TrashPolicyDefault.class, TrashPolicy.class);
-    TrashPolicy trash = ReflectionUtils.newInstance(trashClass, conf);
-    trash.initialize(conf, fs); // initialize TrashPolicy
-    return trash;
+    TrashPolicy trashPolicy = fs.getTrashPolicy(conf);
+    trashPolicy.initialize(conf, fs); // initialize TrashPolicy
+    return trashPolicy;
   }
 }

hadoop-common-project/hadoop-common/src/site/markdown/filesystem/filesystem.md

@@ -636,6 +636,36 @@ The path does not have to exist, but the path does need to be valid and reconcil
 * The path returned is a directory
 
 
+###  `TrashPolicy getTrashPolicy(Configuration conf)`
+
+Get the trash policy implementation used by this FileSystem.
+
+This method allows different FileSystem implementations to use different TrashPolicy
+implementations. This is important in environments where multiple FileSystem schemes
+are used (e.g., HDFS and Ozone), as each may require a distinct TrashPolicy.
+
+#### Preconditions
+
+#### Postconditions
+
+    result = a valid TrashPolicy instance associated with the FileSystem implementation
+
+The default implementation:
+1. Reads the configuration parameter `fs.trash.classname` (defaults to `TrashPolicyDefault`)
+2. Instantiates the specified TrashPolicy class
+3. Initializes the TrashPolicy with the given configuration
+4. Returns the initialized TrashPolicy
+
+#### Implementation Notes
+
+* FileSystem implementations MAY override this method to provide filesystem-specific
+  TrashPolicy implementations. For example, Ozone `getTrashPolicy` can return its custom trash policy,
+  while HDFS can still use `TrashPolicyDefault`.
+* The returned TrashPolicy should not be null.
+* FileSystem implementations with multiple child file systems (e.g. `ViewFileSystem`)
+  should NOT implement this method since the Hadoop trash mechanism should resolve to the underlying filesystem
+  before invoking `getTrashPolicy`.
+
 ## <a name="state_changing_operations"></a> State Changing Operations
 
 ### `boolean mkdirs(Path p, FsPermission permission)`

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

@@ -229,6 +229,8 @@ public BlockStoragePolicySpi getStoragePolicy(final Path src)
     public Collection<? extends BlockStoragePolicySpi> getAllStoragePolicies()
         throws IOException;
 
+    public TrashPolicy getTrashPolicy(Configuration conf);
+
     public Path getTrashRoot(Path path) throws IOException;
 
     public Collection<FileStatus> getTrashRoots(boolean allUsers) throws IOException;

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

@@ -39,6 +39,8 @@
 import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -629,10 +631,23 @@ public void testNonDefaultFS() throws IOException {
   public void testPluggableTrash() throws IOException {
     Configuration conf = new Configuration();
 
-    // Test plugged TrashPolicy
-    conf.setClass("fs.trash.classname", TestTrashPolicy.class, TrashPolicy.class);
-    Trash trash = new Trash(conf);
-    assertTrue(trash.getTrashPolicy().getClass().equals(TestTrashPolicy.class));
+    {
+      // Test plugged TrashPolicy
+      conf.setClass("fs.trash.classname", TestTrashPolicy.class, TrashPolicy.class);
+      Trash trash = new Trash(conf);
+      assertInstanceOf(TestTrashPolicy.class, trash.getTrashPolicy());
+    }
+
+    {
+      // Test FileSystem implementation that implements getTrashPolicy to return custom TrashPolicy
+      // regardless of fs.trash.classname
+      conf.setClass("fs.file.impl", TestLFSWithCustomTrashPolicy.class, FileSystem.class);
+      conf.setBoolean("fs.file.impl.disable.cache", true);
+      FileSystem fs = FileSystem.getLocal(conf);
+      conf.set("fs.defaultFS", fs.getUri().toString());
+      Trash trash = new Trash(fs, conf);
+      assertInstanceOf(TestLFSWithCustomTrashPolicy.CustomTrashPolicy.class, trash.getTrashPolicy());
+    }
   }
 
   @Test
@@ -873,6 +888,24 @@ public void setUri(String uri){
       uriName = URI.create(uri);
     }
   }
+
+  public static class TestLFSWithCustomTrashPolicy extends TestLFS {
+
+    @Override
+    public String getScheme() {
+      return "testlfswithcustomtrashpolicy";
+    }
+
+    @Override
+    public TrashPolicy getTrashPolicy(Configuration conf) {
+      return new CustomTrashPolicy();
+    }
+
+    public static class CustomTrashPolicy extends TrashPolicyDefault {
+    }
+  }
+
+
 
   /**
    *  test same file deletion - multiple time