下面列出了怎么用org.apache.hadoop.mapreduce.JobStatus.State的API类实例代码及写法，或者点击链接到github查看源代码。
@Test(timeout=60000)
public void testJobKillTimeout() throws Exception {
long timeToWaitBeforeHardKill =
10000 + MRJobConfig.DEFAULT_MR_AM_HARD_KILL_TIMEOUT_MS;
conf.setLong(MRJobConfig.MR_AM_HARD_KILL_TIMEOUT_MS,
timeToWaitBeforeHardKill);
clientDelegate = mock(ClientServiceDelegate.class);
doAnswer(
new Answer<ClientServiceDelegate>() {
@Override
public ClientServiceDelegate answer(InvocationOnMock invocation)
throws Throwable {
return clientDelegate;
}
}
).when(clientCache).getClient(any(JobID.class));
when(clientDelegate.getJobStatus(any(JobID.class))).thenReturn(new
org.apache.hadoop.mapreduce.JobStatus(jobId, 0f, 0f, 0f, 0f,
State.RUNNING, JobPriority.HIGH, "tmp", "tmp", "tmp", "tmp"));
long startTimeMillis = System.currentTimeMillis();
yarnRunner.killJob(jobId);
assertTrue("killJob should have waited at least " + timeToWaitBeforeHardKill
+ " ms.", System.currentTimeMillis() - startTimeMillis
>= timeToWaitBeforeHardKill);
}
public static TaskAttemptState toYarn(
org.apache.hadoop.mapred.TaskStatus.State state) {
switch (state) {
case COMMIT_PENDING:
return TaskAttemptState.COMMIT_PENDING;
case FAILED:
case FAILED_UNCLEAN:
return TaskAttemptState.FAILED;
case KILLED:
case KILLED_UNCLEAN:
return TaskAttemptState.KILLED;
case RUNNING:
return TaskAttemptState.RUNNING;
case SUCCEEDED:
return TaskAttemptState.SUCCEEDED;
case UNASSIGNED:
return TaskAttemptState.STARTING;
default:
throw new YarnRuntimeException("Unrecognized State: " + state);
}
}
public static State fromYarn(YarnApplicationState yarnApplicationState,
FinalApplicationStatus finalApplicationStatus) {
switch (yarnApplicationState) {
case NEW:
case NEW_SAVING:
case SUBMITTED:
case ACCEPTED:
return State.PREP;
case RUNNING:
return State.RUNNING;
case FINISHED:
if (finalApplicationStatus == FinalApplicationStatus.SUCCEEDED) {
return State.SUCCEEDED;
} else if (finalApplicationStatus == FinalApplicationStatus.KILLED) {
return State.KILLED;
}
case FAILED:
return State.FAILED;
case KILLED:
return State.KILLED;
}
throw new YarnRuntimeException("Unrecognized application state: " + yarnApplicationState);
}
@Test
public void testEnums() throws Exception {
for (YarnApplicationState applicationState : YarnApplicationState.values()) {
TypeConverter.fromYarn(applicationState, FinalApplicationStatus.FAILED);
}
// ad hoc test of NEW_SAVING, which is newly added
Assert.assertEquals(State.PREP, TypeConverter.fromYarn(
YarnApplicationState.NEW_SAVING, FinalApplicationStatus.FAILED));
for (TaskType taskType : TaskType.values()) {
TypeConverter.fromYarn(taskType);
}
for (JobState jobState : JobState.values()) {
TypeConverter.fromYarn(jobState);
}
for (QueueState queueState : QueueState.values()) {
TypeConverter.fromYarn(queueState);
}
for (TaskState taskState : TaskState.values()) {
TypeConverter.fromYarn(taskState);
}
}
@Test
public void testJobToString() throws IOException, InterruptedException {
Cluster cluster = mock(Cluster.class);
ClientProtocol client = mock(ClientProtocol.class);
when(cluster.getClient()).thenReturn(client);
JobID jobid = new JobID("1014873536921", 6);
JobStatus status = new JobStatus(jobid, 0.0f, 0.0f, 0.0f, 0.0f,
State.FAILED, JobPriority.NORMAL, "root", "TestJobToString",
"job file", "tracking url");
when(client.getJobStatus(jobid)).thenReturn(status);
when(client.getTaskReports(jobid, TaskType.MAP)).thenReturn(
new TaskReport[0]);
when(client.getTaskReports(jobid, TaskType.REDUCE)).thenReturn(
new TaskReport[0]);
when(client.getTaskCompletionEvents(jobid, 0, 10)).thenReturn(
new TaskCompletionEvent[0]);
Job job = Job.getInstance(cluster, status, new JobConf());
Assert.assertNotNull(job.toString());
}
@Test(timeout=60000)
public void testJobKillTimeout() throws Exception {
long timeToWaitBeforeHardKill =
10000 + MRJobConfig.DEFAULT_MR_AM_HARD_KILL_TIMEOUT_MS;
conf.setLong(MRJobConfig.MR_AM_HARD_KILL_TIMEOUT_MS,
timeToWaitBeforeHardKill);
clientDelegate = mock(ClientServiceDelegate.class);
doAnswer(
new Answer<ClientServiceDelegate>() {
@Override
public ClientServiceDelegate answer(InvocationOnMock invocation)
throws Throwable {
return clientDelegate;
}
}
).when(clientCache).getClient(any(JobID.class));
when(clientDelegate.getJobStatus(any(JobID.class))).thenReturn(new
org.apache.hadoop.mapreduce.JobStatus(jobId, 0f, 0f, 0f, 0f,
State.RUNNING, JobPriority.HIGH, "tmp", "tmp", "tmp", "tmp"));
long startTimeMillis = System.currentTimeMillis();
yarnRunner.killJob(jobId);
assertTrue("killJob should have waited at least " + timeToWaitBeforeHardKill
+ " ms.", System.currentTimeMillis() - startTimeMillis
>= timeToWaitBeforeHardKill);
}
public static TaskAttemptState toYarn(
org.apache.hadoop.mapred.TaskStatus.State state) {
switch (state) {
case COMMIT_PENDING:
return TaskAttemptState.COMMIT_PENDING;
case FAILED:
case FAILED_UNCLEAN:
return TaskAttemptState.FAILED;
case KILLED:
case KILLED_UNCLEAN:
return TaskAttemptState.KILLED;
case RUNNING:
return TaskAttemptState.RUNNING;
case SUCCEEDED:
return TaskAttemptState.SUCCEEDED;
case UNASSIGNED:
return TaskAttemptState.STARTING;
default:
throw new YarnRuntimeException("Unrecognized State: " + state);
}
}
public static State fromYarn(YarnApplicationState yarnApplicationState,
FinalApplicationStatus finalApplicationStatus) {
switch (yarnApplicationState) {
case NEW:
case NEW_SAVING:
case SUBMITTED:
case ACCEPTED:
return State.PREP;
case RUNNING:
return State.RUNNING;
case FINISHED:
if (finalApplicationStatus == FinalApplicationStatus.SUCCEEDED) {
return State.SUCCEEDED;
} else if (finalApplicationStatus == FinalApplicationStatus.KILLED) {
return State.KILLED;
}
case FAILED:
return State.FAILED;
case KILLED:
return State.KILLED;
}
throw new YarnRuntimeException("Unrecognized application state: " + yarnApplicationState);
}
@Test
public void testEnums() throws Exception {
for (YarnApplicationState applicationState : YarnApplicationState.values()) {
TypeConverter.fromYarn(applicationState, FinalApplicationStatus.FAILED);
}
// ad hoc test of NEW_SAVING, which is newly added
Assert.assertEquals(State.PREP, TypeConverter.fromYarn(
YarnApplicationState.NEW_SAVING, FinalApplicationStatus.FAILED));
for (TaskType taskType : TaskType.values()) {
TypeConverter.fromYarn(taskType);
}
for (JobState jobState : JobState.values()) {
TypeConverter.fromYarn(jobState);
}
for (QueueState queueState : QueueState.values()) {
TypeConverter.fromYarn(queueState);
}
for (TaskState taskState : TaskState.values()) {
TypeConverter.fromYarn(taskState);
}
}
@Test
public void testJobToString() throws IOException, InterruptedException {
Cluster cluster = mock(Cluster.class);
ClientProtocol client = mock(ClientProtocol.class);
when(cluster.getClient()).thenReturn(client);
JobID jobid = new JobID("1014873536921", 6);
JobStatus status = new JobStatus(jobid, 0.0f, 0.0f, 0.0f, 0.0f,
State.FAILED, JobPriority.NORMAL, "root", "TestJobToString",
"job file", "tracking url");
when(client.getJobStatus(jobid)).thenReturn(status);
when(client.getTaskReports(jobid, TaskType.MAP)).thenReturn(
new TaskReport[0]);
when(client.getTaskReports(jobid, TaskType.REDUCE)).thenReturn(
new TaskReport[0]);
when(client.getTaskCompletionEvents(jobid, 0, 10)).thenReturn(
new TaskCompletionEvent[0]);
Job job = Job.getInstance(cluster, status, new JobConf());
Assert.assertNotNull(job.toString());
}
@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();
}
@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();
}
@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();
}
@Test(timeout=20000)
public void testJobKill() throws Exception {
clientDelegate = mock(ClientServiceDelegate.class);
when(clientDelegate.getJobStatus(any(JobID.class))).thenReturn(new
org.apache.hadoop.mapreduce.JobStatus(jobId, 0f, 0f, 0f, 0f,
State.PREP, JobPriority.HIGH, "tmp", "tmp", "tmp", "tmp"));
when(clientDelegate.killJob(any(JobID.class))).thenReturn(true);
doAnswer(
new Answer<ClientServiceDelegate>() {
@Override
public ClientServiceDelegate answer(InvocationOnMock invocation)
throws Throwable {
return clientDelegate;
}
}
).when(clientCache).getClient(any(JobID.class));
yarnRunner.killJob(jobId);
verify(resourceMgrDelegate).killApplication(appId);
when(clientDelegate.getJobStatus(any(JobID.class))).thenReturn(new
org.apache.hadoop.mapreduce.JobStatus(jobId, 0f, 0f, 0f, 0f,
State.RUNNING, JobPriority.HIGH, "tmp", "tmp", "tmp", "tmp"));
yarnRunner.killJob(jobId);
verify(clientDelegate).killJob(jobId);
when(clientDelegate.getJobStatus(any(JobID.class))).thenReturn(null);
when(resourceMgrDelegate.getApplicationReport(any(ApplicationId.class)))
.thenReturn(
ApplicationReport.newInstance(appId, null, "tmp", "tmp", "tmp",
"tmp", 0, null, YarnApplicationState.FINISHED, "tmp", "tmp",
0l, 0l, FinalApplicationStatus.SUCCEEDED, null, null, 0f,
"tmp", null));
yarnRunner.killJob(jobId);
verify(clientDelegate).killJob(jobId);
}
@Test
public void tesAllJobs() throws Exception {
final ApplicationClientProtocol applicationsManager = Mockito.mock(ApplicationClientProtocol.class);
GetApplicationsResponse allApplicationsResponse = Records
.newRecord(GetApplicationsResponse.class);
List<ApplicationReport> applications = new ArrayList<ApplicationReport>();
applications.add(getApplicationReport(YarnApplicationState.FINISHED,
FinalApplicationStatus.FAILED));
applications.add(getApplicationReport(YarnApplicationState.FINISHED,
FinalApplicationStatus.SUCCEEDED));
applications.add(getApplicationReport(YarnApplicationState.FINISHED,
FinalApplicationStatus.KILLED));
applications.add(getApplicationReport(YarnApplicationState.FAILED,
FinalApplicationStatus.FAILED));
allApplicationsResponse.setApplicationList(applications);
Mockito.when(
applicationsManager.getApplications(Mockito
.any(GetApplicationsRequest.class))).thenReturn(
allApplicationsResponse);
ResourceMgrDelegate resourceMgrDelegate = new ResourceMgrDelegate(
new YarnConfiguration()) {
@Override
protected void serviceStart() throws Exception {
Assert.assertTrue(this.client instanceof YarnClientImpl);
((YarnClientImpl) this.client).setRMClient(applicationsManager);
}
};
JobStatus[] allJobs = resourceMgrDelegate.getAllJobs();
Assert.assertEquals(State.FAILED, allJobs[0].getState());
Assert.assertEquals(State.SUCCEEDED, allJobs[1].getState());
Assert.assertEquals(State.KILLED, allJobs[2].getState());
Assert.assertEquals(State.FAILED, allJobs[3].getState());
}
@Test(timeout = 5000)
public void testTaskID() throws IOException, InterruptedException {
JobID jobid = new JobID("1014873536921", 6);
TaskID tid = new TaskID(jobid, TaskType.MAP, 0);
org.apache.hadoop.mapred.TaskID tid1 =
org.apache.hadoop.mapred.TaskID.downgrade(tid);
org.apache.hadoop.mapred.TaskReport treport =
new org.apache.hadoop.mapred.TaskReport(tid1, 0.0f,
State.FAILED.toString(), null, TIPStatus.FAILED, 100, 100,
new org.apache.hadoop.mapred.Counters());
Assert
.assertEquals(treport.getTaskId(), "task_1014873536921_0006_m_000000");
Assert.assertEquals(treport.getTaskID().toString(),
"task_1014873536921_0006_m_000000");
}
@Test
public void testJobKIll() throws Exception {
Cluster mockCluster = mock(Cluster.class);
CLI cli = spy(new CLI());
doReturn(mockCluster).when(cli).createCluster();
String jobId1 = "job_1234654654_001";
String jobId2 = "job_1234654654_002";
String jobId3 = "job_1234654654_003";
String jobId4 = "job_1234654654_004";
Job mockJob1 = mockJob(mockCluster, jobId1, State.RUNNING);
Job mockJob2 = mockJob(mockCluster, jobId2, State.KILLED);
Job mockJob3 = mockJob(mockCluster, jobId3, State.FAILED);
Job mockJob4 = mockJob(mockCluster, jobId4, State.PREP);
int exitCode1 = cli.run(new String[] { "-kill", jobId1 });
assertEquals(0, exitCode1);
verify(mockJob1, times(1)).killJob();
int exitCode2 = cli.run(new String[] { "-kill", jobId2 });
assertEquals(-1, exitCode2);
verify(mockJob2, times(0)).killJob();
int exitCode3 = cli.run(new String[] { "-kill", jobId3 });
assertEquals(-1, exitCode3);
verify(mockJob3, times(0)).killJob();
int exitCode4 = cli.run(new String[] { "-kill", jobId4 });
assertEquals(0, exitCode4);
verify(mockJob4, times(1)).killJob();
}
@Before
public void setUp() throws IOException {
conf = new Configuration();
clientProtocol = mock(ClientProtocol.class);
Cluster cluster = mock(Cluster.class);
when(cluster.getConf()).thenReturn(conf);
when(cluster.getClient()).thenReturn(clientProtocol);
JobStatus jobStatus = new JobStatus(new JobID("job_000", 1), 0f, 0f, 0f, 0f,
State.RUNNING, JobPriority.HIGH, "tmp-user", "tmp-jobname",
"tmp-jobfile", "tmp-url");
job = Job.getInstance(cluster, jobStatus, conf);
job = spy(job);
}
@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();
}
@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();
}
@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();
}
@Test(timeout=20000)
public void testJobKill() throws Exception {
clientDelegate = mock(ClientServiceDelegate.class);
when(clientDelegate.getJobStatus(any(JobID.class))).thenReturn(new
org.apache.hadoop.mapreduce.JobStatus(jobId, 0f, 0f, 0f, 0f,
State.PREP, JobPriority.HIGH, "tmp", "tmp", "tmp", "tmp"));
when(clientDelegate.killJob(any(JobID.class))).thenReturn(true);
doAnswer(
new Answer<ClientServiceDelegate>() {
@Override
public ClientServiceDelegate answer(InvocationOnMock invocation)
throws Throwable {
return clientDelegate;
}
}
).when(clientCache).getClient(any(JobID.class));
yarnRunner.killJob(jobId);
verify(resourceMgrDelegate).killApplication(appId);
when(clientDelegate.getJobStatus(any(JobID.class))).thenReturn(new
org.apache.hadoop.mapreduce.JobStatus(jobId, 0f, 0f, 0f, 0f,
State.RUNNING, JobPriority.HIGH, "tmp", "tmp", "tmp", "tmp"));
yarnRunner.killJob(jobId);
verify(clientDelegate).killJob(jobId);
when(clientDelegate.getJobStatus(any(JobID.class))).thenReturn(null);
when(resourceMgrDelegate.getApplicationReport(any(ApplicationId.class)))
.thenReturn(
ApplicationReport.newInstance(appId, null, "tmp", "tmp", "tmp",
"tmp", 0, null, YarnApplicationState.FINISHED, "tmp", "tmp",
0l, 0l, FinalApplicationStatus.SUCCEEDED, null, null, 0f,
"tmp", null));
yarnRunner.killJob(jobId);
verify(clientDelegate).killJob(jobId);
}
@Test
public void tesAllJobs() throws Exception {
final ApplicationClientProtocol applicationsManager = Mockito.mock(ApplicationClientProtocol.class);
GetApplicationsResponse allApplicationsResponse = Records
.newRecord(GetApplicationsResponse.class);
List<ApplicationReport> applications = new ArrayList<ApplicationReport>();
applications.add(getApplicationReport(YarnApplicationState.FINISHED,
FinalApplicationStatus.FAILED));
applications.add(getApplicationReport(YarnApplicationState.FINISHED,
FinalApplicationStatus.SUCCEEDED));
applications.add(getApplicationReport(YarnApplicationState.FINISHED,
FinalApplicationStatus.KILLED));
applications.add(getApplicationReport(YarnApplicationState.FAILED,
FinalApplicationStatus.FAILED));
allApplicationsResponse.setApplicationList(applications);
Mockito.when(
applicationsManager.getApplications(Mockito
.any(GetApplicationsRequest.class))).thenReturn(
allApplicationsResponse);
ResourceMgrDelegate resourceMgrDelegate = new ResourceMgrDelegate(
new YarnConfiguration()) {
@Override
protected void serviceStart() throws Exception {
Assert.assertTrue(this.client instanceof YarnClientImpl);
((YarnClientImpl) this.client).setRMClient(applicationsManager);
}
};
JobStatus[] allJobs = resourceMgrDelegate.getAllJobs();
Assert.assertEquals(State.FAILED, allJobs[0].getState());
Assert.assertEquals(State.SUCCEEDED, allJobs[1].getState());
Assert.assertEquals(State.KILLED, allJobs[2].getState());
Assert.assertEquals(State.FAILED, allJobs[3].getState());
}
@Test(timeout = 5000)
public void testTaskID() throws IOException, InterruptedException {
JobID jobid = new JobID("1014873536921", 6);
TaskID tid = new TaskID(jobid, TaskType.MAP, 0);
org.apache.hadoop.mapred.TaskID tid1 =
org.apache.hadoop.mapred.TaskID.downgrade(tid);
org.apache.hadoop.mapred.TaskReport treport =
new org.apache.hadoop.mapred.TaskReport(tid1, 0.0f,
State.FAILED.toString(), null, TIPStatus.FAILED, 100, 100,
new org.apache.hadoop.mapred.Counters());
Assert
.assertEquals(treport.getTaskId(), "task_1014873536921_0006_m_000000");
Assert.assertEquals(treport.getTaskID().toString(),
"task_1014873536921_0006_m_000000");
}
@Test
public void testJobKIll() throws Exception {
Cluster mockCluster = mock(Cluster.class);
CLI cli = spy(new CLI());
doReturn(mockCluster).when(cli).createCluster();
String jobId1 = "job_1234654654_001";
String jobId2 = "job_1234654654_002";
String jobId3 = "job_1234654654_003";
String jobId4 = "job_1234654654_004";
Job mockJob1 = mockJob(mockCluster, jobId1, State.RUNNING);
Job mockJob2 = mockJob(mockCluster, jobId2, State.KILLED);
Job mockJob3 = mockJob(mockCluster, jobId3, State.FAILED);
Job mockJob4 = mockJob(mockCluster, jobId4, State.PREP);
int exitCode1 = cli.run(new String[] { "-kill", jobId1 });
assertEquals(0, exitCode1);
verify(mockJob1, times(1)).killJob();
int exitCode2 = cli.run(new String[] { "-kill", jobId2 });
assertEquals(-1, exitCode2);
verify(mockJob2, times(0)).killJob();
int exitCode3 = cli.run(new String[] { "-kill", jobId3 });
assertEquals(-1, exitCode3);
verify(mockJob3, times(0)).killJob();
int exitCode4 = cli.run(new String[] { "-kill", jobId4 });
assertEquals(0, exitCode4);
verify(mockJob4, times(1)).killJob();
}
@Before
public void setUp() throws IOException {
conf = new Configuration();
clientProtocol = mock(ClientProtocol.class);
Cluster cluster = mock(Cluster.class);
when(cluster.getConf()).thenReturn(conf);
when(cluster.getClient()).thenReturn(clientProtocol);
JobStatus jobStatus = new JobStatus(new JobID("job_000", 1), 0f, 0f, 0f, 0f,
State.RUNNING, JobPriority.HIGH, "tmp-user", "tmp-jobname",
"tmp-jobfile", "tmp-url");
job = Job.getInstance(cluster, jobStatus, conf);
job = spy(job);
}
/** Test that cleanup actually cleans up. */
@Test
public void testAbortJob() throws IOException {
// Setup the sample directory.
generateSampleFiles();
committer.abortJob(mockTaskAttemptContext, State.KILLED);
// Ensure files are deleted by cleanup.
assertThat(!ghfs.exists(outputPath)).isTrue();
assertThat(!ghfs.exists(outputSampleFilePath)).isTrue();
verify(mockCommitter).abortJob(eq(mockTaskAttemptContext), eq(State.KILLED));
}
/** Test that cleanup actually cleans up. */
@Test
public void testAbortJob() throws IOException {
// Setup the sample directory.
generateSampleFiles();
committer.abortJob(mockTaskAttemptContext, State.KILLED);
// Ensure files are deleted by cleanup.
assertThat(!ghfs.exists(outputPath)).isTrue();
assertThat(!ghfs.exists(outputSampleFilePath)).isTrue();
verify(mockCommitter).abortJob(eq(mockTaskAttemptContext), eq(State.KILLED));
}
/** Test to ensure the underlying delegate is being passed the abortJob call. */
@Test
public void testAbortJob() throws IOException {
committer.abortJob(mockTaskAttemptContext, State.KILLED);
// Verify the delegate is being called.
verify(mockCommitter).abortJob(eq(mockTaskAttemptContext), eq(State.KILLED));
}
@Override
public void abortJob(JobContext jobContext, State state) throws IOException {
LOG.info("Abort Job [{0}]", jobContext.getJobID());
Configuration configuration = jobContext.getConfiguration();
Path tableOutput = BlurOutputFormat.getOutputPath(configuration);
makeSureNoEmptyShards(configuration, tableOutput);
FileSystem fileSystem = tableOutput.getFileSystem(configuration);
for (FileStatus fileStatus : fileSystem.listStatus(tableOutput)) {
if (isShard(fileStatus)) {
commitOrAbortJob(jobContext, fileStatus.getPath(), false);
}
}
}