public static BlockWorkerInfo getWorkerWithMostBlocks ( List < BlockWorkerInfo > workers , List < FileBlockInfo > fileBlockInfos ) { // Index workers by their addresses. IndexedSet < BlockWorkerInfo > addressIndexedWorkers = new IndexedSet <> ( WORKER_ADDRESS_INDEX ) ; addressIndexedWorkers . addAll ( workers ) ; // Use ConcurrentMap for putIfAbsent. A regular Map works in Java 8. ConcurrentMap < BlockWorkerInfo , Integer > blocksPerWorker = Maps . newConcurrentMap ( ) ; int maxBlocks = 0 ; BlockWorkerInfo mostBlocksWorker = null ; for ( FileBlockInfo fileBlockInfo : fileBlockInfos ) { for ( BlockLocation location : fileBlockInfo . getBlockInfo ( ) . getLocations ( ) ) { BlockWorkerInfo worker = addressIndexedWorkers . getFirstByField ( WORKER_ADDRESS_INDEX , location . getWorkerAddress ( ) ) ; if ( worker == null ) { // We can only choose workers in the workers list. continue ; } blocksPerWorker . putIfAbsent ( worker , 0 ) ; int newBlockCount = blocksPerWorker . get ( worker ) + 1 ; blocksPerWorker . put ( worker , newBlockCount ) ; if ( newBlockCount > maxBlocks ) { maxBlocks = newBlockCount ; mostBlocksWorker = worker ; } } } return mostBlocksWorker ; }
Returns whichever specified worker stores the most blocks from the block info list .