RBF: The construction of the trash path in the downstream nameservice should be based on src locations.

Author: fuchaohong

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/MountTableResolver.java

@@ -24,6 +24,8 @@
 import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.FEDERATION_MOUNT_TABLE_MAX_CACHE_SIZE_DEFAULT;
 import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.FEDERATION_MOUNT_TABLE_CACHE_ENABLE;
 import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.FEDERATION_MOUNT_TABLE_CACHE_ENABLE_DEFAULT;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_TRASH_PATH_CREATED_BY_MOUNT_POINT;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_TRASH_PATH_CREATED_BY_MOUNT_POINT_DEFAULT;
 import static org.apache.hadoop.hdfs.DFSUtil.isParentEntry;
 
 import java.io.IOException;
@@ -90,6 +92,8 @@ public class MountTableResolver
   private final StateStoreService stateStore;
   /** Interface to the mount table store. */
   private MountTableStore mountTableStore;
+  /** Configuration for the RPC server. */
+  private Configuration conf;
 
   /** If the tree has been initialized. */
   private boolean init = false;
@@ -131,6 +135,7 @@ public MountTableResolver(Configuration conf, StateStoreService store) {
   public MountTableResolver(Configuration conf, Router routerService,
       StateStoreService store) {
     this.router = routerService;
+    this.conf = conf;
     if (store != null) {
       this.stateStore = store;
     } else if (this.router != null) {
@@ -460,9 +465,13 @@ public PathLocation getDestinationForPath(final String path)
         this.getLocCacheAccess().increment();
       }
       if (isTrashPath(path)) {
+        boolean useMountPointCreateTrashPath = conf.getBoolean(
+                DFS_ROUTER_TRASH_PATH_CREATED_BY_MOUNT_POINT,
+                DFS_ROUTER_TRASH_PATH_CREATED_BY_MOUNT_POINT_DEFAULT);
         List<RemoteLocation> remoteLocations = new ArrayList<>();
         for (RemoteLocation remoteLocation : res.getDestinations()) {
-          remoteLocations.add(new RemoteLocation(remoteLocation, path));
+          remoteLocations.add(new RemoteLocation(remoteLocation,
+                  useMountPointCreateTrashPath ? path : getTrashRoot() + "/Current" + remoteLocation.getDest()));
         }
         return new PathLocation(path, remoteLocations,
             res.getDestinationOrder());

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RBFConfigKeys.java

@@ -86,6 +86,10 @@ public class RBFConfigKeys extends CommonConfigurationKeysPublic {
   public static final int
       DFS_ROUTER_METRICS_TOP_NUM_TOKEN_OWNERS_KEY_DEFAULT = 10;
 
+  public static final String DFS_ROUTER_TRASH_PATH_CREATED_BY_MOUNT_POINT =
+          FEDERATION_ROUTER_PREFIX + "trash-path.created-by.mount-point";
+  public static final boolean DFS_ROUTER_TRASH_PATH_CREATED_BY_MOUNT_POINT_DEFAULT = false;
+
   // HDFS Router heartbeat
   public static final String DFS_ROUTER_HEARTBEAT_ENABLE =
       FEDERATION_ROUTER_PREFIX + "heartbeat.enable";

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/resources/hdfs-rbf-default.xml

@@ -673,6 +673,15 @@
     </description>
   </property>
 
+  <property>
+    <name>dfs.federation.router.trash-path.created-by.mount-point</name>
+    <value>false</value>
+    <description>
+      Set to true to use the mount point to create trash path
+      when path is the trail associated with the Trash.
+    </description>
+  </property>
+
   <property>
     <name>dfs.federation.router.keytab.file</name>
     <value></value>

hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterTrash.java

@@ -46,6 +46,7 @@
 import java.util.Collections;
 
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_TRASH_INTERVAL_KEY;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_TRASH_PATH_CREATED_BY_MOUNT_POINT;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertEquals;
@@ -70,6 +71,8 @@ public class TestRouterTrash {
   private static String ns1;
   private static final String MOUNT_POINT = "/home/data";
   private static final String FILE = MOUNT_POINT + "/file1";
+  private static final String DST_PATH = "/home/dst_data";
+  private static final String DST_FILE = DST_PATH + "/file1";
   private static final String TRASH_ROOT = "/user/" + TEST_USER + "/.Trash";
   private static final String CURRENT = "/Current";
 
@@ -276,6 +279,81 @@ public void testDeleteToTrashExistMountPoint() throws IOException,
     assertEquals(2, fileStatuses.length);
   }
 
+  private void deleteToTrashExistMountPoint() throws IOException,
+          URISyntaxException, InterruptedException {
+    MountTable addEntry = MountTable.newInstance(MOUNT_POINT,
+            Collections.singletonMap(ns0, DST_PATH));
+    assertTrue(addMountTable(addEntry));
+
+    // current user client
+    DFSClient client = nnContext.getClient();
+    client.setOwner("/", TEST_USER, TEST_USER);
+    UserGroupInformation ugi = UserGroupInformation.
+            createRemoteUser(TEST_USER);
+    // test user client
+    client = nnContext.getClient(ugi);
+    client.mkdirs(DST_PATH, new FsPermission("777"), true);
+    assertTrue(client.exists(DST_PATH));
+    // create test file
+    client.create(DST_FILE, true);
+
+    Path filePath = new Path(FILE);
+    FileStatus[] fileStatuses = routerFs.listStatus(filePath);
+    assertEquals(1, fileStatuses.length);
+    assertEquals(TEST_USER, fileStatuses[0].getOwner());
+
+    // move to Trash.
+    Configuration routerConf = routerContext.getConf();
+    FileSystem fs =
+            DFSTestUtil.getFileSystemAs(ugi, routerConf);
+    Trash trash = new Trash(fs, routerConf);
+    assertTrue(trash.moveToTrash(filePath));
+  }
+
+  @Test
+  public void testTrashPathStructure() throws IOException,
+          URISyntaxException, InterruptedException {
+    // Trash path created by dst_path by default.
+    deleteToTrashExistMountPoint();
+    FileStatus[] fileStatuses = nnFs.listStatus(
+            new Path(TRASH_ROOT + CURRENT + DST_PATH));
+    assertEquals(1, fileStatuses.length);
+    assertTrue(nnFs.exists(new Path(TRASH_ROOT + CURRENT + DST_FILE)));
+
+    // Re-build and start a federated cluster with dfs.federation.router.trash-path.created-by.mount-point=true.
+    tearDown();
+    cluster = new StateStoreDFSCluster(false, 2);
+    Configuration conf = new RouterConfigBuilder()
+            .stateStore()
+            .admin()
+            .rpc()
+            .http()
+            .build();
+    conf.set(FS_TRASH_INTERVAL_KEY, "100");
+    conf.setBoolean(DFS_ROUTER_TRASH_PATH_CREATED_BY_MOUNT_POINT, true);
+    cluster.addRouterOverrides(conf);
+    cluster.startCluster();
+    cluster.startRouters();
+    cluster.waitClusterUp();
+
+    ns0 = cluster.getNameservices().get(0);
+    ns1 = cluster.getNameservices().get(1);
+
+    routerContext = cluster.getRandomRouter();
+    routerFs = routerContext.getFileSystem();
+    nnContext = cluster.getNamenode(ns0, null);
+    nnFs = nnContext.getFileSystem();
+    Router router = routerContext.getRouter();
+    mountTable = (MountTableResolver) router.getSubclusterResolver();
+
+    // Trash path created by mount_point.
+    deleteToTrashExistMountPoint();
+    fileStatuses = nnFs.listStatus(
+            new Path(TRASH_ROOT + CURRENT + MOUNT_POINT));
+    assertEquals(1, fileStatuses.length);
+    assertTrue(nnFs.exists(new Path(TRASH_ROOT + CURRENT + FILE)));
+  }
+
   @Test
   public void testIsTrashPath() throws IOException {
     UserGroupInformation ugi = UserGroupInformation.getCurrentUser();