void updateLeasedFiles ( SnapshotStorage ssStore ) throws IOException { FSNamesystem fsNamesys = ssStore . getFSNamesystem ( ) ; List < Block > blocksForNN = new ArrayList < Block > ( ) ; leaseUpdateThreadPool = new ThreadPoolExecutor ( 1 , maxLeaseUpdateThreads , 60 , TimeUnit . SECONDS , new LinkedBlockingQueue < Runnable > ( ) ) ; ( ( ThreadPoolExecutor ) leaseUpdateThreadPool ) . allowCoreThreadTimeOut ( true ) ; // Try to update lengths for leases from DN LightWeightLinkedSet < Lease > sortedLeases = fsNamesys . leaseManager . getSortedLeases ( ) ; Iterator < Lease > itr = sortedLeases . iterator ( ) ; while ( itr . hasNext ( ) ) { Lease lease = itr . next ( ) ; for ( String path : lease . getPaths ( ) ) { // Update file lengths using worker threads to increase throughput leaseUpdateThreadPool . execute ( new LeaseUpdateWorker ( conf , path , fsNamesys , blocksForNN ) ) ; } } try { leaseUpdateThreadPool . shutdown ( ) ; // Wait till update tasks finish successfully (max 20 mins?) if ( ! leaseUpdateThreadPool . awaitTermination ( 1200 , TimeUnit . SECONDS ) ) { throw new IOException ( "Updating lease files failed" ) ; } } catch ( InterruptedException e ) { throw new IOException ( "Snapshot creation interrupted while updating leased files" ) ; } // Fetch block lengths for renamed/deleted leases from NN long [ ] blockIds = new long [ blocksForNN . size ( ) ] ; for ( int i = 0 ; i < blocksForNN . size ( ) ; ++ i ) { blockIds [ i ] = blocksForNN . get ( i ) . getBlockId ( ) ; } long [ ] lengths = namenode . getBlockLengths ( blockIds ) ; for ( int i = 0 ; i < blocksForNN . size ( ) ; ++ i ) { if ( lengths [ i ] == - 1 ) { // Couldn't update block length, keep preferred length LOG . error ( "Couldn't update length for block " + blocksForNN . get ( i ) ) ; } else { blocksForNN . get ( i ) . setNumBytes ( lengths [ i ] ) ; } } }
Tries to get the most up to date lengths of files under construction .