hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From w...@apache.org
Subject [23/39] hadoop git commit: HDFS-13215. RBF: Move Router to its own module. Contributed by Wei Yan
Date Tue, 20 Mar 2018 05:18:49 GMT
http://git-wip-us.apache.org/repos/asf/hadoop/blob/6e2b5fa4/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterAdminCLI.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterAdminCLI.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterAdminCLI.java
new file mode 100644
index 0000000..6111c6b
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterAdminCLI.java
@@ -0,0 +1,477 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.federation.router;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.io.ByteArrayOutputStream;
+
+import java.io.PrintStream;
+import java.net.InetSocketAddress;
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.protocol.HdfsConstants;
+import org.apache.hadoop.hdfs.server.federation.RouterConfigBuilder;
+import org.apache.hadoop.hdfs.server.federation.RouterDFSCluster.RouterContext;
+import org.apache.hadoop.hdfs.server.federation.StateStoreDFSCluster;
+import org.apache.hadoop.hdfs.server.federation.resolver.MountTableManager;
+import org.apache.hadoop.hdfs.server.federation.resolver.RemoteLocation;
+import org.apache.hadoop.hdfs.server.federation.resolver.order.DestinationOrder;
+import org.apache.hadoop.hdfs.server.federation.store.StateStoreService;
+import org.apache.hadoop.hdfs.server.federation.store.impl.MountTableStoreImpl;
+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.records.MountTable;
+import org.apache.hadoop.hdfs.tools.federation.RouterAdmin;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.test.GenericTestUtils;
+import org.apache.hadoop.util.ToolRunner;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import com.google.common.base.Supplier;
+
+/**
+ * Tests Router admin commands.
+ */
+public class TestRouterAdminCLI {
+  private static StateStoreDFSCluster cluster;
+  private static RouterContext routerContext;
+  private static StateStoreService stateStore;
+
+  private static RouterAdmin admin;
+  private static RouterClient client;
+
+  private static final String TEST_USER = "test-user";
+
+  private final ByteArrayOutputStream out = new ByteArrayOutputStream();
+  private static final PrintStream OLD_OUT = System.out;
+
+  @BeforeClass
+  public static void globalSetUp() throws Exception {
+    cluster = new StateStoreDFSCluster(false, 1);
+    // Build and start a router with State Store + admin + RPC
+    Configuration conf = new RouterConfigBuilder()
+        .stateStore()
+        .admin()
+        .rpc()
+        .build();
+    cluster.addRouterOverrides(conf);
+
+    // Start routers
+    cluster.startRouters();
+
+    routerContext = cluster.getRandomRouter();
+    Router router = routerContext.getRouter();
+    stateStore = router.getStateStore();
+
+    Configuration routerConf = new Configuration();
+    InetSocketAddress routerSocket = router.getAdminServerAddress();
+    routerConf.setSocketAddr(RBFConfigKeys.DFS_ROUTER_ADMIN_ADDRESS_KEY,
+        routerSocket);
+    admin = new RouterAdmin(routerConf);
+    client = routerContext.getAdminClient();
+  }
+
+  @AfterClass
+  public static void tearDownCluster() {
+    cluster.stopRouter(routerContext);
+    cluster.shutdown();
+    cluster = null;
+  }
+
+  @After
+  public void tearDown() {
+    // set back system out
+    System.setOut(OLD_OUT);
+  }
+
+  @Test
+  public void testAddMountTable() throws Exception {
+    String nsId = "ns0";
+    String src = "/test-addmounttable";
+    String dest = "/addmounttable";
+    String[] argv = new String[] {"-add", src, nsId, dest};
+    assertEquals(0, ToolRunner.run(admin, argv));
+
+    stateStore.loadCache(MountTableStoreImpl.class, true);
+    GetMountTableEntriesRequest getRequest = GetMountTableEntriesRequest
+        .newInstance(src);
+    GetMountTableEntriesResponse getResponse = client.getMountTableManager()
+        .getMountTableEntries(getRequest);
+    MountTable mountTable = getResponse.getEntries().get(0);
+
+    List<RemoteLocation> destinations = mountTable.getDestinations();
+    assertEquals(1, destinations.size());
+
+    assertEquals(src, mountTable.getSourcePath());
+    assertEquals(nsId, destinations.get(0).getNameserviceId());
+    assertEquals(dest, destinations.get(0).getDest());
+    assertFalse(mountTable.isReadOnly());
+
+    // test mount table update behavior
+    dest = dest + "-new";
+    argv = new String[] {"-add", src, nsId, dest, "-readonly"};
+    assertEquals(0, ToolRunner.run(admin, argv));
+    stateStore.loadCache(MountTableStoreImpl.class, true);
+
+    getResponse = client.getMountTableManager()
+        .getMountTableEntries(getRequest);
+    mountTable = getResponse.getEntries().get(0);
+    assertEquals(2, mountTable.getDestinations().size());
+    assertEquals(nsId, mountTable.getDestinations().get(1).getNameserviceId());
+    assertEquals(dest, mountTable.getDestinations().get(1).getDest());
+    assertTrue(mountTable.isReadOnly());
+  }
+
+  @Test
+  public void testAddOrderMountTable() throws Exception {
+    testAddOrderMountTable(DestinationOrder.HASH);
+    testAddOrderMountTable(DestinationOrder.LOCAL);
+    testAddOrderMountTable(DestinationOrder.RANDOM);
+    testAddOrderMountTable(DestinationOrder.HASH_ALL);
+  }
+
+  private void testAddOrderMountTable(DestinationOrder order)
+      throws Exception {
+    final String mnt = "/" + order;
+    final String nsId = "ns0,ns1";
+    final String dest = "/";
+    String[] argv = new String[] {
+        "-add", mnt, nsId, dest, "-order", order.toString()};
+    assertEquals(0, ToolRunner.run(admin, argv));
+
+    // Check the state in the State Store
+    stateStore.loadCache(MountTableStoreImpl.class, true);
+    MountTableManager mountTable = client.getMountTableManager();
+    GetMountTableEntriesRequest request =
+        GetMountTableEntriesRequest.newInstance(mnt);
+    GetMountTableEntriesResponse response =
+        mountTable.getMountTableEntries(request);
+    List<MountTable> entries = response.getEntries();
+    assertEquals(1, entries.size());
+    assertEquals(2, entries.get(0).getDestinations().size());
+    assertEquals(order, response.getEntries().get(0).getDestOrder());
+  }
+
+  @Test
+  public void testListMountTable() throws Exception {
+    String nsId = "ns0";
+    String src = "/test-lsmounttable";
+    String dest = "/lsmounttable";
+    String[] argv = new String[] {"-add", src, nsId, dest};
+    assertEquals(0, ToolRunner.run(admin, argv));
+
+    // re-set system out for testing
+    System.setOut(new PrintStream(out));
+    stateStore.loadCache(MountTableStoreImpl.class, true);
+    argv = new String[] {"-ls", src};
+    assertEquals(0, ToolRunner.run(admin, argv));
+    assertTrue(out.toString().contains(src));
+
+    out.reset();
+    GetMountTableEntriesRequest getRequest = GetMountTableEntriesRequest
+        .newInstance("/");
+    GetMountTableEntriesResponse getResponse = client.getMountTableManager()
+        .getMountTableEntries(getRequest);
+
+    // Test ls command without input path, it will list
+    // mount table under root path.
+    argv = new String[] {"-ls"};
+    assertEquals(0, ToolRunner.run(admin, argv));
+    assertTrue(out.toString().contains(src));
+    String outStr = out.toString();
+    // verify if all the mount table are listed
+    for(MountTable entry: getResponse.getEntries()) {
+      assertTrue(outStr.contains(entry.getSourcePath()));
+    }
+  }
+
+  @Test
+  public void testRemoveMountTable() throws Exception {
+    String nsId = "ns0";
+    String src = "/test-rmmounttable";
+    String dest = "/rmmounttable";
+    String[] argv = new String[] {"-add", src, nsId, dest};
+    assertEquals(0, ToolRunner.run(admin, argv));
+
+    stateStore.loadCache(MountTableStoreImpl.class, true);
+    GetMountTableEntriesRequest getRequest = GetMountTableEntriesRequest
+        .newInstance(src);
+    GetMountTableEntriesResponse getResponse = client.getMountTableManager()
+        .getMountTableEntries(getRequest);
+    // ensure mount table added successfully
+    MountTable mountTable = getResponse.getEntries().get(0);
+    assertEquals(src, mountTable.getSourcePath());
+
+    argv = new String[] {"-rm", src};
+    assertEquals(0, ToolRunner.run(admin, argv));
+
+    stateStore.loadCache(MountTableStoreImpl.class, true);
+    getResponse = client.getMountTableManager()
+        .getMountTableEntries(getRequest);
+    assertEquals(0, getResponse.getEntries().size());
+
+    // remove an invalid mount table
+    String invalidPath = "/invalid";
+    System.setOut(new PrintStream(out));
+    argv = new String[] {"-rm", invalidPath};
+    assertEquals(0, ToolRunner.run(admin, argv));
+    assertTrue(out.toString().contains(
+        "Cannot remove mount point " + invalidPath));
+  }
+
+  @Test
+  public void testMountTableDefaultACL() throws Exception {
+    String[] argv = new String[] {"-add", "/testpath0", "ns0", "/testdir0"};
+    assertEquals(0, ToolRunner.run(admin, argv));
+
+    stateStore.loadCache(MountTableStoreImpl.class, true);
+    GetMountTableEntriesRequest getRequest = GetMountTableEntriesRequest
+        .newInstance("/testpath0");
+    GetMountTableEntriesResponse getResponse = client.getMountTableManager()
+        .getMountTableEntries(getRequest);
+    MountTable mountTable = getResponse.getEntries().get(0);
+
+    UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
+    String group = ugi.getGroups().isEmpty() ? ugi.getShortUserName()
+        : ugi.getPrimaryGroupName();
+    assertEquals(ugi.getShortUserName(), mountTable.getOwnerName());
+    assertEquals(group, mountTable.getGroupName());
+    assertEquals((short) 0755, mountTable.getMode().toShort());
+  }
+
+  @Test
+  public void testMountTablePermissions() throws Exception {
+    // re-set system out for testing
+    System.setOut(new PrintStream(out));
+    // use superuser to add new mount table with only read permission
+    String[] argv = new String[] {"-add", "/testpath2-1", "ns0", "/testdir2-1",
+        "-owner", TEST_USER, "-group", TEST_USER, "-mode", "0455"};
+    assertEquals(0, ToolRunner.run(admin, argv));
+
+    String superUser = UserGroupInformation.
+        getCurrentUser().getShortUserName();
+    // use normal user as current user to test
+    UserGroupInformation remoteUser = UserGroupInformation
+        .createRemoteUser(TEST_USER);
+    UserGroupInformation.setLoginUser(remoteUser);
+
+    // verify read permission by executing other commands
+    verifyExecutionResult("/testpath2-1", true, -1, -1);
+
+    // add new mount table with only write permission
+    argv = new String[] {"-add", "/testpath2-2", "ns0", "/testdir2-2",
+        "-owner", TEST_USER, "-group", TEST_USER, "-mode", "0255"};
+    assertEquals(0, ToolRunner.run(admin, argv));
+    verifyExecutionResult("/testpath2-2", false, 0, 0);
+
+    // set mount table entry with read and write permission
+    argv = new String[] {"-add", "/testpath2-3", "ns0", "/testdir2-3",
+        "-owner", TEST_USER, "-group", TEST_USER, "-mode", "0755"};
+    assertEquals(0, ToolRunner.run(admin, argv));
+    verifyExecutionResult("/testpath2-3", true, 0, 0);
+
+    // set back login user
+    remoteUser = UserGroupInformation.createRemoteUser(superUser);
+    UserGroupInformation.setLoginUser(remoteUser);
+  }
+
+  /**
+   * Verify router admin commands execution result.
+   *
+   * @param mount
+   *          target mount table
+   * @param canRead
+   *          whether can list mount tables under specified mount
+   * @param addCommandCode
+   *          expected return code of add command executed for specified mount
+   * @param rmCommandCode
+   *          expected return code of rm command executed for specified mount
+   * @throws Exception
+   */
+  private void verifyExecutionResult(String mount, boolean canRead,
+      int addCommandCode, int rmCommandCode) throws Exception {
+    String[] argv = null;
+    stateStore.loadCache(MountTableStoreImpl.class, true);
+
+    out.reset();
+    // execute ls command
+    argv = new String[] {"-ls", mount};
+    assertEquals(0, ToolRunner.run(admin, argv));
+    assertEquals(canRead, out.toString().contains(mount));
+
+    // execute add/update command
+    argv = new String[] {"-add", mount, "ns0", mount + "newdir"};
+    assertEquals(addCommandCode, ToolRunner.run(admin, argv));
+
+    stateStore.loadCache(MountTableStoreImpl.class, true);
+    // execute remove command
+    argv = new String[] {"-rm", mount};
+    assertEquals(rmCommandCode, ToolRunner.run(admin, argv));
+  }
+
+  @Test
+  public void testSetAndClearQuota() throws Exception {
+    String nsId = "ns0";
+    String src = "/test-QuotaMounttable";
+    String dest = "/QuotaMounttable";
+    String[] argv = new String[] {"-add", src, nsId, dest};
+    assertEquals(0, ToolRunner.run(admin, argv));
+
+    stateStore.loadCache(MountTableStoreImpl.class, true);
+    GetMountTableEntriesRequest getRequest = GetMountTableEntriesRequest
+        .newInstance(src);
+    GetMountTableEntriesResponse getResponse = client.getMountTableManager()
+        .getMountTableEntries(getRequest);
+    MountTable mountTable = getResponse.getEntries().get(0);
+    RouterQuotaUsage quotaUsage = mountTable.getQuota();
+
+    // verify the default quota set
+    assertEquals(RouterQuotaUsage.QUOTA_USAGE_COUNT_DEFAULT,
+        quotaUsage.getFileAndDirectoryCount());
+    assertEquals(HdfsConstants.QUOTA_DONT_SET, quotaUsage.getQuota());
+    assertEquals(RouterQuotaUsage.QUOTA_USAGE_COUNT_DEFAULT,
+        quotaUsage.getSpaceConsumed());
+    assertEquals(HdfsConstants.QUOTA_DONT_SET, quotaUsage.getSpaceQuota());
+
+    long nsQuota = 50;
+    long ssQuota = 100;
+    argv = new String[] {"-setQuota", src, "-nsQuota", String.valueOf(nsQuota),
+        "-ssQuota", String.valueOf(ssQuota)};
+    assertEquals(0, ToolRunner.run(admin, argv));
+
+    stateStore.loadCache(MountTableStoreImpl.class, true);
+    getResponse = client.getMountTableManager()
+        .getMountTableEntries(getRequest);
+    mountTable = getResponse.getEntries().get(0);
+    quotaUsage = mountTable.getQuota();
+
+    // verify if the quota is set
+    assertEquals(nsQuota, quotaUsage.getQuota());
+    assertEquals(ssQuota, quotaUsage.getSpaceQuota());
+
+    // use quota string for setting ss quota
+    String newSsQuota = "2m";
+    argv = new String[] {"-setQuota", src, "-ssQuota", newSsQuota};
+    assertEquals(0, ToolRunner.run(admin, argv));
+
+    stateStore.loadCache(MountTableStoreImpl.class, true);
+    getResponse = client.getMountTableManager()
+        .getMountTableEntries(getRequest);
+    mountTable = getResponse.getEntries().get(0);
+    quotaUsage = mountTable.getQuota();
+    // verify if ss quota is correctly set
+    assertEquals(2 * 1024 * 1024, quotaUsage.getSpaceQuota());
+
+    // test clrQuota command
+    argv = new String[] {"-clrQuota", src};
+    assertEquals(0, ToolRunner.run(admin, argv));
+
+    stateStore.loadCache(MountTableStoreImpl.class, true);
+    getResponse = client.getMountTableManager()
+        .getMountTableEntries(getRequest);
+    mountTable = getResponse.getEntries().get(0);
+    quotaUsage = mountTable.getQuota();
+
+    // verify if quota unset successfully
+    assertEquals(HdfsConstants.QUOTA_DONT_SET, quotaUsage.getQuota());
+    assertEquals(HdfsConstants.QUOTA_DONT_SET, quotaUsage.getSpaceQuota());
+  }
+
+  @Test
+  public void testManageSafeMode() throws Exception {
+    // ensure the Router become RUNNING state
+    waitState(RouterServiceState.RUNNING);
+    assertFalse(routerContext.getRouter().getRpcServer().isInSafeMode());
+    assertEquals(0, ToolRunner.run(admin,
+        new String[] {"-safemode", "enter"}));
+    // verify state
+    assertEquals(RouterServiceState.SAFEMODE,
+        routerContext.getRouter().getRouterState());
+    assertTrue(routerContext.getRouter().getRpcServer().isInSafeMode());
+
+    System.setOut(new PrintStream(out));
+    assertEquals(0, ToolRunner.run(admin,
+        new String[] {"-safemode", "get"}));
+    assertTrue(out.toString().contains("true"));
+
+    assertEquals(0, ToolRunner.run(admin,
+        new String[] {"-safemode", "leave"}));
+    // verify state
+    assertEquals(RouterServiceState.RUNNING,
+        routerContext.getRouter().getRouterState());
+    assertFalse(routerContext.getRouter().getRpcServer().isInSafeMode());
+
+    out.reset();
+    assertEquals(0, ToolRunner.run(admin,
+        new String[] {"-safemode", "get"}));
+    assertTrue(out.toString().contains("false"));
+  }
+
+  @Test
+  public void testCreateInvalidEntry() throws Exception {
+    String[] argv = new String[] {
+        "-add", "test-createInvalidEntry", "ns0", "/createInvalidEntry"};
+    assertEquals(-1, ToolRunner.run(admin, argv));
+
+    argv = new String[] {
+        "-add", "/test-createInvalidEntry", "ns0", "createInvalidEntry"};
+    assertEquals(-1, ToolRunner.run(admin, argv));
+
+    argv = new String[] {
+        "-add", null, "ns0", "/createInvalidEntry"};
+    assertEquals(-1, ToolRunner.run(admin, argv));
+
+    argv = new String[] {
+        "-add", "/test-createInvalidEntry", "ns0", null};
+    assertEquals(-1, ToolRunner.run(admin, argv));
+
+    argv = new String[] {
+        "-add", "", "ns0", "/createInvalidEntry"};
+    assertEquals(-1, ToolRunner.run(admin, argv));
+
+    argv = new String[] {
+        "-add", "/test-createInvalidEntry", null, "/createInvalidEntry"};
+    assertEquals(-1, ToolRunner.run(admin, argv));
+
+    argv = new String[] {
+        "-add", "/test-createInvalidEntry", "", "/createInvalidEntry"};
+    assertEquals(-1, ToolRunner.run(admin, argv));
+  }
+
+  /**
+   * Wait for the Router transforming to expected state.
+   * @param expectedState Expected Router state.
+   * @throws Exception
+   */
+  private void waitState(final RouterServiceState expectedState)
+      throws Exception {
+    GenericTestUtils.waitFor(new Supplier<Boolean>() {
+      @Override
+      public Boolean get() {
+        return expectedState == routerContext.getRouter().getRouterState();
+      }
+    }, 1000, 30000);
+  }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hadoop/blob/6e2b5fa4/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterHeartbeatService.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterHeartbeatService.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterHeartbeatService.java
new file mode 100644
index 0000000..80f2327
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterHeartbeatService.java
@@ -0,0 +1,146 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.federation.router;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryNTimes;
+import org.apache.curator.test.TestingServer;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CommonConfigurationKeys;
+import org.apache.hadoop.hdfs.server.federation.RouterConfigBuilder;
+import org.apache.hadoop.hdfs.server.federation.store.RouterStore;
+import org.apache.hadoop.hdfs.server.federation.store.StateStoreService;
+import org.apache.hadoop.hdfs.server.federation.store.driver.StateStoreDriver;
+import org.apache.hadoop.hdfs.server.federation.store.driver.impl.StateStoreZooKeeperImpl;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetRouterRegistrationRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.GetRouterRegistrationResponse;
+import org.apache.hadoop.hdfs.server.federation.store.records.RouterState;
+import org.apache.hadoop.hdfs.server.federation.store.records.StateStoreVersion;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils.waitStateStore;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Test cases for router heartbeat service.
+ */
+public class TestRouterHeartbeatService {
+  private Router router;
+  private final String routerId = "router1";
+  private TestingServer testingServer;
+  private CuratorFramework curatorFramework;
+
+  @Before
+  public void setup() throws Exception {
+    router = new Router();
+    router.setRouterId(routerId);
+    Configuration conf = new Configuration();
+    conf.setInt(RBFConfigKeys.DFS_ROUTER_CACHE_TIME_TO_LIVE_MS, 1);
+    Configuration routerConfig =
+        new RouterConfigBuilder(conf).stateStore().build();
+    routerConfig.setLong(RBFConfigKeys.FEDERATION_STORE_CONNECTION_TEST_MS,
+        TimeUnit.HOURS.toMillis(1));
+    routerConfig.setClass(RBFConfigKeys.FEDERATION_STORE_DRIVER_CLASS,
+        StateStoreZooKeeperImpl.class, StateStoreDriver.class);
+
+    testingServer = new TestingServer();
+    String connectStr = testingServer.getConnectString();
+    curatorFramework = CuratorFrameworkFactory.builder()
+        .connectString(connectStr)
+        .retryPolicy(new RetryNTimes(100, 100))
+        .build();
+    curatorFramework.start();
+    routerConfig.set(CommonConfigurationKeys.ZK_ADDRESS, connectStr);
+    router.init(routerConfig);
+    router.start();
+
+
+    waitStateStore(router.getStateStore(), TimeUnit.SECONDS.toMicros(10));
+  }
+
+  @Test
+  public void testStateStoreUnavailable() throws IOException {
+    curatorFramework.close();
+    testingServer.stop();
+    router.getStateStore().stop();
+    // The driver is not ready
+    assertFalse(router.getStateStore().isDriverReady());
+
+    // Do a heartbeat, and no exception thrown out
+    RouterHeartbeatService heartbeatService =
+        new RouterHeartbeatService(router);
+    heartbeatService.updateStateStore();
+  }
+
+  @Test
+  public void testStateStoreAvailable() throws Exception {
+    // The driver is ready
+    StateStoreService stateStore = router.getStateStore();
+    assertTrue(router.getStateStore().isDriverReady());
+    RouterStore routerStore = router.getRouterStateManager();
+
+    // No record about this router
+    stateStore.refreshCaches(true);
+    GetRouterRegistrationRequest request =
+        GetRouterRegistrationRequest.newInstance(routerId);
+    GetRouterRegistrationResponse response =
+        router.getRouterStateManager().getRouterRegistration(request);
+    RouterState routerState = response.getRouter();
+    String id = routerState.getRouterId();
+    StateStoreVersion version = routerState.getStateStoreVersion();
+    assertNull(id);
+    assertNull(version);
+
+    // Do a heartbeat
+    RouterHeartbeatService heartbeatService =
+        new RouterHeartbeatService(router);
+    heartbeatService.updateStateStore();
+
+    // We should have a record
+    stateStore.refreshCaches(true);
+    request = GetRouterRegistrationRequest.newInstance(routerId);
+    response = routerStore.getRouterRegistration(request);
+    routerState = response.getRouter();
+    id = routerState.getRouterId();
+    version = routerState.getStateStoreVersion();
+    assertNotNull(id);
+    assertNotNull(version);
+  }
+
+  @After
+  public void tearDown() throws IOException {
+    if (curatorFramework != null) {
+      curatorFramework.close();
+    }
+    if (testingServer != null) {
+      testingServer.stop();
+    }
+    if (router != null) {
+      router.shutDown();
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/6e2b5fa4/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterMountTable.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterMountTable.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterMountTable.java
new file mode 100644
index 0000000..8702b3c
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterMountTable.java
@@ -0,0 +1,143 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.federation.router;
+
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.io.IOException;
+import java.util.Collections;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.server.federation.RouterConfigBuilder;
+import org.apache.hadoop.hdfs.server.federation.RouterDFSCluster.NamenodeContext;
+import org.apache.hadoop.hdfs.server.federation.RouterDFSCluster.RouterContext;
+import org.apache.hadoop.hdfs.server.federation.StateStoreDFSCluster;
+import org.apache.hadoop.hdfs.server.federation.resolver.MountTableManager;
+import org.apache.hadoop.hdfs.server.federation.resolver.MountTableResolver;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.AddMountTableEntryRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.AddMountTableEntryResponse;
+import org.apache.hadoop.hdfs.server.federation.store.records.MountTable;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+/**
+ * Test a router end-to-end including the MountTable.
+ */
+public class TestRouterMountTable {
+
+  private static StateStoreDFSCluster cluster;
+  private static NamenodeContext nnContext;
+  private static RouterContext routerContext;
+  private static MountTableResolver mountTable;
+
+  @BeforeClass
+  public static void globalSetUp() throws Exception {
+
+    // Build and start a federated cluster
+    cluster = new StateStoreDFSCluster(false, 1);
+    Configuration conf = new RouterConfigBuilder()
+        .stateStore()
+        .admin()
+        .rpc()
+        .build();
+    cluster.addRouterOverrides(conf);
+    cluster.startCluster();
+    cluster.startRouters();
+    cluster.waitClusterUp();
+
+    // Get the end points
+    nnContext = cluster.getRandomNamenode();
+    routerContext = cluster.getRandomRouter();
+    Router router = routerContext.getRouter();
+    mountTable = (MountTableResolver) router.getSubclusterResolver();
+  }
+
+  @AfterClass
+  public static void tearDown() {
+    if (cluster != null) {
+      cluster.stopRouter(routerContext);
+      cluster.shutdown();
+      cluster = null;
+    }
+  }
+
+  @Test
+  public void testReadOnly() throws Exception {
+
+    // Add a read only entry
+    MountTable readOnlyEntry = MountTable.newInstance(
+        "/readonly", Collections.singletonMap("ns0", "/testdir"));
+    readOnlyEntry.setReadOnly(true);
+    assertTrue(addMountTable(readOnlyEntry));
+
+    // Add a regular entry
+    MountTable regularEntry = MountTable.newInstance(
+        "/regular", Collections.singletonMap("ns0", "/testdir"));
+    assertTrue(addMountTable(regularEntry));
+
+    // Create a folder which should show in all locations
+    final FileSystem nnFs = nnContext.getFileSystem();
+    final FileSystem routerFs = routerContext.getFileSystem();
+    assertTrue(routerFs.mkdirs(new Path("/regular/newdir")));
+
+    FileStatus dirStatusNn =
+        nnFs.getFileStatus(new Path("/testdir/newdir"));
+    assertTrue(dirStatusNn.isDirectory());
+    FileStatus dirStatusRegular =
+        routerFs.getFileStatus(new Path("/regular/newdir"));
+    assertTrue(dirStatusRegular.isDirectory());
+    FileStatus dirStatusReadOnly =
+        routerFs.getFileStatus(new Path("/readonly/newdir"));
+    assertTrue(dirStatusReadOnly.isDirectory());
+
+    // It should fail writing into a read only path
+    try {
+      routerFs.mkdirs(new Path("/readonly/newdirfail"));
+      fail("We should not be able to write into a read only mount point");
+    } catch (IOException ioe) {
+      String msg = ioe.getMessage();
+      assertTrue(msg.startsWith(
+          "/readonly/newdirfail is in a read only mount point"));
+    }
+  }
+
+  /**
+   * Add a mount table entry to the mount table through the admin API.
+   * @param entry Mount table entry to add.
+   * @return If it was succesfully added.
+   * @throws IOException Problems adding entries.
+   */
+  private boolean addMountTable(final MountTable entry) throws IOException {
+    RouterClient client = routerContext.getAdminClient();
+    MountTableManager mountTableManager = client.getMountTableManager();
+    AddMountTableEntryRequest addRequest =
+        AddMountTableEntryRequest.newInstance(entry);
+    AddMountTableEntryResponse addResponse =
+        mountTableManager.addMountTableEntry(addRequest);
+
+    // Reload the Router cache
+    mountTable.loadCache(true);
+
+    return addResponse.getStatus();
+  }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hadoop/blob/6e2b5fa4/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterNamenodeMonitoring.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterNamenodeMonitoring.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterNamenodeMonitoring.java
new file mode 100644
index 0000000..eabc0fe
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterNamenodeMonitoring.java
@@ -0,0 +1,143 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.federation.router;
+
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_HA_NAMENODE_ID_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMESERVICE_ID;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_MONITOR_NAMENODE;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Collection;
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.server.federation.RouterConfigBuilder;
+import org.apache.hadoop.hdfs.server.federation.RouterDFSCluster.NamenodeContext;
+import org.apache.hadoop.hdfs.server.federation.RouterDFSCluster.RouterContext;
+import org.apache.hadoop.hdfs.server.federation.StateStoreDFSCluster;
+import org.apache.hadoop.hdfs.server.federation.resolver.FederationNamenodeContext;
+import org.apache.hadoop.hdfs.server.federation.resolver.MembershipNamenodeResolver;
+import org.apache.hadoop.util.Time;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Test namenodes monitor behavior in the Router.
+ */
+public class TestRouterNamenodeMonitoring {
+
+  private static StateStoreDFSCluster cluster;
+  private static RouterContext routerContext;
+  private static MembershipNamenodeResolver resolver;
+
+  private String ns0;
+  private String ns1;
+  private long initializedTime;
+
+  @Before
+  public void setUp() throws Exception {
+    // Build and start a federated cluster with HA enabled
+    cluster = new StateStoreDFSCluster(true, 2);
+    // Enable heartbeat service and local heartbeat
+    Configuration routerConf = new RouterConfigBuilder()
+        .stateStore()
+        .admin()
+        .rpc()
+        .enableLocalHeartbeat(true)
+        .heartbeat()
+        .build();
+
+    // Specify local node (ns0.nn1) to monitor
+    StringBuilder sb = new StringBuilder();
+    ns0 = cluster.getNameservices().get(0);
+    NamenodeContext context = cluster.getNamenodes(ns0).get(1);
+    routerConf.set(DFS_NAMESERVICE_ID, ns0);
+    routerConf.set(DFS_HA_NAMENODE_ID_KEY, context.getNamenodeId());
+
+    // Specify namenodes (ns1.nn0,ns1.nn1) to monitor
+    sb = new StringBuilder();
+    ns1 = cluster.getNameservices().get(1);
+    for (NamenodeContext ctx : cluster.getNamenodes(ns1)) {
+      String suffix = ctx.getConfSuffix();
+      if (sb.length() != 0) {
+        sb.append(",");
+      }
+      sb.append(suffix);
+    }
+    // override with the namenodes: ns1.nn0,ns1.nn1
+    routerConf.set(DFS_ROUTER_MONITOR_NAMENODE, sb.toString());
+
+    cluster.addRouterOverrides(routerConf);
+    cluster.startCluster();
+    cluster.startRouters();
+    cluster.waitClusterUp();
+
+    routerContext = cluster.getRandomRouter();
+    resolver = (MembershipNamenodeResolver) routerContext.getRouter()
+        .getNamenodeResolver();
+    initializedTime = Time.now();
+  }
+
+  @After
+  public void tearDown() {
+    if (cluster != null) {
+      cluster.stopRouter(routerContext);
+      cluster.shutdown();
+      cluster = null;
+    }
+  }
+
+  @Test
+  public void testNamenodeMonitoring() throws Exception {
+    // Set nn0 to active for all nameservices
+    for (String ns : cluster.getNameservices()) {
+      cluster.switchToActive(ns, "nn0");
+      cluster.switchToStandby(ns, "nn1");
+    }
+
+    Collection<NamenodeHeartbeatService> heartbeatServices = routerContext
+        .getRouter().getNamenodeHearbeatServices();
+    // manually trigger the heartbeat
+    for (NamenodeHeartbeatService service : heartbeatServices) {
+      service.periodicInvoke();
+    }
+
+    resolver.loadCache(true);
+    List<? extends FederationNamenodeContext> namespaceInfo0 =
+        resolver.getNamenodesForNameserviceId(ns0);
+    List<? extends FederationNamenodeContext> namespaceInfo1 =
+        resolver.getNamenodesForNameserviceId(ns1);
+
+    // The modified date won't be updated in ns0.nn0 since it isn't
+    // monitored by the Router.
+    assertEquals("nn0", namespaceInfo0.get(1).getNamenodeId());
+    assertTrue(namespaceInfo0.get(1).getDateModified() < initializedTime);
+
+    // other namnodes should be updated as expected
+    assertEquals("nn1", namespaceInfo0.get(0).getNamenodeId());
+    assertTrue(namespaceInfo0.get(0).getDateModified() > initializedTime);
+
+    assertEquals("nn0", namespaceInfo1.get(0).getNamenodeId());
+    assertTrue(namespaceInfo1.get(0).getDateModified() > initializedTime);
+
+    assertEquals("nn1", namespaceInfo1.get(1).getNamenodeId());
+    assertTrue(namespaceInfo1.get(1).getDateModified() > initializedTime);
+  }
+}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/6e2b5fa4/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterQuota.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterQuota.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterQuota.java
new file mode 100644
index 0000000..66a955e
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterQuota.java
@@ -0,0 +1,451 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.federation.router;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.UUID;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CreateFlag;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.QuotaUsage;
+import org.apache.hadoop.hdfs.DFSClient;
+import org.apache.hadoop.hdfs.client.HdfsClientConfigKeys;
+import org.apache.hadoop.hdfs.client.HdfsDataOutputStream;
+import org.apache.hadoop.hdfs.protocol.ClientProtocol;
+import org.apache.hadoop.hdfs.protocol.DSQuotaExceededException;
+import org.apache.hadoop.hdfs.protocol.NSQuotaExceededException;
+import org.apache.hadoop.hdfs.server.federation.RouterConfigBuilder;
+import org.apache.hadoop.hdfs.server.federation.RouterDFSCluster.NamenodeContext;
+import org.apache.hadoop.hdfs.server.federation.RouterDFSCluster.RouterContext;
+import org.apache.hadoop.hdfs.server.federation.StateStoreDFSCluster;
+import org.apache.hadoop.hdfs.server.federation.resolver.MountTableManager;
+import org.apache.hadoop.hdfs.server.federation.resolver.MountTableResolver;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.AddMountTableEntryRequest;
+import org.apache.hadoop.hdfs.server.federation.store.protocol.AddMountTableEntryResponse;
+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.RemoveMountTableEntryResponse;
+import org.apache.hadoop.hdfs.server.federation.store.records.MountTable;
+import org.apache.hadoop.test.GenericTestUtils;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import com.google.common.base.Supplier;
+
+/**
+ * Tests quota behaviors in Router-based Federation.
+ */
+public class TestRouterQuota {
+  private static StateStoreDFSCluster cluster;
+  private static NamenodeContext nnContext1;
+  private static NamenodeContext nnContext2;
+  private static RouterContext routerContext;
+  private static MountTableResolver resolver;
+
+  private static final int BLOCK_SIZE = 512;
+
+  @Before
+  public void setUp() throws Exception {
+
+    // Build and start a federated cluster
+    cluster = new StateStoreDFSCluster(false, 2);
+    Configuration routerConf = new RouterConfigBuilder()
+        .stateStore()
+        .admin()
+        .quota()
+        .rpc()
+        .build();
+    routerConf.set(RBFConfigKeys.DFS_ROUTER_QUOTA_CACHE_UPATE_INTERVAL, "2s");
+
+    // override some hdfs settings that used in testing space quota
+    Configuration hdfsConf = new Configuration(false);
+    hdfsConf.setInt(HdfsClientConfigKeys.DFS_BLOCK_SIZE_KEY, BLOCK_SIZE);
+    hdfsConf.setInt(HdfsClientConfigKeys.DFS_REPLICATION_KEY, 1);
+
+    cluster.addRouterOverrides(routerConf);
+    cluster.addNamenodeOverrides(hdfsConf);
+    cluster.startCluster();
+    cluster.startRouters();
+    cluster.waitClusterUp();
+
+    nnContext1 = cluster.getNamenode(cluster.getNameservices().get(0), null);
+    nnContext2 = cluster.getNamenode(cluster.getNameservices().get(1), null);
+    routerContext = cluster.getRandomRouter();
+    Router router = routerContext.getRouter();
+    resolver = (MountTableResolver) router.getSubclusterResolver();
+  }
+
+  @After
+  public void tearDown() {
+    if (cluster != null) {
+      cluster.stopRouter(routerContext);
+      cluster.shutdown();
+      cluster = null;
+    }
+  }
+
+  @Test
+  public void testNamespaceQuotaExceed() throws Exception {
+    long nsQuota = 3;
+    final FileSystem nnFs1 = nnContext1.getFileSystem();
+    final FileSystem nnFs2 = nnContext2.getFileSystem();
+
+    // Add two mount tables:
+    // /nsquota --> ns0---testdir1
+    // /nsquota/subdir --> ns1---testdir2
+    nnFs1.mkdirs(new Path("/testdir1"));
+    nnFs2.mkdirs(new Path("/testdir2"));
+    MountTable mountTable1 = MountTable.newInstance("/nsquota",
+        Collections.singletonMap("ns0", "/testdir1"));
+
+    mountTable1.setQuota(new RouterQuotaUsage.Builder().quota(nsQuota).build());
+    addMountTable(mountTable1);
+
+    MountTable mountTable2 = MountTable.newInstance("/nsquota/subdir",
+        Collections.singletonMap("ns1", "/testdir2"));
+    mountTable2.setQuota(new RouterQuotaUsage.Builder().quota(nsQuota).build());
+    addMountTable(mountTable2);
+
+    final FileSystem routerFs = routerContext.getFileSystem();
+    GenericTestUtils.waitFor(new Supplier<Boolean>() {
+
+      @Override
+      public Boolean get() {
+        boolean isNsQuotaViolated = false;
+        try {
+          // create new directory to trigger NSQuotaExceededException
+          routerFs.mkdirs(new Path("/nsquota/" + UUID.randomUUID()));
+          routerFs.mkdirs(new Path("/nsquota/subdir/" + UUID.randomUUID()));
+        } catch (NSQuotaExceededException e) {
+          isNsQuotaViolated = true;
+        } catch (IOException ignored) {
+        }
+        return isNsQuotaViolated;
+      }
+    }, 5000, 60000);
+    // mkdir in real FileSystem should be okay
+    nnFs1.mkdirs(new Path("/testdir1/" + UUID.randomUUID()));
+    nnFs2.mkdirs(new Path("/testdir2/" + UUID.randomUUID()));
+  }
+
+  @Test
+  public void testStorageSpaceQuotaaExceed() throws Exception {
+    long ssQuota = 3071;
+    final FileSystem nnFs1 = nnContext1.getFileSystem();
+    final FileSystem nnFs2 = nnContext2.getFileSystem();
+
+    // Add two mount tables:
+    // /ssquota --> ns0---testdir3
+    // /ssquota/subdir --> ns1---testdir4
+    nnFs1.mkdirs(new Path("/testdir3"));
+    nnFs2.mkdirs(new Path("/testdir4"));
+    MountTable mountTable1 = MountTable.newInstance("/ssquota",
+        Collections.singletonMap("ns0", "/testdir3"));
+
+    mountTable1
+        .setQuota(new RouterQuotaUsage.Builder().spaceQuota(ssQuota).build());
+    addMountTable(mountTable1);
+
+    MountTable mountTable2 = MountTable.newInstance("/ssquota/subdir",
+        Collections.singletonMap("ns1", "/testdir4"));
+    mountTable2
+        .setQuota(new RouterQuotaUsage.Builder().spaceQuota(ssQuota).build());
+    addMountTable(mountTable2);
+
+    DFSClient routerClient = routerContext.getClient();
+    routerClient.create("/ssquota/file", true).close();
+    routerClient.create("/ssquota/subdir/file", true).close();
+
+    GenericTestUtils.waitFor(new Supplier<Boolean>() {
+
+      @Override
+      public Boolean get() {
+        boolean isDsQuotaViolated = false;
+        try {
+          // append data to trigger NSQuotaExceededException
+          appendData("/ssquota/file", routerClient, BLOCK_SIZE);
+          appendData("/ssquota/subdir/file", routerClient, BLOCK_SIZE);
+        } catch (DSQuotaExceededException e) {
+          isDsQuotaViolated = true;
+        } catch (IOException ignored) {
+        }
+        return isDsQuotaViolated;
+      }
+    }, 5000, 60000);
+
+    // append data to destination path in real FileSystem should be okay
+    appendData("/testdir3/file", nnContext1.getClient(), BLOCK_SIZE);
+    appendData("/testdir4/file", nnContext2.getClient(), BLOCK_SIZE);
+  }
+
+  /**
+   * Add a mount table entry to the mount table through the admin API.
+   * @param entry Mount table entry to add.
+   * @return If it was successfully added.
+   * @throws IOException Problems adding entries.
+   */
+  private boolean addMountTable(final MountTable entry) throws IOException {
+    RouterClient client = routerContext.getAdminClient();
+    MountTableManager mountTableManager = client.getMountTableManager();
+    AddMountTableEntryRequest addRequest =
+        AddMountTableEntryRequest.newInstance(entry);
+    AddMountTableEntryResponse addResponse =
+        mountTableManager.addMountTableEntry(addRequest);
+
+    // Reload the Router cache
+    resolver.loadCache(true);
+
+    return addResponse.getStatus();
+  }
+
+  /**
+   * Append data in specified file.
+   * @param path Path of file.
+   * @param client DFS Client.
+   * @param dataLen The length of write data.
+   * @throws IOException
+   */
+  private void appendData(String path, DFSClient client, int dataLen)
+      throws IOException {
+    EnumSet<CreateFlag> createFlag = EnumSet.of(CreateFlag.APPEND);
+    HdfsDataOutputStream stream = client.append(path, 1024, createFlag, null,
+        null);
+    byte[] data = new byte[dataLen];
+    stream.write(data);
+    stream.close();
+  }
+
+  @Test
+  public void testSetQuota() throws Exception {
+    long nsQuota = 5;
+    long ssQuota = 100;
+    final FileSystem nnFs1 = nnContext1.getFileSystem();
+    final FileSystem nnFs2 = nnContext2.getFileSystem();
+
+    // Add two mount tables:
+    // /setquota --> ns0---testdir5
+    // /setquota/subdir --> ns1---testdir6
+    nnFs1.mkdirs(new Path("/testdir5"));
+    nnFs2.mkdirs(new Path("/testdir6"));
+    MountTable mountTable1 = MountTable.newInstance("/setquota",
+        Collections.singletonMap("ns0", "/testdir5"));
+    mountTable1
+        .setQuota(new RouterQuotaUsage.Builder().quota(nsQuota)
+        .spaceQuota(ssQuota).build());
+    addMountTable(mountTable1);
+
+    // don't set quota for subpath of mount table
+    MountTable mountTable2 = MountTable.newInstance("/setquota/subdir",
+        Collections.singletonMap("ns1", "/testdir6"));
+    addMountTable(mountTable2);
+
+    RouterQuotaUpdateService updateService = routerContext.getRouter()
+        .getQuotaCacheUpdateService();
+    // ensure setQuota RPC call was invoked
+    updateService.periodicInvoke();
+
+    ClientProtocol client1 = nnContext1.getClient().getNamenode();
+    ClientProtocol client2 = nnContext2.getClient().getNamenode();
+    final QuotaUsage quota1 = client1.getQuotaUsage("/testdir5");
+    final QuotaUsage quota2 = client2.getQuotaUsage("/testdir6");
+
+    assertEquals(nsQuota, quota1.getQuota());
+    assertEquals(ssQuota, quota1.getSpaceQuota());
+    assertEquals(nsQuota, quota2.getQuota());
+    assertEquals(ssQuota, quota2.getSpaceQuota());
+  }
+
+  @Test
+  public void testGetQuota() throws Exception {
+    long nsQuota = 10;
+    long ssQuota = 100;
+    final FileSystem nnFs1 = nnContext1.getFileSystem();
+    final FileSystem nnFs2 = nnContext2.getFileSystem();
+
+    // Add two mount tables:
+    // /getquota --> ns0---/testdir7
+    // /getquota/subdir1 --> ns0---/testdir7/subdir
+    // /getquota/subdir2 --> ns1---/testdir8
+    nnFs1.mkdirs(new Path("/testdir7"));
+    nnFs1.mkdirs(new Path("/testdir7/subdir"));
+    nnFs2.mkdirs(new Path("/testdir8"));
+    MountTable mountTable1 = MountTable.newInstance("/getquota",
+        Collections.singletonMap("ns0", "/testdir7"));
+    mountTable1
+        .setQuota(new RouterQuotaUsage.Builder().quota(nsQuota)
+        .spaceQuota(ssQuota).build());
+    addMountTable(mountTable1);
+
+    MountTable mountTable2 = MountTable.newInstance("/getquota/subdir1",
+        Collections.singletonMap("ns0", "/testdir7/subdir"));
+    addMountTable(mountTable2);
+
+    MountTable mountTable3 = MountTable.newInstance("/getquota/subdir2",
+        Collections.singletonMap("ns1", "/testdir8"));
+    addMountTable(mountTable3);
+
+    // use router client to create new files
+    DFSClient routerClient = routerContext.getClient();
+    routerClient.create("/getquota/file", true).close();
+    routerClient.create("/getquota/subdir1/file", true).close();
+    routerClient.create("/getquota/subdir2/file", true).close();
+
+    ClientProtocol clientProtocol = routerContext.getClient().getNamenode();
+    RouterQuotaUpdateService updateService = routerContext.getRouter()
+        .getQuotaCacheUpdateService();
+    updateService.periodicInvoke();
+    final QuotaUsage quota = clientProtocol.getQuotaUsage("/getquota");
+    // the quota should be aggregated
+    assertEquals(6, quota.getFileAndDirectoryCount());
+  }
+
+  @Test
+  public void testStaleQuotaRemoving() throws Exception {
+    long nsQuota = 20;
+    long ssQuota = 200;
+    String stalePath = "/stalequota";
+    final FileSystem nnFs1 = nnContext1.getFileSystem();
+
+    // Add one mount tables:
+    // /stalequota --> ns0---/testdir9
+    nnFs1.mkdirs(new Path("/testdir9"));
+    MountTable mountTable = MountTable.newInstance(stalePath,
+        Collections.singletonMap("ns0", "/testdir9"));
+    mountTable.setQuota(new RouterQuotaUsage.Builder().quota(nsQuota)
+        .spaceQuota(ssQuota).build());
+    addMountTable(mountTable);
+
+    // Call periodicInvoke to ensure quota for stalePath was
+    // loaded into quota manager.
+    RouterQuotaUpdateService updateService = routerContext.getRouter()
+        .getQuotaCacheUpdateService();
+    updateService.periodicInvoke();
+
+    // use quota manager to get its quota usage and do verification
+    RouterQuotaManager quotaManager = routerContext.getRouter()
+        .getQuotaManager();
+    RouterQuotaUsage quota = quotaManager.getQuotaUsage(stalePath);
+    assertEquals(nsQuota, quota.getQuota());
+    assertEquals(ssQuota, quota.getSpaceQuota());
+
+    // remove stale path entry
+    removeMountTable(stalePath);
+    updateService.periodicInvoke();
+    // the stale entry should be removed and we will get null
+    quota = quotaManager.getQuotaUsage(stalePath);
+    assertNull(quota);
+  }
+
+  /**
+   * Remove a mount table entry to the mount table through the admin API.
+   * @param entry Mount table entry to remove.
+   * @return If it was successfully removed.
+   * @throws IOException Problems removing entries.
+   */
+  private boolean removeMountTable(String path) throws IOException {
+    RouterClient client = routerContext.getAdminClient();
+    MountTableManager mountTableManager = client.getMountTableManager();
+    RemoveMountTableEntryRequest removeRequest = RemoveMountTableEntryRequest
+        .newInstance(path);
+    RemoveMountTableEntryResponse removeResponse = mountTableManager
+        .removeMountTableEntry(removeRequest);
+
+    // Reload the Router cache
+    resolver.loadCache(true);
+    return removeResponse.getStatus();
+  }
+
+  @Test
+  public void testQuotaUpdating() throws Exception {
+    long nsQuota = 30;
+    long ssQuota = 1024;
+    String path = "/updatequota";
+    final FileSystem nnFs1 = nnContext1.getFileSystem();
+
+    // Add one mount table:
+    // /updatequota --> ns0---/testdir10
+    nnFs1.mkdirs(new Path("/testdir10"));
+    MountTable mountTable = MountTable.newInstance(path,
+        Collections.singletonMap("ns0", "/testdir10"));
+    mountTable.setQuota(new RouterQuotaUsage.Builder().quota(nsQuota)
+        .spaceQuota(ssQuota).build());
+    addMountTable(mountTable);
+
+    // Call periodicInvoke to ensure quota  updated in quota manager
+    // and state store.
+    RouterQuotaUpdateService updateService = routerContext.getRouter()
+        .getQuotaCacheUpdateService();
+    updateService.periodicInvoke();
+
+    // verify initial quota value
+    List<MountTable> results = getMountTable(path);
+    MountTable updatedMountTable = !results.isEmpty() ? results.get(0) : null;
+    RouterQuotaUsage quota = updatedMountTable.getQuota();
+    assertEquals(nsQuota, quota.getQuota());
+    assertEquals(ssQuota, quota.getSpaceQuota());
+    assertEquals(1, quota.getFileAndDirectoryCount());
+    assertEquals(0, quota.getSpaceConsumed());
+
+    // mkdir and write a new file
+    final FileSystem routerFs = routerContext.getFileSystem();
+    routerFs.mkdirs(new Path(path + "/" + UUID.randomUUID()));
+    DFSClient routerClient = routerContext.getClient();
+    routerClient.create(path + "/file", true).close();
+    appendData(path + "/file", routerClient, BLOCK_SIZE);
+
+    updateService.periodicInvoke();
+    results = getMountTable(path);
+    updatedMountTable = !results.isEmpty() ? results.get(0) : null;
+    quota = updatedMountTable.getQuota();
+
+    // verify if quota has been updated in state store
+    assertEquals(nsQuota, quota.getQuota());
+    assertEquals(ssQuota, quota.getSpaceQuota());
+    assertEquals(3, quota.getFileAndDirectoryCount());
+    assertEquals(BLOCK_SIZE, quota.getSpaceConsumed());
+  }
+
+  /**
+   * Get the mount table entries of specified path through the admin API.
+   * @param path Mount table entry to get.
+   * @return If it was successfully got.
+   * @throws IOException Problems getting entries.
+   */
+  private List<MountTable> getMountTable(String path) throws IOException {
+    // Reload the Router cache
+    resolver.loadCache(true);
+    RouterClient client = routerContext.getAdminClient();
+    MountTableManager mountTableManager = client.getMountTableManager();
+    GetMountTableEntriesRequest getRequest = GetMountTableEntriesRequest
+        .newInstance(path);
+    GetMountTableEntriesResponse removeResponse = mountTableManager
+        .getMountTableEntries(getRequest);
+
+    return removeResponse.getEntries();
+  }
+}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/6e2b5fa4/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterQuotaManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterQuotaManager.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterQuotaManager.java
new file mode 100644
index 0000000..ce3ee17
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterQuotaManager.java
@@ -0,0 +1,125 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.federation.router;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Set;
+
+import org.apache.hadoop.hdfs.protocol.HdfsConstants;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Tests for class {@link RouterQuotaManager}.
+ */
+public class TestRouterQuotaManager {
+  private static RouterQuotaManager manager;
+
+  @Before
+  public void setup() {
+    manager = new RouterQuotaManager();
+  }
+
+  @After
+  public void cleanup() {
+    manager.clear();
+  }
+
+  @Test
+  public void testGetChildrenPaths() {
+    RouterQuotaUsage quotaUsage = new RouterQuotaUsage.Builder().build();
+    manager.put("/path1", quotaUsage);
+    manager.put("/path2", quotaUsage);
+    manager.put("/path1/subdir", quotaUsage);
+    manager.put("/path1/subdir/subdir", quotaUsage);
+
+    Set<String> childrenPaths = manager.getPaths("/path1");
+    assertEquals(3, childrenPaths.size());
+    assertTrue(childrenPaths.contains("/path1/subdir")
+        && childrenPaths.contains("/path1/subdir/subdir")
+        && childrenPaths.contains("/path1"));
+
+    // test for corner case
+    manager.put("/path3", quotaUsage);
+    manager.put("/path3/subdir", quotaUsage);
+    manager.put("/path3-subdir", quotaUsage);
+
+    childrenPaths = manager.getPaths("/path3");
+    assertEquals(2, childrenPaths.size());
+    // path /path3-subdir should not be returned
+    assertTrue(childrenPaths.contains("/path3")
+        && childrenPaths.contains("/path3/subdir")
+        && !childrenPaths.contains("/path3-subdir"));
+  }
+
+  @Test
+  public void testGetQuotaUsage() {
+    RouterQuotaUsage quotaGet;
+
+    // test case1: get quota with an non-exist path
+    quotaGet = manager.getQuotaUsage("/non-exist-path");
+    assertNull(quotaGet);
+
+    // test case2: get quota from an no-quota set path
+    RouterQuotaUsage.Builder quota = new RouterQuotaUsage.Builder()
+        .quota(HdfsConstants.QUOTA_DONT_SET)
+        .spaceQuota(HdfsConstants.QUOTA_DONT_SET);
+    manager.put("/noQuotaSet", quota.build());
+    quotaGet = manager.getQuotaUsage("/noQuotaSet");
+    // it should return null
+    assertNull(quotaGet);
+
+    // test case3: get quota from an quota-set path
+    quota.quota(1);
+    quota.spaceQuota(HdfsConstants.QUOTA_DONT_SET);
+    manager.put("/hasQuotaSet", quota.build());
+    quotaGet = manager.getQuotaUsage("/hasQuotaSet");
+    assertEquals(1, quotaGet.getQuota());
+    assertEquals(HdfsConstants.QUOTA_DONT_SET, quotaGet.getSpaceQuota());
+
+    // test case4: get quota with an non-exist child path
+    quotaGet = manager.getQuotaUsage("/hasQuotaSet/file");
+    // it will return the nearest ancestor which quota was set
+    assertEquals(1, quotaGet.getQuota());
+    assertEquals(HdfsConstants.QUOTA_DONT_SET, quotaGet.getSpaceQuota());
+
+    // test case5: get quota with an child path which its parent
+    // wasn't quota set
+    quota.quota(HdfsConstants.QUOTA_DONT_SET);
+    quota.spaceQuota(HdfsConstants.QUOTA_DONT_SET);
+    manager.put("/hasQuotaSet/noQuotaSet", quota.build());
+    // here should returns the quota of path /hasQuotaSet
+    // (the nearest ancestor which quota was set)
+    quotaGet = manager.getQuotaUsage("/hasQuotaSet/noQuotaSet/file");
+    assertEquals(1, quotaGet.getQuota());
+    assertEquals(HdfsConstants.QUOTA_DONT_SET, quotaGet.getSpaceQuota());
+
+    // test case6: get quota with an child path which its parent was quota set
+    quota.quota(2);
+    quota.spaceQuota(HdfsConstants.QUOTA_DONT_SET);
+    manager.put("/hasQuotaSet/hasQuotaSet", quota.build());
+    // here should return the quota of path /hasQuotaSet/hasQuotaSet
+    quotaGet = manager.getQuotaUsage("/hasQuotaSet/hasQuotaSet/file");
+    assertEquals(2, quotaGet.getQuota());
+    assertEquals(HdfsConstants.QUOTA_DONT_SET, quotaGet.getSpaceQuota());
+  }
+}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/6e2b5fa4/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterRPCClientRetries.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterRPCClientRetries.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterRPCClientRetries.java
new file mode 100644
index 0000000..61e7657
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterRPCClientRetries.java
@@ -0,0 +1,160 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.federation.router;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CommonConfigurationKeys;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.hdfs.DFSClient;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.hdfs.protocol.ClientProtocol;
+import org.apache.hadoop.hdfs.server.federation.RouterConfigBuilder;
+import org.apache.hadoop.hdfs.server.federation.RouterDFSCluster.NamenodeContext;
+import org.apache.hadoop.hdfs.server.federation.RouterDFSCluster.RouterContext;
+import org.apache.hadoop.hdfs.server.federation.StateStoreDFSCluster;
+import org.apache.hadoop.hdfs.server.federation.metrics.FederationRPCMetrics;
+import org.apache.hadoop.hdfs.server.federation.resolver.FederationNamenodeContext;
+import org.apache.hadoop.hdfs.server.federation.resolver.MembershipNamenodeResolver;
+import org.apache.hadoop.hdfs.server.federation.resolver.NamenodeStatusReport;
+import org.apache.hadoop.ipc.RemoteException;
+import org.apache.hadoop.test.GenericTestUtils;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Test retry behavior of the Router RPC Client.
+ */
+public class TestRouterRPCClientRetries {
+
+  private static StateStoreDFSCluster cluster;
+  private static NamenodeContext nnContext1;
+  private static RouterContext routerContext;
+  private static MembershipNamenodeResolver resolver;
+  private static ClientProtocol routerProtocol;
+
+  @Before
+  public void setUp() throws Exception {
+    // Build and start a federated cluster
+    cluster = new StateStoreDFSCluster(false, 2);
+    Configuration routerConf = new RouterConfigBuilder()
+        .stateStore()
+        .admin()
+        .rpc()
+        .build();
+
+    // reduce IPC client connection retry times and interval time
+    Configuration clientConf = new Configuration(false);
+    clientConf.setInt(
+        CommonConfigurationKeys.IPC_CLIENT_CONNECT_MAX_RETRIES_KEY, 1);
+    clientConf.setInt(
+        CommonConfigurationKeys.IPC_CLIENT_CONNECT_RETRY_INTERVAL_KEY, 100);
+
+    cluster.addRouterOverrides(routerConf);
+    // override some settings for the client
+    cluster.startCluster(clientConf);
+    cluster.startRouters();
+    cluster.waitClusterUp();
+
+    nnContext1 = cluster.getNamenode(cluster.getNameservices().get(0), null);
+    routerContext = cluster.getRandomRouter();
+    resolver = (MembershipNamenodeResolver) routerContext.getRouter()
+        .getNamenodeResolver();
+    routerProtocol = routerContext.getClient().getNamenode();
+  }
+
+  @After
+  public void tearDown() {
+    if (cluster != null) {
+      cluster.stopRouter(routerContext);
+      cluster.shutdown();
+      cluster = null;
+    }
+  }
+
+  @Test
+  public void testRetryWhenAllNameServiceDown() throws Exception {
+    // shutdown the dfs cluster
+    MiniDFSCluster dfsCluster = cluster.getCluster();
+    dfsCluster.shutdown();
+
+    // register an invalid namenode report
+    registerInvalidNameReport();
+
+    // Create a directory via the router
+    String dirPath = "/testRetryWhenClusterisDown";
+    FsPermission permission = new FsPermission("705");
+    try {
+      routerProtocol.mkdirs(dirPath, permission, false);
+      fail("Should have thrown RemoteException error.");
+    } catch (RemoteException e) {
+      String ns0 = cluster.getNameservices().get(0);
+      GenericTestUtils.assertExceptionContains(
+          "No namenode available under nameservice " + ns0, e);
+    }
+
+    // Verify the retry times, it should only retry one time.
+    FederationRPCMetrics rpcMetrics = routerContext.getRouter()
+        .getRpcServer().getRPCMetrics();
+    assertEquals(1, rpcMetrics.getProxyOpRetries());
+  }
+
+  @Test
+  public void testRetryWhenOneNameServiceDown() throws Exception {
+    // shutdown the dfs cluster
+    MiniDFSCluster dfsCluster = cluster.getCluster();
+    dfsCluster.shutdownNameNode(0);
+
+    // register an invalid namenode report
+    registerInvalidNameReport();
+
+    DFSClient client = nnContext1.getClient();
+    // Renew lease for the DFS client, it will succeed.
+    routerProtocol.renewLease(client.getClientName());
+
+    // Verify the retry times, it will retry one time for ns0.
+    FederationRPCMetrics rpcMetrics = routerContext.getRouter()
+        .getRpcServer().getRPCMetrics();
+    assertEquals(1, rpcMetrics.getProxyOpRetries());
+  }
+
+  /**
+   * Register an invalid namenode report.
+   * @throws IOException
+   */
+  private void registerInvalidNameReport() throws IOException {
+    String ns0 = cluster.getNameservices().get(0);
+    List<? extends FederationNamenodeContext> origin = resolver
+        .getNamenodesForNameserviceId(ns0);
+    FederationNamenodeContext nnInfo = origin.get(0);
+    NamenodeStatusReport report = new NamenodeStatusReport(ns0,
+        nnInfo.getNamenodeId(), nnInfo.getRpcAddress(),
+        nnInfo.getServiceAddress(), nnInfo.getLifelineAddress(),
+        nnInfo.getWebAddress());
+    report.setRegistrationValid(false);
+    assertTrue(resolver.registerNamenode(report));
+    resolver.loadCache(true);
+  }
+}


---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org


Mime
View raw message