From aa748c6bca6ee75eec24c27f7fb32eda01b720b6 Mon Sep 17 00:00:00 2001
From: Yiqun Lin <yqlin@apache.org>
Date: Fri, 9 Mar 2018 15:42:57 +0800
Subject: [PATCH] HDFS-13233. RBF: MountTableResolver doesn't return the
 correct mount point of the given path. Contributed by wangzhiyuan.

(cherry picked from commit 122805b43acff2b094bd984fa76dbc8d2e110edd)
---
 .../resolver/MountTableResolver.java          | 13 ++++++++++-
 .../resolver/TestMountTableResolver.java      | 23 +++++++++++++++++++
 .../federation/router/TestRouterQuota.java    |  2 +-
 3 files changed, 36 insertions(+), 2 deletions(-)

diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/MountTableResolver.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/MountTableResolver.java
index 374e3ba85f2..dac6f7f424f 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/MountTableResolver.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/MountTableResolver.java
@@ -521,6 +521,17 @@ public class MountTableResolver
     return this.defaultNameService;
   }
 
+  private boolean isParentEntry(final String path, final String parent) {
+    if (!path.startsWith(parent)) {
+      return false;
+    }
+    if (path.equals(parent)) {
+      return true;
+    }
+    return path.charAt(parent.length()) == Path.SEPARATOR_CHAR
+        || parent.equals(Path.SEPARATOR);
+  }
+
   /**
    * Find the deepest mount point for a path.
    * @param path Path to look for.
@@ -530,7 +541,7 @@ public class MountTableResolver
     readLock.lock();
     try {
       Entry<String, MountTable> entry = this.tree.floorEntry(path);
-      while (entry != null && !path.startsWith(entry.getKey())) {
+      while (entry != null && !isParentEntry(path, entry.getKey())) {
         entry = this.tree.lowerEntry(entry.getKey());
       }
       if (entry == null) {
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/federation/resolver/TestMountTableResolver.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/federation/resolver/TestMountTableResolver.java
index fa2f89c5d19..a09daf0975b 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/federation/resolver/TestMountTableResolver.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/federation/resolver/TestMountTableResolver.java
@@ -178,6 +178,29 @@ public class TestMountTableResolver {
     }
   }
 
+  @Test
+  public void testGetMountPoint() throws IOException {
+    // Check get the mount table entry for a path
+    MountTable mtEntry;
+    mtEntry = mountTable.getMountPoint("/");
+    assertTrue(mtEntry.getSourcePath().equals("/"));
+
+    mtEntry = mountTable.getMountPoint("/user");
+    assertTrue(mtEntry.getSourcePath().equals("/user"));
+
+    mtEntry = mountTable.getMountPoint("/user/a");
+    assertTrue(mtEntry.getSourcePath().equals("/user/a"));
+
+    mtEntry = mountTable.getMountPoint("/user/a/");
+    assertTrue(mtEntry.getSourcePath().equals("/user/a"));
+
+    mtEntry = mountTable.getMountPoint("/user/a/11");
+    assertTrue(mtEntry.getSourcePath().equals("/user/a"));
+
+    mtEntry = mountTable.getMountPoint("/user/a1");
+    assertTrue(mtEntry.getSourcePath().equals("/user"));
+  }
+
   @Test
   public void testGetMountPoints() throws IOException {
 
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterQuota.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterQuota.java
index b7c1c631d37..2b40bbd9248 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterQuota.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterQuota.java
@@ -414,7 +414,7 @@ public class TestRouterQuota {
 
     // mkdir and write a new file
     final FileSystem routerFs = routerContext.getFileSystem();
-    routerFs.mkdirs(new Path(path + UUID.randomUUID()));
+    routerFs.mkdirs(new Path(path + "/" + UUID.randomUUID()));
     DFSClient routerClient = routerContext.getClient();
     routerClient.create(path + "/file", true).close();
     appendData(path + "/file", routerClient, BLOCK_SIZE);