RMResourceFetcher implements ResourceFetcher<AppInfo> { private List<AppInfo> getResource(Constants.ResourceType resourceType, Constants.CompressionType compressionType, Object... parameter) throws Exception { selector.checkUrl(); switch (resourceType) { case COMPLETE_SPARK_JOB: return doFetchApplicationsList(getFinishedJobURL(Constants.JobType.SPARK, parameter), compressionType); case RUNNING_SPARK_JOB: return doFetchRunningApplicationsList(Constants.JobType.SPARK, compressionType, parameter); case RUNNING_MR_JOB: return doFetchRunningApplicationsList(Constants.JobType.MAPREDUCE, compressionType, parameter); case COMPLETE_MR_JOB: return doFetchApplicationsList(getFinishedJobURL(Constants.JobType.MAPREDUCE, parameter), compressionType); case ACCEPTED_JOB: return doFetchAcceptedApplicationList(compressionType, parameter); default: throw new Exception("Not support resourceType :" + resourceType); } } RMResourceFetcher(String[] rmBasePaths); HAURLSelector getSelector(); String getRunningJobURL(Constants.JobType jobType, String startTime, String endTime, String limit); List<AppInfo> getResource(Constants.ResourceType resourceType, Object... parameter); ClusterInfo getClusterInfo();  }
@Test public void testCompleteMrJob() throws Exception { String[] rmBasePaths = new String[]{"http: RMResourceFetcher rmResourceFetcher = new RMResourceFetcher(rmBasePaths); InputStream jsonstream = this.getClass().getResourceAsStream("/mrcompleteapp.json"); mockStatic(InputStreamUtils.class); when(InputStreamUtils.getInputStream("http: when(InputStreamUtils.getInputStream("http: String lastFinishedTime = "1479244718794"; List<AppInfo> appInfos = rmResourceFetcher.getResource(Constants.ResourceType.COMPLETE_MR_JOB, lastFinishedTime); Assert.assertEquals(2, appInfos.size()); Assert.assertEquals("AppInfo{id='application_1326815542473_0001', user='user1', name='insert overwrite table xx.yy...end(Stage-1)', queue='default', state='FINISHED', finalStatus='SUCCEEDED', progress=100.0, trackingUI='History', trackingUrl='http: Assert.assertEquals("AppInfo{id='application_1326815542473_0002', user='user1', name='Sleep job', queue='default', state='FINISHED', finalStatus='SUCCEEDED', progress=100.0, trackingUI='History', trackingUrl='http: }
@Test public void testRunningMrJob() throws Exception { String[] rmBasePaths = new String[]{"http: RMResourceFetcher rmResourceFetcher = new RMResourceFetcher(rmBasePaths); InputStream jsonstream = this.getClass().getResourceAsStream("/mrrunningapp.json"); mockStatic(InputStreamUtils.class); when(InputStreamUtils.getInputStream("http: when(InputStreamUtils.getInputStream("http: List<AppInfo> appInfos = rmResourceFetcher.getResource(Constants.ResourceType.RUNNING_MR_JOB); Assert.assertEquals(2, appInfos.size()); 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("AppInfo{id='application_1479206441898_35341', user='yyy', name='insert overwrite table inter...a.xxx(Stage-3)', queue='yyy', state='RUNNING', finalStatus='UNDEFINED', progress=59.545456, trackingUI='ApplicationMaster', trackingUrl='http: }
@Test public void testSparkRunningJob() throws Exception { String[] rmBasePaths = new String[]{"http: RMResourceFetcher rmResourceFetcher = new RMResourceFetcher(rmBasePaths); InputStream jsonstream = this.getClass().getResourceAsStream("/sparkrunningapp.json"); mockStatic(InputStreamUtils.class); when(InputStreamUtils.getInputStream("http: when(InputStreamUtils.getInputStream("http: List<AppInfo> appInfos = rmResourceFetcher.getResource(Constants.ResourceType.RUNNING_SPARK_JOB); Assert.assertEquals(2, appInfos.size()); Assert.assertEquals("AppInfo{id='application_1479196259121_1041', user='xxx', name='Spark shell', queue='default', state='RUNNING', finalStatus='UNDEFINED', progress=10.0, trackingUI='ApplicationMaster', trackingUrl='http: Assert.assertEquals("AppInfo{id='application_1478655515028_74912', user='xxx', name='spark-thriftserver', queue='default', state='RUNNING', finalStatus='UNDEFINED', progress=10.0, trackingUI='ApplicationMaster', trackingUrl='http: }
@Test public void testCompleteSparkJob() throws Exception { String[] rmBasePaths = new String[]{"http: RMResourceFetcher rmResourceFetcher = new RMResourceFetcher(rmBasePaths); InputStream jsonstream = this.getClass().getResourceAsStream("/sparkcompleteapp.json"); mockStatic(InputStreamUtils.class); long finishedTimeBegin = 1479244718794l; when(InputStreamUtils.getInputStream("http: when(InputStreamUtils.getInputStream("http: List<AppInfo> appInfos = rmResourceFetcher.getResource(Constants.ResourceType.COMPLETE_SPARK_JOB, String.valueOf(finishedTimeBegin)); Assert.assertEquals(2, appInfos.size()); Assert.assertEquals("AppInfo{id='application_1479196259121_1041', user='xxx', name='Spark shell', queue='default', state='FINISHED', finalStatus='SUCCEEDED', progress=10.0, trackingUI='ApplicationMaster', trackingUrl='http: Assert.assertEquals("AppInfo{id='application_1478655515028_74912', user='xxx', name='spark-thriftserver', queue='default', state='FINISHED', finalStatus='SUCCEEDED', progress=10.0, trackingUI='ApplicationMaster', trackingUrl='http: }