private Collection<HadoopTaskInfo> reducerTasks(int[] reducers,HadoopJob job){
  UUID locNodeId=ctx.localNodeId();
  HadoopJobId jobId=job.id();
  JobLocalState state=activeJobs.get(jobId);
  Collection<HadoopTaskInfo> tasks=null;
  if (reducers != null) {
    if (state == null)     state=initState(job.id());
    for (    int rdc : reducers) {
      if (state.addReducer(rdc)) {
        if (log.isDebugEnabled())         log.debug("Submitting REDUCE task for execution [locNodeId=" + locNodeId + ", rdc="+ rdc+ ']');
        HadoopTaskInfo taskInfo=new HadoopTaskInfo(REDUCE,jobId,rdc,0,null);
        if (tasks == null)         tasks=new ArrayList<>();
        tasks.add(taskInfo);
      }
    }
  }
  return tasks;
}

private Collection < HadoopTaskInfo > reducerTasks ( int [ ] reducers , HadoopJob job ) { UUID locNodeId = ctx . localNodeId ( ) ; HadoopJobId jobId = job . id ( ) ; JobLocalState state = activeJobs . get ( jobId ) ; Collection < HadoopTaskInfo > tasks = null ; if ( reducers != null ) { if ( state == null ) state = initState ( job . id ( ) ) ; for ( int rdc : reducers ) { if ( state . addReducer ( rdc ) ) { if ( log . isDebugEnabled ( ) ) log . debug ( "Submitting REDUCE task for execution [locNodeId=" + locNodeId + ", rdc=" + rdc + ']' ) ; HadoopTaskInfo taskInfo = new HadoopTaskInfo ( REDUCE , jobId , rdc , 0 , null ) ; if ( tasks == null ) tasks = new ArrayList < > ( ) ; tasks . add ( taskInfo ) ; } } } return tasks ; }
creates reducer tasks based on job information .
