@Test
public void testFsck() throws Exception {

  MountTable addEntry = MountTable.newInstance("/testdir",
      Collections.singletonMap("ns0", "/testdir"));
  assertTrue(addMountTable(addEntry));
  addEntry = MountTable.newInstance("/testdir2",
      Collections.singletonMap("ns1", "/testdir2"));
  assertTrue(addMountTable(addEntry));
  // create 1 file on ns0
  routerFs.createNewFile(new Path("/testdir/testfile"));
  // create 3 files on ns1
  routerFs.createNewFile(new Path("/testdir2/testfile2"));
  routerFs.createNewFile(new Path("/testdir2/testfile3"));
  routerFs.createNewFile(new Path("/testdir2/testfile4"));
  try (CloseableHttpClient httpClient = HttpClients.createDefault()) {
    // TODO: support https
    HttpGet httpGet = new HttpGet("http://" + webAddress.getHostName() +
            ":" + webAddress.getPort() + "/fsck");
    try (CloseableHttpResponse httpResponse = httpClient.execute(httpGet)) {
      assertEquals(HttpStatus.SC_OK,
          httpResponse.getStatusLine().getStatusCode());
      String out = EntityUtils.toString(
          httpResponse.getEntity(), StandardCharsets.UTF_8);
      LOG.info(out);
      assertTrue(out.contains("Federated FSCK started"));
      // assert 1 file exists in a cluster and 3 files exist
      // in another cluster
      assertTrue(out.contains("Total files:\t1"));
      assertTrue(out.contains("Total files:\t3"));
      assertTrue(out.contains("Federated FSCK ended"));
      int nnCount = 0;
      for (MembershipState nn : memberships) {
        if (nn.getState() == FederationNamenodeServiceState.ACTIVE) {
          assertTrue(out.contains(
              "Checking " + nn + " at " + nn.getWebAddress() + "\n"));
          nnCount++;
        }
      }
      assertEquals(2, nnCount);
    }
    // check if the argument is passed correctly
    httpGet = new HttpGet("http://" + webAddress.getHostName() +
            ":" + webAddress.getPort() + "/fsck?path=/testdir");
    try (CloseableHttpResponse httpResponse = httpClient.execute(httpGet)) {
      assertEquals(HttpStatus.SC_OK,
          httpResponse.getStatusLine().getStatusCode());
      String out = EntityUtils.toString(
          httpResponse.getEntity(), StandardCharsets.UTF_8);
      LOG.info(out);
      assertTrue(out.contains("Federated FSCK started"));
      assertTrue(out.contains("Total files:\t1"));
      // ns1 does not have files under /testdir
      assertFalse(out.contains("Total files:\t3"));
      assertTrue(out.contains("Federated FSCK ended"));
      int nnCount = 0;
      for (MembershipState nn : memberships) {
        if (nn.getState() == FederationNamenodeServiceState.ACTIVE) {
          assertTrue(out.contains(
              "Checking " + nn + " at " + nn.getWebAddress() + "\n"));
          nnCount++;
        }
      }
      assertEquals(2, nnCount);
    }
  }
}