DatasetVolumeChecker { public Set<FsVolumeSpi> checkAllVolumes( final FsDatasetSpi<? extends FsVolumeSpi> dataset) throws InterruptedException { final long gap = timer.monotonicNow() - lastAllVolumesCheck; if (gap < minDiskCheckGapMs) { numSkippedChecks.incrementAndGet(); LOG.trace( "Skipped checking all volumes, time since last check {} is less " + "than the minimum gap between checks ({} ms).", gap, minDiskCheckGapMs); return Collections.emptySet(); } final FsDatasetSpi.FsVolumeReferences references = dataset.getFsVolumeReferences(); if (references.size() == 0) { LOG.warn("checkAllVolumesAsync - no volumes can be referenced"); return Collections.emptySet(); } lastAllVolumesCheck = timer.monotonicNow(); final Set<FsVolumeSpi> healthyVolumes = new HashSet<>(); final Set<FsVolumeSpi> failedVolumes = new HashSet<>(); final Set<FsVolumeSpi> allVolumes = new HashSet<>(); final AtomicLong numVolumes = new AtomicLong(references.size()); final CountDownLatch latch = new CountDownLatch(1); for (int i = 0; i < references.size(); ++i) { final FsVolumeReference reference = references.getReference(i); Optional<ListenableFuture<VolumeCheckResult>> olf = delegateChecker.schedule(reference.getVolume(), IGNORED_CONTEXT); LOG.info("Scheduled health check for volume {}", reference.getVolume()); if (olf.isPresent()) { allVolumes.add(reference.getVolume()); Futures.addCallback(olf.get(), new ResultHandler(reference, healthyVolumes, failedVolumes, numVolumes, new Callback() { @Override public void call(Set<FsVolumeSpi> ignored1, Set<FsVolumeSpi> ignored2) { latch.countDown(); } }), MoreExecutors.directExecutor()); } else { IOUtils.cleanup(null, reference); if (numVolumes.decrementAndGet() == 0) { latch.countDown(); } } } if (!latch.await(maxAllowedTimeForCheckMs, TimeUnit.MILLISECONDS)) { LOG.warn("checkAllVolumes timed out after {} ms", maxAllowedTimeForCheckMs); } numSyncDatasetChecks.incrementAndGet(); synchronized (this) { return new HashSet<>(Sets.difference(allVolumes, healthyVolumes)); } } DatasetVolumeChecker(Configuration conf, Timer timer); Set<FsVolumeSpi> checkAllVolumes(
      final FsDatasetSpi<? extends FsVolumeSpi> dataset); boolean checkVolume(
      final FsVolumeSpi volume,
      Callback callback); void shutdownAndWait(int gracePeriod, TimeUnit timeUnit); long getNumVolumeChecks(); long getNumSyncDatasetChecks(); long getNumSkippedChecks(); static final Logger LOG; }
@Test(timeout = 10000) public void testCheckAllVolumes() throws Exception { LOG.info("Executing {}", testName.getMethodName()); final List<FsVolumeSpi> volumes = makeVolumes( NUM_VOLUMES, expectedVolumeHealth); final FsDatasetSpi<FsVolumeSpi> dataset = makeDataset(volumes); final DatasetVolumeChecker checker = new DatasetVolumeChecker(new HdfsConfiguration(), new FakeTimer()); checker.setDelegateChecker(new DummyChecker()); Set<FsVolumeSpi> failedVolumes = checker.checkAllVolumes(dataset); LOG.info("Got back {} failed volumes", failedVolumes.size()); if (expectedVolumeHealth == null || expectedVolumeHealth == FAILED) { assertThat(failedVolumes.size(), is(NUM_VOLUMES)); } else { assertTrue(failedVolumes.isEmpty()); } for (FsVolumeSpi volume : volumes) { verify(volume, times(1)).check(any()); } }