@Override
  public void checkStatus(File dir) throws DiskErrorException {
    ReadWriteDiskValidatorMetrics metric =
        ReadWriteDiskValidatorMetrics.getMetric(dir.toString());
    Path tmpFile = null;
    try {
      if (!dir.isDirectory()) {
        metric.diskCheckFailed();
        throw new DiskErrorException(dir + " is not a directory!");
      }

      // check the directory presence and permission.
      DiskChecker.checkDir(dir);

      // create a tmp file under the dir
      tmpFile = Files.createTempFile(dir.toPath(), "test", "tmp");

      // write 16 bytes into the tmp file
      byte[] inputBytes = new byte[16];
      RANDOM.nextBytes(inputBytes);
      long startTime = System.nanoTime();
      Files.write(tmpFile, inputBytes);
      long writeLatency = TimeUnit.MICROSECONDS.convert(
          System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
      metric.addWriteFileLatency(writeLatency);

      // read back
      startTime = System.nanoTime();
      byte[] outputBytes = Files.readAllBytes(tmpFile);
      long readLatency = TimeUnit.MICROSECONDS.convert(
          System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
      metric.addReadFileLatency(readLatency);

      // validation
      if (!Arrays.equals(inputBytes, outputBytes)) {
        metric.diskCheckFailed();
        throw new DiskErrorException("Data in file has been corrupted.");
      }
    } catch (IOException e) {
      metric.diskCheckFailed();
      throw new DiskErrorException("Disk Check failed!", e);
    } finally {
      // delete the file
      if (tmpFile != null) {
        try {
          Files.delete(tmpFile);
        } catch (IOException e) {
          metric.diskCheckFailed();
          throw new DiskErrorException("File deletion failed!", e);
        }
      }
    }
  }
@Test
  public void testCheckFailures() throws Throwable {
    ReadWriteDiskValidator readWriteDiskValidator =
        (ReadWriteDiskValidator) DiskValidatorFactory.getInstance(
            ReadWriteDiskValidator.NAME);

    // create a temporary test directory under the system test directory
    File testDir = Files.createTempDirectory(
        Paths.get(System.getProperty("test.build.data")), "test").toFile();

    try {
      Shell.execCommand(Shell.getSetPermissionCommand("000", false,
          testDir.getAbsolutePath()));
    } catch (Exception e){
      testDir.delete();
      throw e;
    }

    try {
      readWriteDiskValidator.checkStatus(testDir);
      fail("Disk check should fail.");
    } catch (DiskErrorException e) {
      assertEquals("Disk Check failed!", e.getMessage());
    }

    MetricsSource source = ms.getSource(
        ReadWriteDiskValidatorMetrics.sourceName(testDir.toString()));
    MetricsCollectorImpl collector = new MetricsCollectorImpl();
    source.getMetrics(collector, true);

    try {
      readWriteDiskValidator.checkStatus(testDir);
      fail("Disk check should fail.");
    } catch (DiskErrorException e) {
      assertEquals("Disk Check failed!", e.getMessage());
    }

    source.getMetrics(collector, true);

    // verify the first metrics record
    MetricsRecords.assertMetric(collector.getRecords().get(0),
        "FailureCount", 1);
    Long lastFailureTime1 = (Long) MetricsRecords.getMetricValueByName(
        collector.getRecords().get(0), "LastFailureTime");

    // verify the second metrics record
    MetricsRecords.assertMetric(collector.getRecords().get(1),
        "FailureCount", 2);
    Long lastFailureTime2 = (Long) MetricsRecords.getMetricValueByName(
        collector.getRecords().get(1), "LastFailureTime");
    assertTrue("The first failure time should be less than the second one",
        lastFailureTime1 < lastFailureTime2);

    testDir.delete();
  }