@Test 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); dataset.createRbw(StorageType.DEFAULT, eb, false); } final String[] dataDirs = conf.get(DFSConfigKeys.DFS_DATANODE_DATA_DIR_KEY).split(","); final String volumePathToRemove = dataDirs[0]; List<StorageLocation> volumesToRemove = new ArrayList<StorageLocation>(); volumesToRemove.add(StorageLocation.parse(volumePathToRemove)); dataset.removeVolumes(volumesToRemove); 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.get(0).getFile(), 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); verify(scanner, times(BLOCK_POOL_IDS.length)) .deleteBlocks(anyString(), any(Block[].class)); }
@Override public void removeVolumes(Collection<StorageLocation> volumes) { Set<String> volumeSet = new HashSet<String>(); for (StorageLocation sl : volumes) { volumeSet.add(sl.getFile().getAbsolutePath()); } Map<String, List<ReplicaInfo>> blkToInvalidate = new HashMap<String, List<ReplicaInfo>>(); List<String> storageToRemove = new ArrayList<String>(); synchronized (this) { for (int idx = 0; idx < dataStorage.getNumStorageDirs(); idx++) { Storage.StorageDirectory sd = dataStorage.getStorageDir(idx); String volume = sd.getRoot().getAbsolutePath(); if (volumeSet.contains(volume)) { LOG.info("Removing " + volume + " from FsDataset."); asyncDiskService.removeVolume(sd.getCurrentDir()); this.volumes.removeVolume(sd.getRoot()); for (String bpid : volumeMap.getBlockPoolList()) { List<ReplicaInfo> blocks = new ArrayList<ReplicaInfo>(); for (Iterator<ReplicaInfo> it = volumeMap.replicas(bpid).iterator(); it.hasNext(); ) { ReplicaInfo block = it.next(); String absBasePath = new File(block.getVolume().getBasePath()).getAbsolutePath(); if (absBasePath.equals(volume)) { blocks.add(block); it.remove(); } } blkToInvalidate.put(bpid, blocks); datanode.getBlockScanner().deleteBlocks(bpid, blocks.toArray(new Block[blocks.size()])); } 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 implements FsDatasetSpi<FsVolumeImpl> { @Override public void removeVolumes(Collection<StorageLocation> volumes) { Set<String> volumeSet = new HashSet<String>(); for (StorageLocation sl : volumes) { volumeSet.add(sl.getFile().getAbsolutePath()); } Map<String, List<ReplicaInfo>> blkToInvalidate = new HashMap<String, List<ReplicaInfo>>(); List<String> storageToRemove = new ArrayList<String>(); synchronized (this) { for (int idx = 0; idx < dataStorage.getNumStorageDirs(); idx++) { Storage.StorageDirectory sd = dataStorage.getStorageDir(idx); String volume = sd.getRoot().getAbsolutePath(); if (volumeSet.contains(volume)) { LOG.info("Removing " + volume + " from FsDataset."); asyncDiskService.removeVolume(sd.getCurrentDir()); this.volumes.removeVolume(sd.getRoot()); for (String bpid : volumeMap.getBlockPoolList()) { List<ReplicaInfo> blocks = new ArrayList<ReplicaInfo>(); for (Iterator<ReplicaInfo> it = volumeMap.replicas(bpid).iterator(); it.hasNext(); ) { ReplicaInfo block = it.next(); String absBasePath = new File(block.getVolume().getBasePath()).getAbsolutePath(); if (absBasePath.equals(volume)) { blocks.add(block); it.remove(); } } blkToInvalidate.put(bpid, blocks); datanode.getBlockScanner().deleteBlocks(bpid, blocks.toArray(new Block[blocks.size()])); } 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 implements FsDatasetSpi<FsVolumeImpl> { @Override public void removeVolumes(Collection<StorageLocation> volumes) { Set<String> volumeSet = new HashSet<String>(); for (StorageLocation sl : volumes) { volumeSet.add(sl.getFile().getAbsolutePath()); } Map<String, List<ReplicaInfo>> blkToInvalidate = new HashMap<String, List<ReplicaInfo>>(); List<String> storageToRemove = new ArrayList<String>(); synchronized (this) { for (int idx = 0; idx < dataStorage.getNumStorageDirs(); idx++) { Storage.StorageDirectory sd = dataStorage.getStorageDir(idx); String volume = sd.getRoot().getAbsolutePath(); if (volumeSet.contains(volume)) { LOG.info("Removing " + volume + " from FsDataset."); asyncDiskService.removeVolume(sd.getCurrentDir()); this.volumes.removeVolume(sd.getRoot()); for (String bpid : volumeMap.getBlockPoolList()) { List<ReplicaInfo> blocks = new ArrayList<ReplicaInfo>(); for (Iterator<ReplicaInfo> it = volumeMap.replicas(bpid).iterator(); it.hasNext(); ) { ReplicaInfo block = it.next(); String absBasePath = new File(block.getVolume().getBasePath()).getAbsolutePath(); if (absBasePath.equals(volume)) { blocks.add(block); it.remove(); } } blkToInvalidate.put(bpid, blocks); datanode.getBlockScanner().deleteBlocks(bpid, blocks.toArray(new Block[blocks.size()])); } 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
      ); }
FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> { @Override public void removeVolumes(Collection<StorageLocation> volumes) { Set<String> volumeSet = new HashSet<String>(); for (StorageLocation sl : volumes) { volumeSet.add(sl.getFile().getAbsolutePath()); } Map<String, List<ReplicaInfo>> blkToInvalidate = new HashMap<String, List<ReplicaInfo>>(); List<String> storageToRemove = new ArrayList<String>(); synchronized (this) { for (int idx = 0; idx < dataStorage.getNumStorageDirs(); idx++) { Storage.StorageDirectory sd = dataStorage.getStorageDir(idx); String volume = sd.getRoot().getAbsolutePath(); if (volumeSet.contains(volume)) { LOG.info("Removing " + volume + " from FsDataset."); asyncDiskService.removeVolume(sd.getCurrentDir()); this.volumes.removeVolume(sd.getRoot()); for (String bpid : volumeMap.getBlockPoolList()) { List<ReplicaInfo> blocks = new ArrayList<ReplicaInfo>(); for (Iterator<ReplicaInfo> it = volumeMap.replicas(bpid).iterator(); it.hasNext(); ) { ReplicaInfo block = it.next(); String absBasePath = new File(block.getVolume().getBasePath()).getAbsolutePath(); if (absBasePath.equals(volume)) { blocks.add(block); it.remove(); } } blkToInvalidate.put(bpid, blocks); datanode.getBlockScanner().deleteBlocks(bpid, blocks.toArray(new Block[blocks.size()])); } 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(Collection<StorageLocation> volumes); @Override // FSDatasetMBean long getDfsUsed(); @Override // FSDatasetMBean long getBlockPoolUsed(String bpid); @Override // FsDatasetSpi boolean hasEnoughResource(); @Override // FSDatasetMBean long getCapacity(); @Override // FSDatasetMBean long getRemaining(); @Override int getNumFailedVolumes(); @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 ckoff); @Override // FsDatasetSpi synchronized ReplicaInPipeline append(ExtendedBlock b,
      long newGS, long expectedBlockLen); @Override // FsDatasetSpi synchronized ReplicaInPipeline recoverAppend(ExtendedBlock b,
      long newGS, long expectedBlockLen); @Override // FsDatasetSpi synchronized String recoverClose(ExtendedBlock b, long newGS,
      long expectedBlockLen); @Override // FsDatasetSpi synchronized ReplicaInPipeline createRbw(StorageType storageType,
      ExtendedBlock b, boolean allowLazyPersist); @Override // FsDatasetSpi synchronized ReplicaInPipeline recoverRbw(ExtendedBlock b,
      long newGS, long minBytesRcvd, long maxBytesRcvd); @Override // FsDatasetSpi synchronized ReplicaInPipeline convertTemporaryToRbw(
      final ExtendedBlock b); @Override // FsDatasetSpi ReplicaInPipeline 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 void 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 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 restoreTrash(String bpid); @Override boolean trashEnabled(String bpid); @Override void setRollingUpgradeMarker(String bpid); @Override void clearRollingUpgradeMarker(String bpid); @Override RollingLogs createRollingLogs(String bpid, String prefix
      ); @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); }
FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> { @Override public void removeVolumes(Collection<StorageLocation> volumes) { Set<String> volumeSet = new HashSet<String>(); for (StorageLocation sl : volumes) { volumeSet.add(sl.getFile().getAbsolutePath()); } Map<String, List<ReplicaInfo>> blkToInvalidate = new HashMap<String, List<ReplicaInfo>>(); List<String> storageToRemove = new ArrayList<String>(); synchronized (this) { for (int idx = 0; idx < dataStorage.getNumStorageDirs(); idx++) { Storage.StorageDirectory sd = dataStorage.getStorageDir(idx); String volume = sd.getRoot().getAbsolutePath(); if (volumeSet.contains(volume)) { LOG.info("Removing " + volume + " from FsDataset."); asyncDiskService.removeVolume(sd.getCurrentDir()); this.volumes.removeVolume(sd.getRoot()); for (String bpid : volumeMap.getBlockPoolList()) { List<ReplicaInfo> blocks = new ArrayList<ReplicaInfo>(); for (Iterator<ReplicaInfo> it = volumeMap.replicas(bpid).iterator(); it.hasNext(); ) { ReplicaInfo block = it.next(); String absBasePath = new File(block.getVolume().getBasePath()).getAbsolutePath(); if (absBasePath.equals(volume)) { blocks.add(block); it.remove(); } } blkToInvalidate.put(bpid, blocks); datanode.getBlockScanner().deleteBlocks(bpid, blocks.toArray(new Block[blocks.size()])); } 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(Collection<StorageLocation> volumes); @Override // FSDatasetMBean long getDfsUsed(); @Override // FSDatasetMBean long getBlockPoolUsed(String bpid); @Override // FsDatasetSpi boolean hasEnoughResource(); @Override // FSDatasetMBean long getCapacity(); @Override // FSDatasetMBean long getRemaining(); @Override int getNumFailedVolumes(); @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 ckoff); @Override // FsDatasetSpi synchronized ReplicaInPipeline append(ExtendedBlock b,
      long newGS, long expectedBlockLen); @Override // FsDatasetSpi synchronized ReplicaInPipeline recoverAppend(ExtendedBlock b,
      long newGS, long expectedBlockLen); @Override // FsDatasetSpi synchronized String recoverClose(ExtendedBlock b, long newGS,
      long expectedBlockLen); @Override // FsDatasetSpi synchronized ReplicaInPipeline createRbw(StorageType storageType,
      ExtendedBlock b, boolean allowLazyPersist); @Override // FsDatasetSpi synchronized ReplicaInPipeline recoverRbw(ExtendedBlock b,
      long newGS, long minBytesRcvd, long maxBytesRcvd); @Override // FsDatasetSpi synchronized ReplicaInPipeline convertTemporaryToRbw(
      final ExtendedBlock b); @Override // FsDatasetSpi ReplicaInPipeline 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 void 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 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 restoreTrash(String bpid); @Override boolean trashEnabled(String bpid); @Override void setRollingUpgradeMarker(String bpid); @Override void clearRollingUpgradeMarker(String bpid); @Override RollingLogs createRollingLogs(String bpid, String prefix
      ); @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);  }