@Test public void testAddVolumes() throws IOException { final int numNewVolumes = 3; final int numExistingVolumes = dataset.getVolumes().size(); final int totalVolumes = numNewVolumes + numExistingVolumes; List<StorageLocation> newLocations = new ArrayList<StorageLocation>(); for (int i = 0; i < numNewVolumes; i++) { String path = BASE_DIR + "/newData" + i; newLocations.add(StorageLocation.parse(path)); when(storage.getStorageDir(numExistingVolumes + i)) .thenReturn(new Storage.StorageDirectory(new File(path))); } when(storage.getNumStorageDirs()).thenReturn(totalVolumes); dataset.addVolumes(newLocations); assertEquals(totalVolumes, dataset.getVolumes().size()); for (int i = 0; i < numNewVolumes; i++) { assertEquals(newLocations.get(i).getFile().getPath(), dataset.getVolumes().get(numExistingVolumes + i).getBasePath()); } }
@Override public synchronized void addVolumes(Collection<StorageLocation> volumes) throws IOException { final Collection<StorageLocation> dataLocations = DataNode.getStorageLocations(this.conf); Map<String, Storage.StorageDirectory> allStorageDirs = new HashMap<String, Storage.StorageDirectory>(); for (int idx = 0; idx < dataStorage.getNumStorageDirs(); idx++) { Storage.StorageDirectory sd = dataStorage.getStorageDir(idx); allStorageDirs.put(sd.getRoot().getAbsolutePath(), sd); } for (StorageLocation vol : volumes) { String key = vol.getFile().getAbsolutePath(); if (!allStorageDirs.containsKey(key)) { LOG.warn("Attempt to add an invalid volume: " + vol.getFile()); } else { addVolume(dataLocations, allStorageDirs.get(key)); } } }
FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> { @Override public synchronized void addVolumes(Collection<StorageLocation> volumes) throws IOException { final Collection<StorageLocation> dataLocations = DataNode.getStorageLocations(this.conf); Map<String, Storage.StorageDirectory> allStorageDirs = new HashMap<String, Storage.StorageDirectory>(); for (int idx = 0; idx < dataStorage.getNumStorageDirs(); idx++) { Storage.StorageDirectory sd = dataStorage.getStorageDir(idx); allStorageDirs.put(sd.getRoot().getAbsolutePath(), sd); } for (StorageLocation vol : volumes) { String key = vol.getFile().getAbsolutePath(); if (!allStorageDirs.containsKey(key)) { LOG.warn("Attempt to add an invalid volume: " + vol.getFile()); } else { addVolume(dataLocations, allStorageDirs.get(key)); } } } }
FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> { @Override public synchronized void addVolumes(Collection<StorageLocation> volumes) throws IOException { final Collection<StorageLocation> dataLocations = DataNode.getStorageLocations(this.conf); Map<String, Storage.StorageDirectory> allStorageDirs = new HashMap<String, Storage.StorageDirectory>(); for (int idx = 0; idx < dataStorage.getNumStorageDirs(); idx++) { Storage.StorageDirectory sd = dataStorage.getStorageDir(idx); allStorageDirs.put(sd.getRoot().getAbsolutePath(), sd); } for (StorageLocation vol : volumes) { String key = vol.getFile().getAbsolutePath(); if (!allStorageDirs.containsKey(key)) { LOG.warn("Attempt to add an invalid volume: " + vol.getFile()); } else { addVolume(dataLocations, allStorageDirs.get(key)); } } } FsDatasetImpl(DataNode datanode, DataStorage storage, Configuration conf
      ); }
FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> { @Override public synchronized void addVolumes(Collection<StorageLocation> volumes) throws IOException { final Collection<StorageLocation> dataLocations = DataNode.getStorageLocations(this.conf); Map<String, Storage.StorageDirectory> allStorageDirs = new HashMap<String, Storage.StorageDirectory>(); for (int idx = 0; idx < dataStorage.getNumStorageDirs(); idx++) { Storage.StorageDirectory sd = dataStorage.getStorageDir(idx); allStorageDirs.put(sd.getRoot().getAbsolutePath(), sd); } for (StorageLocation vol : volumes) { String key = vol.getFile().getAbsolutePath(); if (!allStorageDirs.containsKey(key)) { LOG.warn("Attempt to add an invalid volume: " + vol.getFile()); } else { addVolume(dataLocations, allStorageDirs.get(key)); } } } 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); @Override synchronized void addVolumes(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 String recoverClose(ExtendedBlock b, long newGS,
      long expectedBlockLen); @Override // FsDatasetSpi synchronized ReplicaInPipeline createRbw(StorageType storageType,
      ExtendedBlock b); @Override // FsDatasetSpi synchronized ReplicaInPipeline recoverRbw(ExtendedBlock b,
      long newGS, long minBytesRcvd, long maxBytesRcvd); @Override // FsDatasetSpi synchronized ReplicaInPipeline convertTemporaryToRbw(
      final ExtendedBlock b); @Override // FsDatasetSpi synchronized 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 // FsDatasetSpi List<Long> getCacheReport(String bpid); @Override Map<DatanodeStorage, BlockListAsLongs> getBlockReports(String bpid); @Override synchronized List<FinalizedReplica> getFinalizedBlocks(String bpid); @Override // FsDatasetSpi boolean isValidBlock(ExtendedBlock b); @Override // {@link FsDatasetSpi} boolean isValidRbw(final ExtendedBlock b); @Override // FsDatasetSpi void invalidate(String bpid, Block invalidBlks[]); @Override // FsDatasetSpi void cache(String bpid, long[] blockIds); @Override // FsDatasetSpi void uncache(String bpid, long[] blockIds); 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 RollingLogs createRollingLogs(String bpid, String prefix
      ); @Override void submitBackgroundSyncFileRangeRequest(ExtendedBlock block,
      FileDescriptor fd, long offset, long nbytes, int flags); }
FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> { @Override public synchronized void addVolumes(Collection<StorageLocation> volumes) throws IOException { final Collection<StorageLocation> dataLocations = DataNode.getStorageLocations(this.conf); Map<String, Storage.StorageDirectory> allStorageDirs = new HashMap<String, Storage.StorageDirectory>(); for (int idx = 0; idx < dataStorage.getNumStorageDirs(); idx++) { Storage.StorageDirectory sd = dataStorage.getStorageDir(idx); allStorageDirs.put(sd.getRoot().getAbsolutePath(), sd); } for (StorageLocation vol : volumes) { String key = vol.getFile().getAbsolutePath(); if (!allStorageDirs.containsKey(key)) { LOG.warn("Attempt to add an invalid volume: " + vol.getFile()); } else { addVolume(dataLocations, allStorageDirs.get(key)); } } } 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); @Override synchronized void addVolumes(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 String recoverClose(ExtendedBlock b, long newGS,
      long expectedBlockLen); @Override // FsDatasetSpi synchronized ReplicaInPipeline createRbw(StorageType storageType,
      ExtendedBlock b); @Override // FsDatasetSpi synchronized ReplicaInPipeline recoverRbw(ExtendedBlock b,
      long newGS, long minBytesRcvd, long maxBytesRcvd); @Override // FsDatasetSpi synchronized ReplicaInPipeline convertTemporaryToRbw(
      final ExtendedBlock b); @Override // FsDatasetSpi synchronized 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 // FsDatasetSpi List<Long> getCacheReport(String bpid); @Override Map<DatanodeStorage, BlockListAsLongs> getBlockReports(String bpid); @Override synchronized List<FinalizedReplica> getFinalizedBlocks(String bpid); @Override // FsDatasetSpi boolean isValidBlock(ExtendedBlock b); @Override // {@link FsDatasetSpi} boolean isValidRbw(final ExtendedBlock b); @Override // FsDatasetSpi void invalidate(String bpid, Block invalidBlks[]); @Override // FsDatasetSpi void cache(String bpid, long[] blockIds); @Override // FsDatasetSpi void uncache(String bpid, long[] blockIds); 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 RollingLogs createRollingLogs(String bpid, String prefix
      ); @Override void submitBackgroundSyncFileRangeRequest(ExtendedBlock block,
      FileDescriptor fd, long offset, long nbytes, int flags);  }