BlockManager { @VisibleForTesting public void removeBlocks(List<Long> allBlockIds, final DatanodeDescriptor node) throws IOException { final Map<Long, List<Long>> inodeIdsToBlockMap = INodeUtil.getINodeIdsForBlockIds(allBlockIds, slicerBatchSize, slicerNbThreads, ((FSNamesystem) namesystem).getFSOperationsExecutor()); final List<Long> inodeIds = new ArrayList<>(inodeIdsToBlockMap.keySet()); try { Slicer.slice(inodeIds.size(), slicerBatchSize, slicerNbThreads, ((FSNamesystem) namesystem).getFSOperationsExecutor(), new Slicer.OperationHandler() { @Override public void handle(int startIndex, int endIndex) throws Exception { List<Long> ids = inodeIds.subList(startIndex, endIndex); removeStoredBlocksTx(ids, inodeIdsToBlockMap, node); } }); } catch (Exception ex) { throw new IOException(ex); } } BlockManager(final Namesystem namesystem, final Configuration conf); long getPendingReplicationBlocksCount(); long getUnderReplicatedBlocksCount(); long getCorruptReplicaBlocksCount(); long getScheduledReplicationBlocksCount(); long getPendingDeletionBlocksCount(); long getStartupDelayBlockDeletionInMs(); long getExcessBlocksCount(); long getPostponedMisreplicatedBlocksCount(); BlockStoragePolicy getDefaultStoragePolicy(); BlockStoragePolicy getStoragePolicy(final String policyName); BlockStoragePolicy getStoragePolicy(final byte policyId); BlockStoragePolicy[] getStoragePolicies(); void setBlockPoolId(String blockPoolId); BlockStoragePolicySuite getStoragePolicySuite(); @VisibleForTesting BlockTokenSecretManager getBlockTokenSecretManager(); void activate(Configuration conf); void close(); DatanodeManager getDatanodeManager(); @VisibleForTesting BlockPlacementPolicy getBlockPlacementPolicy(); void setBlockPlacementPolicy(BlockPlacementPolicy newpolicy); int getMaxReplicationStreams(); boolean checkMinReplication(Block block); boolean commitOrCompleteLastBlock(BlockCollection bc,
      Block commitBlock); BlockInfoContiguous forceCompleteBlock(final BlockCollection bc,
      final BlockInfoContiguousUnderConstruction block); LocatedBlock convertLastBlockToUnderConstruction(
      BlockCollection bc, long bytesToRemove); LocatedBlocks createPhantomLocatedBlocks(INodeFile file, byte[] data,
      final boolean isFileUnderConstruction,
      final boolean needBlockToken, FileEncryptionInfo feInfo); LocatedBlocks createLocatedBlocks(final BlockInfoContiguous[] blocks,
      final long fileSizeExcludeBlocksUnderConstruction,
      final boolean isFileUnderConstruction, final long offset,
      final long length, final boolean needBlockToken, FileEncryptionInfo feInfo); ExportedBlockKeys getBlockKeys(); void setBlockToken(final LocatedBlock b,
      final AccessMode mode); DataEncryptionKey generateDataEncryptionKey(); short adjustReplication(short replication); void verifyReplication(String src, short replication,
      String clientName); boolean isSufficientlyReplicated(BlockInfoContiguous b); BlocksWithLocations getBlocks(DatanodeID datanode, long size
      // used in the namenode protocol
  ); void findAndMarkBlockAsCorrupt(final ExtendedBlock blk,
      final DatanodeInfo dn, final String storageID, final String reason); int getUnderReplicatedNotMissingBlocks(); 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); 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[] chooseTarget4ParityRepair(String src,  int numOfReplicas,
      Node clientnode,List<DatanodeStorageInfo> chosen, Set<Node> excludes, long blocksize,
      byte storagePolicyID); List<Integer> checkHashes(final DatanodeID nodeID,
                                   final DatanodeStorage storage,
                                   final BlockReport newReport); boolean processReport(final DatanodeID nodeID,
      final DatanodeStorage storage,
      final BlockReport newReport, BlockReportContext context,
      boolean lastStorageInRpc); void markBlockReplicasAsCorrupt(BlockInfoContiguous block, 
      long oldGenerationStamp, long oldNumBytes, 
      DatanodeStorageInfo[] newStorages); @VisibleForTesting ReportStatistics processReport(final DatanodeStorageInfo storage,final BlockReport report); @VisibleForTesting void removeBlocks(List<Long> allBlockIds, final DatanodeDescriptor node); void removeBlocks(Map<Long,Long> allBlocksAndInodesIds, final DatanodeDescriptor node); synchronized 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 removeStoredBlock(Block block, int sid); void processIncrementalBlockReport(DatanodeRegistration nodeID,
      final StorageReceivedDeletedBlocks blockInfos); NumberReplicas countNodes(Block b); int getActiveBlockCount(); DatanodeStorageInfo[] getStorages(BlockInfoContiguous block); int getTotalBlocks(); void removeBlock(Block block); BlockInfoContiguous getStoredBlock(Block block); void checkReplication(BlockCollection bc); boolean checkBlocksProperlyReplicated(
      String src, BlockInfoContiguous[] blocks); long getMissingBlocksCount(); long getMissingReplOneBlocksCount(); BlockInfoContiguous addBlockCollection(BlockInfoContiguous block, 
      BlockCollection bc); BlockCollection getBlockCollection(Block b); Iterable<DatanodeStorageInfo> getStorages(final Block block); List<DatanodeStorageInfo> storageList(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); String getCorruptReason(BlockInfoContiguous block, DatanodeDescriptor node); int numOfUnderReplicatedBlocks(); void clearQueues(); static LocatedBlock newLocatedBlock(
      ExtendedBlock b, DatanodeStorageInfo[] storages,
      long startOffset, boolean corrupt); BlockInfoContiguous tryToCompleteBlock(final BlockCollection bc,
      final int blkIndex); @VisibleForTesting void processTimedOutPendingBlock(final long timedOutItemId); int getTotalCompleteBlocks(); void shutdown(); int getNumBuckets(); int getBlockFetcherNBThreads(); int getBlockFetcherBucketsPerThread(); int getRemovalBatchSize(); int getRemovalNoThreads(); void blockReportCompleted(final DatanodeID nodeID, DatanodeStorage[] storages, boolean success); static final Logger LOG; static final Logger blockLog; final ExcessReplicasMap excessReplicateMap; final UnderReplicatedBlocks neededReplications; final short maxReplication; final short minReplication; final int defaultReplication; }
@Test public void testRemoveBlocks() throws IOException, InterruptedException, ExecutionException { List<DatanodeDescriptor> testNodes = new ArrayList<>(); testNodes.add(nodes.get(0)); testNodes.add(nodes.get(1)); ExecutorService executor = Executors.newFixedThreadPool(10); List<Future<Object>> futures = new ArrayList<>(); long blockId = 0; List<Long> blockIds = new ArrayList<>(); for (int i = 2; i < 4000 + 2; i++) { for (int j = 0; j < 2; j++) { blockIds.add(blockId); futures.add(executor.submit(new SliceRunner(blockId++, testNodes, i))); } } for (Future<Object> futur : futures) { futur.get(); } executor.shutdown(); executor.awaitTermination(10, TimeUnit.SECONDS); for (int sid : testNodes.get(0).getSidsOnNode()) { checkNbReplicas(sid, blockIds.size()); } for (int sid : testNodes.get(1).getSidsOnNode()) { checkNbReplicas(sid, blockIds.size()); } bm.removeBlocks(blockIds, testNodes.get(0)); for (int sid : testNodes.get(0).getSidsOnNode()) { checkNbReplicas(sid, 0); } for (int sid : testNodes.get(1).getSidsOnNode()) { checkNbReplicas(sid, blockIds.size()); } }