@Override
  public JobDesc create(QualifiedJobId qualifiedJobId, long submitTimeMillis,
      Configuration jobConf) {
    String appId = getAppId(jobConf);
    String version = jobConf.get(Constants.PIG_VERSION_CONF_KEY,
        Constants.UNKNOWN);
    long pigSubmitTimeMillis = jobConf.getLong(Constants.PIG_RUN_CONF_KEY, 0);

    // This means that Constants.PIG_RUN_CONF_KEY was not present (for jobs
    // launched with an older pig version).
    if (pigSubmitTimeMillis == 0) {
      String pigLogfile = jobConf.get(Constants.PIG_LOG_FILE_CONF_KEY);
      if (pigLogfile == null) {
        // Should be rare, but we're seeing this happen occasionally
        // Give up on grouping the jobs within the run together, and treat these as individual runs.
        pigSubmitTimeMillis = submitTimeMillis;
      } else {
        pigSubmitTimeMillis = getScriptStartTimeFromLogfileName(pigLogfile);
      }
    }

    return create(qualifiedJobId, jobConf, appId, version, Framework.PIG,
        pigSubmitTimeMillis);
  }
@ Override public JobDesc create ( QualifiedJobId qualifiedJobId , long submitTimeMillis , Configuration jobConf ) { String appId = getAppId ( jobConf ) ; String version = jobConf . get ( Constants . PIG_VERSION_CONF_KEY , Constants . UNKNOWN ) ; long pigSubmitTimeMillis = jobConf . getLong ( Constants . PIG_RUN_CONF_KEY , 0 ) ; if ( pigSubmitTimeMillis == 0 ) { String pigLogfile = jobConf . get ( Constants . PIG_LOG_FILE_CONF_KEY ) ; if ( pigLogfile == null ) { pigSubmitTimeMillis = submitTimeMillis ; } else { pigSubmitTimeMillis = getScriptStartTimeFromLogfileName ( pigLogfile ) ; } } return create ( qualifiedJobId , jobConf , appId , version , Framework . PIG , pigSubmitTimeMillis ) ; }
/*
(non-Javadoc)

@see
com.twitter.hraven.JobKeyFactoryBase#create(com.twitter.corestorage
.rhaven.QualifiedJobId, long, org.apache.hadoop.conf.Configuration)