@Test public void testGetInvalidateBlocks() throws Exception { final int MAX_BLOCKS = 10; final int REMAINING_BLOCKS = 2; final int MAX_LIMIT = MAX_BLOCKS - REMAINING_BLOCKS; DatanodeDescriptor dd = DFSTestUtil.getLocalDatanodeDescriptor(); ArrayList<Block> blockList = new ArrayList<Block>(MAX_BLOCKS); for (int i=0; i<MAX_BLOCKS; i++) { blockList.add(new Block(i, 0, GenerationStamp.LAST_RESERVED_STAMP)); } dd.addBlocksToBeInvalidated(blockList); Block[] bc = dd.getInvalidateBlocks(MAX_LIMIT); assertEquals(bc.length, MAX_LIMIT); bc = dd.getInvalidateBlocks(MAX_LIMIT); assertEquals(bc.length, REMAINING_BLOCKS); }
public Block[] getInvalidateBlocks(int maxblocks) { synchronized (invalidateBlocks) { Block[] deleteList = invalidateBlocks.pollToArray(new Block[Math.min( invalidateBlocks.size(), maxblocks)]); return deleteList.length == 0 ? null : deleteList; } }
DatanodeDescriptor extends DatanodeInfo { public Block[] getInvalidateBlocks(int maxblocks) { synchronized (invalidateBlocks) { Block[] deleteList = invalidateBlocks.pollToArray(new Block[Math.min( invalidateBlocks.size(), maxblocks)]); return deleteList.length == 0 ? null : deleteList; } } }
DatanodeDescriptor extends DatanodeInfo { public Block[] getInvalidateBlocks(int maxblocks) { synchronized (invalidateBlocks) { Block[] deleteList = invalidateBlocks.pollToArray(new Block[Math.min( invalidateBlocks.size(), maxblocks)]); return deleteList.length == 0 ? null : deleteList; } } DatanodeDescriptor(DatanodeID nodeID);  DatanodeDescriptor(DatanodeID nodeID, 
                            String networkLocation); }
DatanodeDescriptor extends DatanodeInfo { public Block[] getInvalidateBlocks(int maxblocks) { synchronized (invalidateBlocks) { Block[] deleteList = invalidateBlocks.pollToArray(new Block[Math.min( invalidateBlocks.size(), maxblocks)]); return deleteList.length == 0 ? null : deleteList; } } DatanodeDescriptor(DatanodeID nodeID);  DatanodeDescriptor(DatanodeID nodeID, 
                            String networkLocation); int updateBlockReportContext(BlockReportContext context); void clearBlockReportContext(); CachedBlocksList getPendingCached(); CachedBlocksList getCached(); CachedBlocksList getPendingUncached(); @VisibleForTesting DatanodeStorageInfo getStorageInfo(String storageID); StorageReport[] getStorageReports(); void resetBlocks(); void clearBlockQueues(); int numBlocks(); void updateHeartbeat(StorageReport[] reports, long cacheCapacity,
      long cacheUsed, int xceiverCount, int volFailures,
      VolumeFailureSummary volumeFailureSummary); void updateHeartbeatState(StorageReport[] reports, long cacheCapacity,
      long cacheUsed, int xceiverCount, int volFailures,
      VolumeFailureSummary volumeFailureSummary); List<BlockTargetPair> getReplicationCommand(int maxTransfers); BlockInfoContiguousUnderConstruction[] getLeaseRecoveryCommand(int maxTransfers); Block[] getInvalidateBlocks(int maxblocks); long getRemaining(StorageType t, long minSize); int getBlocksScheduled(StorageType t); int getBlocksScheduled(); @Override int hashCode(); @Override boolean equals(Object obj); void setDisallowed(boolean flag); boolean isDisallowed(); int getVolumeFailures(); VolumeFailureSummary getVolumeFailureSummary(); @Override void updateRegInfo(DatanodeID nodeReg); long getBalancerBandwidth(); void setBalancerBandwidth(long bandwidth); @Override String dumpDatanode(); long getLastCachingDirectiveSentTimeMs(); void setLastCachingDirectiveSentTimeMs(long time); boolean checkBlockReportReceived(); }
DatanodeDescriptor extends DatanodeInfo { public Block[] getInvalidateBlocks(int maxblocks) { synchronized (invalidateBlocks) { Block[] deleteList = invalidateBlocks.pollToArray(new Block[Math.min( invalidateBlocks.size(), maxblocks)]); return deleteList.length == 0 ? null : deleteList; } } DatanodeDescriptor(DatanodeID nodeID);  DatanodeDescriptor(DatanodeID nodeID, 
                            String networkLocation); int updateBlockReportContext(BlockReportContext context); void clearBlockReportContext(); CachedBlocksList getPendingCached(); CachedBlocksList getCached(); CachedBlocksList getPendingUncached(); @VisibleForTesting DatanodeStorageInfo getStorageInfo(String storageID); StorageReport[] getStorageReports(); void resetBlocks(); void clearBlockQueues(); int numBlocks(); void updateHeartbeat(StorageReport[] reports, long cacheCapacity,
      long cacheUsed, int xceiverCount, int volFailures,
      VolumeFailureSummary volumeFailureSummary); void updateHeartbeatState(StorageReport[] reports, long cacheCapacity,
      long cacheUsed, int xceiverCount, int volFailures,
      VolumeFailureSummary volumeFailureSummary); List<BlockTargetPair> getReplicationCommand(int maxTransfers); BlockInfoContiguousUnderConstruction[] getLeaseRecoveryCommand(int maxTransfers); Block[] getInvalidateBlocks(int maxblocks); long getRemaining(StorageType t, long minSize); int getBlocksScheduled(StorageType t); int getBlocksScheduled(); @Override int hashCode(); @Override boolean equals(Object obj); void setDisallowed(boolean flag); boolean isDisallowed(); int getVolumeFailures(); VolumeFailureSummary getVolumeFailureSummary(); @Override void updateRegInfo(DatanodeID nodeReg); long getBalancerBandwidth(); void setBalancerBandwidth(long bandwidth); @Override String dumpDatanode(); long getLastCachingDirectiveSentTimeMs(); void setLastCachingDirectiveSentTimeMs(long time); boolean checkBlockReportReceived(); static final Log LOG; static final DatanodeDescriptor[] EMPTY_ARRAY; final DecommissioningStatus decommissioningStatus; public boolean isAlive; public boolean needKeyUpdate; }