BlockManager { public void shutdown() { stopReplicationInitializer(); blocksMap.close(); } BlockManager(final Namesystem namesystem, final FSClusterStats stats,
      final Configuration conf); long getPendingReplicationBlocksCount(); long getUnderReplicatedBlocksCount(); long getCorruptReplicaBlocksCount(); long getScheduledReplicationBlocksCount(); long getPendingDeletionBlocksCount(); long getStartupDelayBlockDeletionInMs(); long getExcessBlocksCount(); long getPostponedMisreplicatedBlocksCount(); int getPendingDataNodeMessageCount(); BlockStoragePolicy getStoragePolicy(final String policyName); BlockStoragePolicy getStoragePolicy(final byte policyId); BlockStoragePolicy[] getStoragePolicies(); void setBlockPoolId(String blockPoolId); @VisibleForTesting BlockTokenSecretManager getBlockTokenSecretManager(); void activate(Configuration conf); void close(); DatanodeManager getDatanodeManager(); @VisibleForTesting BlockPlacementPolicy getBlockPlacementPolicy(); void setBlockPlacementPolicy(BlockPlacementPolicy newpolicy); void metaSave(PrintWriter out); int getMaxReplicationStreams(); boolean checkMinReplication(Block block); boolean commitOrCompleteLastBlock(BlockCollection bc,
      Block commitBlock); BlockInfo forceCompleteBlock(final BlockCollection bc,
      final BlockInfoUnderConstruction block); LocatedBlock convertLastBlockToUnderConstruction(
      BlockCollection bc); LocatedBlocks createLocatedBlocks(final BlockInfo[] blocks,
      final long fileSizeExcludeBlocksUnderConstruction,
      final boolean isFileUnderConstruction, final long offset,
      final long length, final boolean needBlockToken,
      final boolean inSnapshot, FileEncryptionInfo feInfo); ExportedBlockKeys getBlockKeys(); void setBlockToken(final LocatedBlock b,
      final BlockTokenSecretManager.AccessMode mode); DataEncryptionKey generateDataEncryptionKey(); short adjustReplication(short replication); void verifyReplication(String src,
                          short replication,
                          String clientName); boolean isSufficientlyReplicated(BlockInfo b); BlocksWithLocations getBlocks(DatanodeID datanode, long size
      ); void findAndMarkBlockAsCorrupt(final ExtendedBlock blk,
      final DatanodeInfo dn, String storageID, String reason); void setPostponeBlocksFromFuture(boolean postpone); int getUnderReplicatedNotMissingBlocks(); DatanodeStorageInfo[] chooseTarget4WebHDFS(String src,
      DatanodeDescriptor clientnode, Set<Node> excludes, long blocksize); DatanodeStorageInfo[] chooseTarget4AdditionalDatanode(String src,
      int numAdditionalNodes,
      Node clientnode,
      List<DatanodeStorageInfo> chosen,
      Set<Node> excludes,
      long blocksize,
      byte storagePolicyID); DatanodeStorageInfo[] chooseTarget4NewBlock(final String src,
      final int numOfReplicas, final Node client,
      final Set<Node> excludedNodes,
      final long blocksize,
      final List<String> favoredNodes,
      final byte storagePolicyID); boolean processReport(final DatanodeID nodeID,
      final DatanodeStorage storage,
      final BlockListAsLongs newReport, BlockReportContext context,
      boolean lastStorageInRpc); void markBlockReplicasAsCorrupt(BlockInfo block, 
      long oldGenerationStamp, long oldNumBytes, 
      DatanodeStorageInfo[] newStorages); void processQueuedMessagesForBlock(Block b); void processAllPendingDNMessages(); void processMisReplicatedBlocks(); double getReplicationQueuesInitProgress(); void setReplication(final short oldRepl, final short newRepl,
      final String src, final Block... blocks); void removeStoredBlock(Block block, DatanodeDescriptor node); void processIncrementalBlockReport(final DatanodeID nodeID,
      final StorageReceivedDeletedBlocks srdb); NumberReplicas countNodes(Block b); int getActiveBlockCount(); DatanodeStorageInfo[] getStorages(BlockInfo block); int getTotalBlocks(); void removeBlock(Block block); BlockInfo getStoredBlock(Block block); void checkReplication(BlockCollection bc); long getMissingBlocksCount(); BlockInfo addBlockCollection(BlockInfo block, BlockCollection bc); BlockCollection getBlockCollection(Block b); Iterable<DatanodeStorageInfo> getStorages(final Block block); int numCorruptReplicas(Block block); void removeBlockFromMap(Block block); int getCapacity(); long[] getCorruptReplicaBlockIds(int numExpectedBlocks,
                                   Long startingBlockId); Iterator<Block> getCorruptReplicaBlockIterator(); Collection<DatanodeDescriptor> getCorruptReplicas(Block block); int numOfUnderReplicatedBlocks(); void clearQueues(); void shutdown(); void clear(); static final Log blockLog; final Map<String, LightWeightLinkedSet<Block>> excessReplicateMap; final UnderReplicatedBlocks neededReplications; final short maxReplication; final short minReplication; final int defaultReplication; }
@Test public void testStorageWithRemainingCapacity() throws Exception { final Configuration conf = new HdfsConfiguration(); final MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf).build(); FileSystem fs = FileSystem.get(conf); Path file1 = null; try { cluster.waitActive(); final FSNamesystem namesystem = cluster.getNamesystem(); final String poolId = namesystem.getBlockPoolId(); final DatanodeRegistration nodeReg = DataNodeTestUtils.getDNRegistrationForBP(cluster.getDataNodes(). get(0), poolId); final DatanodeDescriptor dd = NameNodeAdapter.getDatanode(namesystem, nodeReg); for(DatanodeStorageInfo storage: dd.getStorageInfos()) { storage.setUtilizationForTesting(65536, 0, 65536, 0); } dd.setRemaining(131072); file1 = new Path("testRemainingStorage.dat"); try { DFSTestUtil.createFile(fs, file1, 102400, 102400, 102400, (short)1, 0x1BAD5EED); } catch (RemoteException re) { GenericTestUtils.assertExceptionContains("nodes instead of " + "minReplication", re); } } finally { assertTrue(fs.exists(file1)); fs.delete(file1, true); assertTrue(!fs.exists(file1)); cluster.shutdown(); } }