ClientServiceDelegate { public JobStatus getJobStatus(JobID oldJobID) throws IOException { org.apache.hadoop.mapreduce.v2.api.records.JobId jobId = TypeConverter.toYarn(oldJobID); GetJobReportRequest request = recordFactory.newRecordInstance(GetJobReportRequest.class); request.setJobId(jobId); JobReport report = ((GetJobReportResponse) invoke("getJobReport", GetJobReportRequest.class, request)).getJobReport(); JobStatus jobStatus = null; if (report != null) { if (StringUtils.isEmpty(report.getJobFile())) { String jobFile = MRApps.getJobFile(conf, report.getUser(), oldJobID); report.setJobFile(jobFile); } String historyTrackingUrl = report.getTrackingUrl(); String url = StringUtils.isNotEmpty(historyTrackingUrl) ? historyTrackingUrl : trackingUrl; jobStatus = TypeConverter.fromYarn(report, url); } return jobStatus; } ClientServiceDelegate(Configuration conf, ResourceMgrDelegate rm,
      JobID jobId, MRClientProtocol historyServerProxy); @VisibleForTesting int getMaxClientRetry(); org.apache.hadoop.mapreduce.Counters getJobCounters(JobID arg0); TaskCompletionEvent[] getTaskCompletionEvents(JobID arg0, int arg1, int arg2); String[] getTaskDiagnostics(org.apache.hadoop.mapreduce.TaskAttemptID arg0); JobStatus getJobStatus(JobID oldJobID); org.apache.hadoop.mapreduce.TaskReport[] getTaskReports(JobID oldJobID, TaskType taskType); boolean killTask(TaskAttemptID taskAttemptID, boolean fail); boolean killJob(JobID oldJobID); LogParams getLogFilePath(JobID oldJobID, TaskAttemptID oldTaskAttemptID); void close();  }
@Test public void testUnknownAppInRM() throws Exception { MRClientProtocol historyServerProxy = mock(MRClientProtocol.class); when(historyServerProxy.getJobReport(getJobReportRequest())).thenReturn( getJobReportResponse()); ClientServiceDelegate clientServiceDelegate = getClientServiceDelegate( historyServerProxy, getRMDelegate()); JobStatus jobStatus = clientServiceDelegate.getJobStatus(oldJobId); Assert.assertNotNull(jobStatus); }
@Test public void testRemoteExceptionFromHistoryServer() throws Exception { MRClientProtocol historyServerProxy = mock(MRClientProtocol.class); when(historyServerProxy.getJobReport(getJobReportRequest())).thenThrow( new IOException("Job ID doesnot Exist")); ResourceMgrDelegate rm = mock(ResourceMgrDelegate.class); when(rm.getApplicationReport(TypeConverter.toYarn(oldJobId).getAppId())) .thenReturn(null); ClientServiceDelegate clientServiceDelegate = getClientServiceDelegate( historyServerProxy, rm); try { clientServiceDelegate.getJobStatus(oldJobId); Assert.fail("Invoke should throw exception after retries."); } catch (IOException e) { Assert.assertTrue(e.getMessage().contains( "Job ID doesnot Exist")); } }
@Test public void testRetriesOnConnectionFailure() throws Exception { MRClientProtocol historyServerProxy = mock(MRClientProtocol.class); when(historyServerProxy.getJobReport(getJobReportRequest())).thenThrow( new RuntimeException("1")).thenThrow(new RuntimeException("2")) .thenReturn(getJobReportResponse()); ResourceMgrDelegate rm = mock(ResourceMgrDelegate.class); when(rm.getApplicationReport(TypeConverter.toYarn(oldJobId).getAppId())) .thenReturn(null); ClientServiceDelegate clientServiceDelegate = getClientServiceDelegate( historyServerProxy, rm); JobStatus jobStatus = clientServiceDelegate.getJobStatus(oldJobId); Assert.assertNotNull(jobStatus); verify(historyServerProxy, times(3)).getJobReport( any(GetJobReportRequest.class)); }
@Test public void testHistoryServerNotConfigured() throws Exception { ClientServiceDelegate clientServiceDelegate = getClientServiceDelegate( null, getRMDelegate()); JobStatus jobStatus = clientServiceDelegate.getJobStatus(oldJobId); Assert.assertEquals("N/A", jobStatus.getUsername()); Assert.assertEquals(JobStatus.State.PREP, jobStatus.getState()); ResourceMgrDelegate rm = mock(ResourceMgrDelegate.class); ApplicationReport applicationReport = getFinishedApplicationReport(); when(rm.getApplicationReport(jobId.getAppId())).thenReturn( applicationReport); clientServiceDelegate = getClientServiceDelegate(null, rm); jobStatus = clientServiceDelegate.getJobStatus(oldJobId); Assert.assertEquals(applicationReport.getUser(), jobStatus.getUsername()); Assert.assertEquals(JobStatus.State.SUCCEEDED, jobStatus.getState()); }
@Test public void testJobReportFromHistoryServer() throws Exception { MRClientProtocol historyServerProxy = mock(MRClientProtocol.class); when(historyServerProxy.getJobReport(getJobReportRequest())).thenReturn( getJobReportResponseFromHistoryServer()); ResourceMgrDelegate rm = mock(ResourceMgrDelegate.class); when(rm.getApplicationReport(TypeConverter.toYarn(oldJobId).getAppId())) .thenReturn(null); ClientServiceDelegate clientServiceDelegate = getClientServiceDelegate( historyServerProxy, rm); JobStatus jobStatus = clientServiceDelegate.getJobStatus(oldJobId); Assert.assertNotNull(jobStatus); Assert.assertEquals("TestJobFilePath", jobStatus.getJobFile()); Assert.assertEquals("http: Assert.assertEquals(1.0f, jobStatus.getMapProgress(), 0.0f); Assert.assertEquals(1.0f, jobStatus.getReduceProgress(), 0.0f); }
@Test public void testRMDownRestoreForJobStatusBeforeGetAMReport() throws IOException { Configuration conf = new YarnConfiguration(); conf.setInt(MRJobConfig.MR_CLIENT_MAX_RETRIES, 3); conf.set(MRConfig.FRAMEWORK_NAME, MRConfig.YARN_FRAMEWORK_NAME); conf.setBoolean(MRJobConfig.JOB_AM_ACCESS_DISABLED, !isAMReachableFromClient); MRClientProtocol historyServerProxy = mock(MRClientProtocol.class); when(historyServerProxy.getJobReport(any(GetJobReportRequest.class))) .thenReturn(getJobReportResponse()); ResourceMgrDelegate rmDelegate = mock(ResourceMgrDelegate.class); try { when(rmDelegate.getApplicationReport(jobId.getAppId())).thenThrow( new java.lang.reflect.UndeclaredThrowableException(new IOException( "Connection refuced1"))).thenThrow( new java.lang.reflect.UndeclaredThrowableException(new IOException( "Connection refuced2"))) .thenReturn(getFinishedApplicationReport()); ClientServiceDelegate clientServiceDelegate = new ClientServiceDelegate( conf, rmDelegate, oldJobId, historyServerProxy); JobStatus jobStatus = clientServiceDelegate.getJobStatus(oldJobId); verify(rmDelegate, times(3)).getApplicationReport( any(ApplicationId.class)); Assert.assertNotNull(jobStatus); } catch (YarnException e) { throw new IOException(e); } }