HDFS-15082. RBF: Check each component length of destination path when add/update mount entry. Contributed by Xiaoqiao He.

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

@@ -270,6 +270,9 @@ public class RBFConfigKeys extends CommonConfigurationKeysPublic {
   public static final String DFS_ROUTER_ADMIN_ENABLE =
       FEDERATION_ROUTER_PREFIX + "admin.enable";
   public static final boolean DFS_ROUTER_ADMIN_ENABLE_DEFAULT = true;
+  public static final String DFS_ROUTER_ADMIN_MAX_COMPONENT_LENGTH_KEY =
+      FEDERATION_ROUTER_PREFIX + "fs-limits.max-component-length";
+  public static final int DFS_ROUTER_ADMIN_MAX_COMPONENT_LENGTH_DEFAULT = 0;
 
   // HDFS Router-based federation web
   public static final String DFS_ROUTER_HTTP_ENABLE =

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

@@ -32,9 +32,11 @@ import java.util.Set;
 import com.google.common.base.Preconditions;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.StorageType;
 import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.hdfs.DFSUtil;
+import org.apache.hadoop.hdfs.protocol.FSLimitException.PathComponentTooLongException;
 import org.apache.hadoop.hdfs.protocol.HdfsConstants;
 import org.apache.hadoop.hdfs.protocol.HdfsFileStatus;
 import org.apache.hadoop.hdfs.protocol.proto.RouterProtocolProtos.RouterAdminProtocolService;
@@ -123,6 +125,7 @@ public class RouterAdminServer extends AbstractService
   private static String superGroup;
   private static boolean isPermissionEnabled;
   private boolean iStateStoreCache;
+  private final long maxComponentLength;
 
   public RouterAdminServer(Configuration conf, Router router)
       throws IOException {
@@ -177,6 +180,10 @@ public class RouterAdminServer extends AbstractService
     router.setAdminServerAddress(this.adminAddress);
     iStateStoreCache =
         router.getSubclusterResolver() instanceof StateStoreCache;
+    // The mount table destination path length limit keys.
+    this.maxComponentLength = (int) conf.getLongBytes(
+        RBFConfigKeys.DFS_ROUTER_ADMIN_MAX_COMPONENT_LENGTH_KEY,
+        RBFConfigKeys.DFS_ROUTER_ADMIN_MAX_COMPONENT_LENGTH_DEFAULT);
 
     GenericRefreshProtocolServerSideTranslatorPB genericRefreshXlator =
         new GenericRefreshProtocolServerSideTranslatorPB(this);
@@ -249,6 +256,50 @@ public class RouterAdminServer extends AbstractService
     }
   }
 
+  /**
+   * Verify each component name of a destination path for fs limit.
+   *
+   * @param destPath destination path name of mount point.
+   * @throws PathComponentTooLongException destination path name is too long.
+   */
+  void verifyMaxComponentLength(String destPath)
+      throws PathComponentTooLongException {
+    if (maxComponentLength <= 0) {
+      return;
+    }
+    if (destPath == null) {
+      return;
+    }
+    String[] components = destPath.split(Path.SEPARATOR);
+    for (String component : components) {
+      int length = component.length();
+      if (length > maxComponentLength) {
+        PathComponentTooLongException e = new PathComponentTooLongException(
+            maxComponentLength, length, destPath, component);
+        throw e;
+      }
+    }
+  }
+
+  /**
+   * Verify each component name of every destination path of mount table
+   * for fs limit.
+   *
+   * @param mountTable mount point.
+   * @throws PathComponentTooLongException destination path name is too long.
+   */
+  void verifyMaxComponentLength(MountTable mountTable)
+      throws PathComponentTooLongException {
+    if (mountTable != null) {
+      List<RemoteLocation> dests = mountTable.getDestinations();
+      if (dests != null && !dests.isEmpty()) {
+        for (RemoteLocation dest : dests) {
+          verifyMaxComponentLength(dest.getDest());
+        }
+      }
+    }
+  }
+
   @Override
   protected void serviceInit(Configuration configuration) throws Exception {
     this.conf = configuration;
@@ -272,6 +323,9 @@ public class RouterAdminServer extends AbstractService
   @Override
   public AddMountTableEntryResponse addMountTableEntry(
       AddMountTableEntryRequest request) throws IOException {
+    // Checks max component length limit.
+    MountTable mountTable = request.getEntry();
+    verifyMaxComponentLength(mountTable);
     return getMountTableStore().addMountTableEntry(request);
   }
 
@@ -280,6 +334,8 @@ public class RouterAdminServer extends AbstractService
       UpdateMountTableEntryRequest request) throws IOException {
     MountTable updateEntry = request.getEntry();
     MountTable oldEntry = null;
+    // Checks max component length limit.
+    verifyMaxComponentLength(updateEntry);
     if (this.router.getSubclusterResolver() instanceof MountTableResolver) {
       MountTableResolver mResolver =
           (MountTableResolver) this.router.getSubclusterResolver();

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

@@ -263,6 +263,18 @@
     </description>
   </property>
 
+  <property>
+    <name>dfs.federation.router.fs-limits.max-component-length</name>
+    <value>0</value>
+    <description>
+      Defines the maximum number of bytes in UTF-8 encoding in each
+      component of a path at Router side. A value of 0 will disable the check.
+      Support multiple size unit suffix(case insensitive). It is act as
+      configuration dfs.namenode.fs-limits.max-component-length at NameNode
+      side.
+    </description>
+  </property>
+
   <property>
     <name>dfs.federation.router.file.resolver.client.class</name>
     <value>org.apache.hadoop.hdfs.server.federation.resolver.MountTableResolver</value>

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

@@ -53,6 +53,8 @@ import org.apache.hadoop.hdfs.server.federation.store.protocol.AddMountTableEntr
 import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesRequest;
 import org.apache.hadoop.hdfs.server.federation.store.protocol.GetMountTableEntriesResponse;
 import org.apache.hadoop.hdfs.server.federation.store.protocol.RemoveMountTableEntryRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.UpdateMountTableEntryRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.UpdateMountTableEntryResponse;
 import org.apache.hadoop.hdfs.server.federation.store.records.MountTable;
 import org.apache.hadoop.security.AccessControlException;
 import org.apache.hadoop.security.UserGroupInformation;
@@ -90,6 +92,7 @@ public class TestRouterMountTable {
         .admin()
         .rpc()
         .build();
+    conf.setInt(RBFConfigKeys.DFS_ROUTER_ADMIN_MAX_COMPONENT_LENGTH_KEY, 20);
     cluster.addRouterOverrides(conf);
     cluster.startCluster();
     cluster.startRouters();
@@ -189,6 +192,55 @@ public class TestRouterMountTable {
     return addResponse.getStatus();
   }
 
+  /**
+   * Update a mount table entry to the mount table through the admin API.
+   * @param entry Mount table entry to update.
+   * @return If it was successfully update.
+   * @throws IOException Problems adding entries.
+   */
+  private boolean updateMountTable(final MountTable entry) throws IOException {
+    RouterClient client = routerContext.getAdminClient();
+    MountTableManager mountTableManager = client.getMountTableManager();
+    UpdateMountTableEntryRequest updateRequest =
+        UpdateMountTableEntryRequest.newInstance(entry);
+    UpdateMountTableEntryResponse updateResponse =
+        mountTableManager.updateMountTableEntry(updateRequest);
+
+    // Reload the Router cache
+    mountTable.loadCache(true);
+
+    return updateResponse.getStatus();
+  }
+
+  /**
+   * Verify that the maximum number of bytes in each component of a
+   * destination path.
+   */
+  @Test
+  public void testMountPointLimit() throws Exception {
+    // Add mount table entry
+    MountTable addEntry = MountTable.newInstance("/testdir-shortlength",
+        Collections.singletonMap("ns0", "/testdir-shortlength"));
+    assertTrue(addMountTable(addEntry));
+
+    final MountTable longAddEntry = MountTable.newInstance(
+        "/testdir-verylonglength",
+        Collections.singletonMap("ns0", "/testdir-verylonglength"));
+    LambdaTestUtils.intercept(IOException.class,
+        "The maximum path component name limit of testdir-verylonglength in "
+            + "directory /testdir-verylonglength is exceeded",
+        () -> addMountTable(longAddEntry));
+
+    final MountTable updateEntry = MountTable.newInstance(
+        "/testdir-shortlength",
+        Collections.singletonMap("ns0", "/testdir-shortlength-change-to-long"));
+    LambdaTestUtils.intercept(IOException.class,
+        "The maximum path component name limit of " +
+            "testdir-shortlength-change-to-long in directory " +
+            "/testdir-shortlength-change-to-long is exceeded",
+        () -> updateMountTable(updateEntry));
+  }
+
   /**
    * Verify that the file/dir listing contains correct date/time information.
    */