MRJobParser implements Runnable { public MRJobParser(MRRunningJobConfig.EndpointConfig endpointConfig, MRRunningJobConfig.EagleServiceConfig eagleServiceConfig, AppInfo app, Map<String, JobExecutionAPIEntity> mrJobMap, MRRunningJobManager runningJobManager, List<String> configKeys, Config config) { this.app = app; if (mrJobMap == null) { this.mrJobEntityMap = new HashMap<>(); } else { this.mrJobEntityMap = mrJobMap; } this.mrJobConfigs = new HashMap<>(); this.mrJobEntityCreationHandler = new MRJobEntityCreationHandler(eagleServiceConfig); this.commonTags.put(MRJobTagName.SITE.toString(), endpointConfig.site); this.commonTags.put(MRJobTagName.USER.toString(), app.getUser()); this.commonTags.put(MRJobTagName.JOB_QUEUE.toString(), app.getQueue()); this.runningJobManager = runningJobManager; this.parserStatus = ParserStatus.FINISHED; this.finishedTaskIds = new HashSet<>(); this.configKeys = configKeys; this.config = config; } MRJobParser(MRRunningJobConfig.EndpointConfig endpointConfig,
                       MRRunningJobConfig.EagleServiceConfig eagleServiceConfig,
                       AppInfo app, Map<String, JobExecutionAPIEntity> mrJobMap,
                       MRRunningJobManager runningJobManager,
                       List<String> configKeys,
                       Config config); void setAppInfo(AppInfo app); ParserStatus status(); void setStatus(ParserStatus status); @Override void run();  }
@Test public void testMRJobParser() throws Exception { setupMock(); mockInputJobSteam("/mrjob_30784.json", JOB_URL); mockInputJobSteam("/jobcounts_30784.json", JOB_COUNT_URL); mockGetConnection("/mrconf_30784.xml"); Assert.assertTrue(curator.checkExists().forPath(ZKROOT) == null); Assert.assertTrue(curator.checkExists().forPath(ZK_JOB_PATH) == null); List<String> confKeyKeys = makeConfKeyKeys(mrRunningJobConfig); InputStream previousmrrunningapp = this.getClass().getResourceAsStream("/previousmrrunningapp.json"); AppsWrapper appsWrapper = OBJ_MAPPER.readValue(previousmrrunningapp, AppsWrapper.class); List<AppInfo> appInfos = appsWrapper.getApps().getApp(); AppInfo app1 = appInfos.get(0); Map<String, JobExecutionAPIEntity> mrJobs = null; MRRunningJobManager runningJobManager = new MRRunningJobManager(mrRunningJobConfig.getZkStateConfig()); RMResourceFetcher resourceFetcher = new RMResourceFetcher(mrRunningJobConfig.getEndpointConfig().rmUrls); MRJobParser mrJobParser = new MRJobParser(mrRunningJobConfig.getEndpointConfig(), mrRunningJobConfig.getEagleServiceConfig(), app1, mrJobs, runningJobManager, confKeyKeys, config); Map<String, JobExecutionAPIEntity> jobIdToJobExecutionAPIEntity = getMrJobs(mrJobParser); Map<String, JobConfig> jobIdToJobConfig = getMrJobConfigs(mrJobParser); MRJobEntityCreationHandler mrJobEntityCreationHandler = getMrJobEntityCreationHandler(mrJobParser); List<TaggedLogAPIEntity> entities = getMrJobEntityCreationHandlerEntities(mrJobEntityCreationHandler); Assert.assertTrue(jobIdToJobExecutionAPIEntity.isEmpty()); Assert.assertTrue(jobIdToJobConfig.isEmpty()); Assert.assertTrue(entities.isEmpty()); Assert.assertTrue(mrJobParser.status() == MRJobParser.ParserStatus.FINISHED); mrJobParser.run(); Assert.assertTrue(jobIdToJobExecutionAPIEntity.size() == 1); JobExecutionAPIEntity jobExecutionAPIEntity = jobIdToJobExecutionAPIEntity.get(JOB_ID); Assert.assertEquals("AppInfo{id='application_1479206441898_30784', user='xxx', name='oozie:launcher:T=shell:W=wf_co_xxx_xxx_v3:A=extract_org_data:ID=0002383-161115184801730-oozie-oozi-W', queue='xxx', state='RUNNING', finalStatus='UNDEFINED', progress=95.0, trackingUI='ApplicationMaster', trackingUrl='http: Assert.assertEquals("RUNNING", jobExecutionAPIEntity.getCurrentState()); Assert.assertEquals("RUNNING", jobExecutionAPIEntity.getInternalState()); Assert.assertEquals("prefix:null, timestamp:1479328221694, humanReadableDate:2016-11-16 20:30:21,694, tags: jobName=oozie:launcher:T=shell:W=wf_co_xxx_xxx_v3:A=extract_org_data:ID=0002383-161115184801730-oozie-oozi-W,jobId=job_1479206441898_30784,site=sandbox,jobDefId=eagletest,jobType=HIVE,user=xxx,queue=xxx,, encodedRowkey:null", jobExecutionAPIEntity.toString()); Assert.assertTrue(mrJobParser.status() == MRJobParser.ParserStatus.FINISHED); Assert.assertEquals("{eagle.job.name=eagletest, hive.optimize.skewjoin.compiletime=false, hive.query.string=insert overwrite table xxxx}", jobExecutionAPIEntity.getJobConfig().toString()); Assert.assertTrue(jobIdToJobConfig.size() == 1); Assert.assertEquals("{eagle.job.name=eagletest, hive.optimize.skewjoin.compiletime=false, hive.query.string=insert overwrite table xxxx}", jobIdToJobConfig.get(JOB_ID).toString()); Assert.assertTrue(curator.checkExists().forPath(ZK_JOB_PATH) != null); Assert.assertTrue(curator.checkExists().forPath(ZK_APP_PATH) != null); Assert.assertTrue(curator.checkExists().forPath(ZKROOT) != null); Assert.assertEquals(DATA_FROM_ZK, new String(curator.getData().forPath(ZK_JOB_PATH), "UTF-8")); Assert.assertTrue(entities.isEmpty()); verify(client, times(1)).create(any()); }