private ScanStatus resplitPartiallyCompleteTasks ( ScanStatus status ) { boolean anyUpdated = false ; int nextTaskId = - 1 ; for ( ScanRangeStatus complete : status . getCompleteScanRanges ( ) ) { if ( complete . getResplitRange ( ) . isPresent ( ) ) { // This task only partially completed; there are still more data to scan. if ( nextTaskId == - 1 ) { nextTaskId = getNextTaskId ( status ) ; } ScanRange resplitRange = complete . getResplitRange ( ) . get ( ) ; // Resplit the un-scanned portion into new ranges List < ScanRange > subRanges = resplit ( complete . getPlacement ( ) , resplitRange , status . getOptions ( ) . getRangeScanSplitSize ( ) ) ; // Create new tasks for each subrange that are immediately available for being queued. List < ScanRangeStatus > resplitStatuses = Lists . newArrayListWithCapacity ( subRanges . size ( ) ) ; for ( ScanRange subRange : subRanges ) { resplitStatuses . add ( new ScanRangeStatus ( nextTaskId ++ , complete . getPlacement ( ) , subRange , complete . getBatchId ( ) , complete . getBlockedByBatchId ( ) , complete . getConcurrencyId ( ) ) ) ; } _scanStatusDAO . resplitScanRangeTask ( status . getScanId ( ) , complete . getTaskId ( ) , resplitStatuses ) ; anyUpdated = true ; } } if ( ! anyUpdated ) { return status ; } // Slightly inefficient to reload but less risky than trying to keep the DAO and in-memory object in sync return _scanStatusDAO . getScanStatus ( status . getScanId ( ) ) ; }
Checks whether any completed tasks returned before scanning the entire range . If so then the unscanned ranges are resplit and new tasks are created from them .