Author: omalley
Date: Tue Mar 8 05:59:48 2011
New Revision: 1079244
URL: http://svn.apache.org/viewvc?rev=1079244&view=rev
Log:
commit 754898ddd5a747c642deb08d9ed55f88832c6f2b
Author: Greg Roelofs <[email protected]>
Date: Wed Jan 12 12:59:05 2011 -0800
UberTask/MR-1220 "v10" changes: minor comment fixes to JIP and 10 more unit
tests fixed.
Modified:
hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/JobInProgress.java
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestJobHistory.java
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestJobRetire.java
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestKillSubProcesses.java
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestMiniMRLocalFS.java
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestMiniMRWithDFS.java
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestMultipleLevelCaching.java
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestNodeRefresh.java
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapreduce/TestNoJobSetupCleanup.java
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/security/authorize/TestServiceLevelAuthorization.java
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/tools/rumen/TestRumenJobTraces.java
Modified:
hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/JobInProgress.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/JobInProgress.java?rev=1079244&r1=1079243&r2=1079244&view=diff
==============================================================================
---
hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/JobInProgress.java
(original)
+++
hadoop/mapreduce/branches/yahoo-merge/src/java/org/apache/hadoop/mapred/JobInProgress.java
Tue Mar 8 05:59:48 2011
@@ -1265,6 +1265,7 @@ public class JobInProgress {
}
// Tell the job to fail the relevant task
+ // [this is the private failedTask()]
failedTask(tip, taskid, status, taskTracker,
wasRunning, wasComplete, wasAttemptRunning);
@@ -3332,7 +3333,7 @@ public class JobInProgress {
// We do it once per TIP and that too for the task that fails the TIP
if (!wasFailed && tip.isFailed()) {
//
- // Allow upto 'mapFailuresPercent' of map tasks to fail or
+ // Allow up to 'mapFailuresPercent' of map tasks to fail or
// 'reduceFailuresPercent' of reduce tasks to fail
//
boolean killJob = tip.isJobCleanupTask() || tip.isJobSetupTask() ? true :
@@ -3576,7 +3577,8 @@ public class JobInProgress {
LOG.info("Too many fetch-failures for output of task: " + mapTaskId
+ " ... killing it");
- failedTask(tip, mapTaskId, "Too many fetch-failures",
+ // [this is the public failedTask()]
+ failedTask(tip, mapTaskId, "Too many fetch-failures",
(tip.isMapTask() ? TaskStatus.Phase.MAP :
TaskStatus.Phase.REDUCE),
TaskStatus.State.FAILED, trackerName);
Modified:
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestJobHistory.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestJobHistory.java?rev=1079244&r1=1079243&r2=1079244&view=diff
==============================================================================
---
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestJobHistory.java
(original)
+++
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestJobHistory.java
Tue Mar 8 05:59:48 2011
@@ -208,10 +208,13 @@ public class TestJobHistory extends Test
if (type.equals(TaskType.REDUCE) &&
status.equals(TaskStatus.State.SUCCEEDED.toString())) {
long shuffleFinishTime = attempt.getShuffleFinishTime();
- assertTrue(startTime < shuffleFinishTime);
+ assertTrue("invalid shuffle finish time",
+ startTime < shuffleFinishTime);
long sortFinishTime = attempt.getSortFinishTime();
- assertTrue(shuffleFinishTime < sortFinishTime);
+ assertTrue(
+ "invalid sort finish time or shuffle didn't finish before sort",
+ shuffleFinishTime < sortFinishTime);
}
else if (type.equals(TaskType.MAP) &&
status.equals(TaskStatus.State.SUCCEEDED.toString())) {
@@ -665,6 +668,10 @@ public class TestJobHistory extends Test
// run the TCs
conf = mr.createJobConf();
+ // sort-vs-shuffle-time check in
validateTaskAttemptLevelKeyValuesFormat()
+ // is invalid for uberized jobs, so disable:
+ conf.setBoolean(JobContext.JOB_UBERTASK_ENABLE, false);
+
FileSystem fs = FileSystem.get(conf);
// clean up
fs.delete(new Path("succeed"), true);
@@ -757,6 +764,10 @@ public class TestJobHistory extends Test
// run the TCs
conf = mr.createJobConf();
+ // sort-vs-shuffle-time check in
validateTaskAttemptLevelKeyValuesFormat()
+ // is invalid for uberized jobs, so disable:
+ conf.setBoolean(JobContext.JOB_UBERTASK_ENABLE, false);
+
FileSystem fs = FileSystem.get(conf);
// clean up
fs.delete(new Path(TEST_ROOT_DIR + "/succeed"), true);
Modified:
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestJobRetire.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestJobRetire.java?rev=1079244&r1=1079243&r2=1079244&view=diff
==============================================================================
---
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestJobRetire.java
(original)
+++
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestJobRetire.java
Tue Mar 8 05:59:48 2011
@@ -218,6 +218,7 @@ public class TestJobRetire extends TestC
Path inDir = new Path(testDir, "in-1");
Path outDir = new Path(testDir, "out-1");
JobConf jConf = mr.createJobConf();
+ jConf.setBoolean(JobContext.JOB_UBERTASK_ENABLE, false);
FileInputFormat.setInputPaths(jConf, new Path[] {inDir});
FileOutputFormat.setOutputPath(jConf, outDir);
SleepJob sleepJob = new SleepJob();
@@ -232,7 +233,7 @@ public class TestJobRetire extends TestC
for (int i = 0; i < 100 && (jip.finishedMaps() < 1); i++) {
UtilsForTests.waitFor(1000);
}
- assertEquals(jip.finishedMaps(), 1);
+ assertEquals("wrong number of finished maps", jip.finishedMaps(), 1);
// start a tracker that will wait
LOG.info("Adding a waiting tracker");
Modified:
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestKillSubProcesses.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestKillSubProcesses.java?rev=1079244&r1=1079243&r2=1079244&view=diff
==============================================================================
---
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestKillSubProcesses.java
(original)
+++
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestKillSubProcesses.java
Tue Mar 8 05:59:48 2011
@@ -108,11 +108,13 @@ public class TestKillSubProcesses extend
conf.setJobName("testfailjobsubprocesses");
conf.setMapperClass(FailingMapperWithChildren.class);
-
- // We don't want to run the failing map task 4 times. So we run it once and
- // check if all the subprocesses are killed properly.
+
+ // We don't want to run the failing map task 4 times [and the test doesn't
+ // work correctly in that case anyway]. So we run it once and check if all
+ // the subprocesses are killed properly.
conf.setMaxMapAttempts(1);
-
+ conf.setMaxReduceAttempts(1); // for uberized jobs
+
RunningJob job = runJobAndSetProcessHandle(jt, conf);
signalTask(signalFile.toString(), conf);
validateKillingSubprocesses(job, conf);
@@ -156,18 +158,23 @@ public class TestKillSubProcesses extend
pid = null;
jobClient = new JobClient(conf);
-
- // get the taskAttemptID of the map task and use it to get the pid
- // of map task
- TaskReport[] mapReports = jobClient.getMapTaskReports(job.getID());
-
JobInProgress jip = jt.getJob(job.getID());
- for(TaskReport tr : mapReports) {
+ boolean isUber = jip.isUber();
+
+ // get the taskAttemptID of the map task (or ubertask) and use it to get
+ // the task's pid
+ TaskReport[] taskReports = isUber
+ ? jobClient.getReduceTaskReports(job.getID())
+ : jobClient.getMapTaskReports(job.getID());
+
+ String taskName = isUber? "ubertask" : "map task";
+
+ for (TaskReport tr : taskReports) {
TaskInProgress tip = jip.getTaskInProgress(tr.getTaskID());
// for this tip, get active tasks of all attempts
- while(tip.getActiveTasks().size() == 0) {
- //wait till the activeTasks Tree is built
+ while (tip.getActiveTasks().size() == 0) {
+ // wait until the activeTasks Tree is built
try {
Thread.sleep(500);
} catch (InterruptedException ie) {
@@ -179,7 +186,7 @@ public class TestKillSubProcesses extend
for (Iterator<TaskAttemptID> it =
tip.getActiveTasks().keySet().iterator(); it.hasNext();) {
TaskAttemptID id = it.next();
- LOG.info("taskAttemptID of map task is " + id);
+ LOG.info("taskAttemptID of " + taskName + " is " + id);
while(pid == null) {
pid = mr.getTaskTrackerRunner(0).getTaskTracker().getPid(id);
@@ -189,10 +196,11 @@ public class TestKillSubProcesses extend
} catch(InterruptedException e) {}
}
}
- LOG.info("pid of map task is " + pid);
- //Checking if the map task is alive
- assertTrue("Map is no more alive", isAlive(pid));
- LOG.info("The map task is alive before Job completion, as expected.");
+ LOG.info("pid of " + taskName + " is " + pid);
+ //Checking if the task is alive
+ assertTrue(taskName + " is no longer alive", isAlive(pid));
+ LOG.info("The " + taskName +
+ " is alive before Job completion, as expected.");
}
}
@@ -220,8 +228,8 @@ public class TestKillSubProcesses extend
childPid = TestProcfsBasedProcessTree.getPidFromPidFile(
scriptDirName + "/childPidFile" + i);
LOG.info("pid of the descendant process at level " + i +
- "in the subtree of processes(with the map task as the root)" +
- " is " + childPid);
+ "in the subtree of processes(with the " + taskName +
+ " as the root) is " + childPid);
assertTrue("Unexpected: The subprocess at level " + i +
" in the subtree is not alive before Job completion",
isAlive(childPid));
@@ -301,6 +309,10 @@ public class TestKillSubProcesses extend
conf.get(JobConf.MAPRED_REDUCE_TASK_JAVA_OPTS,
conf.get(JobConf.MAPRED_TASK_JAVA_OPTS)) +
" -Dtest.build.data=" + BASE_TEST_ROOT_DIR);
+ conf.set(JobConf.MAPRED_UBER_TASK_JAVA_OPTS,
+ conf.get(JobConf.MAPRED_UBER_TASK_JAVA_OPTS,
+ conf.get(JobConf.MAPRED_TASK_JAVA_OPTS)) +
+ " -Dtest.build.data=" + BASE_TEST_ROOT_DIR);
return UtilsForTests.runJob(conf, inDir, outDir);
}
@@ -397,12 +409,12 @@ public class TestKillSubProcesses extend
// Set executable permissions on the script.
new File(scriptPath.toUri().getPath()).setExecutable(true);
- LOG.info("Calling script from map task : " + shellScript);
- Runtime.getRuntime()
- .exec(shellScript + " " + numLevelsOfSubProcesses);
-
- String childPid =
TestProcfsBasedProcessTree.getPidFromPidFile(scriptDirName
- + "/childPidFile" + 0);
+ LOG.info("Calling script from map task with " + numLevelsOfSubProcesses
+ + " levels of subprocesses: " + shellScript);
+ Runtime.getRuntime().exec(shellScript + " " + numLevelsOfSubProcesses);
+
+ String childPid = TestProcfsBasedProcessTree.getPidFromPidFile(
+ scriptDirName + "/childPidFile" + 0);
while (childPid == null) {
LOG.warn(scriptDirName + "/childPidFile" + 0 + " is null;
Sleeping...");
try {
@@ -414,6 +426,12 @@ public class TestKillSubProcesses extend
childPid = TestProcfsBasedProcessTree.getPidFromPidFile(scriptDirName
+ "/childPidFile" + 0);
}
+ if (childPid != null) {
+ LOG.info("Successfully found childPid (" + childPid + ") in " +
+ scriptDirName + "/childPidFile" + 0);
+ } else {
+ LOG.warn("Script failed: " + scriptDirName + "/childPidFile" + 0);
+ }
}
}
@@ -428,7 +446,7 @@ public class TestKillSubProcesses extend
fs = FileSystem.getLocal(conf);
runChildren(conf);
} catch (Exception e) {
- LOG.warn("Exception in configure: " +
+ LOG.warn("Exception in MapperWithChildren.configure: " +
StringUtils.stringifyException(e));
}
}
@@ -469,13 +487,13 @@ public class TestKillSubProcesses extend
Thread.sleep(1000);
}
} catch (InterruptedException e) {
- LOG.warn("Exception in KillMapperWithChild.map:" + e);
+ LOG.warn("Exception in KillMapperWithChild.map: " + e);
}
}
}
/**
- * Mapper that fails when recieves a signal. Signal is existence of a file.
+ * Mapper that fails when it receives a signal. Signal is existence of a
file.
*/
static class FailingMapperWithChildren extends MapperWithChildren {
public void configure(JobConf conf) {
@@ -493,7 +511,7 @@ public class TestKillSubProcesses extend
}
} catch (InterruptedException ie) {
System.out.println("Interrupted while the map was waiting for "
- + " the signal.");
+ + "the signal.");
break;
}
}
Modified:
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestMiniMRLocalFS.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestMiniMRLocalFS.java?rev=1079244&r1=1079243&r2=1079244&view=diff
==============================================================================
---
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestMiniMRLocalFS.java
(original)
+++
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestMiniMRLocalFS.java
Tue Mar 8 05:59:48 2011
@@ -62,14 +62,16 @@ public class TestMiniMRLocalFS extends T
MiniMRCluster mr = null;
try {
mr = new MiniMRCluster(2, "file:///", 3);
- // make cleanup inline sothat validation of existence of these
directories
- // can be done
+ // make cleanup inline so that validation of existence of these
+ // directories can be done
mr.setInlineCleanupThreads();
TestMiniMRWithDFS.runPI(mr, mr.createJobConf());
- // run the wordcount example with caching
+ // run the wordcount example with caching but no uberization (want to
+ // verify counts of subtasks)
JobConf job = mr.createJobConf();
+ job.setBoolean(JobContext.JOB_UBERTASK_ENABLE, false);
TestResult ret = MRCaching.launchMRCache(TEST_ROOT_DIR + "/wc/input",
TEST_ROOT_DIR + "/wc/output",
TEST_ROOT_DIR + "/cachedir",
Modified:
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestMiniMRWithDFS.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestMiniMRWithDFS.java?rev=1079244&r1=1079243&r2=1079244&view=diff
==============================================================================
---
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestMiniMRWithDFS.java
(original)
+++
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestMiniMRWithDFS.java
Tue Mar 8 05:59:48 2011
@@ -108,10 +108,10 @@ public class TestMiniMRWithDFS extends T
* <br/>
* <br/>
*
- * For e.g., if we want to check the existence of *only* the directories for
- * user1's tasks job1-attempt1, job1-attempt2, job2-attempt1, we pass user1
as
- * user, {job1, job1, job2, job3} as jobIds and {attempt1, attempt2,
attempt1,
- * attempt3} as taskDirs.
+ * For example, if we want to check the existence of *only* the directories
+ * for user1's tasks job1-attempt1, job1-attempt2, job2-attempt1, we pass
+ * user1 as user, {job1, job1, job2, job3} as jobIds and {attempt1, attempt2,
+ * attempt1, attempt3} as taskDirs.
*
* @param mr the map-reduce cluster
* @param user the name of the job-owner
@@ -205,9 +205,15 @@ public class TestMiniMRWithDFS extends T
}
}
- public static void runPI(MiniMRCluster mr, JobConf jobconf)
- throws IOException, InterruptedException, ClassNotFoundException {
- LOG.info("runPI");
+ public static void runPI(MiniMRCluster mr, JobConf jobconf)
+ throws IOException, InterruptedException, ClassNotFoundException {
+ runPI(mr, jobconf, false);
+ }
+
+ public static void runPI(MiniMRCluster mr, JobConf jobconf, boolean uberize)
+ throws IOException, InterruptedException, ClassNotFoundException {
+ LOG.info("runPI (" + (uberize? "" : "non-") + "uberized)");
+ jobconf.setBoolean(JobContext.JOB_UBERTASK_ENABLE, uberize);
double estimate = org.apache.hadoop.examples.QuasiMonteCarlo.estimatePi(
NUM_MAPS, NUM_SAMPLES, jobconf).doubleValue();
double error = Math.abs(Math.PI - estimate);
@@ -216,14 +222,24 @@ public class TestMiniMRWithDFS extends T
checkTaskDirectories(mr, userName, new String[] {}, new String[] {});
}
- public static void runWordCount(MiniMRCluster mr, JobConf jobConf)
+ public static void runWordCount(MiniMRCluster mr, JobConf jobConf)
throws IOException {
- LOG.info("runWordCount");
- // Run a word count example
- // Keeping tasks that match this pattern
- String pattern =
- TaskAttemptID.getTaskAttemptIDsPattern(null, null, TaskType.MAP, 1,
null);
+ runWordCount(mr, jobConf, false);
+ }
+
+ public static void runWordCount(MiniMRCluster mr, JobConf jobConf,
+ boolean uberize)
+ throws IOException {
+ LOG.info("runWordCount (" + (uberize? "" : "non-") + "uberized)");
+ // Run a word-count example
+ // Keep task files/dirs that match one of these patterns:
+ String pattern = uberize
+ ? TaskAttemptID.getTaskAttemptIDsPattern(null, null, TaskType.REDUCE,
+ 0, null)
+ : TaskAttemptID.getTaskAttemptIDsPattern(null, null, TaskType.MAP,
+ 1, null);
jobConf.setKeepTaskFilesPattern(pattern);
+ jobConf.setBoolean(JobContext.JOB_UBERTASK_ENABLE, uberize);
TestResult result;
final Path inDir = new Path("./wc/input");
final Path outDir = new Path("./wc/output");
@@ -232,14 +248,17 @@ public class TestMiniMRWithDFS extends T
assertEquals("The\t1\nbrown\t1\nfox\t2\nhas\t1\nmany\t1\n" +
"quick\t1\nred\t1\nsilly\t1\nsox\t1\n", result.output);
JobID jobid = result.job.getID();
- TaskAttemptID taskid = new TaskAttemptID(
- new TaskID(jobid, TaskType.MAP, 1),0);
+ // taskid must match configured keep-pattern above:
+ TaskAttemptID taskid = uberize
+ ? new TaskAttemptID(new TaskID(jobid, TaskType.REDUCE, 0), 0)
+ : new TaskAttemptID(new TaskID(jobid, TaskType.MAP, 1), 0);
String userName = UserGroupInformation.getLoginUser().getUserName();
-
+
checkTaskDirectories(mr, userName, new String[] { jobid.toString() },
new String[] { taskid.toString() });
// test with maps=0
jobConf = mr.createJobConf();
+ jobConf.setBoolean(JobContext.JOB_UBERTASK_ENABLE, uberize);
input = "owen is oom";
result = launchWordCount(jobConf, inDir, outDir, input, 0, 1);
assertEquals("is\t1\noom\t1\nowen\t1\n", result.output);
@@ -254,8 +273,8 @@ public class TestMiniMRWithDFS extends T
// add the correction factor of 234 as the input split is also streamed
assertEquals(input.length() + rawSplitBytesRead, hdfsRead);
- // Run a job with input and output going to localfs even though the
- // default fs is hdfs.
+ // Run a 1-map/1-reduce job with input and output going to localfs even
+ // though the default fs is hdfs.
{
FileSystem localfs = FileSystem.getLocal(jobConf);
String TEST_ROOT_DIR =
@@ -274,7 +293,35 @@ public class TestMiniMRWithDFS extends T
}
}
- public void testWithDFS()
+ public void testWithDFSNonUber()
+ throws IOException, InterruptedException, ClassNotFoundException {
+ MiniDFSCluster dfs = null;
+ MiniMRCluster mr = null;
+ FileSystem fileSys = null;
+ try {
+ final int taskTrackers = 4;
+
+ Configuration conf = new Configuration();
+ dfs = new MiniDFSCluster(conf, 4, true, null);
+ fileSys = dfs.getFileSystem();
+ mr = new MiniMRCluster(taskTrackers, fileSys.getUri().toString(), 1);
+ // make cleanup inline so that validation of existence of these
+ // directories can be done
+ mr.setInlineCleanupThreads();
+
+ runPI(mr, mr.createJobConf(), false);
+ // regardless of order, if try to run two wordcounts here, second one
+ // gets hosed by leftover job dirs from first (i.e., triggers "Unexpected
+ // file job_xxx_0002 found" assertion in verifyContents())
+ runWordCount(mr, mr.createJobConf(), false);
+ } finally {
+ if (dfs != null) { dfs.shutdown(); }
+ if (mr != null) { mr.shutdown();
+ }
+ }
+ }
+
+ public void testWithDFSUber()
throws IOException, InterruptedException, ClassNotFoundException {
MiniDFSCluster dfs = null;
MiniMRCluster mr = null;
@@ -286,19 +333,22 @@ public class TestMiniMRWithDFS extends T
dfs = new MiniDFSCluster(conf, 4, true, null);
fileSys = dfs.getFileSystem();
mr = new MiniMRCluster(taskTrackers, fileSys.getUri().toString(), 1);
- // make cleanup inline sothat validation of existence of these
directories
- // can be done
+ // make cleanup inline so that validation of existence of these
+ // directories can be done
mr.setInlineCleanupThreads();
- runPI(mr, mr.createJobConf());
- runWordCount(mr, mr.createJobConf());
+ runPI(mr, mr.createJobConf(), true);
+ // regardless of order, if try to run two wordcounts here, second one
+ // gets hosed by leftover job dirs from first (i.e., triggers "Unexpected
+ // file job_xxx_0002 found" assertion in verifyContents())
+ runWordCount(mr, mr.createJobConf(), true);
} finally {
if (dfs != null) { dfs.shutdown(); }
if (mr != null) { mr.shutdown();
}
}
}
-
+
public void testWithDFSWithDefaultPort() throws IOException {
MiniDFSCluster dfs = null;
MiniMRCluster mr = null;
Modified:
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestMultipleLevelCaching.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestMultipleLevelCaching.java?rev=1079244&r1=1079243&r2=1079244&view=diff
==============================================================================
---
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestMultipleLevelCaching.java
(original)
+++
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestMultipleLevelCaching.java
Tue Mar 8 05:59:48 2011
@@ -154,14 +154,12 @@ public class TestMultipleLevelCaching ex
}
RunningJob job = launchJob(jobConf, in, out, numMaps, jobName);
Counters counters = job.getCounters();
- assertEquals("Number of local maps",
- counters.getCounter(JobCounter.OTHER_LOCAL_MAPS), otherLocalMaps);
- assertEquals("Number of Data-local maps",
- counters.getCounter(JobCounter.DATA_LOCAL_MAPS),
- dataLocalMaps);
- assertEquals("Number of Rack-local maps",
- counters.getCounter(JobCounter.RACK_LOCAL_MAPS),
- rackLocalMaps);
+ assertEquals("Number of local maps", otherLocalMaps,
+ counters.getCounter(JobCounter.OTHER_LOCAL_MAPS));
+ assertEquals("Number of data-local maps", dataLocalMaps,
+ counters.getCounter(JobCounter.DATA_LOCAL_MAPS));
+ assertEquals("Number of rack-local maps", rackLocalMaps,
+ counters.getCounter(JobCounter.RACK_LOCAL_MAPS));
mr.waitUntilIdle();
mr.shutdown();
}
@@ -180,6 +178,8 @@ public class TestMultipleLevelCaching ex
jobConf.setNumMapTasks(numMaps);
jobConf.setNumReduceTasks(0);
jobConf.setJar("build/test/mapred/testjar/testjob.jar");
+ // can't check sub-maptask details if uberized:
+ jobConf.setBoolean(JobContext.JOB_UBERTASK_ENABLE, false);
return JobClient.runJob(jobConf);
}
}
Modified:
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestNodeRefresh.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestNodeRefresh.java?rev=1079244&r1=1079243&r2=1079244&view=diff
==============================================================================
---
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestNodeRefresh.java
(original)
+++
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapred/TestNodeRefresh.java
Tue Mar 8 05:59:48 2011
@@ -419,9 +419,9 @@ public class TestNodeRefresh extends Tes
2, jt.getClusterStatus(false).getTaskTrackers());
// validate blacklisted count
assertEquals("Blacklisted tracker count mismatch",
- 0, jt.getClusterStatus(false).getBlacklistedTrackers());
+ 0, jt.getClusterStatus(false).getBlacklistedTrackers());
- // run a failing job to blacklist the tracker
+ // run a failing job to (heuristically) blacklist the tracker
JobConf jConf = mr.createJobConf();
jConf.set(MRJobConfig.MAX_TASK_FAILURES_PER_TRACKER, "1");
jConf.setJobName("test-job-fail-once");
@@ -429,7 +429,9 @@ public class TestNodeRefresh extends Tes
jConf.setReducerClass(IdentityReducer.class);
jConf.setNumMapTasks(1);
jConf.setNumReduceTasks(0);
-
+ // disable uber-mode since don't want failing task to kill the whole job
+ jConf.setBoolean(JobContext.JOB_UBERTASK_ENABLE, false);
+
RunningJob job =
UtilsForTests.runJob(jConf, new Path("in"), new Path("out"));
job.waitForCompletion();
@@ -440,7 +442,7 @@ public class TestNodeRefresh extends Tes
1, jt.getClusterStatus(false).getTaskTrackers());
// validate blacklisted count
assertEquals("Blacklisted tracker count mismatch",
- 1, jt.getClusterStatus(false).getBlacklistedTrackers());
+ 1, jt.getClusterStatus(false).getBlacklistedTrackers());
// find the tracker to decommission
String hostToDecommission =
Modified:
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapreduce/TestNoJobSetupCleanup.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapreduce/TestNoJobSetupCleanup.java?rev=1079244&r1=1079243&r2=1079244&view=diff
==============================================================================
---
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapreduce/TestNoJobSetupCleanup.java
(original)
+++
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/mapreduce/TestNoJobSetupCleanup.java
Tue Mar 8 05:59:48 2011
@@ -49,10 +49,14 @@ public class TestNoJobSetupCleanup exten
job.setOutputFormatClass(MyOutputFormat.class);
job.waitForCompletion(true);
assertTrue(job.isSuccessful());
- assertTrue(job.getTaskReports(TaskType.JOB_SETUP).length == 0);
- assertTrue(job.getTaskReports(TaskType.JOB_CLEANUP).length == 0);
- assertTrue(job.getTaskReports(TaskType.MAP).length == numMaps);
- assertTrue(job.getTaskReports(TaskType.REDUCE).length == numReds);
+ assertEquals("wrong number of setup tasks", 0,
+ job.getTaskReports(TaskType.JOB_SETUP).length);
+ assertEquals("wrong number of cleanup tasks", 0,
+ job.getTaskReports(TaskType.JOB_CLEANUP).length);
+ assertEquals("wrong number of map tasks", numMaps,
+ job.getTaskReports(TaskType.MAP).length);
+ assertEquals("wrong number of reduce tasks", numReds,
+ job.getTaskReports(TaskType.REDUCE).length);
FileSystem fs = FileSystem.get(conf);
assertTrue("Job output directory doesn't exit!", fs.exists(outDir));
FileStatus[] list = fs.listStatus(outDir, new OutputFilter());
@@ -65,17 +69,20 @@ public class TestNoJobSetupCleanup exten
public void testNoJobSetupCleanup() throws Exception {
try {
Configuration conf = createJobConf();
+
+ // explicitly checking for counts of subtasks, so can't uberize:
+ conf.setBoolean(JobContext.JOB_UBERTASK_ENABLE, false);
// run a job without job-setup and cleanup
submitAndValidateJob(conf, 1, 1);
- // run a map only job.
+ // run a map-only job
submitAndValidateJob(conf, 1, 0);
// run empty job without job setup and cleanup
submitAndValidateJob(conf, 0, 0);
- // run empty job without job setup and cleanup, with non-zero reduces
+ // run a reduce-only job without job setup and cleanup
submitAndValidateJob(conf, 0, 1);
} finally {
tearDown();
Modified:
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/security/authorize/TestServiceLevelAuthorization.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/security/authorize/TestServiceLevelAuthorization.java?rev=1079244&r1=1079243&r2=1079244&view=diff
==============================================================================
---
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/security/authorize/TestServiceLevelAuthorization.java
(original)
+++
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/security/authorize/TestServiceLevelAuthorization.java
Tue Mar 8 05:59:48 2011
@@ -41,7 +41,14 @@ import org.apache.hadoop.util.StringUtil
import junit.framework.TestCase;
public class TestServiceLevelAuthorization extends TestCase {
+
public void testServiceLevelAuthorization() throws Exception {
+ // run both non-uberized and uberized
+ runServiceLevelAuthorization(false);
+ runServiceLevelAuthorization(true);
+ }
+
+ private void runServiceLevelAuthorization(boolean uberize) throws Exception {
MiniDFSCluster dfs = null;
MiniMRCluster mr = null;
FileSystem fileSys = null;
@@ -58,7 +65,8 @@ public class TestServiceLevelAuthorizati
// Start the mini clusters
dfs = new MiniDFSCluster(conf, slaves, true, null);
- // Ensure that the protocols authorized on the name node are only the
HDFS protocols.
+ // Ensure that the protocols authorized on the name node are only the
HDFS
+ // protocols.
Set<Class<?>> protocolsWithAcls =
NameNodeAdapter.getRpcServer(dfs.getNameNode())
.getServiceAuthorizationManager().getProtocolsWithAcls();
Service[] hdfsServices = new HDFSPolicyProvider().getServices();
@@ -86,20 +94,20 @@ public class TestServiceLevelAuthorizati
if (hdfsServices.length != protocolsWithAcls.size())
fail("there should be an entry for every HDFS service in the protocols
with ACLs map");
- // make cleanup inline sothat validation of existence of these
directories
- // can be done
+ // make cleanup inline so that validation of existence of these
+ // directories can be done
mr.setInlineCleanupThreads();
// Run examples
- TestMiniMRWithDFS.runPI(mr, mr.createJobConf(mrConf));
- TestMiniMRWithDFS.runWordCount(mr, mr.createJobConf(mrConf));
+ TestMiniMRWithDFS.runPI(mr, mr.createJobConf(mrConf), uberize);
+ TestMiniMRWithDFS.runWordCount(mr, mr.createJobConf(mrConf), uberize);
} finally {
if (dfs != null) { dfs.shutdown(); }
if (mr != null) { mr.shutdown();
}
}
}
-
+
private static final String DUMMY_ACL = "nouser nogroup";
private static final String UNKNOWN_USER = "dev,null";
Modified:
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/tools/rumen/TestRumenJobTraces.java
URL:
http://svn.apache.org/viewvc/hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/tools/rumen/TestRumenJobTraces.java?rev=1079244&r1=1079243&r2=1079244&view=diff
==============================================================================
---
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/tools/rumen/TestRumenJobTraces.java
(original)
+++
hadoop/mapreduce/branches/yahoo-merge/src/test/mapred/org/apache/hadoop/tools/rumen/TestRumenJobTraces.java
Tue Mar 8 05:59:48 2011
@@ -35,6 +35,7 @@ import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.JobTracker;
import org.apache.hadoop.mapred.MiniMRCluster;
import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.JobID;
import org.apache.hadoop.mapreduce.MRJobConfig;
import org.apache.hadoop.mapreduce.MapReduceTestUtil;
@@ -237,7 +238,7 @@ public class TestRumenJobTraces {
// Validate the events seen by history parser from
// history file v20-single-input-log.gz
- validateSeenHistoryEvents(seenEvents, goldLines);
+ validateSeenHistoryEvents(seenEvents, goldLinesV20Parser);
} finally {
if (parser != null) {
parser.close();
@@ -397,12 +398,27 @@ public class TestRumenJobTraces {
assertTrue("Missing input file " + inputPath2,
options.inputs.contains(inputPath2));
}
-
+
/**
* Test if {@link CurrentJHParser} can read events from current JH files.
*/
@Test
public void testCurrentJHParser() throws Exception {
+ // test both non-uberized and uberized versions
+ runCurrentJHParser(false);
+ runCurrentJHParser(true);
+ }
+
+ private void runCurrentJHParser(boolean isUber) throws Exception {
+ // The list of history events expected when parsing the job's
+ // history log file
+ final String[] goldLinesCurrentParserNonUber = new String[] {
+ JSE, JPCE, JIE, JSCE, TSE, ASE, MFE, TFE, TSE, ASE, RFE, TFE, JFE
+ };
+ final String[] goldLinesCurrentParserUber = new String[] {
+ JSE, JPCE, JIE, JSCE, TSE, ASE, RFE, TFE, JFE
+ };
+
final Configuration conf = new Configuration();
final FileSystem lfs = FileSystem.getLocal(conf);
@@ -421,9 +437,8 @@ public class TestRumenJobTraces {
new JobConf(conf));
JobTracker tracker = mrCluster.getJobTrackerRunner().getJobTracker();
JobHistory history = tracker.getJobHistory();
-
-
- // run a job
+
+ // run the job
Path inDir = new Path(tempDir, "input");
Path outDir = new Path(tempDir, "output");
JobHistoryParser parser = null;
@@ -432,6 +447,7 @@ public class TestRumenJobTraces {
try {
JobConf jConf = mrCluster.createJobConf();
+ jConf.setBoolean(JobContext.JOB_UBERTASK_ENABLE, isUber);
// construct a job with 1 map and 1 reduce task.
Job job = MapReduceTestUtil.createJob(jConf, inDir, outDir, 1, 1);
// disable setup/cleanup
@@ -475,25 +491,20 @@ public class TestRumenJobTraces {
// Check against the gold standard
System.out.println("testCurrentJHParser validating using gold std ");
- // The list of history events expected when parsing the above job's
- // history log file
- String[] goldLinesExpected = new String[] {
- JSE, JPCE, JIE, JSCE, TSE, ASE, MFE, TFE, TSE, ASE, RFE, TFE, JFE
- };
-
- validateSeenHistoryEvents(seenEvents, goldLinesExpected);
+ validateSeenHistoryEvents(seenEvents,
+ isUber? goldLinesCurrentParserUber : goldLinesCurrentParserNonUber);
} finally {
// stop the MR cluster
mrCluster.shutdown();
if (ris != null) {
- ris.close();
+ ris.close();
}
if (parser != null) {
parser.close();
}
- // cleanup the filesystem
+ // clean up the filesystem
lfs.delete(tempDir, true);
}
}
@@ -826,7 +837,7 @@ public class TestRumenJobTraces {
// The expected job history events(in order) when parsing
// the job history file v20-single-input-log.gz
- final static String[] goldLines = new String[] {
+ final static String[] goldLinesV20Parser = new String[] {
JSE, JPCE, JSCE, JIE, JICE, TSE, ASE, AFE, MFE, TUE, TFE, JSCE, TSE,
TSE, TSE, TSE, TSE, TSE, TSE, TSE, TSE, TSE, TSE, TSE, TSE, TSE, TSE,
TSE, TSE, TSE, TSE, TSE, ASE, AFE, MFE, TUE, TFE, ASE, AFE, MFE, TUE,
