@Test public void testRunIdParsing() throws Exception { Configuration c = new Configuration(); c.set("mapred.job.name", jobName1); c.set("cascading.app.id", "93543E9B29A4CD14E0556849A93E171B"); c.set("cascading.flow.id", "0A2CEB3C5DFB905802EB96D1AE0C04D5"); c.set("user.name", "testuser"); ScaldingJobDescFactory factory = new ScaldingJobDescFactory(); QualifiedJobId jobId = new QualifiedJobId("test@local", "job_201206010000_0001"); long now = System.currentTimeMillis(); LOG.info("Using "+now+" as submit time"); long expectedMonthStart = DateUtil.getMonthStart(now); LOG.info("Month start set to "+expectedMonthStart); JobDesc desc = factory.create(jobId, now, c); LOG.info("Set runId to "+desc.getRunId()); assertTrue(expectedMonthStart <= desc.getRunId()); long monthEnd = expectedMonthStart + DateUtil.MONTH_IN_MILLIS; assertTrue("runId should be less than "+monthEnd, monthEnd > desc.getRunId()); Configuration c2 = new Configuration(c); c2.set(Constants.CASCADING_RUN_CONF_KEY, Long.toString(now)); desc = factory.create(jobId, now, c2); assertEquals(now, desc.getRunId()); }
@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); }    }