1. 包列表
  2. 类列表
  3. org.apache.hadoop.mapreduce.v2.app.commit.CommitterEventHandler

Java 类org.apache.hadoop.mapreduce.v2.app.commit.CommitterEventHandler 实例源码

项目：hadoop    文件：TestJobImpl.java    
@Test(timeout=20000)
public void testCommitJobFailsJob() throws Exception {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  CyclicBarrier syncBarrier = new CyclicBarrier(2);
  OutputCommitter committer = new TestingOutputCommitter(syncBarrier, false);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  JobImpl job = createRunningStubbedJob(conf, dispatcher, 2, null);
  completeJobTasks(job);
  assertJobState(job, JobStateInternal.COMMITTING);

  // let the committer fail and verify the job fails
  syncBarrier.await();
  assertJobState(job, JobStateInternal.FAILED);
  dispatcher.stop();
  commitHandler.stop();
}
项目：hadoop    文件：TestJobImpl.java    
@Test(timeout=20000)
public void testKilledDuringCommit() throws Exception {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  CyclicBarrier syncBarrier = new CyclicBarrier(2);
  OutputCommitter committer = new WaitingOutputCommitter(syncBarrier, true);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  JobImpl job = createRunningStubbedJob(conf, dispatcher, 2, null);
  completeJobTasks(job);
  assertJobState(job, JobStateInternal.COMMITTING);

  syncBarrier.await();
  job.handle(new JobEvent(job.getID(), JobEventType.JOB_KILL));
  assertJobState(job, JobStateInternal.KILLED);
  dispatcher.stop();
  commitHandler.stop();
}
项目：hadoop    文件：TestJobImpl.java    
@Test
public void testTransitionsAtFailed() throws IOException {
  Configuration conf = new Configuration();
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();

  OutputCommitter committer = mock(OutputCommitter.class);
  doThrow(new IOException("forcefail"))
    .when(committer).setupJob(any(JobContext.class));
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  AppContext mockContext = mock(AppContext.class);
  when(mockContext.hasSuccessfullyUnregistered()).thenReturn(false);
  JobImpl job = createStubbedJob(conf, dispatcher, 2, mockContext);
  JobId jobId = job.getID();
  job.handle(new JobEvent(jobId, JobEventType.JOB_INIT));
  assertJobState(job, JobStateInternal.INITED);
  job.handle(new JobStartEvent(jobId));
  assertJobState(job, JobStateInternal.FAILED);

  job.handle(new JobEvent(jobId, JobEventType.JOB_TASK_COMPLETED));
  assertJobState(job, JobStateInternal.FAILED);
  job.handle(new JobEvent(jobId, JobEventType.JOB_TASK_ATTEMPT_COMPLETED));
  assertJobState(job, JobStateInternal.FAILED);
  job.handle(new JobEvent(jobId, JobEventType.JOB_MAP_TASK_RESCHEDULED));
  assertJobState(job, JobStateInternal.FAILED);
  job.handle(new JobEvent(jobId, JobEventType.JOB_TASK_ATTEMPT_FETCH_FAILURE));
  assertJobState(job, JobStateInternal.FAILED);
  Assert.assertEquals(JobState.RUNNING, job.getState());
  when(mockContext.hasSuccessfullyUnregistered()).thenReturn(true);
  Assert.assertEquals(JobState.FAILED, job.getState());

  dispatcher.stop();
  commitHandler.stop();
}
项目：hadoop    文件：TestJobImpl.java    
private static CommitterEventHandler createCommitterEventHandler(
    Dispatcher dispatcher, OutputCommitter committer) {
  final SystemClock clock = new SystemClock();
  AppContext appContext = mock(AppContext.class);
  when(appContext.getEventHandler()).thenReturn(
      dispatcher.getEventHandler());
  when(appContext.getClock()).thenReturn(clock);
  RMHeartbeatHandler heartbeatHandler = new RMHeartbeatHandler() {
    @Override
    public long getLastHeartbeatTime() {
      return clock.getTime();
    }
    @Override
    public void runOnNextHeartbeat(Runnable callback) {
      callback.run();
    }
  };
  ApplicationAttemptId id = 
    ConverterUtils.toApplicationAttemptId("appattempt_1234567890000_0001_0");
  when(appContext.getApplicationID()).thenReturn(id.getApplicationId());
  when(appContext.getApplicationAttemptId()).thenReturn(id);
  CommitterEventHandler handler =
      new CommitterEventHandler(appContext, committer, heartbeatHandler);
  dispatcher.register(CommitterEventType.class, handler);
  return handler;
}
项目：aliyun-oss-hadoop-fs    文件：TestJobImpl.java    
@Test(timeout=20000)
public void testCommitJobFailsJob() throws Exception {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  CyclicBarrier syncBarrier = new CyclicBarrier(2);
  OutputCommitter committer = new TestingOutputCommitter(syncBarrier, false);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  JobImpl job = createRunningStubbedJob(conf, dispatcher, 2, null);
  completeJobTasks(job);
  assertJobState(job, JobStateInternal.COMMITTING);

  // let the committer fail and verify the job fails
  syncBarrier.await();
  assertJobState(job, JobStateInternal.FAILED);
  dispatcher.stop();
  commitHandler.stop();
}
项目：aliyun-oss-hadoop-fs    文件：TestJobImpl.java    
@Test(timeout=20000)
public void testKilledDuringCommit() throws Exception {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  CyclicBarrier syncBarrier = new CyclicBarrier(2);
  OutputCommitter committer = new WaitingOutputCommitter(syncBarrier, true);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  JobImpl job = createRunningStubbedJob(conf, dispatcher, 2, null);
  completeJobTasks(job);
  assertJobState(job, JobStateInternal.COMMITTING);

  syncBarrier.await();
  job.handle(new JobEvent(job.getID(), JobEventType.JOB_KILL));
  assertJobState(job, JobStateInternal.KILLED);
  dispatcher.stop();
  commitHandler.stop();
}
项目：aliyun-oss-hadoop-fs    文件：TestJobImpl.java    
private static CommitterEventHandler createCommitterEventHandler(
    Dispatcher dispatcher, OutputCommitter committer) {
  final SystemClock clock = new SystemClock();
  AppContext appContext = mock(AppContext.class);
  when(appContext.getEventHandler()).thenReturn(
      dispatcher.getEventHandler());
  when(appContext.getClock()).thenReturn(clock);
  RMHeartbeatHandler heartbeatHandler = new RMHeartbeatHandler() {
    @Override
    public long getLastHeartbeatTime() {
      return clock.getTime();
    }
    @Override
    public void runOnNextHeartbeat(Runnable callback) {
      callback.run();
    }
  };
  ApplicationAttemptId id = 
    ConverterUtils.toApplicationAttemptId("appattempt_1234567890000_0001_0");
  when(appContext.getApplicationID()).thenReturn(id.getApplicationId());
  when(appContext.getApplicationAttemptId()).thenReturn(id);
  CommitterEventHandler handler =
      new CommitterEventHandler(appContext, committer, heartbeatHandler);
  dispatcher.register(CommitterEventType.class, handler);
  return handler;
}
项目：big-c    文件：TestJobImpl.java    
@Test(timeout=20000)
public void testCommitJobFailsJob() throws Exception {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  CyclicBarrier syncBarrier = new CyclicBarrier(2);
  OutputCommitter committer = new TestingOutputCommitter(syncBarrier, false);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  JobImpl job = createRunningStubbedJob(conf, dispatcher, 2, null);
  completeJobTasks(job);
  assertJobState(job, JobStateInternal.COMMITTING);

  // let the committer fail and verify the job fails
  syncBarrier.await();
  assertJobState(job, JobStateInternal.FAILED);
  dispatcher.stop();
  commitHandler.stop();
}
项目：big-c    文件：TestJobImpl.java    
@Test(timeout=20000)
public void testKilledDuringCommit() throws Exception {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  CyclicBarrier syncBarrier = new CyclicBarrier(2);
  OutputCommitter committer = new WaitingOutputCommitter(syncBarrier, true);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  JobImpl job = createRunningStubbedJob(conf, dispatcher, 2, null);
  completeJobTasks(job);
  assertJobState(job, JobStateInternal.COMMITTING);

  syncBarrier.await();
  job.handle(new JobEvent(job.getID(), JobEventType.JOB_KILL));
  assertJobState(job, JobStateInternal.KILLED);
  dispatcher.stop();
  commitHandler.stop();
}
项目：big-c    文件：TestJobImpl.java    
private static CommitterEventHandler createCommitterEventHandler(
    Dispatcher dispatcher, OutputCommitter committer) {
  final SystemClock clock = new SystemClock();
  AppContext appContext = mock(AppContext.class);
  when(appContext.getEventHandler()).thenReturn(
      dispatcher.getEventHandler());
  when(appContext.getClock()).thenReturn(clock);
  RMHeartbeatHandler heartbeatHandler = new RMHeartbeatHandler() {
    @Override
    public long getLastHeartbeatTime() {
      return clock.getTime();
    }
    @Override
    public void runOnNextHeartbeat(Runnable callback) {
      callback.run();
    }
  };
  ApplicationAttemptId id = 
    ConverterUtils.toApplicationAttemptId("appattempt_1234567890000_0001_0");
  when(appContext.getApplicationID()).thenReturn(id.getApplicationId());
  when(appContext.getApplicationAttemptId()).thenReturn(id);
  CommitterEventHandler handler =
      new CommitterEventHandler(appContext, committer, heartbeatHandler);
  dispatcher.register(CommitterEventType.class, handler);
  return handler;
}
项目：hadoop-2.6.0-cdh5.4.3    文件：TestJobImpl.java    
@Test(timeout=20000)
public void testCommitJobFailsJob() throws Exception {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  CyclicBarrier syncBarrier = new CyclicBarrier(2);
  OutputCommitter committer = new TestingOutputCommitter(syncBarrier, false);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  JobImpl job = createRunningStubbedJob(conf, dispatcher, 2, null);
  completeJobTasks(job);
  assertJobState(job, JobStateInternal.COMMITTING);

  // let the committer fail and verify the job fails
  syncBarrier.await();
  assertJobState(job, JobStateInternal.FAILED);
  dispatcher.stop();
  commitHandler.stop();
}
项目：hadoop-2.6.0-cdh5.4.3    文件：TestJobImpl.java    
@Test(timeout=20000)
public void testKilledDuringCommit() throws Exception {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  CyclicBarrier syncBarrier = new CyclicBarrier(2);
  OutputCommitter committer = new WaitingOutputCommitter(syncBarrier, true);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  JobImpl job = createRunningStubbedJob(conf, dispatcher, 2, null);
  completeJobTasks(job);
  assertJobState(job, JobStateInternal.COMMITTING);

  syncBarrier.await();
  job.handle(new JobEvent(job.getID(), JobEventType.JOB_KILL));
  assertJobState(job, JobStateInternal.KILLED);
  dispatcher.stop();
  commitHandler.stop();
}
项目：hadoop-2.6.0-cdh5.4.3    文件：TestJobImpl.java    
private static CommitterEventHandler createCommitterEventHandler(
    Dispatcher dispatcher, OutputCommitter committer) {
  final SystemClock clock = new SystemClock();
  AppContext appContext = mock(AppContext.class);
  when(appContext.getEventHandler()).thenReturn(
      dispatcher.getEventHandler());
  when(appContext.getClock()).thenReturn(clock);
  RMHeartbeatHandler heartbeatHandler = new RMHeartbeatHandler() {
    @Override
    public long getLastHeartbeatTime() {
      return clock.getTime();
    }
    @Override
    public void runOnNextHeartbeat(Runnable callback) {
      callback.run();
    }
  };
  ApplicationAttemptId id = 
    ConverterUtils.toApplicationAttemptId("appattempt_1234567890000_0001_0");
  when(appContext.getApplicationID()).thenReturn(id.getApplicationId());
  when(appContext.getApplicationAttemptId()).thenReturn(id);
  CommitterEventHandler handler =
      new CommitterEventHandler(appContext, committer, heartbeatHandler);
  dispatcher.register(CommitterEventType.class, handler);
  return handler;
}
项目：hadoop-plus    文件：TestJobImpl.java    
@Test(timeout=20000)
public void testCommitJobFailsJob() throws Exception {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  CyclicBarrier syncBarrier = new CyclicBarrier(2);
  OutputCommitter committer = new TestingOutputCommitter(syncBarrier, false);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  JobImpl job = createRunningStubbedJob(conf, dispatcher, 2);
  completeJobTasks(job);
  assertJobState(job, JobStateInternal.COMMITTING);

  // let the committer fail and verify the job fails
  syncBarrier.await();
  assertJobState(job, JobStateInternal.FAILED);
  dispatcher.stop();
  commitHandler.stop();
}
项目：hadoop-plus    文件：TestJobImpl.java    
@Test(timeout=20000)
public void testCheckJobCompleteSuccess() throws Exception {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  CyclicBarrier syncBarrier = new CyclicBarrier(2);
  OutputCommitter committer = new TestingOutputCommitter(syncBarrier, true);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  JobImpl job = createRunningStubbedJob(conf, dispatcher, 2);
  completeJobTasks(job);
  assertJobState(job, JobStateInternal.COMMITTING);

  // let the committer complete and verify the job succeeds
  syncBarrier.await();
  assertJobState(job, JobStateInternal.SUCCEEDED);
  dispatcher.stop();
  commitHandler.stop();
}
项目：hadoop-plus    文件：TestJobImpl.java    
@Test(timeout=20000)
public void testRebootedDuringCommit() throws Exception {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  CyclicBarrier syncBarrier = new CyclicBarrier(2);
  OutputCommitter committer = new WaitingOutputCommitter(syncBarrier, true);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  JobImpl job = createRunningStubbedJob(conf, dispatcher, 2);
  completeJobTasks(job);
  assertJobState(job, JobStateInternal.COMMITTING);

  syncBarrier.await();
  job.handle(new JobEvent(job.getID(), JobEventType.JOB_AM_REBOOT));
  assertJobState(job, JobStateInternal.REBOOT);
  dispatcher.stop();
  commitHandler.stop();
}
项目：hadoop-plus    文件：TestJobImpl.java    
@Test(timeout=20000)
public void testKilledDuringCommit() throws Exception {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  CyclicBarrier syncBarrier = new CyclicBarrier(2);
  OutputCommitter committer = new WaitingOutputCommitter(syncBarrier, true);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  JobImpl job = createRunningStubbedJob(conf, dispatcher, 2);
  completeJobTasks(job);
  assertJobState(job, JobStateInternal.COMMITTING);

  syncBarrier.await();
  job.handle(new JobEvent(job.getID(), JobEventType.JOB_KILL));
  assertJobState(job, JobStateInternal.KILLED);
  dispatcher.stop();
  commitHandler.stop();
}
项目：hadoop-plus    文件：TestJobImpl.java    
private static CommitterEventHandler createCommitterEventHandler(
    Dispatcher dispatcher, OutputCommitter committer) {
  final SystemClock clock = new SystemClock();
  AppContext appContext = mock(AppContext.class);
  when(appContext.getEventHandler()).thenReturn(
      dispatcher.getEventHandler());
  when(appContext.getClock()).thenReturn(clock);
  RMHeartbeatHandler heartbeatHandler = new RMHeartbeatHandler() {
    @Override
    public long getLastHeartbeatTime() {
      return clock.getTime();
    }
    @Override
    public void runOnNextHeartbeat(Runnable callback) {
      callback.run();
    }
  };
  ApplicationAttemptId id = 
    ConverterUtils.toApplicationAttemptId("appattempt_1234567890000_0001_0");
  when(appContext.getApplicationID()).thenReturn(id.getApplicationId());
  when(appContext.getApplicationAttemptId()).thenReturn(id);
  CommitterEventHandler handler =
      new CommitterEventHandler(appContext, committer, heartbeatHandler);
  dispatcher.register(CommitterEventType.class, handler);
  return handler;
}
项目：FlexMap    文件：TestJobImpl.java    
@Test(timeout=20000)
public void testCommitJobFailsJob() throws Exception {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  CyclicBarrier syncBarrier = new CyclicBarrier(2);
  OutputCommitter committer = new TestingOutputCommitter(syncBarrier, false);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  JobImpl job = createRunningStubbedJob(conf, dispatcher, 2, null);
  completeJobTasks(job);
  assertJobState(job, JobStateInternal.COMMITTING);

  // let the committer fail and verify the job fails
  syncBarrier.await();
  assertJobState(job, JobStateInternal.FAILED);
  dispatcher.stop();
  commitHandler.stop();
}
项目：FlexMap    文件：TestJobImpl.java    
@Test(timeout=20000)
public void testKilledDuringCommit() throws Exception {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  CyclicBarrier syncBarrier = new CyclicBarrier(2);
  OutputCommitter committer = new WaitingOutputCommitter(syncBarrier, true);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  JobImpl job = createRunningStubbedJob(conf, dispatcher, 2, null);
  completeJobTasks(job);
  assertJobState(job, JobStateInternal.COMMITTING);

  syncBarrier.await();
  job.handle(new JobEvent(job.getID(), JobEventType.JOB_KILL));
  assertJobState(job, JobStateInternal.KILLED);
  dispatcher.stop();
  commitHandler.stop();
}
项目：FlexMap    文件：TestJobImpl.java    
private static CommitterEventHandler createCommitterEventHandler(
    Dispatcher dispatcher, OutputCommitter committer) {
  final SystemClock clock = new SystemClock();
  AppContext appContext = mock(AppContext.class);
  when(appContext.getEventHandler()).thenReturn(
      dispatcher.getEventHandler());
  when(appContext.getClock()).thenReturn(clock);
  RMHeartbeatHandler heartbeatHandler = new RMHeartbeatHandler() {
    @Override
    public long getLastHeartbeatTime() {
      return clock.getTime();
    }
    @Override
    public void runOnNextHeartbeat(Runnable callback) {
      callback.run();
    }
  };
  ApplicationAttemptId id = 
    ConverterUtils.toApplicationAttemptId("appattempt_1234567890000_0001_0");
  when(appContext.getApplicationID()).thenReturn(id.getApplicationId());
  when(appContext.getApplicationAttemptId()).thenReturn(id);
  CommitterEventHandler handler =
      new CommitterEventHandler(appContext, committer, heartbeatHandler);
  dispatcher.register(CommitterEventType.class, handler);
  return handler;
}
项目：hops    文件：TestJobImpl.java    
@Test(timeout=20000)
public void testCommitJobFailsJob() throws Exception {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  CyclicBarrier syncBarrier = new CyclicBarrier(2);
  OutputCommitter committer = new TestingOutputCommitter(syncBarrier, false);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  JobImpl job = createRunningStubbedJob(conf, dispatcher, 2, null);
  completeJobTasks(job);
  assertJobState(job, JobStateInternal.COMMITTING);

  // let the committer fail and verify the job fails
  syncBarrier.await();
  assertJobState(job, JobStateInternal.FAILED);
  dispatcher.stop();
  commitHandler.stop();
}
项目：hops    文件：TestJobImpl.java    
@Test(timeout=20000)
public void testKilledDuringCommit() throws Exception {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  CyclicBarrier syncBarrier = new CyclicBarrier(2);
  OutputCommitter committer = new WaitingOutputCommitter(syncBarrier, true);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  JobImpl job = createRunningStubbedJob(conf, dispatcher, 2, null);
  completeJobTasks(job);
  assertJobState(job, JobStateInternal.COMMITTING);

  syncBarrier.await();
  job.handle(new JobEvent(job.getID(), JobEventType.JOB_KILL));
  assertJobState(job, JobStateInternal.KILLED);
  dispatcher.stop();
  commitHandler.stop();
}
项目：hops    文件：TestJobImpl.java    
private static CommitterEventHandler createCommitterEventHandler(
    Dispatcher dispatcher, OutputCommitter committer) {
  final SystemClock clock = new SystemClock();
  AppContext appContext = mock(AppContext.class);
  when(appContext.getEventHandler()).thenReturn(
      dispatcher.getEventHandler());
  when(appContext.getClock()).thenReturn(clock);
  RMHeartbeatHandler heartbeatHandler = new RMHeartbeatHandler() {
    @Override
    public long getLastHeartbeatTime() {
      return clock.getTime();
    }
    @Override
    public void runOnNextHeartbeat(Runnable callback) {
      callback.run();
    }
  };
  ApplicationAttemptId id = ApplicationAttemptId.fromString(
      "appattempt_1234567890000_0001_0");
  when(appContext.getApplicationID()).thenReturn(id.getApplicationId());
  when(appContext.getApplicationAttemptId()).thenReturn(id);
  CommitterEventHandler handler =
      new CommitterEventHandler(appContext, committer, heartbeatHandler);
  dispatcher.register(CommitterEventType.class, handler);
  return handler;
}
项目：hadoop-TCP    文件：TestJobImpl.java    
@Test(timeout=20000)
public void testCommitJobFailsJob() throws Exception {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  CyclicBarrier syncBarrier = new CyclicBarrier(2);
  OutputCommitter committer = new TestingOutputCommitter(syncBarrier, false);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  JobImpl job = createRunningStubbedJob(conf, dispatcher, 2, null);
  completeJobTasks(job);
  assertJobState(job, JobStateInternal.COMMITTING);

  // let the committer fail and verify the job fails
  syncBarrier.await();
  assertJobState(job, JobStateInternal.FAILED);
  dispatcher.stop();
  commitHandler.stop();
}
项目：hadoop-TCP    文件：TestJobImpl.java    
@Test(timeout=20000)
public void testKilledDuringCommit() throws Exception {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  CyclicBarrier syncBarrier = new CyclicBarrier(2);
  OutputCommitter committer = new WaitingOutputCommitter(syncBarrier, true);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  JobImpl job = createRunningStubbedJob(conf, dispatcher, 2, null);
  completeJobTasks(job);
  assertJobState(job, JobStateInternal.COMMITTING);

  syncBarrier.await();
  job.handle(new JobEvent(job.getID(), JobEventType.JOB_KILL));
  assertJobState(job, JobStateInternal.KILLED);
  dispatcher.stop();
  commitHandler.stop();
}
项目：hadoop-TCP    文件：TestJobImpl.java    
private static CommitterEventHandler createCommitterEventHandler(
    Dispatcher dispatcher, OutputCommitter committer) {
  final SystemClock clock = new SystemClock();
  AppContext appContext = mock(AppContext.class);
  when(appContext.getEventHandler()).thenReturn(
      dispatcher.getEventHandler());
  when(appContext.getClock()).thenReturn(clock);
  RMHeartbeatHandler heartbeatHandler = new RMHeartbeatHandler() {
    @Override
    public long getLastHeartbeatTime() {
      return clock.getTime();
    }
    @Override
    public void runOnNextHeartbeat(Runnable callback) {
      callback.run();
    }
  };
  ApplicationAttemptId id = 
    ConverterUtils.toApplicationAttemptId("appattempt_1234567890000_0001_0");
  when(appContext.getApplicationID()).thenReturn(id.getApplicationId());
  when(appContext.getApplicationAttemptId()).thenReturn(id);
  CommitterEventHandler handler =
      new CommitterEventHandler(appContext, committer, heartbeatHandler);
  dispatcher.register(CommitterEventType.class, handler);
  return handler;
}
项目：hardfs    文件：TestJobImpl.java    
@Test(timeout=20000)
public void testCommitJobFailsJob() throws Exception {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  CyclicBarrier syncBarrier = new CyclicBarrier(2);
  OutputCommitter committer = new TestingOutputCommitter(syncBarrier, false);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  JobImpl job = createRunningStubbedJob(conf, dispatcher, 2, null);
  completeJobTasks(job);
  assertJobState(job, JobStateInternal.COMMITTING);

  // let the committer fail and verify the job fails
  syncBarrier.await();
  assertJobState(job, JobStateInternal.FAILED);
  dispatcher.stop();
  commitHandler.stop();
}
项目：hardfs    文件：TestJobImpl.java    
@Test(timeout=20000)
public void testKilledDuringCommit() throws Exception {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  CyclicBarrier syncBarrier = new CyclicBarrier(2);
  OutputCommitter committer = new WaitingOutputCommitter(syncBarrier, true);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  JobImpl job = createRunningStubbedJob(conf, dispatcher, 2, null);
  completeJobTasks(job);
  assertJobState(job, JobStateInternal.COMMITTING);

  syncBarrier.await();
  job.handle(new JobEvent(job.getID(), JobEventType.JOB_KILL));
  assertJobState(job, JobStateInternal.KILLED);
  dispatcher.stop();
  commitHandler.stop();
}
项目：hardfs    文件：TestJobImpl.java    
private static CommitterEventHandler createCommitterEventHandler(
    Dispatcher dispatcher, OutputCommitter committer) {
  final SystemClock clock = new SystemClock();
  AppContext appContext = mock(AppContext.class);
  when(appContext.getEventHandler()).thenReturn(
      dispatcher.getEventHandler());
  when(appContext.getClock()).thenReturn(clock);
  RMHeartbeatHandler heartbeatHandler = new RMHeartbeatHandler() {
    @Override
    public long getLastHeartbeatTime() {
      return clock.getTime();
    }
    @Override
    public void runOnNextHeartbeat(Runnable callback) {
      callback.run();
    }
  };
  ApplicationAttemptId id = 
    ConverterUtils.toApplicationAttemptId("appattempt_1234567890000_0001_0");
  when(appContext.getApplicationID()).thenReturn(id.getApplicationId());
  when(appContext.getApplicationAttemptId()).thenReturn(id);
  CommitterEventHandler handler =
      new CommitterEventHandler(appContext, committer, heartbeatHandler);
  dispatcher.register(CommitterEventType.class, handler);
  return handler;
}
项目：hadoop-on-lustre2    文件：TestJobImpl.java    
@Test(timeout=20000)
public void testCommitJobFailsJob() throws Exception {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  CyclicBarrier syncBarrier = new CyclicBarrier(2);
  OutputCommitter committer = new TestingOutputCommitter(syncBarrier, false);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  JobImpl job = createRunningStubbedJob(conf, dispatcher, 2, null);
  completeJobTasks(job);
  assertJobState(job, JobStateInternal.COMMITTING);

  // let the committer fail and verify the job fails
  syncBarrier.await();
  assertJobState(job, JobStateInternal.FAILED);
  dispatcher.stop();
  commitHandler.stop();
}
项目：hadoop-on-lustre2    文件：TestJobImpl.java    
@Test(timeout=20000)
public void testKilledDuringCommit() throws Exception {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  CyclicBarrier syncBarrier = new CyclicBarrier(2);
  OutputCommitter committer = new WaitingOutputCommitter(syncBarrier, true);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  JobImpl job = createRunningStubbedJob(conf, dispatcher, 2, null);
  completeJobTasks(job);
  assertJobState(job, JobStateInternal.COMMITTING);

  syncBarrier.await();
  job.handle(new JobEvent(job.getID(), JobEventType.JOB_KILL));
  assertJobState(job, JobStateInternal.KILLED);
  dispatcher.stop();
  commitHandler.stop();
}
项目：hadoop-on-lustre2    文件：TestJobImpl.java    
private static CommitterEventHandler createCommitterEventHandler(
    Dispatcher dispatcher, OutputCommitter committer) {
  final SystemClock clock = new SystemClock();
  AppContext appContext = mock(AppContext.class);
  when(appContext.getEventHandler()).thenReturn(
      dispatcher.getEventHandler());
  when(appContext.getClock()).thenReturn(clock);
  RMHeartbeatHandler heartbeatHandler = new RMHeartbeatHandler() {
    @Override
    public long getLastHeartbeatTime() {
      return clock.getTime();
    }
    @Override
    public void runOnNextHeartbeat(Runnable callback) {
      callback.run();
    }
  };
  ApplicationAttemptId id = 
    ConverterUtils.toApplicationAttemptId("appattempt_1234567890000_0001_0");
  when(appContext.getApplicationID()).thenReturn(id.getApplicationId());
  when(appContext.getApplicationAttemptId()).thenReturn(id);
  CommitterEventHandler handler =
      new CommitterEventHandler(appContext, committer, heartbeatHandler);
  dispatcher.register(CommitterEventType.class, handler);
  return handler;
}
项目：hadoop    文件：TestJobImpl.java    
@Test
public void testJobNoTasks() {
  Configuration conf = new Configuration();
  conf.setInt(MRJobConfig.NUM_REDUCES, 0);
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  conf.set(MRJobConfig.WORKFLOW_ID, "testId");
  conf.set(MRJobConfig.WORKFLOW_NAME, "testName");
  conf.set(MRJobConfig.WORKFLOW_NODE_NAME, "testNodeName");
  conf.set(MRJobConfig.WORKFLOW_ADJACENCY_PREFIX_STRING + "key1", "value1");
  conf.set(MRJobConfig.WORKFLOW_ADJACENCY_PREFIX_STRING + "key2", "value2");
  conf.set(MRJobConfig.WORKFLOW_TAGS, "tag1,tag2");


  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  OutputCommitter committer = mock(OutputCommitter.class);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  JobSubmittedEventHandler jseHandler = new JobSubmittedEventHandler("testId",
      "testName", "testNodeName", "\"key2\"=\"value2\" \"key1\"=\"value1\" ",
      "tag1,tag2");
  dispatcher.register(EventType.class, jseHandler);
  JobImpl job = createStubbedJob(conf, dispatcher, 0, null);
  job.handle(new JobEvent(job.getID(), JobEventType.JOB_INIT));
  assertJobState(job, JobStateInternal.INITED);
  job.handle(new JobStartEvent(job.getID()));
  assertJobState(job, JobStateInternal.SUCCEEDED);
  dispatcher.stop();
  commitHandler.stop();
  try {
    Assert.assertTrue(jseHandler.getAssertValue());
  } catch (InterruptedException e) {
    Assert.fail("Workflow related attributes are not tested properly");
  }
}
项目：hadoop    文件：TestJobImpl.java    
@Test(timeout=20000)
public void testCheckJobCompleteSuccess() throws Exception {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  CyclicBarrier syncBarrier = new CyclicBarrier(2);
  OutputCommitter committer = new TestingOutputCommitter(syncBarrier, true);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  JobImpl job = createRunningStubbedJob(conf, dispatcher, 2, null);
  completeJobTasks(job);
  assertJobState(job, JobStateInternal.COMMITTING);

  // let the committer complete and verify the job succeeds
  syncBarrier.await();
  assertJobState(job, JobStateInternal.SUCCEEDED);

  job.handle(new JobEvent(job.getID(),
      JobEventType.JOB_TASK_ATTEMPT_COMPLETED));
  assertJobState(job, JobStateInternal.SUCCEEDED);

  job.handle(new JobEvent(job.getID(), 
      JobEventType.JOB_MAP_TASK_RESCHEDULED));
  assertJobState(job, JobStateInternal.SUCCEEDED);

  dispatcher.stop();
  commitHandler.stop();
}
项目：hadoop    文件：TestJobImpl.java    
@Test(timeout=20000)
public void testRebootedDuringCommit() throws Exception {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  conf.setInt(MRJobConfig.MR_AM_MAX_ATTEMPTS, 2);
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  CyclicBarrier syncBarrier = new CyclicBarrier(2);
  OutputCommitter committer = new WaitingOutputCommitter(syncBarrier, true);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  AppContext mockContext = mock(AppContext.class);
  when(mockContext.isLastAMRetry()).thenReturn(true);
  when(mockContext.hasSuccessfullyUnregistered()).thenReturn(false);
  JobImpl job = createRunningStubbedJob(conf, dispatcher, 2, mockContext);
  completeJobTasks(job);
  assertJobState(job, JobStateInternal.COMMITTING);

  syncBarrier.await();
  job.handle(new JobEvent(job.getID(), JobEventType.JOB_AM_REBOOT));
  assertJobState(job, JobStateInternal.REBOOT);
  // return the external state as ERROR since this is last retry.
  Assert.assertEquals(JobState.RUNNING, job.getState());
  when(mockContext.hasSuccessfullyUnregistered()).thenReturn(true);
  Assert.assertEquals(JobState.ERROR, job.getState());

  dispatcher.stop();
  commitHandler.stop();
}
项目：hadoop    文件：TestJobImpl.java    
@Test(timeout=20000)
public void testKilledDuringSetup() throws Exception {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  OutputCommitter committer = new StubbedOutputCommitter() {
    @Override
    public synchronized void setupJob(JobContext jobContext)
        throws IOException {
      while (!Thread.interrupted()) {
        try {
          wait();
        } catch (InterruptedException e) {
        }
      }
    }
  };
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  JobImpl job = createStubbedJob(conf, dispatcher, 2, null);
  JobId jobId = job.getID();
  job.handle(new JobEvent(jobId, JobEventType.JOB_INIT));
  assertJobState(job, JobStateInternal.INITED);
  job.handle(new JobStartEvent(jobId));
  assertJobState(job, JobStateInternal.SETUP);

  job.handle(new JobEvent(job.getID(), JobEventType.JOB_KILL));
  assertJobState(job, JobStateInternal.KILLED);
  dispatcher.stop();
  commitHandler.stop();
}
项目：hadoop    文件：TestJobImpl.java    
@Test
public void testAbortJobCalledAfterKillingTasks() throws IOException {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  conf.set(MRJobConfig.MR_AM_COMMITTER_CANCEL_TIMEOUT_MS, "1000");
  InlineDispatcher dispatcher = new InlineDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  OutputCommitter committer = Mockito.mock(OutputCommitter.class);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();
  JobImpl job = createRunningStubbedJob(conf, dispatcher, 2, null);

  //Fail one task. This should land the JobImpl in the FAIL_WAIT state
  job.handle(new JobTaskEvent(
    MRBuilderUtils.newTaskId(job.getID(), 1, TaskType.MAP),
    TaskState.FAILED));
  //Verify abort job hasn't been called
  Mockito.verify(committer, Mockito.never())
    .abortJob((JobContext) Mockito.any(), (State) Mockito.any());
  assertJobState(job, JobStateInternal.FAIL_WAIT);

  //Verify abortJob is called once and the job failed
  Mockito.verify(committer, Mockito.timeout(2000).times(1))
    .abortJob((JobContext) Mockito.any(), (State) Mockito.any());
  assertJobState(job, JobStateInternal.FAILED);

  dispatcher.stop();
}
项目：hadoop    文件：TestJobImpl.java    
@Test (timeout=10000)
public void testFailAbortDoesntHang() throws IOException {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  conf.set(MRJobConfig.MR_AM_COMMITTER_CANCEL_TIMEOUT_MS, "1000");

  DrainDispatcher dispatcher = new DrainDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  OutputCommitter committer = Mockito.mock(OutputCommitter.class);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();
  //Job has only 1 mapper task. No reducers
  conf.setInt(MRJobConfig.NUM_REDUCES, 0);
  conf.setInt(MRJobConfig.MAP_MAX_ATTEMPTS, 1);
  JobImpl job = createRunningStubbedJob(conf, dispatcher, 1, null);

  //Fail / finish all the tasks. This should land the JobImpl directly in the
  //FAIL_ABORT state
  for(Task t: job.tasks.values()) {
    TaskImpl task = (TaskImpl) t;
    task.handle(new TaskEvent(task.getID(), TaskEventType.T_SCHEDULE));
    for(TaskAttempt ta: task.getAttempts().values()) {
      task.handle(new TaskTAttemptEvent(ta.getID(),
        TaskEventType.T_ATTEMPT_FAILED));
    }
  }

  dispatcher.await();
  //Verify abortJob is called once and the job failed
  Mockito.verify(committer, Mockito.timeout(2000).times(1))
    .abortJob((JobContext) Mockito.any(), (State) Mockito.any());
  assertJobState(job, JobStateInternal.FAILED);

  dispatcher.stop();
}
项目：hadoop    文件：TestJobImpl.java    
@Test(timeout=20000)
public void testKilledDuringKillAbort() throws Exception {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  OutputCommitter committer = new StubbedOutputCommitter() {
    @Override
    public synchronized void abortJob(JobContext jobContext, State state)
        throws IOException {
      while (!Thread.interrupted()) {
        try {
          wait();
        } catch (InterruptedException e) {
        }
      }
    }
  };
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  JobImpl job = createStubbedJob(conf, dispatcher, 2, null);
  JobId jobId = job.getID();
  job.handle(new JobEvent(jobId, JobEventType.JOB_INIT));
  assertJobState(job, JobStateInternal.INITED);
  job.handle(new JobStartEvent(jobId));
  assertJobState(job, JobStateInternal.SETUP);

  job.handle(new JobEvent(jobId, JobEventType.JOB_KILL));
  assertJobState(job, JobStateInternal.KILL_ABORT);

  job.handle(new JobEvent(jobId, JobEventType.JOB_KILL));
  assertJobState(job, JobStateInternal.KILLED);
  dispatcher.stop();
  commitHandler.stop();
}