DirectoryScanner implements Runnable { 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()); } DirectoryScanner(FsDatasetSpi<?> dataset, Configuration conf); @VisibleForTesting void setRetainDiffs(boolean b); @Override void run(); @VisibleForTesting void reconcile(); @VisibleForTesting Collection<ScanInfoVolumeReport> getVolumeReports();  }
@Test(timeout = 600000) public void testDirectoryScanner() throws Exception { for (int parallelism = 1; parallelism < 3; parallelism++) { runTest(parallelism); } }