private List < ClaimedTask > claimMigrationRangeTasks ( int max ) { try { Date claimTime = new Date ( ) ; List < ScanRangeTask > migrationRangeTasks = _workflow . claimScanRangeTasks ( max , QUEUE_CLAIM_TTL ) ; if ( migrationRangeTasks . isEmpty ( ) ) { return ImmutableList . of ( ) ; } List < ClaimedTask > newlyClaimedTasks = Lists . newArrayListWithCapacity ( migrationRangeTasks . size ( ) ) ; for ( ScanRangeTask task : migrationRangeTasks ) { final ClaimedTask claimedTask = new ClaimedTask ( task , claimTime ) ; // Record that the task is claimed locally boolean alreadyClaimed = _claimedTasks . putIfAbsent ( task . getId ( ) , claimedTask ) != null ; if ( alreadyClaimed ) { _log . warn ( "Workflow returned migration range task that is already claimed: {}" , task ) ; // Do not acknowledge the task, let it expire naturally. Eventually it should come up again // after the previous claim has been released. } else { _log . info ( "Claimed migration range task: {}" , task ) ; newlyClaimedTasks . add ( claimedTask ) ; // Schedule a follow-up to ensure the scanning service assigns it a thread // in a reasonable amount of time. _backgroundService . schedule ( new Runnable ( ) { @ Override public void run ( ) { validateClaimedTaskHasStarted ( claimedTask ) ; } } , CLAIM_START_TIMEOUT . toMillis ( ) , TimeUnit . MILLISECONDS ) ; } } return newlyClaimedTasks ; } catch ( Exception e ) { _log . error ( "Failed to start next available migration range" , e ) ; return ImmutableList . of ( ) ; } }
Claims migration range tasks that have been queued by the leader and are ready to scan .