JobEndNotifier { public static void localRunnerNotification(JobConf conf, JobStatus status) { JobEndStatusInfo notification = createNotification(conf, status); if (notification != null) { do { try { int code = httpNotification(notification.getUri(), notification.getTimeout()); if (code != 200) { throw new IOException("Invalid response status code: " + code); } else { break; } } catch (IOException ioex) { LOG.error("Notification error [" + notification.getUri() + "]", ioex); } catch (Exception ex) { LOG.error("Notification error [" + notification.getUri() + "]", ex); } try { Thread.sleep(notification.getRetryInterval()); } catch (InterruptedException iex) { LOG.error("Notification retry error [" + notification + "]", iex); } } while (notification.configureForRetry()); } }  static void localRunnerNotification(JobConf conf, JobStatus status);  }
@Test public void testLocalJobRunnerUriSubstitution() throws InterruptedException { JobStatus jobStatus = createTestJobStatus( "job_20130313155005308_0001", JobStatus.SUCCEEDED); JobConf jobConf = createTestJobConf( new Configuration(), 0, baseUrl + "jobend?jobid=$jobId&status=$jobStatus"); JobEndNotifier.localRunnerNotification(jobConf, jobStatus); assertEquals(1, JobEndServlet.calledTimes); assertEquals("jobid=job_20130313155005308_0001&status=SUCCEEDED", JobEndServlet.requestUri.getQuery()); }
@Test public void testLocalJobRunnerRetryCount() throws InterruptedException { int retryAttempts = 3; JobStatus jobStatus = createTestJobStatus( "job_20130313155005308_0001", JobStatus.SUCCEEDED); JobConf jobConf = createTestJobConf( new Configuration(), retryAttempts, baseUrl + "fail"); JobEndNotifier.localRunnerNotification(jobConf, jobStatus); assertEquals(retryAttempts + 1, FailServlet.calledTimes); }
@Test public void testNotificationTimeout() throws InterruptedException { Configuration conf = new Configuration(); conf.setInt("mapreduce.job.end-notification.timeout", 1000); JobStatus jobStatus = createTestJobStatus( "job_20130313155005308_0001", JobStatus.SUCCEEDED); JobConf jobConf = createTestJobConf( conf, 0, baseUrl + "delay"); long startTime = System.currentTimeMillis(); JobEndNotifier.localRunnerNotification(jobConf, jobStatus); long elapsedTime = System.currentTimeMillis() - startTime; assertEquals(1, DelayServlet.calledTimes); assertTrue(elapsedTime < 2000); }