Configures all the chain elements for the task.

@param jobConf chain job's JobConf.
424845
public void configure(JobConf jobConf) {
    String prefix = getPrefix(isMap);
    chainJobConf = jobConf;
    SerializationFactory serializationFactory =
      new SerializationFactory(chainJobConf);
    int index = jobConf.getInt(prefix + CHAIN_MAPPER_SIZE, 0);
    for (int i = 0; i < index; i++) {
      Class<? extends Mapper> klass =
        jobConf.getClass(prefix + CHAIN_MAPPER_CLASS + i, null, Mapper.class);
      JobConf mConf =
        getChainElementConf(jobConf, prefix + CHAIN_MAPPER_CONFIG + i);
      Mapper mapper = ReflectionUtils.newInstance(klass, mConf);
      mappers.add(mapper);

      if (mConf.getBoolean(MAPPER_BY_VALUE, true)) {
        mappersKeySerialization.add(serializationFactory.getSerialization(
          mConf.getClass(MAPPER_OUTPUT_KEY_CLASS, null)));
        mappersValueSerialization.add(serializationFactory.getSerialization(
          mConf.getClass(MAPPER_OUTPUT_VALUE_CLASS, null)));
      } else {
        mappersKeySerialization.add(null);
        mappersValueSerialization.add(null);
      }
    }
    Class<? extends Reducer> klass =
      jobConf.getClass(prefix + CHAIN_REDUCER_CLASS, null, Reducer.class);
    if (klass != null) {
      JobConf rConf =
        getChainElementConf(jobConf, prefix + CHAIN_REDUCER_CONFIG);
      reducer = ReflectionUtils.newInstance(klass, rConf);
      if (rConf.getBoolean(REDUCER_BY_VALUE, true)) {
        reducerKeySerialization = serializationFactory
          .getSerialization(rConf.getClass(REDUCER_OUTPUT_KEY_CLASS, null));
        reducerValueSerialization = serializationFactory
          .getSerialization(rConf.getClass(REDUCER_OUTPUT_VALUE_CLASS, null));
      } else {
        reducerKeySerialization = null;
        reducerValueSerialization = null;
      }
    }
  }
1123480
@Override
    public void setJobQueue(int jobId, Queue queue)
    {
        JqmClientFactory.getClient().setJobQueue(jobId, queue);
    }
163624
public void setJobs(java.util.Collection<Job> jobs) {
        if (jobs == null) {
            this.jobs = null;
            return;
        }

        this.jobs = new java.util.ArrayList<Job>(jobs);
    }
850467
public void setJobSummaries(java.util.Collection<JobSummary> jobSummaries) {
        if (jobSummaries == null) {
            this.jobSummaries = null;
            return;
        }

        this.jobSummaries = new java.util.ArrayList<JobSummary>(jobSummaries);
    }
1308915
private void configureCustomFields()
   {
      CustomFieldContainer customFields = m_projectFile.getCustomFields();

      // If the caller hasn't already supplied a value for this field
      if (m_activityIDField == null)
      {
         m_activityIDField = (TaskField) customFields.getFieldByAlias(FieldTypeClass.TASK, "Code");
         if (m_activityIDField == null)
         {
            m_activityIDField = TaskField.WBS;
         }
      }

      // If the caller hasn't already supplied a value for this field
      if (m_activityTypeField == null)
      {
         m_activityTypeField = (TaskField) customFields.getFieldByAlias(FieldTypeClass.TASK, "Activity Type");
      }
   }
1182935
private static void setJobConf(Object theObject, Configuration conf) {
        //If JobConf and JobConfigurable are in classpath, AND
        //theObject is of type JobConfigurable AND
        //conf is of type JobConf then
        //invoke configure on theObject
        try {
            Class<?> jobConfClass = conf.getClassByName("org.apache.hadoop.mapred.JobConf");
            Class<?> jobConfigurableClass = conf.getClassByName("org.apache.hadoop.mapred.JobConfigurable");
            if (jobConfClass.isAssignableFrom(conf.getClass())
                    && jobConfigurableClass.isAssignableFrom(theObject.getClass())) {
                Method configureMethod = jobConfigurableClass.getMethod("configure", jobConfClass);
                configureMethod.invoke(theObject, conf);
            }
        } catch (ClassNotFoundException e) {
            //JobConf/JobConfigurable not in classpath. no need to configure
        } catch (Exception e) {
            throw new RuntimeException("Error in configuring object", e);
        }
    }
362393
@Override
  protected void configure() {

    // Bind and expose LogWriter (a bit hacky, but needed by MapReduce for now)
    bind(LogWriter.class).to(LocalLogWriter.class);
    expose(LogWriter.class);

    // Bind ServiceAnnouncer for procedure.
    bind(ServiceAnnouncer.class).to(DiscoveryServiceAnnouncer.class);

    // For Binding queue stuff
    bind(QueueReaderFactory.class).in(Scopes.SINGLETON);

    // Bind ProgramRunner
    MapBinder<ProgramRunnerFactory.Type, ProgramRunner> runnerFactoryBinder =
      MapBinder.newMapBinder(binder(), ProgramRunnerFactory.Type.class, ProgramRunner.class);
    runnerFactoryBinder.addBinding(ProgramRunnerFactory.Type.FLOW).to(FlowProgramRunner.class);
    runnerFactoryBinder.addBinding(ProgramRunnerFactory.Type.FLOWLET).to(FlowletProgramRunner.class);

    bind(ProgramRunnerFactory.class).to(InMemoryFlowProgramRunnerFactory.class).in(Scopes.SINGLETON);
    // Note: Expose for test cases. Need to refactor test cases.
    expose(ProgramRunnerFactory.class);

    // For binding DataSet transaction stuff
    install(new DataFabricFacadeModule());
  }
974036
public static void init(String jobId) {
        JobContext parent = current_.get();
        JobContext ctx = new JobContext(parent);
        current_.set(ctx);
        // don't call setJobId(String)
        // as it will trigger listeners -- TODO fix me
        ctx.jobId = jobId;
        if (null == parent) {
            Act.eventBus().trigger(new JobContextInitialized(ctx));
        }
    }
65001
public void setJobDefinitions(java.util.Collection<String> jobDefinitions) {
        if (jobDefinitions == null) {
            this.jobDefinitions = null;
            return;
        }

        this.jobDefinitions = new java.util.ArrayList<String>(jobDefinitions);
    }
498851
@Override
  void initializeJob(JobID jobid) {
    for(String localDir : this.mapredLocalDirs) {
      File jobDirectory = new File(localDir, 
          TaskTracker.getLocalJobDir(jobid.toString()));
      if(!jobDirectory.exists()) {
        if(!jobDirectory.mkdir()) {
          LOG.warn("Unable to create job cache directory : " 
              + jobDirectory.getPath());
          continue;
        }
      }
      //Should be recursive because the jar and work folders might be 
      //present under the job cache directory
      changeDirectoryPermissions(
          jobDirectory.getPath(), FILE_PERMISSIONS, true);
    }
  }
509315
@Override
    public GenThread[] prepare(JobConf conf, Text key, Text value)
        throws IOException {
      this.rtc = new RunTimeConstants();
      super.prepare(conf, key, value, rtc);
      Path basePath = new Path(key.toString());
      LOG.info("base path is " + basePath);
      Path checksumPath = null;
      FileSystem fs = FileSystem.newInstance(conf);
      if (value.toString().length() != 0) {
        checksumPath = new Path(value.toString());
      }
      HashMap<String, Long> checksumMap = null;
      boolean verifyChecksum = false;
      if (fs.exists(checksumPath)) {
        LOG.info("checksum path is " + checksumPath);
        verifyChecksum = true;
        checksumMap = new HashMap<String, Long>();
        SequenceFile.Reader reader = null;
        try {
          reader = new SequenceFile.Reader(fs, checksumPath, conf);
          Writable dir = (Writable) ReflectionUtils.newInstance(
            reader.getKeyClass(), conf);
          Writable checksum = (Writable) ReflectionUtils.newInstance(
            reader.getValueClass(), conf);
          while(reader.next(dir, checksum)) {
            LOG.info("dir: " + dir.toString() + " checksum: " + checksum);
            checksumMap.put(
                fs.makeQualified(new Path(dir.toString())).toUri().getPath(),
                Long.parseLong(checksum.toString()));
          }
        } catch(Exception e) {
          LOG.error(e);
          throw new IOException(e);
        } finally {
          IOUtils.closeStream(reader);
        }
      }
      
      FileStatus[] baseDirs = fs.listStatus(basePath);
      if (rtc.nthreads != baseDirs.length) {
        throw new IOException("Number of directory under " + basePath + 
            "(" + baseDirs.length + ") doesn't match number of threads " + 
            "(" + rtc.nthreads + ").");
      }
      
      GenReaderThread[] threads = new GenReaderThread[(int)rtc.nthreads];

      for (int i=0; i < rtc.nthreads; i++) {
        long checksum = 0;
        if (verifyChecksum) {
          String basePathStr = baseDirs[i].getPath().toUri().getPath();
          checksum = checksumMap.get(basePathStr);
        }
        threads[i] = new GenReaderThread(conf, baseDirs[i].getPath(), 
            checksum, verifyChecksum, rtc);
      }
      return threads;
    }
978266
public void setWorkteams(java.util.Collection<Workteam> workteams) {
        if (workteams == null) {
            this.workteams = null;
            return;
        }

        this.workteams = new java.util.ArrayList<Workteam>(workteams);
    }
460404
public void run(Configuration conf, Path[] inputPaths, Path outputPath,
      int numMapTasks, Shard[] shards) throws IOException {
    JobConf jobConf =
        createJob(conf, inputPaths, outputPath, numMapTasks, shards);
    JobClient.runJob(jobConf);
  }
714602
@SuppressWarnings("deprecation")
  private static JobConf createJobConfForCopyByChunk(Configuration conf) {
    JobConf jobconf = new JobConf(conf, DistCp.class);
    jobconf.setJobName(NAME);

    // turn off speculative execution, because DFS doesn't handle
    // multiple writers to the same file.
    jobconf.setReduceSpeculativeExecution(false);

    jobconf.setMapOutputKeyClass(Text.class);
    jobconf.setMapOutputValueClass(IntWritable.class);
    jobconf.setOutputKeyClass(Text.class);
    jobconf.setOutputValueClass(Text.class);

    jobconf.setInputFormat(CopyByChunkInputFormat.class);
    jobconf.setMapperClass(CopyFilesByChunkMapper.class);
    jobconf.setReducerClass(CopyFilesByChunkReducer.class);
    
    // Prevent the reducer from starting until all maps are done.
    jobconf.setInt("mapred.job.rushreduce.reduce.threshold", 0);
    jobconf.setFloat("mapred.reduce.slowstart.completed.maps", 1.0f);
    
    return jobconf;
  }
603830
@Override
	public void configure(Configuration parameters) {

		// enforce sequential configure() calls
		synchronized (CONFIGURE_MUTEX) {
			// configure MR OutputFormat if necessary
			if (this.mapredOutputFormat instanceof Configurable) {
				((Configurable) this.mapredOutputFormat).setConf(this.jobConf);
			} else if (this.mapredOutputFormat instanceof JobConfigurable) {
				((JobConfigurable) this.mapredOutputFormat).configure(this.jobConf);
			}
		}
	}
1135293
@SuppressWarnings("deprecation")
  private void localizeTaskConfiguration(TaskTracker tracker, JobConf ttConf,
      String workDir, Task t, JobID jobID) throws IOException {
    Path jobFile = new Path(t.getJobFile());
    FileSystem systemFS = tracker.systemFS;
    this.localizedJobFile = new Path(workDir, jobID + ".xml");
    LOG.info("Localizing CJT configuration from " + jobFile + " to " +
        localizedJobFile);
    systemFS.copyToLocalFile(jobFile, localizedJobFile);
    JobConf localJobConf = new JobConf(localizedJobFile);
    boolean modified = Task.saveStaticResolutions(localJobConf);
    if (modified) {
      FSDataOutputStream out = new FSDataOutputStream(
        new FileOutputStream(localizedJobFile.toUri().getPath()));
      try {
        localJobConf.writeXml(out);
      } catch (IOException e) {
        out.close();
        throw e;
      }
    }
    // Add the values from the job conf to the configuration of this runner
    this.conf.addResource(localizedJobFile);
  }
520114
public void setJobTypes(final Map<String, ? extends Class<?>> jobTypes) {
        checkJobTypes(jobTypes);
        this.jobTypes.clear();
        this.jobTypes.putAll(jobTypes);
    }