Merge -r 1357722:1357723 from trunk to branch. FIXES: MAPREDUCE-4355
git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-2@1357728 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
parent
d222e91bfa
commit
c25a45dc52
|
@ -20,6 +20,8 @@ Release 2.0.1-alpha - UNRELEASED
|
||||||
MAPREDUCE-4253. Tests for mapreduce-client-core are lying under
|
MAPREDUCE-4253. Tests for mapreduce-client-core are lying under
|
||||||
mapreduce-client-jobclient (Tsuyoshi Ozawa via harsh)
|
mapreduce-client-jobclient (Tsuyoshi Ozawa via harsh)
|
||||||
|
|
||||||
|
MAPREDUCE-4355. Add RunningJob.getJobStatus() (kkambatl via tucu)
|
||||||
|
|
||||||
OPTIMIZATIONS
|
OPTIMIZATIONS
|
||||||
|
|
||||||
BUG FIXES
|
BUG FIXES
|
||||||
|
|
|
@ -441,6 +441,14 @@ public class JobClient extends CLI {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public JobStatus getJobStatus() throws IOException {
|
||||||
|
try {
|
||||||
|
return JobStatus.downgrade(job.getStatus());
|
||||||
|
} catch (InterruptedException ie) {
|
||||||
|
throw new IOException(ie);
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
|
@ -149,8 +149,16 @@ public interface RunningJob {
|
||||||
public int getJobState() throws IOException;
|
public int getJobState() throws IOException;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Kill the running job. Blocks until all job tasks have been
|
* Returns a snapshot of the current status, {@link JobStatus}, of the Job.
|
||||||
* killed as well. If the job is no longer running, it simply returns.
|
* Need to call again for latest information.
|
||||||
|
*
|
||||||
|
* @throws IOException
|
||||||
|
*/
|
||||||
|
public JobStatus getJobStatus() throws IOException;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Kill the running job. Blocks until all job tasks have been killed as well.
|
||||||
|
* If the job is no longer running, it simply returns.
|
||||||
*
|
*
|
||||||
* @throws IOException
|
* @throws IOException
|
||||||
*/
|
*/
|
||||||
|
|
|
@ -18,16 +18,31 @@
|
||||||
|
|
||||||
package org.apache.hadoop.mapred;
|
package org.apache.hadoop.mapred;
|
||||||
|
|
||||||
import static org.junit.Assert.*;
|
import static org.junit.Assert.assertEquals;
|
||||||
|
import static org.junit.Assert.assertNull;
|
||||||
|
import static org.mockito.Mockito.mock;
|
||||||
|
import static org.mockito.Mockito.verify;
|
||||||
|
import static org.mockito.Mockito.when;
|
||||||
|
|
||||||
import java.util.List;
|
import java.io.File;
|
||||||
|
import java.io.IOException;
|
||||||
|
|
||||||
|
import org.apache.hadoop.conf.Configuration;
|
||||||
|
import org.apache.hadoop.fs.FSDataOutputStream;
|
||||||
|
import org.apache.hadoop.fs.FileSystem;
|
||||||
|
import org.apache.hadoop.fs.Path;
|
||||||
|
import org.apache.hadoop.mapred.lib.IdentityMapper;
|
||||||
|
import org.apache.hadoop.mapred.lib.IdentityReducer;
|
||||||
import org.apache.hadoop.mapreduce.Job;
|
import org.apache.hadoop.mapreduce.Job;
|
||||||
import org.junit.Test;
|
import org.junit.Test;
|
||||||
import static org.mockito.Mockito.*;
|
|
||||||
|
|
||||||
|
|
||||||
public class TestNetworkedJob {
|
public class TestNetworkedJob {
|
||||||
|
private static String TEST_ROOT_DIR = new File(System.getProperty(
|
||||||
|
"test.build.data", "/tmp")).toURI().toString().replace(' ', '+');
|
||||||
|
private static Path testDir = new Path(TEST_ROOT_DIR + "/test_mini_mr_local");
|
||||||
|
private static Path inFile = new Path(testDir, "in");
|
||||||
|
private static Path outDir = new Path(testDir, "out");
|
||||||
|
|
||||||
@SuppressWarnings("deprecation")
|
@SuppressWarnings("deprecation")
|
||||||
@Test
|
@Test
|
||||||
|
@ -41,4 +56,53 @@ public class TestNetworkedJob {
|
||||||
//verification
|
//verification
|
||||||
verify(mockJob).getCounters();
|
verify(mockJob).getCounters();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void testGetJobStatus() throws IOException, InterruptedException,
|
||||||
|
ClassNotFoundException {
|
||||||
|
MiniMRClientCluster mr = null;
|
||||||
|
FileSystem fileSys = null;
|
||||||
|
|
||||||
|
try {
|
||||||
|
mr = MiniMRClientClusterFactory.create(this.getClass(), 2,
|
||||||
|
new Configuration());
|
||||||
|
|
||||||
|
JobConf job = new JobConf(mr.getConfig());
|
||||||
|
|
||||||
|
fileSys = FileSystem.get(job);
|
||||||
|
fileSys.delete(testDir, true);
|
||||||
|
FSDataOutputStream out = fileSys.create(inFile, true);
|
||||||
|
out.writeBytes("This is a test file");
|
||||||
|
out.close();
|
||||||
|
|
||||||
|
FileInputFormat.setInputPaths(job, inFile);
|
||||||
|
FileOutputFormat.setOutputPath(job, outDir);
|
||||||
|
|
||||||
|
job.setInputFormat(TextInputFormat.class);
|
||||||
|
job.setOutputFormat(TextOutputFormat.class);
|
||||||
|
|
||||||
|
job.setMapperClass(IdentityMapper.class);
|
||||||
|
job.setReducerClass(IdentityReducer.class);
|
||||||
|
job.setNumReduceTasks(0);
|
||||||
|
|
||||||
|
JobClient client = new JobClient(mr.getConfig());
|
||||||
|
RunningJob rj = client.submitJob(job);
|
||||||
|
JobID jobId = rj.getID();
|
||||||
|
|
||||||
|
// The following asserts read JobStatus twice and ensure the returned
|
||||||
|
// JobStatus objects correspond to the same Job.
|
||||||
|
assertEquals("Expected matching JobIDs", jobId, client.getJob(jobId)
|
||||||
|
.getJobStatus().getJobID());
|
||||||
|
assertEquals("Expected matching startTimes", rj.getJobStatus()
|
||||||
|
.getStartTime(), client.getJob(jobId).getJobStatus()
|
||||||
|
.getStartTime());
|
||||||
|
} finally {
|
||||||
|
if (fileSys != null) {
|
||||||
|
fileSys.delete(testDir, true);
|
||||||
|
}
|
||||||
|
if (mr != null) {
|
||||||
|
mr.stop();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in New Issue