@Test public void testAppIdParsing() throws Exception { String username = "testuser"; Configuration c = new Configuration(); c.set("mapred.job.name", jobName1); c.set("user.name", username); c.set("cascading.app.id", "93543E9B29A4CD14E0556849A93E171B"); c.set("cascading.flow.id", "0A2CEB3C5DFB905802EB96D1AE0C04D5"); ScaldingJobDescFactory factory = new ScaldingJobDescFactory(); QualifiedJobId jobId = new QualifiedJobId("test@local", "job_201206010000_0001"); long now = System.currentTimeMillis(); JobDesc desc = factory.create(jobId, now, c); assertJobDesc(desc, jobId, username, "93543E9B29A4CD14E0556849A93E171B"); c.set("mapred.job.name", jobName2); desc = factory.create(jobId, now, c); assertJobDesc(desc, jobId, username, expectedAppId2); c.set("mapred.job.name", jobName3); desc = factory.create(jobId, now, c); assertJobDesc(desc, jobId, username, expectedAppId3); c.set("mapred.job.name", jobName4); desc = factory.create(jobId, now, c); assertJobDesc(desc, jobId, username, expectedAppId4); c.set("mapred.job.name", jobName5); desc = factory.create(jobId, now, c); assertJobDesc(desc, jobId, username, expectedAppId5); }
@Override JobDesc create(QualifiedJobId qualifiedJobId, long submitTimeMillis, Configuration jobConf) { String appId = getAppId(jobConf); if (Constants.UNKNOWN.equals(appId)) { appId = cleanAppId(jobConf.get(Constants.CASCADING_APP_ID_CONF_KEY)); } String version = jobConf.get(Constants.CASCADING_VERSION_CONF_KEY); long scaldingSubmitTimeMillis = getFlowSubmitTimeMillis(jobConf, submitTimeMillis); return create(qualifiedJobId, jobConf, appId, version, Framework.SCALDING, scaldingSubmitTimeMillis); }
ScaldingJobDescFactory extends JobDescFactoryBase { @Override JobDesc create(QualifiedJobId qualifiedJobId, long submitTimeMillis, Configuration jobConf) { String appId = getAppId(jobConf); if (Constants.UNKNOWN.equals(appId)) { appId = cleanAppId(jobConf.get(Constants.CASCADING_APP_ID_CONF_KEY)); } String version = jobConf.get(Constants.CASCADING_VERSION_CONF_KEY); long scaldingSubmitTimeMillis = getFlowSubmitTimeMillis(jobConf, submitTimeMillis); return create(qualifiedJobId, jobConf, appId, version, Framework.SCALDING, scaldingSubmitTimeMillis); } }
ScaldingJobDescFactory extends JobDescFactoryBase { @Override JobDesc create(QualifiedJobId qualifiedJobId, long submitTimeMillis, Configuration jobConf) { String appId = getAppId(jobConf); if (Constants.UNKNOWN.equals(appId)) { appId = cleanAppId(jobConf.get(Constants.CASCADING_APP_ID_CONF_KEY)); } String version = jobConf.get(Constants.CASCADING_VERSION_CONF_KEY); long scaldingSubmitTimeMillis = getFlowSubmitTimeMillis(jobConf, submitTimeMillis); return create(qualifiedJobId, jobConf, appId, version, Framework.SCALDING, scaldingSubmitTimeMillis); }  }
ScaldingJobDescFactory extends JobDescFactoryBase { @Override JobDesc create(QualifiedJobId qualifiedJobId, long submitTimeMillis, Configuration jobConf) { String appId = getAppId(jobConf); if (Constants.UNKNOWN.equals(appId)) { appId = cleanAppId(jobConf.get(Constants.CASCADING_APP_ID_CONF_KEY)); } String version = jobConf.get(Constants.CASCADING_VERSION_CONF_KEY); long scaldingSubmitTimeMillis = getFlowSubmitTimeMillis(jobConf, submitTimeMillis); return create(qualifiedJobId, jobConf, appId, version, Framework.SCALDING, scaldingSubmitTimeMillis); }   }
ScaldingJobDescFactory extends JobDescFactoryBase { @Override JobDesc create(QualifiedJobId qualifiedJobId, long submitTimeMillis, Configuration jobConf) { String appId = getAppId(jobConf); if (Constants.UNKNOWN.equals(appId)) { appId = cleanAppId(jobConf.get(Constants.CASCADING_APP_ID_CONF_KEY)); } String version = jobConf.get(Constants.CASCADING_VERSION_CONF_KEY); long scaldingSubmitTimeMillis = getFlowSubmitTimeMillis(jobConf, submitTimeMillis); return create(qualifiedJobId, jobConf, appId, version, Framework.SCALDING, scaldingSubmitTimeMillis); }    }