public void fsck() {
    final long startTime = Time.monotonicNow();
    try {
      String warnMsg = "Now FSCK to DFSRouter is unstable feature. " +
          "There may be incompatible changes between releases.";
      LOG.warn(warnMsg);
      out.println(warnMsg);
      String msg = "Federated FSCK started by " +
          UserGroupInformation.getCurrentUser() + " from " + remoteAddress +
          " at " + new Date();
      LOG.info(msg);
      out.println(msg);

      // Check each Namenode in the federation
      StateStoreService stateStore = router.getStateStore();
      MembershipStore membership =
          stateStore.getRegisteredRecordStore(MembershipStore.class);
      GetNamenodeRegistrationsRequest request =
          GetNamenodeRegistrationsRequest.newInstance();
      GetNamenodeRegistrationsResponse response =
          membership.getNamenodeRegistrations(request);
      List<MembershipState> memberships = response.getNamenodeMemberships();
      Collections.sort(memberships);
      for (MembershipState nn : memberships) {
        if (nn.getState() == FederationNamenodeServiceState.ACTIVE) {
          try {
            String webAddress = nn.getWebAddress();
            out.write("Checking " + nn + " at " + webAddress + "\n");
            remoteFsck(nn);
          } catch (IOException ioe) {
            out.println("Cannot query " + nn + ": " + ioe.getMessage() + "\n");
          }
        }
      }

      out.println("Federated FSCK ended at " + new Date() + " in "
          + (Time.monotonicNow() - startTime + " milliseconds"));
    } catch (Exception e) {
      String errMsg = "Fsck " + e.getMessage();
      LOG.warn(errMsg, e);
      out.println("Federated FSCK ended at " + new Date() + " in "
          + (Time.monotonicNow() - startTime + " milliseconds"));
      out.println(e.getMessage());
      out.print("\n\n" + errMsg);
    } finally {
      out.close();
    }
  }
@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);
      }
    }
  }