FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> { @Override public void shutdown() { fsRunning = false; if (lazyWriter != null) { ((LazyWriter) lazyWriter.getRunnable()).stop(); lazyWriter.interrupt(); } if (mbeanName != null) { MBeans.unregister(mbeanName); mbeanName = null; } if (asyncDiskService != null) { asyncDiskService.shutdown(); } if (asyncLazyPersistService != null) { asyncLazyPersistService.shutdown(); } if(volumes != null) { volumes.shutdown(); } if (lazyWriter != null) { try { lazyWriter.join(); } catch (InterruptedException ie) { LOG.warn("FsDatasetImpl.shutdown ignoring InterruptedException " + "from LazyWriter.join"); } } cacheManager.shutdown(); } FsDatasetImpl(DataNode datanode, DataStorage storage, Configuration conf
      ); @Override // FsDatasetSpi FsVolumeReferences getFsVolumeReferences(); @Override DatanodeStorage getStorage(final String storageUuid); @Override // FsDatasetSpi StorageReport[] getStorageReports(String bpid); @Override FsVolumeImpl getVolume(final ExtendedBlock b); @Override // FsDatasetSpi Block getStoredBlock(String bpid, long blkid); @Override Set<? extends Replica> deepCopyReplica(String bpid); @Override // FsDatasetSpi LengthInputStream getMetaDataInputStream(ExtendedBlock b); @Override AutoCloseableLock acquireDatasetLock(); @Override AutoCloseableLock acquireDatasetReadLock(); @VisibleForTesting FsVolumeImpl createFsVolume(String storageUuid,
      Storage.StorageDirectory sd,
      final StorageLocation location); @Override void addVolume(final StorageLocation location,
      final List<NamespaceInfo> nsInfos); @Override void removeVolumes(
      final Collection<StorageLocation> storageLocsToRemove,
      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 void getMetrics(MetricsCollector collector, boolean all); @Override // FSDatasetMBean long getNumBlocksCached(); @Override // FsDatasetSpi long getLength(ExtendedBlock b); @Override // FsDatasetSpi InputStream getBlockInputStream(ExtendedBlock b,
      long seekOffset); @Override // FsDatasetSpi ReplicaInputStreams getTmpInputStreams(ExtendedBlock b,
      long blkOffset, long metaOffset); @Override ReplicaInfo moveBlockAcrossStorage(ExtendedBlock block,
      StorageType targetStorageType, String targetStorageId); @Override ReplicaInfo moveBlockAcrossVolumes(ExtendedBlock block, FsVolumeSpi
      destination); @Override // FsDatasetSpi ReplicaHandler append(ExtendedBlock b,
      long newGS, long expectedBlockLen); @Override // FsDatasetSpi ReplicaHandler recoverAppend(
      ExtendedBlock b, long newGS, long expectedBlockLen); @Override // FsDatasetSpi Replica recoverClose(ExtendedBlock b, long newGS,
      long expectedBlockLen); @Override // FsDatasetSpi ReplicaHandler createRbw(
      StorageType storageType, String storageId, ExtendedBlock b,
      boolean allowLazyPersist); @Override // FsDatasetSpi ReplicaHandler recoverRbw(
      ExtendedBlock b, long newGS, long minBytesRcvd, long maxBytesRcvd); @Override // FsDatasetSpi ReplicaInPipeline convertTemporaryToRbw(
      final ExtendedBlock b); @Override // FsDatasetSpi ReplicaHandler createTemporary(StorageType storageType,
      String storageId, ExtendedBlock b, boolean isTransfer); @Override // FsDatasetSpi void adjustCrcChannelPosition(ExtendedBlock b, ReplicaOutputStreams streams, 
      int checksumSize); @Override // FsDatasetSpi void finalizeBlock(ExtendedBlock b, boolean fsyncDir); @Override // FsDatasetSpi void unfinalizeBlock(ExtendedBlock b); @Override // FsDatasetSpi List<Long> getCacheReport(String bpid); @Override Map<DatanodeStorage, BlockListAsLongs> getBlockReports(String bpid); @Override List<ReplicaInfo> getSortedFinalizedBlocks(String bpid); @Override // FsDatasetSpi 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 boolean contains(final ExtendedBlock block); @Override // FsDatasetSpi void handleVolumeFailures(Set<FsVolumeSpi> failedVolumes); @Override // FsDatasetSpi String toString(); @Override // FsDatasetSpi void shutdown(); @Override // FSDatasetMBean String getStorageInfo(); @Override void checkAndUpdate(String bpid, ScanInfo scanInfo); @Override // FsDatasetSpi @Deprecated ReplicaInfo getReplica(String bpid, long blockId); @Override String getReplicaString(String bpid, long blockId); @Override // FsDatasetSpi ReplicaRecoveryInfo initReplicaRecovery(RecoveringBlock rBlock); @Override // FsDatasetSpi Replica updateReplicaUnderRecovery(
                                    final ExtendedBlock oldBlock,
                                    final long recoveryId,
                                    final long newBlockId,
                                    final long newlength); @Override // FsDatasetSpi long getReplicaVisibleLength(final ExtendedBlock block); @Override void addBlockPool(String bpid, Configuration conf); @VisibleForTesting static void setBlockPoolId(String bpid); @Override void shutdownBlockPool(String bpid); @Override Map<String, Object> getVolumeInfoMap(); @Override //FsDatasetSpi void deleteBlockPool(String bpid, boolean force); @Override // FsDatasetSpi BlockLocalPathInfo getBlockLocalPathInfo(ExtendedBlock block); @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,
      ReplicaOutputStreams outs, 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); @VisibleForTesting void evictLazyPersistBlocks(long bytesNeeded); @VisibleForTesting int getNonPersistentReplicas(); @VisibleForTesting void setTimer(Timer newTimer); int getBPServiceCount(); int getVolumeCount();  }
@Test public void testAddVolumeWithSameStorageUuid() throws IOException { HdfsConfiguration config = new HdfsConfiguration(); MiniDFSCluster cluster = new MiniDFSCluster.Builder(config) .numDataNodes(1).build(); try { cluster.waitActive(); assertTrue(cluster.getDataNodes().get(0).isConnectedToNN( cluster.getNameNode().getServiceRpcAddress())); MiniDFSCluster.DataNodeProperties dn = cluster.stopDataNode(0); File vol0 = cluster.getStorageDir(0, 0); File vol1 = cluster.getStorageDir(0, 1); Storage.StorageDirectory sd0 = new Storage.StorageDirectory(vol0); Storage.StorageDirectory sd1 = new Storage.StorageDirectory(vol1); FileUtils.copyFile(sd0.getVersionFile(), sd1.getVersionFile()); cluster.restartDataNode(dn, true); cluster.waitActive(); assertFalse(cluster.getDataNodes().get(0).isConnectedToNN( cluster.getNameNode().getServiceRpcAddress())); } finally { cluster.shutdown(); } }