FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> { @Override public void removeVolumes(Set<File> volumesToRemove, boolean clearFailure) { 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."); asyncDiskService.removeVolume(sd.getCurrentDir()); volumes.removeVolume(absRoot, clearFailure); 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(); } 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
      ); @Override // FsDatasetSpi List<FsVolumeImpl> getVolumes(); @Override DatanodeStorage getStorage(final String storageUuid); @Override // FsDatasetSpi StorageReport[] getStorageReports(String bpid); @Override synchronized FsVolumeImpl getVolume(final ExtendedBlock b); @Override // FsDatasetSpi synchronized Block getStoredBlock(String bpid, long blkid); @Override // FsDatasetSpi LengthInputStream getMetaDataInputStream(ExtendedBlock b); @VisibleForTesting FsVolumeImpl createFsVolume(String storageUuid, File currentDir,
      StorageType storageType); @Override void addVolume(final StorageLocation location,
      final List<NamespaceInfo> nsInfos); @Override void removeVolumes(Set<File> volumesToRemove, boolean clearFailure); @Override // FSDatasetMBean long getDfsUsed(); @Override // FSDatasetMBean long getBlockPoolUsed(String bpid); @Override // FsDatasetSpi boolean hasEnoughResource(); @Override // FSDatasetMBean long getCapacity(); @Override // FSDatasetMBean long getRemaining(); @Override // FSDatasetMBean int getNumFailedVolumes(); @Override // FSDatasetMBean String[] getFailedStorageLocations(); @Override // FSDatasetMBean long getLastVolumeFailureDate(); @Override // FSDatasetMBean long getEstimatedCapacityLostTotal(); @Override // FsDatasetSpi VolumeFailureSummary getVolumeFailureSummary(); @Override // FSDatasetMBean long getCacheUsed(); @Override // FSDatasetMBean long getCacheCapacity(); @Override // FSDatasetMBean long getNumBlocksFailedToCache(); @Override // FSDatasetMBean long getNumBlocksFailedToUncache(); @Override // FSDatasetMBean long getNumBlocksCached(); @Override // FsDatasetSpi long getLength(ExtendedBlock b); @Override // FsDatasetSpi InputStream getBlockInputStream(ExtendedBlock b,
      long seekOffset); @Override // FsDatasetSpi synchronized ReplicaInputStreams getTmpInputStreams(ExtendedBlock b,
      long blkOffset, long metaOffset); @Override ReplicaInfo moveBlockAcrossStorage(ExtendedBlock block,
      StorageType targetStorageType); @Override // FsDatasetSpi synchronized ReplicaHandler append(ExtendedBlock b,
      long newGS, long expectedBlockLen); @Override // FsDatasetSpi synchronized ReplicaHandler recoverAppend(
      ExtendedBlock b, long newGS, long expectedBlockLen); @Override // FsDatasetSpi synchronized String recoverClose(ExtendedBlock b, long newGS,
      long expectedBlockLen); @Override // FsDatasetSpi synchronized ReplicaHandler createRbw(
      StorageType storageType, ExtendedBlock b, boolean allowLazyPersist); @Override // FsDatasetSpi synchronized ReplicaHandler recoverRbw(
      ExtendedBlock b, long newGS, long minBytesRcvd, long maxBytesRcvd); @Override // FsDatasetSpi synchronized ReplicaInPipeline convertTemporaryToRbw(
      final ExtendedBlock b); @Override // FsDatasetSpi ReplicaHandler createTemporary(
      StorageType storageType, ExtendedBlock b); @Override // FsDatasetSpi void adjustCrcChannelPosition(ExtendedBlock b, ReplicaOutputStreams streams, 
      int checksumSize); @Override // FsDatasetSpi synchronized void finalizeBlock(ExtendedBlock b); @Override // FsDatasetSpi synchronized void unfinalizeBlock(ExtendedBlock b); @Override Map<DatanodeStorage, BlockListAsLongs> getBlockReports(String bpid); @Override // FsDatasetSpi List<Long> getCacheReport(String bpid); @Override synchronized List<FinalizedReplica> getFinalizedBlocks(String bpid); @Override synchronized List<FinalizedReplica> getFinalizedBlocksOnPersistentStorage(String bpid); void checkBlock(ExtendedBlock b, long minLength, ReplicaState state); @Override // FsDatasetSpi boolean isValidBlock(ExtendedBlock b); @Override // {@link FsDatasetSpi} boolean isValidRbw(final ExtendedBlock b); @Override // FsDatasetSpi void invalidate(String bpid, Block invalidBlks[]); void invalidate(String bpid, ReplicaInfo block); @Override // FsDatasetSpi void cache(String bpid, long[] blockIds); @Override // FsDatasetSpi void uncache(String bpid, long[] blockIds); @Override boolean isCached(String bpid, long blockId); @Override // FsDatasetSpi synchronized boolean contains(final ExtendedBlock block); @Override // FsDatasetSpi Set<File> checkDataDir(); @Override // FsDatasetSpi String toString(); @Override // FsDatasetSpi void shutdown(); @Override // FSDatasetMBean String getStorageInfo(); @Override void checkAndUpdate(String bpid, long blockId, File diskFile,
      File diskMetaFile, FsVolumeSpi vol); @Override // FsDatasetSpi @Deprecated ReplicaInfo getReplica(String bpid, long blockId); @Override synchronized String getReplicaString(String bpid, long blockId); @Override // FsDatasetSpi synchronized ReplicaRecoveryInfo initReplicaRecovery(
      RecoveringBlock rBlock); @Override // FsDatasetSpi synchronized String updateReplicaUnderRecovery(
                                    final ExtendedBlock oldBlock,
                                    final long recoveryId,
                                    final long newBlockId,
                                    final long newlength); @Override // FsDatasetSpi synchronized long getReplicaVisibleLength(final ExtendedBlock block); @Override void addBlockPool(String bpid, Configuration conf); @Override synchronized void shutdownBlockPool(String bpid); @Override Map<String, Object> getVolumeInfoMap(); @Override //FsDatasetSpi synchronized void deleteBlockPool(String bpid, boolean force); @Override // FsDatasetSpi BlockLocalPathInfo getBlockLocalPathInfo(ExtendedBlock block); @Override // FsDatasetSpi HdfsBlocksMetadata getHdfsBlocksMetadata(String poolId,
      long[] blockIds); @Override void enableTrash(String bpid); @Override void clearTrash(String bpid); @Override boolean trashEnabled(String bpid); @Override void setRollingUpgradeMarker(String bpid); @Override void clearRollingUpgradeMarker(String bpid); @Override void onCompleteLazyPersist(String bpId, long blockId,
      long creationTime, File[] savedFiles, FsVolumeImpl targetVolume); @Override void onFailLazyPersist(String bpId, long blockId); @Override void submitBackgroundSyncFileRangeRequest(ExtendedBlock block,
      FileDescriptor fd, long offset, long nbytes, int flags); @Override void setPinning(ExtendedBlock block); @Override boolean getPinning(ExtendedBlock block); @Override boolean isDeletingBlock(String bpid, long blockId); void removeDeletedBlocks(String bpid, Set<Long> blockIds);  }
@Test(timeout = 30000) public void testRemoveVolumes() throws IOException { final int NUM_BLOCKS = 100; for (int i = 0; i < NUM_BLOCKS; i++) { String bpid = BLOCK_POOL_IDS[NUM_BLOCKS % BLOCK_POOL_IDS.length]; ExtendedBlock eb = new ExtendedBlock(bpid, i); try (ReplicaHandler replica = dataset.createRbw(StorageType.DEFAULT, eb, false)) { } } final String[] dataDirs = conf.get(DFSConfigKeys.DFS_DATANODE_DATA_DIR_KEY).split(","); final String volumePathToRemove = dataDirs[0]; Set<File> volumesToRemove = new HashSet<>(); volumesToRemove.add(StorageLocation.parse(volumePathToRemove).getFile()); dataset.removeVolumes(volumesToRemove, true); int expectedNumVolumes = dataDirs.length - 1; assertEquals("The volume has been removed from the volumeList.", expectedNumVolumes, dataset.getVolumes().size()); assertEquals("The volume has been removed from the storageMap.", expectedNumVolumes, dataset.storageMap.size()); try { dataset.asyncDiskService.execute(volumesToRemove.iterator().next(), new Runnable() { @Override public void run() {} }); fail("Expect RuntimeException: the volume has been removed from the " + "AsyncDiskService."); } catch (RuntimeException e) { GenericTestUtils.assertExceptionContains("Cannot find root", e); } int totalNumReplicas = 0; for (String bpid : dataset.volumeMap.getBlockPoolList()) { totalNumReplicas += dataset.volumeMap.size(bpid); } assertEquals("The replica infos on this volume has been removed from the " + "volumeMap.", NUM_BLOCKS / NUM_INIT_VOLUMES, totalNumReplicas); }