@SuppressWarnings("unchecked") protected void handleJobCommit(CommitterJobCommitEvent event) { try { touchz(startCommitFile); jobCommitStarted(); waitForValidCommitWindow(); committer.commitJob(event.getJobContext()); touchz(endCommitSuccessFile); context.getEventHandler().handle( new JobCommitCompletedEvent(event.getJobID())); } catch (Exception e) { try { touchz(endCommitFailureFile); } catch (Exception e2) { LOG.error("could not create failure file.", e2); } LOG.error("Could not commit job", e); context.getEventHandler().handle( new JobCommitFailedEvent(event.getJobID(), StringUtils.stringifyException(e))); } finally { jobCommitEnded(); } }
@Override public void transition(JobImpl job, JobEvent event) { JobCommitFailedEvent jcfe = (JobCommitFailedEvent)event; job.addDiagnostic("Job commit failed: " + jcfe.getMessage()); job.eventHandler.handle(new CommitterJobAbortEvent(job.jobId, job.jobContext, org.apache.hadoop.mapreduce.JobStatus.State.FAILED)); }
@Test public void testFailure() throws Exception { AppContext mockContext = mock(AppContext.class); OutputCommitter mockCommitter = mock(OutputCommitter.class); Clock mockClock = mock(Clock.class); CommitterEventHandler handler = new CommitterEventHandler(mockContext, mockCommitter, new TestingRMHeartbeatHandler()); YarnConfiguration conf = new YarnConfiguration(); conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir); JobContext mockJobContext = mock(JobContext.class); ApplicationAttemptId attemptid = ConverterUtils.toApplicationAttemptId("appattempt_1234567890000_0001_0"); JobId jobId = TypeConverter.toYarn( TypeConverter.fromYarn(attemptid.getApplicationId())); WaitForItHandler waitForItHandler = new WaitForItHandler(); when(mockContext.getApplicationID()).thenReturn(attemptid.getApplicationId()); when(mockContext.getApplicationAttemptId()).thenReturn(attemptid); when(mockContext.getEventHandler()).thenReturn(waitForItHandler); when(mockContext.getClock()).thenReturn(mockClock); doThrow(new YarnRuntimeException("Intentional Failure")).when(mockCommitter) .commitJob(any(JobContext.class)); handler.init(conf); handler.start(); try { handler.handle(new CommitterJobCommitEvent(jobId, mockJobContext)); String user = UserGroupInformation.getCurrentUser().getShortUserName(); Path startCommitFile = MRApps.getStartJobCommitFile(conf, user, jobId); Path endCommitSuccessFile = MRApps.getEndJobCommitSuccessFile(conf, user, jobId); Path endCommitFailureFile = MRApps.getEndJobCommitFailureFile(conf, user, jobId); Event e = waitForItHandler.getAndClearEvent(); assertNotNull(e); assertTrue(e instanceof JobCommitFailedEvent); FileSystem fs = FileSystem.get(conf); assertTrue(fs.exists(startCommitFile)); assertFalse(fs.exists(endCommitSuccessFile)); assertTrue(fs.exists(endCommitFailureFile)); verify(mockCommitter).commitJob(any(JobContext.class)); } finally { handler.stop(); } }
@Test public void testFailure() throws Exception { AppContext mockContext = mock(AppContext.class); OutputCommitter mockCommitter = mock(OutputCommitter.class); Clock mockClock = mock(Clock.class); CommitterEventHandler handler = new CommitterEventHandler(mockContext, mockCommitter, new TestingRMHeartbeatHandler()); YarnConfiguration conf = new YarnConfiguration(); conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir); JobContext mockJobContext = mock(JobContext.class); ApplicationAttemptId attemptid = ApplicationAttemptId.fromString("appattempt_1234567890000_0001_0"); JobId jobId = TypeConverter.toYarn( TypeConverter.fromYarn(attemptid.getApplicationId())); WaitForItHandler waitForItHandler = new WaitForItHandler(); when(mockContext.getApplicationID()).thenReturn(attemptid.getApplicationId()); when(mockContext.getApplicationAttemptId()).thenReturn(attemptid); when(mockContext.getEventHandler()).thenReturn(waitForItHandler); when(mockContext.getClock()).thenReturn(mockClock); doThrow(new YarnRuntimeException("Intentional Failure")).when(mockCommitter) .commitJob(any(JobContext.class)); handler.init(conf); handler.start(); try { handler.handle(new CommitterJobCommitEvent(jobId, mockJobContext)); String user = UserGroupInformation.getCurrentUser().getShortUserName(); Path startCommitFile = MRApps.getStartJobCommitFile(conf, user, jobId); Path endCommitSuccessFile = MRApps.getEndJobCommitSuccessFile(conf, user, jobId); Path endCommitFailureFile = MRApps.getEndJobCommitFailureFile(conf, user, jobId); Event e = waitForItHandler.getAndClearEvent(); assertNotNull(e); assertTrue(e instanceof JobCommitFailedEvent); FileSystem fs = FileSystem.get(conf); assertTrue(fs.exists(startCommitFile)); assertFalse(fs.exists(endCommitSuccessFile)); assertTrue(fs.exists(endCommitFailureFile)); verify(mockCommitter).commitJob(any(JobContext.class)); } finally { handler.stop(); } }