public DirectoryScanner(FsDatasetSpi<?> dataset, Configuration conf) {
    this.dataset = dataset;
    this.stats = new HashMap<>(DEFAULT_MAP_SIZE);
    int interval = (int) conf.getTimeDuration(
        DFSConfigKeys.DFS_DATANODE_DIRECTORYSCAN_INTERVAL_KEY,
        DFSConfigKeys.DFS_DATANODE_DIRECTORYSCAN_INTERVAL_DEFAULT,
        TimeUnit.SECONDS);

    scanPeriodMsecs = TimeUnit.SECONDS.toMillis(interval);

    int throttle = conf.getInt(
        DFSConfigKeys.DFS_DATANODE_DIRECTORYSCAN_THROTTLE_LIMIT_MS_PER_SEC_KEY,
        DFSConfigKeys.DFS_DATANODE_DIRECTORYSCAN_THROTTLE_LIMIT_MS_PER_SEC_DEFAULT);

    if (throttle >= TimeUnit.SECONDS.toMillis(1)) {
      LOG.warn(
          "{} set to value above 1000 ms/sec. Assuming default value of {}",
          DFSConfigKeys.DFS_DATANODE_DIRECTORYSCAN_THROTTLE_LIMIT_MS_PER_SEC_KEY,
          DFSConfigKeys.DFS_DATANODE_DIRECTORYSCAN_THROTTLE_LIMIT_MS_PER_SEC_DEFAULT);
      throttle =
          DFSConfigKeys.DFS_DATANODE_DIRECTORYSCAN_THROTTLE_LIMIT_MS_PER_SEC_DEFAULT;
    }

    throttleLimitMsPerSec = throttle;

    int threads =
        conf.getInt(DFSConfigKeys.DFS_DATANODE_DIRECTORYSCAN_THREADS_KEY,
            DFSConfigKeys.DFS_DATANODE_DIRECTORYSCAN_THREADS_DEFAULT);

    reportCompileThreadPool =
        Executors.newFixedThreadPool(threads, new Daemon.DaemonFactory());

    masterThread =
        new ScheduledThreadPoolExecutor(1, new Daemon.DaemonFactory());

    int reconcileBatchSize =
        conf.getInt(DFSConfigKeys.
                DFS_DATANODE_RECONCILE_BLOCKS_BATCH_SIZE,
            DFSConfigKeys.
                DFS_DATANODE_RECONCILE_BLOCKS_BATCH_SIZE_DEFAULT);

    if (reconcileBatchSize <= 0) {
      LOG.warn("Invalid value configured for " +
              "dfs.datanode.reconcile.blocks.batch.size, " +
              "should be greater than 0, Using default.");
      reconcileBatchSize =
          DFSConfigKeys.
              DFS_DATANODE_RECONCILE_BLOCKS_BATCH_SIZE_DEFAULT;
    }

    reconcileBlocksBatchSize = reconcileBatchSize;

    long reconcileBatchInterval =
        conf.getTimeDuration(DFSConfigKeys.
                DFS_DATANODE_RECONCILE_BLOCKS_BATCH_INTERVAL,
            DFSConfigKeys.
                DFS_DATANODE_RECONCILE_BLOCKS_BATCH_INTERVAL_DEFAULT,
            TimeUnit.MILLISECONDS);

    if (reconcileBatchInterval <= 0) {
      LOG.warn("Invalid value configured for " +
              "dfs.datanode.reconcile.blocks.batch.interval, " +
              "should be greater than 0, Using default.");
      reconcileBatchInterval =
          DFSConfigKeys.
              DFS_DATANODE_RECONCILE_BLOCKS_BATCH_INTERVAL_DEFAULT;
    }

    reconcileBlocksBatchInterval = reconcileBatchInterval;
  }
@Test(timeout = 600000)
  public void testDirectoryScanner() throws Exception {
    // Run the test with and without parallel scanning
    for (int parallelism = 1; parallelism < 3; parallelism++) {
      runTest(parallelism);
    }
  }