@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 testReadWriteDiskValidator()
      throws DiskErrorException, InterruptedException {
    int count = 100;
    File testDir = new File(System.getProperty("test.build.data"));
    ReadWriteDiskValidator readWriteDiskValidator =
        (ReadWriteDiskValidator) DiskValidatorFactory.getInstance(
            ReadWriteDiskValidator.NAME);

    for (int i = 0; i < count; i++) {
      readWriteDiskValidator.checkStatus(testDir);
    }

    ReadWriteDiskValidatorMetrics metric =
        ReadWriteDiskValidatorMetrics.getMetric(testDir.toString());
    Assert.assertEquals("The count number of estimator in MutableQuantiles"
        + "metrics of file read is not right",
        metric.getFileReadQuantiles()[0].getEstimator().getCount(), count);

    Assert.assertEquals("The count number of estimator in MutableQuantiles"
        + "metrics of file write is not right",
        metric.getFileWriteQuantiles()[0].getEstimator().getCount(),
        count);

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

    MetricsRecords.assertMetric(collector.getRecords().get(0),
        "FailureCount", 0);
    MetricsRecords.assertMetric(collector.getRecords().get(0),
        "LastFailureTime", (long)0);

    // All MutableQuantiles haven't rolled over yet because the minimum
    // interval is 1 hours, so we just test if these metrics exist.
    MetricsRecords.assertMetricNotNull(collector.getRecords().get(0),
        "WriteLatency3600sNumOps");
    MetricsRecords.assertMetricNotNull(collector.getRecords().get(0),
        "WriteLatency3600s50thPercentileLatencyMicros");
    MetricsRecords.assertMetricNotNull(collector.getRecords().get(0),
        "WriteLatency86400sNumOps");
    MetricsRecords.assertMetricNotNull(collector.getRecords().get(0),
        "WriteLatency864000sNumOps");

    MetricsRecords.assertMetricNotNull(collector.getRecords().get(0),
        "ReadLatency3600sNumOps");
    MetricsRecords.assertMetricNotNull(collector.getRecords().get(0),
        "ReadLatency3600s50thPercentileLatencyMicros");
    MetricsRecords.assertMetricNotNull(collector.getRecords().get(0),
        "ReadLatency86400sNumOps");
    MetricsRecords.assertMetricNotNull(collector.getRecords().get(0),
        "ReadLatency864000sNumOps");
  }