class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> {

  @Override
  public void removeVolumes(Set<File> volumesToRemove, boolean clearFailure) {
    // Make sure that all volumes are absolute path.
    for (File vol : volumesToRemove) {
      Preconditions.checkArgument(vol.isAbsolute(),
          String.format("%s is not absolute path.", vol.getPath()));
    }

    Map<String, List<ReplicaInfo>> blkToInvalidate = new HashMap<>();
    List<String> storageToRemove = new ArrayList<>();
    synchronized (this) {
      for (int idx = 0; idx < dataStorage.getNumStorageDirs(); idx++) {
        Storage.StorageDirectory sd = dataStorage.getStorageDir(idx);
        final File absRoot = sd.getRoot().getAbsoluteFile();
        if (volumesToRemove.contains(absRoot)) {
          LOG.info("Removing " + absRoot + " from FsDataset.");

          // Disable the volume from the service.
          asyncDiskService.removeVolume(sd.getCurrentDir());
          volumes.removeVolume(absRoot, clearFailure);

          // Removed all replica information for the blocks on the volume.
          // Unlike updating the volumeMap in addVolume(), this operation does
          // not scan disks.
          for (String bpid : volumeMap.getBlockPoolList()) {
            List<ReplicaInfo> blocks = new ArrayList<>();
            for (Iterator<ReplicaInfo> it = volumeMap.replicas(bpid).iterator();
                 it.hasNext(); ) {
              ReplicaInfo block = it.next();
              final File absBasePath =
                  new File(block.getVolume().getBasePath()).getAbsoluteFile();
              if (absBasePath.equals(absRoot)) {
                blocks.add(block);
                it.remove();
              }
            }
            blkToInvalidate.put(bpid, blocks);
          }

          storageToRemove.add(sd.getStorageUuid());
        }
      }
      setupAsyncLazyPersistThreads();
    }

    // Call this outside the lock.
    for (Map.Entry<String, List<ReplicaInfo>> entry :
        blkToInvalidate.entrySet()) {
      String bpid = entry.getKey();
      List<ReplicaInfo> blocks = entry.getValue();
      for (ReplicaInfo block : blocks) {
        invalidate(bpid, block);
      }
    }

    synchronized (this) {
      for(String storageUuid : storageToRemove) {
        storageMap.remove(storageUuid);
      }
    }
  }

    FsDatasetImpl(DataNode datanode, DataStorage storage, Configuration conf
      );

}