class HddsVolumeChecker {

  public Set<HddsVolume> checkAllVolumes(Collection<HddsVolume> volumes)
      throws InterruptedException {
    final long gap = timer.monotonicNow() - lastAllVolumesCheck;
    if (gap < minDiskCheckGapMs) {
      numSkippedChecks.incrementAndGet();
      if (LOG.isTraceEnabled()) {
        LOG.trace(
            "Skipped checking all volumes, time since last check {} is less " +
                "than the minimum gap between checks ({} ms).",
            gap, minDiskCheckGapMs);
      }
      return Collections.emptySet();
    }

    lastAllVolumesCheck = timer.monotonicNow();
    final Set<HddsVolume> healthyVolumes = new HashSet<>();
    final Set<HddsVolume> failedVolumes = new HashSet<>();
    final Set<HddsVolume> allVolumes = new HashSet<>();

    final AtomicLong numVolumes = new AtomicLong(volumes.size());
    final CountDownLatch latch = new CountDownLatch(1);

    for (HddsVolume v : volumes) {
      Optional<ListenableFuture<VolumeCheckResult>> olf =
          delegateChecker.schedule(v, null);
      LOG.info("Scheduled health check for volume {}", v);
      if (olf.isPresent()) {
        allVolumes.add(v);
        Futures.addCallback(olf.get(),
            new ResultHandler(v, healthyVolumes, failedVolumes,
                numVolumes, (ignored1, ignored2) -> latch.countDown()),
            MoreExecutors.directExecutor());
      } else {
        if (numVolumes.decrementAndGet() == 0) {
          latch.countDown();
        }
      }
    }

    // Wait until our timeout elapses, after which we give up on
    // the remaining volumes.
    if (!latch.await(maxAllowedTimeForCheckMs, TimeUnit.MILLISECONDS)) {
      LOG.warn("checkAllVolumes timed out after {} ms",
          maxAllowedTimeForCheckMs);
    }

    numAllVolumeChecks.incrementAndGet();
    synchronized (this) {
      // All volumes that have not been detected as healthy should be
      // considered failed. This is a superset of 'failedVolumes'.
      //
      // Make a copy under the mutex as Sets.difference() returns a view
      // of a potentially changing set.
      return new HashSet<>(Sets.difference(allVolumes, healthyVolumes));
    }
  }

  public  HddsVolumeChecker(ConfigurationSource conf, Timer timer);

}