private void runPendingTasks()
  {
    runPendingTasksExec.submit(
        new Callable<Void>()
        {
          @Override
          public Void call()
          {
            try {
              // make a copy of the pending tasks because tryAssignTask may delete tasks from pending and move them
              // into running status
              List<RemoteTaskRunnerWorkItem> copy = Lists.newArrayList(pendingTasks.values());
              sortByInsertionTime(copy);

              for (RemoteTaskRunnerWorkItem taskRunnerWorkItem : copy) {
                String taskId = taskRunnerWorkItem.getTaskId();
                if (tryAssignTasks.putIfAbsent(taskId, taskId) == null) {
                  try {
                    //this can still be null due to race from explicit task shutdown request
                    //or if another thread steals and completes this task right after this thread makes copy
                    //of pending tasks. See https://github.com/apache/incubator-druid/issues/2842 .
                    Task task = pendingTaskPayloads.get(taskId);
                    if (task != null && tryAssignTask(task, taskRunnerWorkItem)) {
                      pendingTaskPayloads.remove(taskId);
                    }
                  }
                  catch (Exception e) {
                    log.makeAlert(e, "Exception while trying to assign task")
                       .addData("taskId", taskRunnerWorkItem.getTaskId())
                       .emit();
                    RemoteTaskRunnerWorkItem workItem = pendingTasks.remove(taskId);
                    if (workItem != null) {
                      taskComplete(workItem, null, TaskStatus.failure(taskId));
                    }
                  }
                  finally {
                    tryAssignTasks.remove(taskId);
                  }
                }
              }
            }
            catch (Exception e) {
              log.makeAlert(e, "Exception in running pending tasks").emit();
            }

            return null;
          }
        }
    );
  }
private void runPendingTasks ( ) { runPendingTasksExec . submit ( new Callable < Void > ( ) { @ Override public Void call ( ) { try { List < RemoteTaskRunnerWorkItem > copy = Lists . newArrayList ( pendingTasks . values ( ) ) ; sortByInsertionTime ( copy ) ; for ( RemoteTaskRunnerWorkItem taskRunnerWorkItem : copy ) { String taskId = taskRunnerWorkItem . getTaskId ( ) ; if ( tryAssignTasks . putIfAbsent ( taskId , taskId ) == null ) { try { Task task = pendingTaskPayloads . get ( taskId ) ; if ( task != null && tryAssignTask ( task , taskRunnerWorkItem ) ) { pendingTaskPayloads . remove ( taskId ) ; } } catch ( Exception e ) { log . makeAlert ( e , "Exception while trying to assign task" ) . addData ( "taskId" , taskRunnerWorkItem . getTaskId ( ) ) . emit ( ) ; RemoteTaskRunnerWorkItem workItem = pendingTasks . remove ( taskId ) ; if ( workItem != null ) { taskComplete ( workItem , null , TaskStatus . failure ( taskId ) ) ; } } finally { tryAssignTasks . remove ( taskId ) ; } } } } catch ( Exception e ) { log . makeAlert ( e , "Exception in running pending tasks" ) . emit ( ) ; } return null ; } } ) ; }
This method uses a multi-threaded executor to extract all pending tasks and attempt to run them. Any tasks that
are successfully assigned to a worker will be moved from pendingTasks to runningTasks. This method is thread-safe.
This method should be run each time there is new worker capacity or if new tasks are assigned.